| 1 | from __future__ import annotations
|
| 2 |
|
| 3 | import sys
|
| 4 | import unittest
|
| 5 | from pathlib import Path
|
| 6 |
|
| 7 | sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
| 8 |
|
| 9 | from oakbench.cachectl import _run_purge_command, cache_fields
|
| 10 |
|
| 11 |
|
| 12 | def _ok_purge() -> dict:
|
| 13 | return {"attempted": True, "ok": True, "method": "darwin_purge", "error": None}
|
| 14 |
|
| 15 |
|
| 16 | def _failed_purge() -> dict:
|
| 17 | return {"attempted": True, "ok": False, "method": "linux_drop_caches", "error": "PermissionError: denied"}
|
| 18 |
|
| 19 |
|
| 20 | class CacheFieldsHonestyTest(unittest.TestCase):
|
| 21 | def test_unset_request_is_unknown(self) -> None:
|
| 22 | fields = cache_fields(None, None)
|
| 23 | self.assertEqual(fields["cache_state"], "unknown")
|
| 24 | self.assertEqual(fields["cache_state_reason"], "cache_state_unset")
|
| 25 |
|
| 26 | def test_cold_with_successful_purge_is_cold(self) -> None:
|
| 27 | fields = cache_fields("cold", _ok_purge())
|
| 28 | self.assertEqual(fields["cache_state"], "cold")
|
| 29 | self.assertIn("darwin_purge", fields["cache_state_reason"])
|
| 30 |
|
| 31 | def test_cold_with_failed_purge_is_unknown_with_reason(self) -> None:
|
| 32 | fields = cache_fields("cold", _failed_purge())
|
| 33 | self.assertEqual(fields["cache_state"], "unknown")
|
| 34 | self.assertIn("purge_failed", fields["cache_state_reason"])
|
| 35 | self.assertIn("PermissionError", fields["cache_state_reason"])
|
| 36 |
|
| 37 | def test_cold_without_any_purge_attempt_is_unknown(self) -> None:
|
| 38 | fields = cache_fields("cold", None)
|
| 39 | self.assertEqual(fields["cache_state"], "unknown")
|
| 40 | self.assertEqual(fields["cache_state_reason"], "purge_not_attempted")
|
| 41 |
|
| 42 | def test_warm_requires_explicit_warming_claim(self) -> None:
|
| 43 | silent = cache_fields("warm", None)
|
| 44 | self.assertEqual(silent["cache_state"], "unknown")
|
| 45 | self.assertEqual(silent["cache_state_reason"], "warm_requested_without_warming_claim")
|
| 46 |
|
| 47 | claimed = cache_fields("warm", None, warmed_by_scenario=True)
|
| 48 | self.assertEqual(claimed["cache_state"], "warm")
|
| 49 | self.assertEqual(claimed["cache_state_reason"], "explicitly_warmed_by_scenario")
|
| 50 |
|
| 51 | def test_unrecognized_request_is_unknown_not_echoed_as_state(self) -> None:
|
| 52 | fields = cache_fields("lukewarm", None)
|
| 53 | self.assertEqual(fields["cache_state"], "unknown")
|
| 54 | self.assertIn("lukewarm", fields["cache_state_reason"])
|
| 55 |
|
| 56 | def test_successful_purge_evidence_beats_failed_warm_logic_isolation(self) -> None:
|
| 57 | # A purge result never upgrades a warm request to cold.
|
| 58 | fields = cache_fields("warm", _ok_purge())
|
| 59 | self.assertEqual(fields["cache_state"], "unknown")
|
| 60 |
|
| 61 |
|
| 62 | class PurgeCommandWrapperTest(unittest.TestCase):
|
| 63 | # Exercise the subprocess wrapper without actually purging the test host.
|
| 64 | def test_successful_command_returns_no_error(self) -> None:
|
| 65 | self.assertIsNone(_run_purge_command([sys.executable, "-c", "pass"]))
|
| 66 |
|
| 67 | def test_nonzero_exit_is_reported_not_raised(self) -> None:
|
| 68 | error = _run_purge_command([sys.executable, "-c", "import sys; sys.exit(3)"])
|
| 69 | self.assertIsNotNone(error)
|
| 70 | self.assertIn("exit 3", error)
|
| 71 |
|
| 72 | def test_missing_binary_is_reported_not_raised(self) -> None:
|
| 73 | error = _run_purge_command(["/nonexistent/oakbench-purge-binary"])
|
| 74 | self.assertIsNotNone(error)
|
| 75 | self.assertIn("Error", error)
|
| 76 |
|
| 77 |
|
| 78 | if __name__ == "__main__":
|
| 79 | unittest.main()
|