From a8ae85c0a467514b6d40e898f79f6d5a09997d62 Mon Sep 17 00:00:00 2001 From: Jan Nedbal Date: Tue, 14 Apr 2026 19:33:41 +0200 Subject: [PATCH] tests: add mcp wrapper test suite --- tests/mcp/__init__.py | 0 tests/mcp/conftest.py | 32 +++++ tests/mcp/test_cli_runner.py | 86 +++++++++++++ tests/mcp/test_tool_dispatch.py | 211 ++++++++++++++++++++++++++++++++ tests/mcp/test_tool_schema.py | 140 +++++++++++++++++++++ 5 files changed, 469 insertions(+) create mode 100644 tests/mcp/__init__.py create mode 100644 tests/mcp/conftest.py create mode 100644 tests/mcp/test_cli_runner.py create mode 100644 tests/mcp/test_tool_dispatch.py create mode 100644 tests/mcp/test_tool_schema.py diff --git a/tests/mcp/__init__.py b/tests/mcp/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/mcp/conftest.py b/tests/mcp/conftest.py new file mode 100644 index 0000000..f4e8f46 --- /dev/null +++ b/tests/mcp/conftest.py @@ -0,0 +1,32 @@ +"""Fixtures for the MCP wrapper test suite.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass + + +@dataclass +class FakeProc: + stdout: str + stderr: str = "" + returncode: int = 0 + + +def make_runner(envelope: dict | None = None, *, stdout: str | None = None, stderr: str = "", returncode: int = 0): + """Return a callable compatible with :func:`subprocess.run`. + + Records the most recent ``cmd`` on ``runner.calls``. + """ + calls: list[list[str]] = [] + + def runner(cmd, capture_output=True, text=True, check=False): # noqa: ARG001 + calls.append(list(cmd)) + if stdout is not None: + s = stdout + else: + s = json.dumps(envelope if envelope is not None else {"ok": True, "command": "x", "status": "ok"}) + return FakeProc(stdout=s, stderr=stderr, returncode=returncode) + + runner.calls = calls # type: ignore[attr-defined] + return runner diff --git a/tests/mcp/test_cli_runner.py b/tests/mcp/test_cli_runner.py new file mode 100644 index 0000000..1ae44a2 --- /dev/null +++ b/tests/mcp/test_cli_runner.py @@ -0,0 +1,86 @@ +"""Tests for :mod:`metin_release_mcp.cli_runner`.""" + +from __future__ import annotations + +import json + +import pytest + +from metin_release_mcp import cli_runner +from metin_release_mcp.errors import CliNotFoundError, CliUnparseableOutputError + +from .conftest import make_runner + + +def test_run_cli_success_parses_envelope(): + runner = make_runner({"ok": True, "command": "release inspect", "status": "inspected", "stats": {"file_count": 3}}) + result = cli_runner.run_cli( + ["release", "inspect", "--json", "--source", "/tmp/x"], + binary="/fake/metin-release", + runner=runner, + ) + assert result.ok is True + assert result.envelope["status"] == "inspected" + assert result.envelope["stats"]["file_count"] == 3 + assert result.returncode == 0 + assert runner.calls[0][0] == "/fake/metin-release" + assert "--json" in runner.calls[0] + + +def test_run_cli_error_envelope_passed_through(): + err = { + "ok": False, + "command": "release sign", + "status": "failed", + "error": {"code": "key_permission", "message": "key must be mode 600"}, + } + runner = make_runner(err, stderr="error: key_permission\n", returncode=3) + result = cli_runner.run_cli(["release", "sign"], binary="/fake/mr", runner=runner) + assert result.ok is False + assert result.returncode == 3 + assert result.envelope["error"]["code"] == "key_permission" + assert "key_permission" in result.stderr + + +def test_run_cli_unparseable_output_raises(): + runner = make_runner(stdout="not-json-at-all", returncode=0) + with pytest.raises(CliUnparseableOutputError): + cli_runner.run_cli(["release", "inspect"], binary="/fake/mr", runner=runner) + + +def test_run_cli_empty_stdout_raises(): + runner = make_runner(stdout=" \n", stderr="boom", returncode=2) + with pytest.raises(CliUnparseableOutputError): + cli_runner.run_cli(["release", "inspect"], binary="/fake/mr", runner=runner) + + +def test_run_cli_non_object_json_raises(): + runner = make_runner(stdout=json.dumps([1, 2, 3])) + with pytest.raises(CliUnparseableOutputError): + cli_runner.run_cli(["release", "inspect"], binary="/fake/mr", runner=runner) + + +def test_run_cli_file_not_found_raises(): + def boom(*args, **kwargs): + raise FileNotFoundError("no such file") + + with pytest.raises(CliNotFoundError): + cli_runner.run_cli(["release", "inspect"], binary="/does/not/exist", runner=boom) + + +def test_resolve_binary_env_override(monkeypatch): + monkeypatch.setenv("METIN_RELEASE_BINARY", "/custom/path/metin-release") + assert cli_runner.resolve_binary() == "/custom/path/metin-release" + + +def test_resolve_binary_which_fallback(monkeypatch): + monkeypatch.delenv("METIN_RELEASE_BINARY", raising=False) + monkeypatch.setattr(cli_runner.shutil, "which", lambda name: "/usr/bin/metin-release") + assert cli_runner.resolve_binary() == "/usr/bin/metin-release" + + +def test_resolve_binary_missing_raises(monkeypatch): + monkeypatch.delenv("METIN_RELEASE_BINARY", raising=False) + monkeypatch.setattr(cli_runner.shutil, "which", lambda name: None) + with pytest.raises(CliNotFoundError): + cli_runner.resolve_binary() diff --git a/tests/mcp/test_tool_dispatch.py b/tests/mcp/test_tool_dispatch.py new file mode 100644 index 0000000..b2a3d2d --- /dev/null +++ b/tests/mcp/test_tool_dispatch.py @@ -0,0 +1,211 @@ +"""Dispatch tests — input dict to CLI argv translation.""" + +from __future__ import annotations + +import json + +import pytest + +from metin_release_mcp import cli_runner, server +from metin_release_mcp.tool_defs import TOOLS_BY_NAME, build_cli_args + +from .conftest import make_runner + + +def test_inspect_maps_source_to_flag(): + spec = TOOLS_BY_NAME["release_inspect"] + argv = build_cli_args(spec, {"source": "/tmp/client"}) + assert argv == ["release", "inspect", "--json", "--source", "/tmp/client"] + + +def test_build_manifest_full_flags(): + spec = TOOLS_BY_NAME["release_build_manifest"] + argv = build_cli_args( + spec, + { + "source": "/src", + "version": "2026.04.14-1", + "out": "/out/manifest.json", + "previous": "2026.04.13-1", + "notes": "/notes.md", + "launcher": "Metin2Launcher.exe", + "created_at": "2026-04-14T00:00:00Z", + }, + ) + # Required args present in order declared + assert argv[0:3] == ["release", "build-manifest", "--json"] + assert "--source" in argv and argv[argv.index("--source") + 1] == "/src" + assert "--version" in argv and argv[argv.index("--version") + 1] == "2026.04.14-1" + assert "--out" in argv and argv[argv.index("--out") + 1] == "/out/manifest.json" + assert "--previous" in argv and argv[argv.index("--previous") + 1] == "2026.04.13-1" + assert "--created-at" in argv + assert argv[argv.index("--created-at") + 1] == "2026-04-14T00:00:00Z" + + +def test_build_manifest_omits_missing_optionals(): + spec = TOOLS_BY_NAME["release_build_manifest"] + argv = build_cli_args( + spec, + {"source": "/src", "version": "v1", "out": "/m.json"}, + ) + assert "--previous" not in argv + assert "--notes" not in argv + assert "--launcher" not in argv + assert "--created-at" not in argv + + +def test_none_valued_optional_is_omitted(): + spec = TOOLS_BY_NAME["release_build_manifest"] + argv = build_cli_args( + spec, + {"source": "/s", "version": "v", "out": "/m", "previous": None, "notes": None}, + ) + assert "--previous" not in argv + assert "--notes" not in argv + + +def test_boolean_flag_true_adds_flag_without_value(): + spec = TOOLS_BY_NAME["release_upload_blobs"] + argv = build_cli_args( + spec, + { + "release_dir": "/rel", + "rsync_target": "user@host:/srv/updates", + "dry_run": True, + "yes": True, + }, + ) + assert "--dry-run" in argv + assert "--yes" in argv + # No value should follow --dry-run + i = argv.index("--dry-run") + # next element either end of list or another flag + assert i == len(argv) - 1 or argv[i + 1].startswith("--") + + +def test_boolean_flag_false_omits_flag(): + spec = TOOLS_BY_NAME["release_upload_blobs"] + argv = build_cli_args( + spec, + {"release_dir": "/rel", "rsync_target": "tgt", "dry_run": False, "yes": False}, + ) + assert "--dry-run" not in argv + assert "--yes" not in argv + + +def test_path_with_spaces_passes_through_as_single_argv_element(): + spec = TOOLS_BY_NAME["release_inspect"] + argv = build_cli_args(spec, {"source": "/tmp/has spaces/client root"}) + # argv is a list — no shell involved, so spaces stay in one element + assert "/tmp/has spaces/client root" in argv + assert argv.count("--source") == 1 + + +def test_diff_remote_numeric_timeout_serialised_as_string(): + spec = TOOLS_BY_NAME["release_diff_remote"] + argv = build_cli_args( + spec, + {"manifest": "/m.json", "base_url": "https://x/", "timeout": 7.5}, + ) + assert argv[argv.index("--timeout") + 1] == "7.5" + + +def test_publish_mixes_booleans_and_strings(): + spec = TOOLS_BY_NAME["release_publish"] + argv = build_cli_args( + spec, + { + "source": "/src", + "version": "v1", + "out": "/out", + "key": "/k", + "rsync_target": "t", + "base_url": "https://x/", + "public_key": "deadbeef", + "force": True, + "dry_run_upload": True, + "yes": False, + }, + ) + assert "--force" in argv + assert "--dry-run-upload" in argv + assert "--yes" not in argv + + +# --------------------------------------------------------------------------- +# server.dispatch integration against a stubbed CLI runner +# --------------------------------------------------------------------------- + + +@pytest.fixture +def stub_runner(monkeypatch): + def install(envelope=None, *, stdout=None, stderr="", returncode=0): + runner = make_runner(envelope, stdout=stdout, stderr=stderr, returncode=returncode) + monkeypatch.setattr( + server, + "run_cli", + lambda argv_tail: cli_runner.run_cli( + argv_tail, binary="/fake/metin-release", runner=runner + ), + ) + return runner + + return install + + +def test_dispatch_success_returns_envelope(stub_runner): + runner = stub_runner( + {"ok": True, "command": "release inspect", "status": "inspected", "stats": {"file_count": 42}} + ) + envelope, stderr = server.dispatch("release_inspect", {"source": "/tmp/x"}) + assert envelope["ok"] is True + assert envelope["stats"]["file_count"] == 42 + # the runner saw --json + assert "--json" in runner.calls[0] + assert "--source" in runner.calls[0] + assert "/tmp/x" in runner.calls[0] + + +def test_dispatch_cli_error_envelope_passes_through(stub_runner): + err = { + "ok": False, + "command": "release sign", + "status": "failed", + "error": {"code": "key_permission", "message": "mode 644"}, + } + stub_runner(err, stderr="error\n", returncode=3) + envelope, stderr = server.dispatch( + "release_sign", {"manifest": "/m.json", "key": "/k"} + ) + assert envelope == err + assert "error" in stderr + + +def test_dispatch_cli_binary_missing_returns_wrapper_error(monkeypatch): + def boom(argv_tail): + from metin_release_mcp.errors import CliNotFoundError + + raise CliNotFoundError("metin-release binary not found") + + monkeypatch.setattr(server, "run_cli", boom) + envelope, stderr = server.dispatch("release_inspect", {"source": "/x"}) + assert envelope["ok"] is False + assert envelope["error"]["code"] == "cli_not_found" + + +def test_dispatch_invalid_input_returns_wrapper_error(): + envelope, _ = server.dispatch("release_inspect", {}) + assert envelope["ok"] is False + assert envelope["error"]["code"] == "invalid_tool_input" + + +def test_dispatch_unparseable_output_returns_wrapper_error(monkeypatch): + def boom(argv_tail): + from metin_release_mcp.errors import CliUnparseableOutputError + + raise CliUnparseableOutputError("not json") + + monkeypatch.setattr(server, "run_cli", boom) + envelope, _ = server.dispatch("release_inspect", {"source": "/x"}) + assert envelope["ok"] is False + assert envelope["error"]["code"] == "cli_unparseable_output" diff --git a/tests/mcp/test_tool_schema.py b/tests/mcp/test_tool_schema.py new file mode 100644 index 0000000..aa9ad8e --- /dev/null +++ b/tests/mcp/test_tool_schema.py @@ -0,0 +1,140 @@ +"""Schema tests — each MCP tool must mirror its CLI subcommand exactly.""" + +from __future__ import annotations + +import argparse + +import pytest + +from metin_release.commands import ( + build_manifest, + diff_remote, + inspect, + promote, + publish, + sign, + upload_blobs, + verify_public, +) +from metin_release_mcp.errors import UnknownToolError +from metin_release_mcp.server import dispatch +from metin_release_mcp.tool_defs import ( + TOOL_SPECS, + TOOLS_BY_NAME, + build_cli_args, + json_schema, +) + + +EXPECTED_TOOL_NAMES = { + "release_inspect", + "release_build_manifest", + "release_sign", + "release_diff_remote", + "release_upload_blobs", + "release_promote", + "release_verify_public", + "release_publish", +} + + +def test_tool_catalogue_is_exactly_the_phase_one_set(): + assert {t.name for t in TOOL_SPECS} == EXPECTED_TOOL_NAMES + assert len(TOOL_SPECS) == 8 + + +def test_every_tool_has_release_subcommand_and_description(): + for spec in TOOL_SPECS: + assert spec.subcommand[0] == "release" + assert spec.description.strip() + schema = json_schema(spec) + assert schema["type"] == "object" + assert schema["additionalProperties"] is False + + +def _argparse_flags_for(add_parser_fn) -> dict[str, dict]: + """Introspect a command's argparse signature. + + Returns a mapping from flag name (e.g. ``--source``) to a dict with + ``required`` and ``kind`` ("boolean" for store_true, else "value"). + """ + parser = argparse.ArgumentParser() + sub = parser.add_subparsers() + sp = add_parser_fn(sub) + out: dict[str, dict] = {} + for action in sp._actions: + if not action.option_strings: + continue + if action.dest in ("help",): + continue + flag = action.option_strings[0] + is_bool = isinstance(action, argparse._StoreTrueAction) + out[flag] = {"required": action.required, "kind": "boolean" if is_bool else "value"} + return out + + +_COMMAND_MODULES = { + "release_inspect": inspect, + "release_build_manifest": build_manifest, + "release_sign": sign, + "release_diff_remote": diff_remote, + "release_upload_blobs": upload_blobs, + "release_promote": promote, + "release_verify_public": verify_public, + "release_publish": publish, +} + + +@pytest.mark.parametrize("tool_name", sorted(EXPECTED_TOOL_NAMES)) +def test_schema_mirrors_argparse(tool_name): + spec = TOOLS_BY_NAME[tool_name] + mod = _COMMAND_MODULES[tool_name] + argparse_flags = _argparse_flags_for(mod.add_parser) + + spec_flags = {f.cli_flag: f for f in spec.fields} + assert set(spec_flags) == set(argparse_flags), ( + f"{tool_name}: spec flags {set(spec_flags)} != argparse flags {set(argparse_flags)}" + ) + for flag, info in argparse_flags.items(): + field = spec_flags[flag] + assert field.required == info["required"], f"{tool_name} {flag} required mismatch" + if info["kind"] == "boolean": + assert field.kind == "boolean", f"{tool_name} {flag} expected boolean" + else: + assert field.kind != "boolean", f"{tool_name} {flag} should not be boolean" + + +@pytest.mark.parametrize("tool_name", sorted(EXPECTED_TOOL_NAMES)) +def test_required_fields_in_json_schema(tool_name): + spec = TOOLS_BY_NAME[tool_name] + schema = json_schema(spec) + required_in_spec = {f.name for f in spec.fields if f.required} + required_in_schema = set(schema.get("required", [])) + assert required_in_schema == required_in_spec + + +def test_unknown_tool_rejected_by_dispatch(): + with pytest.raises(UnknownToolError): + dispatch("nope_not_a_tool", {}) + + +def test_unknown_tool_name_not_in_catalogue(): + assert "release_rollback" not in TOOLS_BY_NAME + assert "erp_reserve" not in TOOLS_BY_NAME + assert "m2pack_build" not in TOOLS_BY_NAME + + +def test_build_cli_args_rejects_missing_required(): + from metin_release_mcp.errors import InvalidToolInputError + + spec = TOOLS_BY_NAME["release_inspect"] + with pytest.raises(InvalidToolInputError): + build_cli_args(spec, {}) + + +def test_build_cli_args_rejects_unknown_fields(): + from metin_release_mcp.errors import InvalidToolInputError + + spec = TOOLS_BY_NAME["release_inspect"] + with pytest.raises(InvalidToolInputError): + build_cli_args(spec, {"source": "/x", "nonsense": 1})