diff --git a/tests/test_guard_race_2026_05_18.py b/tests/test_guard_race_2026_05_18.py index 61d2063..3bd2138 100644 --- a/tests/test_guard_race_2026_05_18.py +++ b/tests/test_guard_race_2026_05_18.py @@ -88,12 +88,20 @@ def _fake_popen(cmd_parts, **kwargs): fake_pid = 900_000 + worker_index return _DummyProc(fake_pid) + def _fake_is_process_alive(pid: int) -> bool: + """Model the spawned fake guard as alive for the contention check.""" + return pid > 0 + # Mock subprocess.Popen so no real daemon is spawned, and mock # find_claude_pid so we don't fail when no Claude is running. with ( _patch("cozempic.guard.subprocess.Popen", side_effect=_fake_popen), _patch("cozempic.guard.find_claude_pid", return_value=12345), _patch("cozempic.guard._cleanup_legacy_pid"), + _patch("cozempic.guard._is_guard_running_for_session", return_value=None), + _patch( + "cozempic.spawn_lock._is_process_alive", side_effect=_fake_is_process_alive + ), ): # Barrier sync: both children pile up here and release at the same instant try: diff --git a/tests/test_spawn_lock.py b/tests/test_spawn_lock.py index cea93ea..5fa1b0f 100644 --- a/tests/test_spawn_lock.py +++ b/tests/test_spawn_lock.py @@ -13,6 +13,7 @@ import multiprocessing as mp import os import sys +import tempfile import time import unittest from pathlib import Path @@ -62,10 +63,18 @@ def _fake_popen(cmd_parts, **kwargs): return _DummyProc(900_000 + worker_index) return _real_popen(cmd_parts, **kwargs) + def _fake_is_process_alive(pid: int) -> bool: + """Model the spawned fake guard as alive for the contention check.""" + return pid > 0 + with ( _patch("cozempic.guard.subprocess.Popen", side_effect=_fake_popen), _patch("cozempic.guard.find_claude_pid", return_value=12345), _patch("cozempic.guard._cleanup_legacy_pid"), + _patch("cozempic.guard._is_guard_running_for_session", return_value=None), + _patch( + "cozempic.spawn_lock._is_process_alive", side_effect=_fake_is_process_alive + ), ): try: barrier_handle.wait(timeout=10.0) @@ -433,6 +442,31 @@ def test_env_var_unset_uses_default(self): self.assertEqual(self._reload_with_env(None), _DEFAULT_FRESH) +class TestFreshClaimProtection(unittest.TestCase): + """The race fixtures must not be the only coverage of fresh claims.""" + + def test_fresh_dead_pid_is_not_reclaimed(self): + """A freshly written dead PID is a peer claim, not stale state.""" + from cozempic import spawn_lock + from cozempic.spawn_lock import DaemonAlreadyStarting, DaemonSpawnClaim + + with tempfile.TemporaryDirectory() as tmp_dir: + pid_file = Path(tmp_dir) / "guard.pid" + pid_file.write_text("900000\n", encoding="utf-8") + fresh_window = spawn_lock._FRESH_PIDFILE_SECONDS + fresh_now = pid_file.stat().st_mtime + fresh_window / 2 + claim = DaemonSpawnClaim("test-session", pid_file) + + with ( + patch("cozempic.spawn_lock._is_process_alive", return_value=False), + patch("cozempic.spawn_lock.time.time", return_value=fresh_now), + self.assertRaises(DaemonAlreadyStarting) as raised, + ): + claim.__enter__() + + self.assertEqual(raised.exception.holder_pid, 900000) + + class TestSymlinkDefense(unittest.TestCase): """Round-3 C1 regression test: the post-Popen ``.pid.tmp`` write must NOT follow symlinks. A local user with write access to ``/tmp`` (a