test: MCP PID lock tests
Some checks failed
Contributor Attribution Check / check-attribution (pull_request) Failing after 58s
Docker Build and Publish / build-and-push (pull_request) Has been skipped
Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 1m2s
Tests / e2e (pull_request) Successful in 2m15s
Tests / test (pull_request) Failing after 53m34s

Part of #734
This commit is contained in:
2026-04-15 02:51:02 +00:00
parent c4aad087d4
commit 590b601b5c

View File

@@ -0,0 +1,75 @@
"""Tests for MCP PID file lock (#734)."""
import os
import sys
import tempfile
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
# Override MCP_DIR for testing
import tools.mcp_pid_lock as lock_mod
_test_dir = Path(tempfile.mkdtemp())
lock_mod._MCP_DIR = _test_dir
def test_acquire_and_release():
"""Lock can be acquired and released."""
pid = lock_mod.acquire_lock("test_server")
assert pid == os.getpid()
assert lock_mod.is_locked("test_server")
lock_mod.release_lock("test_server")
assert not lock_mod.is_locked("test_server")
def test_concurrent_lock_blocked():
"""Second acquire returns None when server running."""
lock_mod.acquire_lock("test_concurrent")
result = lock_mod.acquire_lock("test_concurrent")
assert result is None
lock_mod.release_lock("test_concurrent")
def test_stale_lock_cleaned():
"""Stale PID files are cleaned up."""
# Write a fake stale PID
pid_file = _test_dir / "stale.pid"
pid_file.write_text("99999999")
assert not lock_mod.is_locked("stale")
assert not pid_file.exists()
def test_list_locks():
"""list_locks returns only active locks."""
lock_mod.acquire_lock("list_test")
locks = lock_mod.list_locks()
assert "list_test" in locks
assert locks["list_test"] == os.getpid()
lock_mod.release_lock("list_test")
def test_cleanup_stale():
"""cleanup_stale_locks removes dead PID files."""
(_test_dir / "dead1.pid").write_text("99999998")
(_test_dir / "dead2.pid").write_text("99999999")
count = lock_mod.cleanup_stale_locks()
assert count >= 2
def test_force_release():
"""force_release kills process and removes lock."""
lock_mod.acquire_lock("force_test")
assert lock_mod.is_locked("force_test")
lock_mod.force_release("force_test")
assert not lock_mod.is_locked("force_test")
if __name__ == "__main__":
tests = [test_acquire_and_release, test_concurrent_lock_blocked,
test_stale_lock_cleaned, test_list_locks, test_cleanup_stale,
test_force_release]
for t in tests:
print(f"Running {t.__name__}...")
t()
print(" PASS")
print("\nAll tests passed.")