F....F..FF. [100%] =================================== FAILURES =================================== _______________ test_watcher_fires_shutdown_when_marker_appears ________________ tmp_path = PosixPath('/tmp/nc-r8-gw.qo0yHY/test_watcher_fires_shutdown_wh0') monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x79b4370c7350> 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 ../.worktrees/nutricoach-v140-impl/tests/gateway/test_planned_stop_watcher.py:88: AssertionError ______________ test_watcher_fires_only_once_when_marker_persists _______________ tmp_path = PosixPath('/tmp/nc-r8-gw.qo0yHY/test_watcher_fires_only_once_w0') monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x79b4370d4770> 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 ../.worktrees/nutricoach-v140-impl/tests/gateway/test_planned_stop_watcher.py:244: AssertionError ____________ test_watcher_cleans_up_stale_marker_and_keeps_running _____________ tmp_path = PosixPath('/tmp/nc-r8-gw.qo0yHY/test_watcher_cleans_up_stale_m0') monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x79b4370d6720> 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/nc-r8-gw.qo0yHY/test_watcher_cleans_up_stale_m0/.gateway-planned-stop.json').exists ../.worktrees/nutricoach-v140-impl/tests/gateway/test_planned_stop_watcher.py:368: AssertionError ________ test_planned_stop_marker_targets_self_probe_is_non_destructive ________ tmp_path = PosixPath('/tmp/nc-r8-gw.qo0yHY/test_planned_stop_marker_targe0') monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x79b4370dbce0> 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 ../.worktrees/nutricoach-v140-impl/tests/gateway/test_planned_stop_watcher.py:381: AssertionError =========================== short test summary info ============================ FAILED ../.worktrees/nutricoach-v140-impl/tests/gateway/test_planned_stop_watcher.py::test_watcher_fires_shutdown_when_marker_appears FAILED ../.worktrees/nutricoach-v140-impl/tests/gateway/test_planned_stop_watcher.py::test_watcher_fires_only_once_when_marker_persists FAILED ../.worktrees/nutricoach-v140-impl/tests/gateway/test_planned_stop_watcher.py::test_watcher_cleans_up_stale_marker_and_keeps_running FAILED ../.worktrees/nutricoach-v140-impl/tests/gateway/test_planned_stop_watcher.py::test_planned_stop_marker_targets_self_probe_is_non_destructive 4 failed, 7 passed in 5.14s