| 1 | """Monotonic settle discipline: fake clocks only, never real waiting."""
|
| 2 |
|
| 3 | from __future__ import annotations
|
| 4 |
|
| 5 | import sys
|
| 6 | import unittest
|
| 7 | from pathlib import Path
|
| 8 |
|
| 9 | sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
| 10 |
|
| 11 | from oakbench import platform_clock # noqa: E402
|
| 12 |
|
| 13 |
|
| 14 | class FakeClock:
|
| 15 | def __init__(self, start: float = 100.0) -> None:
|
| 16 | self.now = start
|
| 17 |
|
| 18 | def __call__(self) -> float:
|
| 19 | return self.now
|
| 20 |
|
| 21 | def sleep(self, seconds: float) -> None:
|
| 22 | self.now += seconds
|
| 23 |
|
| 24 |
|
| 25 | class SettleTest(unittest.TestCase):
|
| 26 | def test_settles_on_third_poll_with_exact_fields(self) -> None:
|
| 27 | clock = FakeClock()
|
| 28 | results = iter([False, False, True])
|
| 29 | result = platform_clock.settle(
|
| 30 | lambda: next(results),
|
| 31 | interval_s=2.0,
|
| 32 | timeout_s=60.0,
|
| 33 | clock=clock,
|
| 34 | sleep=clock.sleep,
|
| 35 | )
|
| 36 | self.assertTrue(result["settled"])
|
| 37 | self.assertEqual(result["elapsed_ms"], 4000.0)
|
| 38 | self.assertEqual(result["observed_wait_ms"], 4000.0)
|
| 39 | self.assertEqual(result["poll_iterations"], 3)
|
| 40 | self.assertEqual(result["poll_interval_s"], 2.0)
|
| 41 | self.assertEqual(result["poll_quantization_ms"], 2000.0)
|
| 42 |
|
| 43 | def test_timeout_means_unmeasured_settle_duration(self) -> None:
|
| 44 | # ADR-0002: a timed-out settle did not complete. settled is False and
|
| 45 | # elapsed_ms (duration to the settled state) is null/unmeasured; the
|
| 46 | # wall time actually spent watching is kept as observed_wait_ms.
|
| 47 | clock = FakeClock()
|
| 48 | result = platform_clock.settle(
|
| 49 | lambda: False,
|
| 50 | interval_s=2.0,
|
| 51 | timeout_s=5.0,
|
| 52 | clock=clock,
|
| 53 | sleep=clock.sleep,
|
| 54 | )
|
| 55 | self.assertFalse(result["settled"])
|
| 56 | self.assertIsNone(result["elapsed_ms"])
|
| 57 | self.assertEqual(result["observed_wait_ms"], 4000.0)
|
| 58 | self.assertEqual(result["poll_iterations"], 3)
|
| 59 | self.assertEqual(result["poll_quantization_ms"], 2000.0)
|
| 60 |
|
| 61 | def test_never_starts_a_poll_past_the_timeout(self) -> None:
|
| 62 | clock = FakeClock()
|
| 63 | calls = {"count": 0}
|
| 64 |
|
| 65 | def poll() -> bool:
|
| 66 | calls["count"] += 1
|
| 67 | return False
|
| 68 |
|
| 69 | platform_clock.settle(poll, interval_s=10.0, timeout_s=10.0, clock=clock, sleep=clock.sleep)
|
| 70 | # Polls at t=0 and t=10; t=20 would exceed the timeout.
|
| 71 | self.assertEqual(calls["count"], 2)
|
| 72 |
|
| 73 | def test_immediate_settle_is_iteration_one_elapsed_zero(self) -> None:
|
| 74 | clock = FakeClock()
|
| 75 | result = platform_clock.settle(
|
| 76 | lambda: True, interval_s=10.0, timeout_s=30.0, clock=clock, sleep=clock.sleep
|
| 77 | )
|
| 78 | self.assertTrue(result["settled"])
|
| 79 | self.assertEqual(result["elapsed_ms"], 0.0)
|
| 80 | self.assertEqual(result["poll_iterations"], 1)
|
| 81 |
|
| 82 | def test_rejects_nonpositive_interval(self) -> None:
|
| 83 | with self.assertRaises(ValueError):
|
| 84 | platform_clock.settle(lambda: True, interval_s=0.0, timeout_s=1.0)
|
| 85 |
|
| 86 | def test_quantization_is_interval_times_thousand(self) -> None:
|
| 87 | clock = FakeClock()
|
| 88 | result = platform_clock.settle(
|
| 89 | lambda: True, interval_s=10.0, timeout_s=60.0, clock=clock, sleep=clock.sleep
|
| 90 | )
|
| 91 | self.assertEqual(result["poll_quantization_ms"], 10000.0)
|
| 92 |
|
| 93 |
|
| 94 | class MonotonicMsTest(unittest.TestCase):
|
| 95 | def test_returns_float_milliseconds(self) -> None:
|
| 96 | first = platform_clock.monotonic_ms()
|
| 97 | second = platform_clock.monotonic_ms()
|
| 98 | self.assertIsInstance(first, float)
|
| 99 | self.assertGreaterEqual(second, first)
|
| 100 |
|
| 101 |
|
| 102 | class BoundedCrossClockTest(unittest.TestCase):
|
| 103 | def test_marks_value_descriptive_only(self) -> None:
|
| 104 | wrapped = platform_clock.bounded_cross_clock(1234.5)
|
| 105 | self.assertEqual(
|
| 106 | wrapped,
|
| 107 | {"value_ms": 1234.5, "cross_clock": True, "descriptive_only": True},
|
| 108 | )
|
| 109 |
|
| 110 |
|
| 111 | if __name__ == "__main__":
|
| 112 | unittest.main()
|