Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions tests/test_guard_race_2026_05_18.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
34 changes: 34 additions & 0 deletions tests/test_spawn_lock.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
import multiprocessing as mp
import os
import sys
import tempfile
import time
import unittest
from pathlib import Path
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down