Skip to content

Commit ddf1681

Browse files
committed
test: Add comprehensive macOS process management tests
1 parent 4e27059 commit ddf1681

File tree

1 file changed

+110
-0
lines changed

1 file changed

+110
-0
lines changed

tests/test_process_manager_macos.py

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
# tests/test_process_manager_macos.py
2+
import pytest
3+
import platform
4+
import signal
5+
import time
6+
import os
7+
import subprocess
8+
from typing import Generator
9+
10+
pytestmark = [
11+
pytest.mark.skipif(
12+
platform.system() != "Darwin",
13+
reason="These tests only run on macOS"
14+
),
15+
pytest.mark.macos,
16+
pytest.mark.slow
17+
]
18+
19+
@pytest.fixture
20+
def process_manager():
21+
from mcp_shell_server.process_manager import ProcessManager
22+
pm = ProcessManager()
23+
yield pm
24+
pm.cleanup_all()
25+
26+
def get_process_status(pid: int) -> str:
27+
"""Get process status using ps command."""
28+
try:
29+
ps = subprocess.run(
30+
['ps', '-o', 'stat=', '-p', str(pid)],
31+
capture_output=True,
32+
text=True
33+
)
34+
return ps.stdout.strip()
35+
except subprocess.CalledProcessError:
36+
return ""
37+
38+
def test_zombie_process_cleanup(process_manager):
39+
"""Test that background processes don't become zombies."""
40+
cmd = ["sh", "-c", "sleep 0.5 & wait"]
41+
process = process_manager.start_process(cmd)
42+
43+
# Wait for the background process to finish
44+
time.sleep(1)
45+
46+
# Get process status
47+
status = get_process_status(process.pid)
48+
49+
# Verify process is either gone or not zombie (Z state)
50+
assert 'Z' not in status, f"Process {process.pid} is zombie (status: {status})"
51+
52+
def test_process_timeout(process_manager):
53+
"""Test process timeout functionality."""
54+
# Start a process that should timeout
55+
cmd = ["sleep", "10"]
56+
process = process_manager.start_process(cmd, timeout=1)
57+
58+
# Wait slightly longer than the timeout
59+
time.sleep(1.5)
60+
61+
# Verify process was terminated
62+
assert not process.is_running()
63+
assert process.returncode is not None
64+
65+
def test_multiple_process_cleanup(process_manager):
66+
"""Test cleanup of multiple processes."""
67+
# Start multiple background processes
68+
processes = [
69+
process_manager.start_process(["sleep", "2"])
70+
for _ in range(3)
71+
]
72+
73+
# Give them a moment to start
74+
time.sleep(0.1)
75+
76+
# Verify they're all running
77+
assert all(p.is_running() for p in processes)
78+
79+
# Cleanup
80+
process_manager.cleanup_all()
81+
82+
# Give cleanup a moment to complete
83+
time.sleep(0.1)
84+
85+
# Verify all processes are gone
86+
for p in processes:
87+
status = get_process_status(p.pid)
88+
assert status == "", f"Process {p.pid} still exists with status: {status}"
89+
90+
def test_process_group_termination(process_manager):
91+
"""Test that entire process group is terminated."""
92+
# Create a process that spawns children
93+
cmd = ["sh", "-c", "sleep 10 & sleep 10 & sleep 10 & wait"]
94+
process = process_manager.start_process(cmd)
95+
96+
# Give processes time to start
97+
time.sleep(0.5)
98+
99+
# Kill the main process
100+
process.kill()
101+
102+
# Wait a moment for cleanup
103+
time.sleep(0.5)
104+
105+
# Check if any processes from the group remain
106+
ps = subprocess.run(
107+
["pgrep", "-g", str(process.pid)],
108+
capture_output=True
109+
)
110+
assert ps.returncode != 0, "Process group still exists"

0 commit comments

Comments
 (0)