From 62bdf6cfc17470d1e0f641e25150a8cd055febfc Mon Sep 17 00:00:00 2001 From: Paradox <159635503+i-am-paradox@users.noreply.github.com> Date: Fri, 9 Oct 2026 04:24:07 +0530 Subject: [PATCH 1/2] test(mcp): resolve baseline toml path and restore mcp server integration tests (fixes #527) - Resolve openssf-baseline.toml via importlib.resources.files without dead-code fallbacks - Assert baseline TOML file presence so missing configs fail tests instead of skipping silently - Mark test_mcp_server.py with pytest.mark.integration so CI integration test suite runs the module - Update expected tools count to 22 in test_server_starts_and_lists_tools (features 025 and 041) - Assert control domain directly in test_audit_with_tags_filter against baseline controls table - Add test_get_pending_data_and_confirm_project_data round trip Fixes #527 Assisted-by: Antigravity:eni-gemini-2.5-pro Signed-off-by: Paradox <159635503+i-am-paradox@users.noreply.github.com> --- tests/darnit/server/test_factory.py | 27 +++---- tests/integration/test_mcp_server.py | 102 +++++++++++++++++---------- 2 files changed, 75 insertions(+), 54 deletions(-) diff --git a/tests/darnit/server/test_factory.py b/tests/darnit/server/test_factory.py index ef2910f6..f95324bc 100644 --- a/tests/darnit/server/test_factory.py +++ b/tests/darnit/server/test_factory.py @@ -145,23 +145,16 @@ def test_refused_handler_is_reported_from_dict(self, caplog): def test_openssf_baseline_toml(self): """Test loading the actual openssf-baseline.toml file.""" # Find the openssf-baseline.toml file - baseline_path = ( - Path(__file__).parent.parent.parent.parent - / "packages" - / "darnit-baseline" - / "openssf-baseline.toml" - ) - if baseline_path.exists(): - # Should be able to create server from it - # Note: This test may fail if darnit_baseline tools have import errors - try: - server = create_server(str(baseline_path)) - assert server.name == "openssf-baseline" - except ImportError: - # Skip if darnit_baseline not installed - pytest.skip("darnit_baseline not installed") - else: - pytest.skip("openssf-baseline.toml not found") + from importlib.resources import files + + baseline_path = Path(str(files("darnit_baseline") / "openssf-baseline.toml")) + assert baseline_path.is_file(), f"openssf-baseline.toml not found at {baseline_path}" + try: + server = create_server(str(baseline_path)) + assert server.name == "openssf-baseline" + except ImportError: + # Skip if darnit_baseline not installed + pytest.skip("darnit_baseline not installed") def _shipped_framework_configs() -> list[tuple[str, Path]]: diff --git a/tests/integration/test_mcp_server.py b/tests/integration/test_mcp_server.py index 87162368..5ccd6d0b 100644 --- a/tests/integration/test_mcp_server.py +++ b/tests/integration/test_mcp_server.py @@ -5,24 +5,30 @@ """ import json +import tomllib +from importlib.resources import files from pathlib import Path import pytest -# Configure pytest-asyncio -pytestmark = pytest.mark.asyncio(loop_scope="function") +# Configure pytest-asyncio and mark as integration test +pytestmark = [ + pytest.mark.asyncio(loop_scope="function"), + pytest.mark.integration, +] from mcp import StdioServerParameters from mcp.client.session import ClientSession from mcp.client.stdio import stdio_client # Path to the openssf-baseline.toml config -BASELINE_TOML = ( - Path(__file__).parent.parent.parent - / "packages" - / "darnit-baseline" - / "openssf-baseline.toml" -) +BASELINE_TOML = Path(str(files("darnit_baseline") / "openssf-baseline.toml")) + + +@pytest.fixture(scope="module", autouse=True) +def require_baseline_toml(): + """Ensure openssf-baseline.toml is present; fail if missing so tests do not go dark silently.""" + assert BASELINE_TOML.is_file(), f"openssf-baseline.toml not found at {BASELINE_TOML}" @pytest.fixture @@ -30,9 +36,7 @@ def test_repo(tmp_path): """Create a minimal test repository for auditing.""" # Create basic repo structure (tmp_path / ".git").mkdir() - (tmp_path / ".git" / "config").write_text( - "[remote \"origin\"]\n\turl = https://github.com/test-org/test-repo.git\n" - ) + (tmp_path / ".git" / "config").write_text('[remote "origin"]\n\turl = https://github.com/test-org/test-repo.git\n') (tmp_path / ".git" / "HEAD").write_text("ref: refs/heads/main\n") # Create a README @@ -50,9 +54,6 @@ class TestMCPServerIntegration: @pytest.mark.asyncio async def test_server_starts_and_lists_tools(self): """Test that the server starts and exposes tools.""" - if not BASELINE_TOML.exists(): - pytest.skip("openssf-baseline.toml not found") - server_params = StdioServerParameters( command="uv", args=["run", "darnit", "serve", str(BASELINE_TOML)], @@ -68,7 +69,7 @@ async def test_server_starts_and_lists_tools(self): tools_result = await session.list_tools() tool_names = [t.name for t in tools_result.tools] - # Verify all 18 expected tools are present + # Verify all 22 expected tools are present expected_tools = [ # Audit "audit_openssf_baseline", @@ -94,19 +95,22 @@ async def test_server_starts_and_lists_tools(self): # Org & Test Repository "list_org_repos", "create_test_repository", + # Harness Loop (Feature 025) + "run_next_action", + "submit_action_result", + # Judgments & Candidates (Feature 041) + "submit_judgment", + "confirm_pass_candidate", ] for tool in expected_tools: assert tool in tool_names, f"Missing tool: {tool}" - # Should have exactly 18 tools - assert len(tool_names) == 18, f"Expected 18 tools, got {len(tool_names)}: {tool_names}" + # Should have exactly 22 tools + assert len(tool_names) == 22, f"Expected 22 tools, got {len(tool_names)}: {tool_names}" @pytest.mark.asyncio async def test_list_available_checks(self): """Test calling the list_available_checks tool.""" - if not BASELINE_TOML.exists(): - pytest.skip("openssf-baseline.toml not found") - server_params = StdioServerParameters( command="uv", args=["run", "darnit", "serve", str(BASELINE_TOML)], @@ -136,9 +140,6 @@ async def test_list_available_checks(self): @pytest.mark.asyncio async def test_audit_on_test_repo(self, test_repo): """Test running an audit on a test repository.""" - if not BASELINE_TOML.exists(): - pytest.skip("openssf-baseline.toml not found") - server_params = StdioServerParameters( command="uv", args=["run", "darnit", "serve", str(BASELINE_TOML)], @@ -179,9 +180,6 @@ async def test_audit_on_test_repo(self, test_repo): @pytest.mark.asyncio async def test_audit_with_tags_filter(self, test_repo): """Test running an audit with tags filtering.""" - if not BASELINE_TOML.exists(): - pytest.skip("openssf-baseline.toml not found") - server_params = StdioServerParameters( command="uv", args=["run", "darnit", "serve", str(BASELINE_TOML)], @@ -215,17 +213,18 @@ async def test_audit_with_tags_filter(self, test_repo): assert "results" in audit_result results = audit_result["results"] - # All results should be from VM domain + # All returned controls should belong to the VM domain + baseline_controls = tomllib.loads(BASELINE_TOML.read_text(encoding="utf-8")).get("controls", {}) for r in results: control_id = r.get("id", "") - assert "VM" in control_id, f"Expected VM domain control, got {control_id}" + assert control_id in baseline_controls, f"Unknown control: {control_id}" + ctrl = baseline_controls[control_id] + domain = ctrl.get("domain") or ctrl.get("tags", {}).get("domain") + assert domain == "VM", f"Expected VM domain control, got {control_id} with domain {domain}" @pytest.mark.asyncio async def test_get_project_config_no_config(self, test_repo): """Test get_project_config when no config exists.""" - if not BASELINE_TOML.exists(): - pytest.skip("openssf-baseline.toml not found") - server_params = StdioServerParameters( command="uv", args=["run", "darnit", "serve", str(BASELINE_TOML)], @@ -249,6 +248,41 @@ async def test_get_project_config_no_config(self, test_repo): # Should indicate no config found assert "No .project.yaml found" in content.text or "init_project_config" in content.text + @pytest.mark.asyncio + async def test_get_pending_data_and_confirm_project_data(self, test_repo): + """Test get_pending_data and confirm_project_data round trip.""" + + server_params = StdioServerParameters( + command="uv", + args=["run", "darnit", "serve", str(BASELINE_TOML)], + env=None, + ) + + async with stdio_client(server_params) as (read, write): + async with ClientSession(read, write) as session: + await session.initialize() + + # Call get_pending_data + pending_res = await session.call_tool( + "get_pending_data", + {"local_path": str(test_repo)}, + ) + assert pending_res.content + assert "AskUserQuestion" in pending_res.content[0].text + + # Call confirm_project_data + confirm_res = await session.call_tool( + "confirm_project_data", + { + "local_path": str(test_repo), + "owner": "test-org", + "repo": "test-repo", + "security_contact": "security@test-org.com", + }, + ) + assert confirm_res.content + assert "security_contact: confirmed" in confirm_res.content[0].text + class TestMCPServerToolDescriptions: """Test that tool descriptions are properly exposed.""" @@ -256,9 +290,6 @@ class TestMCPServerToolDescriptions: @pytest.mark.asyncio async def test_tool_descriptions_are_set(self): """Test that tools have descriptions from TOML.""" - if not BASELINE_TOML.exists(): - pytest.skip("openssf-baseline.toml not found") - server_params = StdioServerParameters( command="uv", args=["run", "darnit", "serve", str(BASELINE_TOML)], @@ -288,9 +319,6 @@ class TestMCPServerErrorHandling: @pytest.mark.asyncio async def test_audit_nonexistent_path(self): """Test audit with a non-existent path returns error gracefully.""" - if not BASELINE_TOML.exists(): - pytest.skip("openssf-baseline.toml not found") - server_params = StdioServerParameters( command="uv", args=["run", "darnit", "serve", str(BASELINE_TOML)], From a939fce1dd1342cf05fc868010ea885d7ecf51a3 Mon Sep 17 00:00:00 2001 From: Paradox <159635503+i-am-paradox@users.noreply.github.com> Date: Sun, 11 Oct 2026 04:20:47 +0530 Subject: [PATCH 2/2] test(mcp): isolate subprocess environment and verify context round trip - Provide mcp_server_params fixture with HOME and XDG pointed into tmp_path - Launch darnit binary directly to avoid uv cache requirements under isolated HOME - Symlink existing tree-sitter language pack cache to prevent network fetches in tests - Isolate test_repo directory and assert security_contact presence and removal in round trip - Add non-empty assertion on results in test_audit_with_tags_filter - Move importlib.resources.files to top of test_factory.py and drop ImportError skip Assisted-by: Antigravity:eni-gemini-2.5-pro Signed-off-by: Paradox <159635503+i-am-paradox@users.noreply.github.com> --- tests/darnit/server/test_factory.py | 24 ++--- tests/integration/test_mcp_server.py | 155 ++++++++++++++------------- 2 files changed, 89 insertions(+), 90 deletions(-) diff --git a/tests/darnit/server/test_factory.py b/tests/darnit/server/test_factory.py index f95324bc..897dd7f0 100644 --- a/tests/darnit/server/test_factory.py +++ b/tests/darnit/server/test_factory.py @@ -3,6 +3,7 @@ import asyncio import logging import tomllib +from importlib.resources import files from pathlib import Path import pytest @@ -68,31 +69,31 @@ def test_file_not_found(self): def test_loads_from_toml_file(self, tmp_path): """Test loading server from TOML file.""" config_path = tmp_path / "test.toml" - config_path.write_text(''' + config_path.write_text(""" [mcp] name = "from-file-server" [mcp.tools.get_logger] handler = "darnit.core.logging:get_logger" description = "Logger" -''') +""") server = create_server(str(config_path)) assert server.name == "from-file-server" def test_loads_path_object(self, tmp_path): """Test loading server from Path object.""" config_path = tmp_path / "test.toml" - config_path.write_text(''' + config_path.write_text(""" [mcp] name = "path-server" -''') +""") server = create_server(config_path) # Pass Path directly assert server.name == "path-server" def test_handles_invalid_handler(self, tmp_path, caplog): """A missing handler is skipped with a warning; the other tools still load.""" config_path = tmp_path / "test.toml" - config_path.write_text(''' + config_path.write_text(""" [mcp] name = "test-server" @@ -103,7 +104,7 @@ def test_handles_invalid_handler(self, tmp_path, caplog): [mcp.tools.invalid_tool] handler = "darnit.nonexistent_module:func" description = "Invalid tool" -''') +""") with caplog.at_level(logging.WARNING): server = create_server(str(config_path)) tools = {tool.name for tool in asyncio.run(server.list_tools())} @@ -144,17 +145,10 @@ def test_refused_handler_is_reported_from_dict(self, caplog): def test_openssf_baseline_toml(self): """Test loading the actual openssf-baseline.toml file.""" - # Find the openssf-baseline.toml file - from importlib.resources import files - baseline_path = Path(str(files("darnit_baseline") / "openssf-baseline.toml")) assert baseline_path.is_file(), f"openssf-baseline.toml not found at {baseline_path}" - try: - server = create_server(str(baseline_path)) - assert server.name == "openssf-baseline" - except ImportError: - # Skip if darnit_baseline not installed - pytest.skip("darnit_baseline not installed") + server = create_server(str(baseline_path)) + assert server.name == "openssf-baseline" def _shipped_framework_configs() -> list[tuple[str, Path]]: diff --git a/tests/integration/test_mcp_server.py b/tests/integration/test_mcp_server.py index 5ccd6d0b..c18ace7a 100644 --- a/tests/integration/test_mcp_server.py +++ b/tests/integration/test_mcp_server.py @@ -5,6 +5,7 @@ """ import json +import sys import tomllib from importlib.resources import files from pathlib import Path @@ -19,7 +20,7 @@ from mcp import StdioServerParameters from mcp.client.session import ClientSession -from mcp.client.stdio import stdio_client +from mcp.client.stdio import get_default_environment, stdio_client # Path to the openssf-baseline.toml config BASELINE_TOML = Path(str(files("darnit_baseline") / "openssf-baseline.toml")) @@ -31,36 +32,59 @@ def require_baseline_toml(): assert BASELINE_TOML.is_file(), f"openssf-baseline.toml not found at {BASELINE_TOML}" +@pytest.fixture +def mcp_server_params(tmp_path): + """Build isolated environment and StdioServerParameters launching darnit directly.""" + # Preserve existing tree-sitter language pack cache so the spawned server doesn't download over network + real_home = Path.home() + for rel in ("Library/Caches/tree-sitter-language-pack", ".cache/tree-sitter-language-pack"): + src = real_home / rel + if src.exists(): + dest = tmp_path / rel + dest.parent.mkdir(parents=True, exist_ok=True) + try: + dest.symlink_to(src) + except OSError: + pass + + env = get_default_environment() + env["HOME"] = str(tmp_path) + env["XDG_CONFIG_HOME"] = str(tmp_path / ".config") + env["XDG_DATA_HOME"] = str(tmp_path / ".local" / "share") + darnit_bin = Path(sys.executable).with_name("darnit") + return StdioServerParameters( + command=str(darnit_bin), + args=["serve", str(BASELINE_TOML)], + env=env, + ) + + @pytest.fixture def test_repo(tmp_path): """Create a minimal test repository for auditing.""" + repo = tmp_path / "test-repo" + repo.mkdir(parents=True, exist_ok=True) # Create basic repo structure - (tmp_path / ".git").mkdir() - (tmp_path / ".git" / "config").write_text('[remote "origin"]\n\turl = https://github.com/test-org/test-repo.git\n') - (tmp_path / ".git" / "HEAD").write_text("ref: refs/heads/main\n") + (repo / ".git").mkdir() + (repo / ".git" / "config").write_text('[remote "origin"]\n\turl = https://github.com/test-org/test-repo.git\n') + (repo / ".git" / "HEAD").write_text("ref: refs/heads/main\n") # Create a README - (tmp_path / "README.md").write_text("# Test Repository\n\nA test repo for integration testing.\n") + (repo / "README.md").write_text("# Test Repository\n\nA test repo for integration testing.\n") # Create a LICENSE - (tmp_path / "LICENSE").write_text("MIT License\n\nCopyright 2024 Test Org\n") + (repo / "LICENSE").write_text("MIT License\n\nCopyright 2024 Test Org\n") - return tmp_path + return repo class TestMCPServerIntegration: """Integration tests that start the MCP server and call tools.""" @pytest.mark.asyncio - async def test_server_starts_and_lists_tools(self): + async def test_server_starts_and_lists_tools(self, mcp_server_params): """Test that the server starts and exposes tools.""" - server_params = StdioServerParameters( - command="uv", - args=["run", "darnit", "serve", str(BASELINE_TOML)], - env=None, - ) - - async with stdio_client(server_params) as (read, write): + async with stdio_client(mcp_server_params) as (read, write): async with ClientSession(read, write) as session: # Initialize the session await session.initialize() @@ -109,15 +133,9 @@ async def test_server_starts_and_lists_tools(self): assert len(tool_names) == 22, f"Expected 22 tools, got {len(tool_names)}: {tool_names}" @pytest.mark.asyncio - async def test_list_available_checks(self): + async def test_list_available_checks(self, mcp_server_params): """Test calling the list_available_checks tool.""" - server_params = StdioServerParameters( - command="uv", - args=["run", "darnit", "serve", str(BASELINE_TOML)], - env=None, - ) - - async with stdio_client(server_params) as (read, write): + async with stdio_client(mcp_server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() @@ -138,15 +156,9 @@ async def test_list_available_checks(self): assert len(checks["level1"]) > 0 @pytest.mark.asyncio - async def test_audit_on_test_repo(self, test_repo): + async def test_audit_on_test_repo(self, test_repo, mcp_server_params): """Test running an audit on a test repository.""" - server_params = StdioServerParameters( - command="uv", - args=["run", "darnit", "serve", str(BASELINE_TOML)], - env=None, - ) - - async with stdio_client(server_params) as (read, write): + async with stdio_client(mcp_server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() @@ -178,15 +190,9 @@ async def test_audit_on_test_repo(self, test_repo): assert len(audit_result["results"]) > 0 @pytest.mark.asyncio - async def test_audit_with_tags_filter(self, test_repo): + async def test_audit_with_tags_filter(self, test_repo, mcp_server_params): """Test running an audit with tags filtering.""" - server_params = StdioServerParameters( - command="uv", - args=["run", "darnit", "serve", str(BASELINE_TOML)], - env=None, - ) - - async with stdio_client(server_params) as (read, write): + async with stdio_client(mcp_server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() @@ -212,6 +218,7 @@ async def test_audit_with_tags_filter(self, test_repo): # Should have results assert "results" in audit_result results = audit_result["results"] + assert results # All returned controls should belong to the VM domain baseline_controls = tomllib.loads(BASELINE_TOML.read_text(encoding="utf-8")).get("controls", {}) @@ -223,15 +230,9 @@ async def test_audit_with_tags_filter(self, test_repo): assert domain == "VM", f"Expected VM domain control, got {control_id} with domain {domain}" @pytest.mark.asyncio - async def test_get_project_config_no_config(self, test_repo): + async def test_get_project_config_no_config(self, test_repo, mcp_server_params): """Test get_project_config when no config exists.""" - server_params = StdioServerParameters( - command="uv", - args=["run", "darnit", "serve", str(BASELINE_TOML)], - env=None, - ) - - async with stdio_client(server_params) as (read, write): + async with stdio_client(mcp_server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() @@ -249,26 +250,27 @@ async def test_get_project_config_no_config(self, test_repo): assert "No .project.yaml found" in content.text or "init_project_config" in content.text @pytest.mark.asyncio - async def test_get_pending_data_and_confirm_project_data(self, test_repo): + async def test_get_pending_data_and_confirm_project_data(self, test_repo, mcp_server_params): """Test get_pending_data and confirm_project_data round trip.""" - - server_params = StdioServerParameters( - command="uv", - args=["run", "darnit", "serve", str(BASELINE_TOML)], - env=None, - ) - - async with stdio_client(server_params) as (read, write): + async with stdio_client(mcp_server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() - # Call get_pending_data + # Call get_pending_data before confirm pending_res = await session.call_tool( "get_pending_data", - {"local_path": str(test_repo)}, + { + "local_path": str(test_repo), + "owner": "test-org", + "repo": "test-repo", + }, ) assert pending_res.content - assert "AskUserQuestion" in pending_res.content[0].text + content_before = pending_res.content[0].text + assert "AskUserQuestion" in content_before + parsed_before = json.loads(content_before.split("\n---\n", 1)[-1]) + keys_before = [q["key"] for q in parsed_before.get("questions", [])] + assert "security_contact" in keys_before # Call confirm_project_data confirm_res = await session.call_tool( @@ -283,20 +285,29 @@ async def test_get_pending_data_and_confirm_project_data(self, test_repo): assert confirm_res.content assert "security_contact: confirmed" in confirm_res.content[0].text + # Call get_pending_data after confirm; security_contact should now be gone + after_res = await session.call_tool( + "get_pending_data", + { + "local_path": str(test_repo), + "owner": "test-org", + "repo": "test-repo", + }, + ) + assert after_res.content + content_after = after_res.content[0].text + parsed_after = json.loads(content_after.split("\n---\n", 1)[-1]) + keys_after = [q["key"] for q in parsed_after.get("questions", [])] + assert "security_contact" not in keys_after + class TestMCPServerToolDescriptions: """Test that tool descriptions are properly exposed.""" @pytest.mark.asyncio - async def test_tool_descriptions_are_set(self): + async def test_tool_descriptions_are_set(self, mcp_server_params): """Test that tools have descriptions from TOML.""" - server_params = StdioServerParameters( - command="uv", - args=["run", "darnit", "serve", str(BASELINE_TOML)], - env=None, - ) - - async with stdio_client(server_params) as (read, write): + async with stdio_client(mcp_server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() @@ -317,15 +328,9 @@ class TestMCPServerErrorHandling: """Test error handling in the MCP server.""" @pytest.mark.asyncio - async def test_audit_nonexistent_path(self): + async def test_audit_nonexistent_path(self, mcp_server_params): """Test audit with a non-existent path returns error gracefully.""" - server_params = StdioServerParameters( - command="uv", - args=["run", "darnit", "serve", str(BASELINE_TOML)], - env=None, - ) - - async with stdio_client(server_params) as (read, write): + async with stdio_client(mcp_server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize()