ConfigManager V2

tests/v2/test_stop_graceful.py

@@ -0,0 +1,43 @@
+import asyncio
+import threading
+import time
+
+from config_manager.v2 import ConfigManagerV2
+
+
+class BlockingApp(ConfigManagerV2):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.started_event = threading.Event()
+        self.active_event = threading.Event()
+        self.calls = 0
+
+    def execute(self) -> None:
+        self.calls += 1
+        self.active_event.set()
+        self.started_event.set()
+        time.sleep(0.2)
+        self.active_event.clear()
+
+
+def test_stop_waits_for_active_execute_and_prevents_next_run(tmp_path):
+    async def scenario() -> None:
+        cfg = tmp_path / "config.yaml"
+        cfg.write_text("work_interval: 0.05\nupdate_interval: 0.05\n", encoding="utf-8")
+
+        app = BlockingApp(str(cfg))
+        runner = asyncio.create_task(app.start())
+
+        started = await asyncio.to_thread(app.started_event.wait, 1.0)
+        assert started is True
+
+        await app.stop()
+        await runner
+
+        assert app.active_event.is_set() is False
+        assert app.calls == 1
+
+        await asyncio.sleep(0.15)
+        assert app.calls == 1
+
+    asyncio.run(scenario())