File size: 2,247 Bytes
bbc65cd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | import time
import unittest
from unittest.mock import patch
import monitor
class MonitorTests(unittest.TestCase):
def setUp(self) -> None:
monitor._cache.clear()
monitor._cache_ts.clear()
monitor._sidecar_refresh_in_flight = False
def tearDown(self) -> None:
monitor._cache.clear()
monitor._cache_ts.clear()
monitor._sidecar_refresh_in_flight = False
def test_get_sidecar_status_returns_fresh_cache_without_thread(self) -> None:
expected = {"running": True, "phase": "testing"}
monitor._cache["sidecar_status"] = expected
monitor._cache_ts["sidecar_status"] = time.time()
with patch("monitor.threading.Thread") as thread_cls:
result = monitor.get_sidecar_status()
self.assertEqual(result, expected)
thread_cls.assert_not_called()
def test_get_sidecar_status_returns_stale_value_and_starts_background_refresh(self) -> None:
stale = {"running": True, "phase": "implementation"}
monitor._cache["sidecar_status"] = stale
monitor._cache_ts["sidecar_status"] = 0
started = []
class FakeThread:
def __init__(self, target=None, args=(), daemon=None, name=None):
self.target = target
self.args = args
self.daemon = daemon
self.name = name
def start(self) -> None:
started.append((self.target, self.args, self.daemon, self.name))
with patch("monitor.threading.Thread", FakeThread):
result = monitor.get_sidecar_status()
self.assertEqual(result, stale)
self.assertEqual(len(started), 1)
self.assertTrue(monitor._sidecar_refresh_in_flight)
def test_get_sidecar_status_only_starts_one_refresh_thread(self) -> None:
monitor._cache["sidecar_status"] = {"running": False}
monitor._cache_ts["sidecar_status"] = 0
monitor._sidecar_refresh_in_flight = True
with patch("monitor.threading.Thread") as thread_cls:
result = monitor.get_sidecar_status()
self.assertEqual(result, {"running": False})
thread_cls.assert_not_called()
if __name__ == "__main__":
unittest.main()
|