#!/usr/bin/env python3 """Golden test for tools.py — the MCP tool catalogue generator. Runnable standalone (`python3 test_tools.py`, non-zero exit on failure) AND under pytest (functions named `test_*`). It reads the COMMITTED `hanzo.yaml` and asserts the properties the catalogue promises, WITHOUT depending on any committed output (dist/ is generated, never committed): • name IS the id the operation is SERVED under — verbatim, every tool, no exception. This is the whole contract. A generator that synthesises names from paths is a THIRD naming rule, and a third rule is the disease this file exists to prevent — as is publishing `publish.py` rule 6's codegen suffix, which the MCP server does not answer to; • one name, one tool — a duplicate id makes the second tool unreachable; • callable verbs only — OPTIONS/TRACE are transport machinery, never tools; • self-contained schemas — every `$ref` a tool emits resolves inside that tool's OWN `$defs`. A pointer into `#/components` would leave the catalogue depending on a document the client does not ship; • total — every named operation in the document becomes a tool. A projection that silently drops operations is how a client ends up with 25 tools while the API has thousands; • deterministic — two generations are byte-identical, which is what the `--check` drift gate rests on. """ import json import os import tempfile import yaml import tools DOC = yaml.safe_load(open(tools.DOCUMENT)) CATALOGUE = tools.catalogue() def _ids(): """(served id, METHOD path) for every callable operation in the document.""" for path, item in (DOC.get("paths") or {}).items(): if not isinstance(item, dict): continue for method in tools.VERBS: op = item.get(method) if isinstance(op, dict) and tools.wire(op): yield tools.wire(op), f"{method.upper()} {path}" def test_name_is_the_operation_id(): # Every tool name is an id the document actually publishes — verbatim. published = {oid for oid, _ in _ids()} for t in CATALOGUE: assert t["name"] in published, \ f"tool {t['name']!r} is not an operationId in the document — a name was invented" def test_no_codegen_suffix_becomes_a_tool_name(): """Rule 6's suffix is a fact about Go, and the MCP server does not answer to it. `publish.py` renames an operationId when two collide under the identity a generator uses, so `DELETE /v1/o11y/sessions` is published as `DeleteSession_2` and served as `DeleteSession`. Shipped verbatim, that put a name in three client distributions — npm `@hanzo/mcp`, PyPI `hanzo-mcp`, crates.io `hanzo-mcp` — that nothing can be told to run, which is exactly the defect the catalogue exists to remove and the one shape of it a static artifact CAN refuse: the document itself records what it renamed. """ got = {t["name"] for t in CATALOGUE} wrong = [] for path, item in (DOC.get("paths") or {}).items(): if not isinstance(item, dict): continue for method in tools.VERBS: op = item.get(method) if not isinstance(op, dict) or not op.get("x-id"): continue if op["x-id"] not in got: wrong.append(f"{method.upper()} {path}: served as {op['x-id']!r}, no tool") if op["operationId"] in got: wrong.append(f"{method.upper()} {path}: published the codegen name " f"{op['operationId']!r} as a tool") assert not wrong, "\n".join(wrong) def test_names_unique_and_sorted(): names = [t["name"] for t in CATALOGUE] assert len(names) == len(set(names)), "duplicate tool names" assert names == sorted(names), "catalogue is not sorted by name" def test_total(): # Nothing named in the document is dropped on the floor. published = {oid for oid, _ in _ids()} got = {t["name"] for t in CATALOGUE} missing = published - got assert not missing, f"{len(missing)} operations produced no tool: {sorted(missing)[:10]}" def test_no_transport_verbs(): # OPTIONS/TRACE exist in the document and must never become tools. got = {t["name"] for t in CATALOGUE} leaked = [] for path, item in (DOC.get("paths") or {}).items(): if not isinstance(item, dict): continue for method in ("options", "trace", "head"): op = item.get(method) if isinstance(op, dict) and tools.wire(op) in got: leaked.append(f"{method.upper()} {path}") assert not leaked, f"transport verbs became tools: {leaked[:10]}" def test_schemas_are_self_contained(): # Every $ref resolves inside the tool's own $defs. Nothing points at # #/components, which the client does not ship. def refs(node): if isinstance(node, dict): r = node.get("$ref") if isinstance(r, str): yield r for v in node.values(): yield from refs(v) elif isinstance(node, list): for v in node: yield from refs(v) dangling, foreign = [], [] for t in CATALOGUE: s = t["inputSchema"] have = set(s.get("$defs") or {}) for r in refs(s): if not r.startswith("#/$defs/"): foreign.append((t["name"], r)) elif r[len("#/$defs/"):] not in have: dangling.append((t["name"], r)) assert not foreign, f"{len(foreign)} refs escape the tool: {foreign[:5]}" assert not dangling, f"{len(dangling)} refs resolve to nothing: {dangling[:5]}" def test_input_schema_shape(): for t in CATALOGUE: s = t["inputSchema"] assert isinstance(s, dict), t["name"] assert s.get("type") == "object", f"{t['name']}: inputSchema is not an object schema" assert isinstance(s.get("properties"), dict), f"{t['name']}: no properties map" for k in (s.get("required") or []): assert k in s["properties"], f"{t['name']}: required {k!r} is not a property" def test_deterministic(): a = tools.render(tools.catalogue()) b = tools.render(tools.catalogue()) assert a == b, "two generations differ" with tempfile.TemporaryDirectory() as d: p = os.path.join(d, "tools.json") n = tools.generate(p) assert n == len(CATALOGUE), f"generate() reported {n}, catalogue has {len(CATALOGUE)}" assert open(p).read() == a, "written bytes differ from render()" assert json.load(open(p))["count"] == len(CATALOGUE) def test_duplicate_id_is_refused(): # The one thing that must fail loudly rather than silently shadow a tool. doc = {"paths": {"/a": {"get": {"operationId": "dup"}}, "/b": {"get": {"operationId": "dup"}}}} with tempfile.TemporaryDirectory() as d: p = os.path.join(d, "doc.yaml") open(p, "w").write(yaml.safe_dump(doc)) try: tools.catalogue(p) except SystemExit as e: assert "dup" in str(e) return raise AssertionError("a duplicate operationId was accepted") def main(): tests = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)] failed = 0 for t in tests: try: t() print(f"ok {t.__name__}") except AssertionError as e: failed += 1 print(f"FAIL {t.__name__}: {e}") print(f"\n{len(tests) - failed}/{len(tests)} passed") return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(main())