From 1c7337e8874f4a916fddd62127bdbf80769b5f8e Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 16 Sep 2026 16:03:30 +0000 Subject: [PATCH] test: raise coverage for runner, SSE errors, CLI headers, reports (#64) Add OpenAICompatibleRunner urlopen tests, SSE timeout/empty-endpoint/ POST-and-notification auth failures via FakeMcpSseServer modes, CLI --header/--sse checks, report markdown helpers, and docs/coverage.md waivers for remaining cli/mcp_capture edges. Co-authored-by: Abhinaysai Kamineni --- docs/coverage.md | 15 +++ tests/fixtures/fake_mcp_sse_server.py | 41 ++++++- tests/test_cli.py | 51 +++++++++ tests/test_mcp_capture.py | 43 +++++++ tests/test_report_render.py | 95 ++++++++++++++++ tests/test_runner.py | 158 ++++++++++++++++++++++++++ 6 files changed, 400 insertions(+), 3 deletions(-) create mode 100644 docs/coverage.md create mode 100644 tests/test_report_render.py diff --git a/docs/coverage.md b/docs/coverage.md new file mode 100644 index 0000000..79cc23b --- /dev/null +++ b/docs/coverage.md @@ -0,0 +1,15 @@ +# Coverage targets (#64) + +CI runs `pytest --cov=tool_semantics`. After this work: + +- **Target:** no module below **80%** without an explicit waiver below. +- Hotspots historically: `runner.py` (HTTP adapter), `mcp_capture.py` (SSE + error branches), `cli.py` (header / SSE UX), `report.py` (model/stability + renderers). + +## Waivers + +| Module | Notes | +| --- | --- | +| `cli.py` | Remaining gaps are interactive Rich error paths and rare provenance races; covered by happy-path + header/SSE CLI tests. Raise further when adding commands. | +| `mcp_capture.py` | Deep stdio/HTTP notification and protocol-edge branches; SSE timeout / empty endpoint / POST auth are covered. Full matrix needs more fixture modes. | diff --git a/tests/fixtures/fake_mcp_sse_server.py b/tests/fixtures/fake_mcp_sse_server.py index fb9956b..54a9aaf 100644 --- a/tests/fixtures/fake_mcp_sse_server.py +++ b/tests/fixtures/fake_mcp_sse_server.py @@ -24,8 +24,20 @@ class FakeMcpSseServer: - def __init__(self, *, require_auth: bool = False) -> None: + def __init__( + self, + *, + require_auth: bool = False, + hang_without_endpoint: bool = False, + empty_endpoint: bool = False, + post_fail_auth: bool = False, + notify_fail_auth: bool = False, + ) -> None: self.require_auth = require_auth + self.hang_without_endpoint = hang_without_endpoint + self.empty_endpoint = empty_endpoint + self.post_fail_auth = post_fail_auth + self.notify_fail_auth = notify_fail_auth self._httpd: ThreadingHTTPServer | None = None self._thread: threading.Thread | None = None self.base_url = "" @@ -33,6 +45,10 @@ def __init__(self, *, require_auth: bool = False) -> None: def start(self) -> None: require_auth = self.require_auth + hang_without_endpoint = self.hang_without_endpoint + empty_endpoint = self.empty_endpoint + post_fail_auth = self.post_fail_auth + notify_fail_auth = self.notify_fail_auth class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" @@ -63,15 +79,28 @@ def do_GET(self) -> None: # noqa: N802 self.send_header("Cache-Control", "no-cache") self.send_header("Connection", "keep-alive") self.end_headers() + if hang_without_endpoint: + while getattr(self.server, "is_running", False): + threading.Event().wait(0.1) + return host, port = self.server.server_address # type: ignore[attr-defined] - endpoint = f"http://{host}:{port}/message" - self.wfile.write(f"event: endpoint\ndata: {endpoint}\n\n".encode()) + if empty_endpoint: + self.wfile.write(b"event: endpoint\ndata: \n\n") + else: + endpoint = f"http://{host}:{port}/message" + self.wfile.write(f"event: endpoint\ndata: {endpoint}\n\n".encode()) self.wfile.flush() # Hold the SSE connection open until server shutdown. while getattr(self.server, "is_running", False): threading.Event().wait(0.1) def do_POST(self) -> None: # noqa: N802 + if post_fail_auth and self.headers.get("Authorization") != "Bearer secret-token": + self.send_response(401) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"error":"unauthorized"}') + return if self._unauthorized(): return if urlparse(self.path).path != "/message": @@ -89,6 +118,12 @@ def do_POST(self) -> None: # noqa: N802 method = message.get("method") request_id = message.get("id") if method == "notifications/initialized": + if notify_fail_auth: + self.send_response(401) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"error":"unauthorized"}') + return self.send_response(202) self.end_headers() return diff --git a/tests/test_cli.py b/tests/test_cli.py index 7a90e9e..9e34631 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -192,3 +192,54 @@ def test_compare_verbose_and_config_ignore(tmp_path: Path) -> None: assert "Changes=" in result.stderr assert result.exit_code == 0 assert "compatible" in result.stdout.lower() + + +def test_capture_mcp_header_parse_errors() -> None: + result = runner.invoke( + app, + [ + "capture-mcp", + "-o", + "/tmp/x.json", + "--header", + "NoColon", + "--http", + "http://127.0.0.1:9/", + ], + ) + assert result.exit_code == 2 + assert "Invalid --header" in result.stdout + + result = runner.invoke( + app, + [ + "capture-mcp", + "-o", + "/tmp/x.json", + "--header", + ": value", + "--http", + "http://127.0.0.1:9/", + ], + ) + assert result.exit_code == 2 + assert "Invalid --header name" in result.stdout + + +def test_capture_mcp_sse_cli_happy_path(tmp_path: Path) -> None: + from fixtures.fake_mcp_sse_server import FakeMcpSseServer + + server = FakeMcpSseServer() + server.start() + try: + out = tmp_path / "sse.json" + result = runner.invoke( + app, + ["capture-mcp", "-o", str(out), "--sse", server.sse_url], + ) + assert result.exit_code == 0, result.stdout + payload = json.loads(out.read_text(encoding="utf-8")) + assert payload["protocol"] == "mcp-sse" + assert payload["metadata"]["transport"] == "sse" + finally: + server.stop() diff --git a/tests/test_mcp_capture.py b/tests/test_mcp_capture.py index 28be0fc..68ef9ac 100644 --- a/tests/test_mcp_capture.py +++ b/tests/test_mcp_capture.py @@ -178,3 +178,46 @@ def test_capture_mcp_remote_auth_does_not_fallback() -> None: def test_capture_mcp_remote_total_failure() -> None: with pytest.raises(McpCaptureError, match=r"\[unsupported_server\].*Remote MCP capture failed"): capture_mcp_remote("http://127.0.0.1:9/no-mcp-here", timeout=1.0) + + +def test_capture_mcp_sse_timeout_waiting_for_endpoint() -> None: + server = FakeMcpSseServer(hang_without_endpoint=True) + server.start() + try: + with pytest.raises(McpCaptureError, match="[Tt]imeout|endpoint"): + capture_mcp_sse(server.sse_url, timeout=0.3) + finally: + server.stop() + + +def test_capture_mcp_sse_empty_endpoint_event() -> None: + server = FakeMcpSseServer(empty_endpoint=True) + server.start() + try: + with pytest.raises(McpCaptureError, match="empty URL|endpoint"): + capture_mcp_sse(server.sse_url, timeout=2.0) + finally: + server.stop() + + +def test_capture_mcp_sse_post_auth_failure() -> None: + server = FakeMcpSseServer(post_fail_auth=True) + server.start() + try: + with pytest.raises(McpCaptureError, match="POST failed with HTTP 401"): + capture_mcp_sse(server.sse_url, timeout=5.0) + finally: + server.stop() + + +def test_capture_mcp_sse_notification_auth_failure() -> None: + server = FakeMcpSseServer(notify_fail_auth=True) + server.start() + try: + with pytest.raises( + McpCaptureError, + match="notification.*401|authentication/HTTP error 401", + ): + capture_mcp_sse(server.sse_url, timeout=5.0) + finally: + server.stop() diff --git a/tests/test_report_render.py b/tests/test_report_render.py new file mode 100644 index 0000000..e8560b1 --- /dev/null +++ b/tests/test_report_render.py @@ -0,0 +1,95 @@ +"""Tests for model-probe / stability report renderers (#64).""" + +from __future__ import annotations + +import json + +from tool_semantics.probes import ( + ModelProbeOutcome, + ModelProbeReport, + ModelProbeResult, + ProbeMetrics, + StabilityProbeSummary, + StabilityReport, + TrialDetail, +) +from tool_semantics.report import ( + render_model_probe_report_markdown, + render_stability_json, + render_stability_markdown, +) +from tool_semantics.runner import RunnerMetadata + + +def test_render_model_probe_report_markdown() -> None: + report = ModelProbeReport( + results=[ + ModelProbeResult( + probe_id="ok", + passed=True, + message="Selected search_issues.", + selected_tool="search_issues", + outcome=ModelProbeOutcome.OK, + tool_selection_correct=True, + arguments_valid=True, + runner=RunnerMetadata(provider="fake", model="m"), + ), + ModelProbeResult( + probe_id="miss", + passed=False, + message="Model returned no tool call.", + outcome=ModelProbeOutcome.MISSING_DATA, + tool_selection_correct=False, + ), + ] + ) + md = render_model_probe_report_markdown(report) + assert "Probe metrics" in md + assert "Per-probe results" in md + assert "`ok`" in md + assert "`miss`" in md + assert "search_issues" in md + + +def test_render_stability_json_and_markdown() -> None: + report = StabilityReport( + trial_count=2, + seed=3, + summaries=[ + StabilityProbeSummary( + probe_id="search", + trials=[ + TrialDetail( + trial_index=0, + selected_tool="search_issues", + arguments={"query": "x"}, + passed=True, + outcome=ModelProbeOutcome.OK, + message="ok", + ), + TrialDetail( + trial_index=1, + selected_tool="search_issues", + arguments={"query": "x"}, + passed=True, + outcome=ModelProbeOutcome.OK, + message="ok", + ), + ], + stability_score=1.0, + unstable=False, + deterministic_failure=False, + aggregate_passed=True, + message="Stable across trials.", + ) + ], + metrics=ProbeMetrics(probe_count=2, evaluated_count=2, tool_selection_accuracy=1.0), + ) + payload = json.loads(render_stability_json(report)) + assert payload["trial_count"] == 2 + assert payload["seed"] == 3 + assert payload["summaries"][0]["probe_id"] == "search" + md = render_stability_markdown(report) + assert "Probe stability report" in md + assert "Aggregate metrics" in md + assert "`search`" in md diff --git a/tests/test_runner.py b/tests/test_runner.py index 1e35bf3..f6b85d1 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -1,3 +1,6 @@ +import json +from io import BytesIO + import pytest from tool_semantics.runner import ( @@ -46,6 +49,161 @@ def test_fake_runner_respects_max_calls() -> None: runner.complete(system="s", user="u", tools=[], config=RunnerConfig(max_calls=0)) +def test_fake_runner_exhausts_scripted_responses() -> None: + runner = FakeModelRunner([]) + with pytest.raises(RunnerError, match="no remaining"): + runner.complete(system="s", user="u", tools=[]) + + def test_openai_compatible_runner_requires_credentials() -> None: with pytest.raises(ValueError, match="api_key"): OpenAICompatibleRunner(model="gpt-test", api_key="") + with pytest.raises(ValueError, match="model"): + OpenAICompatibleRunner(model="", api_key="k") + + +def test_openai_compatible_runner_success(monkeypatch: pytest.MonkeyPatch) -> None: + payload = { + "choices": [ + { + "message": { + "content": None, + "tool_calls": [ + { + "function": { + "name": "search_issues", + "arguments": '{"query": "bugs"}', + } + } + ], + } + } + ] + } + + class FakeResponse: + def read(self) -> bytes: + return json.dumps(payload).encode("utf-8") + + def __enter__(self) -> "FakeResponse": + return self + + def __exit__(self, *args: object) -> None: + return None + + monkeypatch.setattr( + "tool_semantics.runner.urllib.request.urlopen", + lambda *args, **kwargs: FakeResponse(), + ) + runner = OpenAICompatibleRunner(model="gpt-test", api_key="sk-test", base_url="http://example/") + assert runner.metadata.model == "gpt-test" + completion = runner.complete( + system="sys", + user="find bugs", + tools=[{"type": "function", "function": {"name": "search_issues"}}], + config=RunnerConfig(seed=9, max_retries=0), + ) + assert completion.tool_calls[0].name == "search_issues" + assert completion.tool_calls[0].arguments == {"query": "bugs"} + assert completion.metadata.run_config["seed"] == 9 + + +def test_openai_compatible_runner_http_error(monkeypatch: pytest.MonkeyPatch) -> None: + import urllib.error + + def boom(*args: object, **kwargs: object) -> None: + raise urllib.error.HTTPError( + "http://example/chat/completions", + 500, + "err", + hdrs=None, # type: ignore[arg-type] + fp=BytesIO(b"upstream failed"), + ) + + monkeypatch.setattr("tool_semantics.runner.urllib.request.urlopen", boom) + monkeypatch.setattr("tool_semantics.runner.time.sleep", lambda *_: None) + runner = OpenAICompatibleRunner(model="gpt-test", api_key="sk-test") + with pytest.raises(RunnerError, match="failed after retries"): + runner.complete( + system="s", + user="u", + tools=[], + config=RunnerConfig(max_retries=1), + ) + + +def test_openai_compatible_runner_parse_malformed_tool_args( + monkeypatch: pytest.MonkeyPatch, +) -> None: + payload = { + "choices": [ + { + "message": { + "tool_calls": [ + {"function": {"name": "echo", "arguments": "{not-json"}}, + {"function": {"name": "", "arguments": "{}"}}, + "skip-me", + ] + } + } + ] + } + + class FakeResponse: + def read(self) -> bytes: + return json.dumps(payload).encode("utf-8") + + def __enter__(self) -> "FakeResponse": + return self + + def __exit__(self, *args: object) -> None: + return None + + monkeypatch.setattr( + "tool_semantics.runner.urllib.request.urlopen", + lambda *args, **kwargs: FakeResponse(), + ) + runner = OpenAICompatibleRunner(model="gpt-test", api_key="sk-test") + completion = runner.complete(system="s", user="u", tools=[], config=RunnerConfig(max_retries=0)) + assert len(completion.tool_calls) == 1 + assert completion.tool_calls[0].name == "echo" + assert completion.tool_calls[0].arguments == {} + + +def test_openai_compatible_runner_parse_dict_and_non_dict_args( + monkeypatch: pytest.MonkeyPatch, +) -> None: + payload = { + "choices": [ + { + "message": { + "content": "ok", + "tool_calls": [ + {"function": {"name": "echo", "arguments": {"text": "hi"}}}, + {"function": {"name": "bad", "arguments": ["not", "a", "dict"]}}, + ], + } + } + ] + } + + class FakeResponse: + def read(self) -> bytes: + return json.dumps(payload).encode("utf-8") + + def __enter__(self) -> "FakeResponse": + return self + + def __exit__(self, *args: object) -> None: + return None + + monkeypatch.setattr( + "tool_semantics.runner.urllib.request.urlopen", + lambda *args, **kwargs: FakeResponse(), + ) + runner = OpenAICompatibleRunner(model="gpt-test", api_key="sk-test") + completion = runner.complete(system="s", user="u", tools=[], config=RunnerConfig(max_retries=0)) + assert completion.text == "ok" + assert len(completion.tool_calls) == 2 + assert completion.tool_calls[0].arguments == {"text": "hi"} + assert completion.tool_calls[1].arguments == {}