F....F..FF. [100%] =================================== FAILURES =================================== _______________ test_watcher_fires_shutdown_when_marker_appears ________________ tmp_path = PosixPath('/tmp/nc9-f227674b3b/test_watcher_fires_shutdown_wh0') monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x76601581aa80> def test_watcher_fires_shutdown_when_marker_appears(tmp_path, monkeypatch): """When a marker targeting THIS process exists, fire the shutdown handler.""" marker = tmp_path / ".gateway-planned-stop.json" # Patch the marker-path resolver so the watcher polls our temp location. monkeypatch.setattr(status_mod, "_get_planned_stop_marker_path", lambda: marker) runner = _FakeRunner(running=True, draining=False) loop = _make_loop_capturing_calls() shutdown_handler = MagicMock(name="shutdown_signal_handler") stop_event = threading.Event() # Drop a self-targeting marker before the thread starts. _write_self_marker(marker) watcher = threading.Thread( target=_run_planned_stop_watcher, args=(stop_event, runner, loop, shutdown_handler), kwargs={"poll_interval": 0.05}, daemon=True, ) watcher.start() watcher.join(timeout=2.0) > assert not watcher.is_alive(), "Watcher should exit after firing" E AssertionError: Watcher should exit after firing E assert not True E + where True = is_alive() E + where is_alive = .is_alive tests/gateway/test_planned_stop_watcher.py:88: AssertionError ______________ test_watcher_fires_only_once_when_marker_persists _______________ tmp_path = PosixPath('/tmp/nc9-f227674b3b/test_watcher_fires_only_once_w0') monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x76601582d8b0> def test_watcher_fires_only_once_when_marker_persists(tmp_path, monkeypatch): """Marker file existing for multiple polls must NOT spam the handler. The watcher fires once and exits its loop (the shutdown handler is responsible for consuming the marker on its own thread). If we re-fired on every tick, the handler would be invoked dozens of times before the gateway actually shuts down. """ marker = tmp_path / ".gateway-planned-stop.json" _write_self_marker(marker) monkeypatch.setattr(status_mod, "_get_planned_stop_marker_path", lambda: marker) runner = _FakeRunner(running=True, draining=False) loop = _make_loop_capturing_calls() stop_event = threading.Event() watcher = threading.Thread( target=_run_planned_stop_watcher, args=(stop_event, runner, loop, MagicMock()), kwargs={"poll_interval": 0.05}, daemon=True, ) watcher.start() # Let the watcher tick several times — but it should exit after the first fire. watcher.join(timeout=1.0) > assert not watcher.is_alive() E assert not True E + where True = is_alive() E + where is_alive = .is_alive tests/gateway/test_planned_stop_watcher.py:244: AssertionError ____________ test_watcher_cleans_up_stale_marker_and_keeps_running _____________ tmp_path = PosixPath('/tmp/nc9-f227674b3b/test_watcher_cleans_up_stale_m0') monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x76601582f710> def test_watcher_cleans_up_stale_marker_and_keeps_running(tmp_path, monkeypatch): """A marker older than the TTL is unlinked and never fires shutdown.""" marker = tmp_path / ".gateway-planned-stop.json" # Self-targeting but backdated past the TTL: must be treated as dead. _write_self_marker(marker, stale=True) monkeypatch.setattr(status_mod, "_get_planned_stop_marker_path", lambda: marker) runner = _FakeRunner(running=True, draining=False) loop = _make_loop_capturing_calls() shutdown_handler = MagicMock(name="shutdown_signal_handler") stop_event = threading.Event() watcher = threading.Thread( target=_run_planned_stop_watcher, args=(stop_event, runner, loop, shutdown_handler), kwargs={"poll_interval": 0.05}, daemon=True, ) watcher.start() time.sleep(0.3) stop_event.set() watcher.join(timeout=2.0) assert not watcher.is_alive() assert loop._captured == [], "Stale marker must not fire shutdown" shutdown_handler.assert_not_called() > assert not marker.exists(), "Stale marker should have been cleaned up" E AssertionError: Stale marker should have been cleaned up E assert not True E + where True = exists() E + where exists = PosixPath('/tmp/nc9-f227674b3b/test_watcher_cleans_up_stale_m0/.gateway-planned-stop.json').exists tests/gateway/test_planned_stop_watcher.py:368: AssertionError ________ test_planned_stop_marker_targets_self_probe_is_non_destructive ________ tmp_path = PosixPath('/tmp/nc9-f227674b3b/test_planned_stop_marker_targe0') monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7660158308c0> def test_planned_stop_marker_targets_self_probe_is_non_destructive(tmp_path, monkeypatch): """The probe returns True for a self-marker WITHOUT unlinking it. The shutdown handler performs the authoritative consume on its own thread, so the watcher's probe must leave a matching marker intact. """ marker = tmp_path / ".gateway-planned-stop.json" _write_self_marker(marker) monkeypatch.setattr(status_mod, "_get_planned_stop_marker_path", lambda: marker) > assert status_mod.planned_stop_marker_targets_self() is True E assert False is True E + where False = () E + where = status_mod.planned_stop_marker_targets_self tests/gateway/test_planned_stop_watcher.py:381: AssertionError =========================== short test summary info ============================ FAILED tests/gateway/test_planned_stop_watcher.py::test_watcher_fires_shutdown_when_marker_appears FAILED tests/gateway/test_planned_stop_watcher.py::test_watcher_fires_only_once_when_marker_persists FAILED tests/gateway/test_planned_stop_watcher.py::test_watcher_cleans_up_stale_marker_and_keeps_running FAILED tests/gateway/test_planned_stop_watcher.py::test_planned_stop_marker_targets_self_probe_is_non_destructive 4 failed, 7 passed in 5.15s