tests: add mcp wrapper test suite
This commit is contained in:
0
tests/mcp/__init__.py
Normal file
0
tests/mcp/__init__.py
Normal file
32
tests/mcp/conftest.py
Normal file
32
tests/mcp/conftest.py
Normal file
@@ -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
|
||||
86
tests/mcp/test_cli_runner.py
Normal file
86
tests/mcp/test_cli_runner.py
Normal file
@@ -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()
|
||||
211
tests/mcp/test_tool_dispatch.py
Normal file
211
tests/mcp/test_tool_dispatch.py
Normal file
@@ -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"
|
||||
140
tests/mcp/test_tool_schema.py
Normal file
140
tests/mcp/test_tool_schema.py
Normal file
@@ -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})
|
||||
Reference in New Issue
Block a user