Build Unblocker MCP

import pytest import time import json from unittest.mock import MagicMock, patch from build_unblocker_mcp.server import mcp, main from build_unblocker_mcp.config import DEFAULT_PROCESSES # Mock psutil.Process object for testing class MockProcess: def __init__(self, pid, name, create_time): self.info = {'pid': pid, 'name': name, 'create_time': create_time} self._cpu_percent = 0.0 self._killed = False def cpu_percent(self, interval=None): return self._cpu_percent def set_cpu_percent(self, percent): self._cpu_percent = percent def create_time(self): return self.info['create_time'] def kill(self): self._killed = True def is_killed(self): return self._killed @pytest.fixture def mock_psutil_process_iter(mocker): """Fixture to mock psutil.process_iter.""" mock_procs = [] mock_iter = mocker.patch("psutil.process_iter") mock_iter.return_value = mock_procs return mock_procs import pytest # Import pytest for async test functions def test_unblock_build_no_processes_killed(mock_psutil_process_iter): """Test that no processes are killed when none meet the criteria.""" current_time = time.time() # Add a process that is too young mock_psutil_process_iter.append(MockProcess(1, "cl.exe", current_time - 10)) # idle_seconds default is 90 # Add a process with high CPU high_cpu_proc = MockProcess(2, "link.exe", current_time - 100) high_cpu_proc.set_cpu_percent(10.0) mock_psutil_process_iter.append(high_cpu_proc) # Add a process not in the default list mock_psutil_process_iter.append(MockProcess(3, "notepad.exe", current_time - 100)) @pytest.mark.asyncio async def test_unblock_build_no_processes_killed(mock_psutil_process_iter): """Test that no processes are killed when none meet the criteria.""" current_time = time.time() # Add a process that is too young mock_psutil_process_iter.append(MockProcess(1, "cl.exe", current_time - 10)) # idle_seconds default is 90 # Add a process with high CPU high_cpu_proc = MockProcess(2, "link.exe", current_time - 100) high_cpu_proc.set_cpu_percent(10.0) mock_psutil_process_iter.append(high_cpu_proc) # Add a process not in the default list mock_psutil_process_iter.append(MockProcess(3, "notepad.exe", current_time - 100)) result = await mcp.call_tool(name="unblock_build", arguments={}) # Call the tool using call_tool # The call_tool method returns a list of content objects, so we need to extract the result dictionary result_dict = result[0].text if result and result[0].type == "text" else "{}" result = json.loads(result_dict) assert result['examined'] == 2 # cl.exe and link.exe assert result['killed'] == 0 assert result['killed_processes'] == [] assert result['dry_run'] is False assert result['idle_seconds_threshold'] == 90 assert result['process_names_monitored'] == DEFAULT_PROCESSES @pytest.mark.asyncio async def test_unblock_build_processes_killed(mock_psutil_process_iter): """Test that processes meeting the criteria are killed.""" current_time = time.time() # Add a process that should be killed killable_proc_1 = MockProcess(1, "cl.exe", current_time - 100) killable_proc_1.set_cpu_percent(0.5) mock_psutil_process_iter.append(killable_proc_1) # Add another process that should be killed killable_proc_2 = MockProcess(2, "link.exe", current_time - 120) killable_proc_2.set_cpu_percent(0.1) mock_psutil_process_iter.append(killable_proc_2) # Add a process that should not be killed (high CPU) high_cpu_proc = MockProcess(3, "msbuild.exe", current_time - 100) high_cpu_proc.set_cpu_percent(5.0) mock_psutil_process_iter.append(high_cpu_proc) result = await mcp.call_tool(name="unblock_build", arguments={}) result_dict = result[0].text if result and result[0].type == "text" else "{}" result = json.loads(result_dict) current_time = time.time() # Add a process that should be killed killable_proc_1 = MockProcess(1, "cl.exe", current_time - 100) killable_proc_1.set_cpu_percent(0.5) mock_psutil_process_iter.append(killable_proc_1) # Add another process that should be killed killable_proc_2 = MockProcess(2, "link.exe", current_time - 120) killable_proc_2.set_cpu_percent(0.1) mock_psutil_process_iter.append(killable_proc_2) # Add a process that should not be killed (high CPU) high_cpu_proc = MockProcess(3, "msbuild.exe", current_time - 100) high_cpu_proc.set_cpu_percent(5.0) mock_psutil_process_iter.append(high_cpu_proc) result = mcp.call_tool(name="unblock_build", arguments={}) result_dict = result[0].text if result and result[0].type == "text" else "{}" result = json.loads(result_dict) assert result['examined'] == 3 assert result['killed'] == 2 assert len(result['killed_processes']) == 2 killed_pids = [p['pid'] for p in result['killed_processes']] assert 1 in killed_pids assert 2 in killed_pids assert killable_proc_1.is_killed() is True assert killable_proc_2.is_killed() is True assert high_cpu_proc.is_killed() is False @pytest.mark.asyncio async def test_unblock_build_dry_run(mock_psutil_process_iter, capsys): """Test that dry_run prevents killing processes but logs.""" current_time = time.time() # Add a process that would be killed killable_proc = MockProcess(1, "cl.exe", current_time - 100) killable_proc.set_cpu_percent(0.5) mock_psutil_process_iter.append(killable_proc) result = await mcp.call_tool(name="unblock_build", arguments={"dry_run": True}) result_dict = result[0].text if result and result[0].type == "text" else "{}" result = json.loads(result_dict) current_time = time.time() # Add a process that would be killed killable_proc = MockProcess(1, "cl.exe", current_time - 100) killable_proc.set_cpu_percent(0.5) mock_psutil_process_iter.append(killable_proc) result = mcp.call_tool(name="unblock_build", arguments={"dry_run": True}) result_dict = result[0].text if result and result[0].type == "text" else "{}" result = json.loads(result_dict) assert result['examined'] == 1 assert result['killed'] == 0 assert result['dry_run'] is True assert killable_proc.is_killed() is False captured = capsys.readouterr() assert "Dry run: Would kill process cl.exe" in captured.stderr @pytest.mark.asyncio async def test_unblock_build_custom_process_names(mock_psutil_process_iter): """Test using custom process names.""" current_time = time.time() custom_processes = ["mybuild.exe", "anotherproc.exe"] # Add a process from the custom list that should be killed killable_proc = MockProcess(1, "mybuild.exe", current_time - 100) killable_proc.set_cpu_percent(0.5) mock_psutil_process_iter.append(killable_proc) # Add a process from the default list (should not be examined) mock_psutil_process_iter.append(MockProcess(2, "cl.exe", current_time - 100)) result = await mcp.call_tool(name="unblock_build", arguments={"process_names": custom_processes}) result_dict = result[0].text if result and result[0].type == "text" else "{}" result = json.loads(result_dict) current_time = time.time() custom_processes = ["mybuild.exe", "anotherproc.exe"] # Add a process from the custom list that should be killed killable_proc = MockProcess(1, "mybuild.exe", current_time - 100) killable_proc.set_cpu_percent(0.5) mock_psutil_process_iter.append(killable_proc) # Add a process from the default list (should not be examined) mock_psutil_process_iter.append(MockProcess(2, "cl.exe", current_time - 100)) result = mcp.call_tool(name="unblock_build", arguments={"process_names": custom_processes}) result_dict = result[0].text if result and result[0].type == "text" else "{}" result = json.loads(result_dict) assert result['examined'] == 1 assert result['killed'] == 1 assert len(result['killed_processes']) == 1 assert result['killed_processes'][0]['pid'] == 1 assert result['process_names_monitored'] == custom_processes assert killable_proc.is_killed() is True @patch('sys.argv', ['unblock-build-mcp', '--help']) @patch('sys.exit') @pytest.mark.asyncio # Mark as async test async def test_cli_help(mock_exit, capsys): """Test that the CLI --help command runs without error.""" # This test primarily checks that the FastMCP CLI setup doesn't crash on --help # The actual help output content is handled by FastMCP main() captured = capsys.readouterr() # We expect sys.exit(0) to be called by FastMCP's help handler mock_exit.assert_called_once_with(0) # Basic check for some output, not the full help text assert captured.stdout or captured.stderr # This test primarily checks that the FastMCP CLI setup doesn't crash on --help # The actual help output content is handled by FastMCP main() captured = capsys.readouterr() # We expect sys.exit(0) to be called by FastMCP's help handler mock_exit.assert_called_once_with(0) # Basic check for some output, not the full help text assert captured.stdout or captured.stderr