test: add ConnectorBench MCP contract conformance suite

tests/connectorbench.py

@@ -0,0 +1,105 @@
+"""Reusable, dependency-light conformance checks for MCP tool catalogs.
+
+ConnectorBench validates the contract exposed by a server's list_tools result.
+It intentionally does not invoke tools: inspection must be safe for connectors
+whose calls can mutate a model, launch a solver, or consume a commercial license.
+"""
+from __future__ import annotations
+
+import json
+import re
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+
+_NAME = re.compile(r"^[A-Za-z0-9_.-]+$")
+
+
+def _field(value: Any, name: str, default: Any = None) -> Any:
+    """Read an MCP model field from either an object or a plain mapping."""
+    if isinstance(value, Mapping):
+        return value.get(name, default)
+    return getattr(value, name, default)
+
+
+def inspect_tool_contracts(tools: Sequence[Any]) -> dict[str, Any]:
+    """Inspect an MCP tool catalog and return a stable JSON-serializable report.
+
+    Errors indicate malformed or ambiguous contracts. Missing descriptions are
+    warnings because MCP permits them, but they reduce discoverability.
+    No connector tool is called and no optional engine is imported here.
+    """
+    errors: list[dict[str, str]] = []
+    warnings: list[dict[str, str]] = []
+    seen: set[str] = set()
+    catalog: list[dict[str, Any]] = []
+
+    def issue(target: str, code: str, message: str, *, warning: bool = False) -> None:
+        item = {"target": target, "code": code, "message": message}
+        (warnings if warning else errors).append(item)
+
+    for index, tool in enumerate(tools):
+        name = _field(tool, "name")
+        target = name if isinstance(name, str) and name else f"tools[{index}]"
+
+        if not isinstance(name, str) or not name.strip():
+            issue(target, "invalid_name", "Tool name must be a non-empty string.")
+        elif not _NAME.fullmatch(name):
+            issue(target, "invalid_name", "Tool name may contain only letters, digits, '.', '_' and '-'.")
+        elif name in seen:
+            issue(target, "duplicate_name", f"Tool name {name!r} is registered more than once.")
+        else:
+            seen.add(name)
+
+        description = _field(tool, "description")
+        if not isinstance(description, str) or not description.strip():
+            issue(
+                target,
+                "missing_description",
+                "Tool has no non-empty description; agents may struggle to discover its purpose.",
+                warning=True,
+            )
+
+        # Keep schema validity independent from catalog-level name errors.
+        schema_error_count = len(errors)
+        schema = _field(tool, "inputSchema", _field(tool, "input_schema"))
+        if not isinstance(schema, Mapping):
+            issue(target, "invalid_input_schema", "Tool inputSchema must be a JSON Schema object.")
+            catalog.append({"name": target, "schema_valid": False})
+            continue
+
+        if schema.get("type") not in (None, "object"):
+            issue(target, "invalid_schema_root_type", "Tool inputSchema root type must be 'object'.")
+
+        properties = schema.get("properties", {})
+        if not isinstance(properties, Mapping):
+            issue(target, "invalid_schema_properties", "inputSchema.properties must be an object.")
+            properties = {}
+
+        required = schema.get("required", [])
+        if not isinstance(required, list) or any(not isinstance(key, str) for key in required):
+            issue(target, "invalid_schema_required", "inputSchema.required must be a list of strings.")
+            required = []
+        # JSON Schema permits required names to be constrained by composed schemas.
+        try:
+            json.dumps(schema, allow_nan=False)
+        except (TypeError, ValueError):
+            issue(target, "non_json_schema", "inputSchema must be JSON-serializable without NaN or Infinity.")
+
+        catalog.append(
+            {
+                "name": target,
+                "schema_valid": len(errors) == schema_error_count,
+                "property_count": len(properties),
+                "required_count": len(required),
+            }
+        )
+
+    return {
+        "schema_version": 1,
+        "tool_count": len(tools),
+        "tools": catalog,
+        "errors": errors,
+        "warnings": warnings,
+        "passed": not errors,
+    }

tests/server_entry_smoke.py

@@ -124,6 +124,13 @@ def record_run(self, *args, **kwargs):
     tools = asyncio.run(server.list_tools())
     assert tools, f"{tool_name}: server registered no tools"
 
+    # Contract checks are read-only: they inspect catalog metadata but never
+    # invoke a model-mutating tool or launch a vendor engine.
+    from connectorbench import inspect_tool_contracts
+
+    report = inspect_tool_contracts(tools)
+    assert report["passed"], f"{tool_name}: ConnectorBench failures: {report['errors']}"
+
 
 if __name__ == "__main__":
     main(sys.argv[1])

tests/test_connectorbench.py

@@ -0,0 +1,78 @@
+"""Unit tests for ConnectorBench's safe, read-only MCP catalog checks."""
+from __future__ import annotations
+
+from connectorbench import inspect_tool_contracts
+
+
+def _tool(name="solve", schema=None, description="Run a solver"):
+    return {
+        "name": name,
+        "description": description,
+        "inputSchema": schema if schema is not None else {
+            "type": "object",
+            "properties": {"case": {"type": "string"}},
+            "required": ["case"],
+        },
+    }
+
+
+def test_valid_catalog_passes_and_is_json_serializable():
+    report = inspect_tool_contracts([_tool(), _tool("inspect", {"type": "object"})])
+
+    assert report["passed"] is True
+    assert report["tool_count"] == 2
+    assert report["errors"] == []
+    assert report["tools"][0]["required_count"] == 1
+
+
+def test_duplicate_tool_names_are_reported():
+    report = inspect_tool_contracts([_tool(), _tool()])
+
+    assert report["passed"] is False
+    assert any(item["code"] == "duplicate_name" for item in report["errors"])
+    # Duplicate registration is a catalog error, not a schema error.
+    assert all(item["schema_valid"] for item in report["tools"])
+
+
+def test_required_properties_can_be_defined_by_composed_schema():
+    report = inspect_tool_contracts([
+        _tool(schema={
+            "type": "object",
+            "allOf": [{"properties": {"bus_id": {"type": "string"}}}],
+            "required": ["bus_id"],
+        })
+    ])
+
+    assert report["passed"] is True
+    assert report["errors"] == []
+
+
+def test_schema_root_must_be_an_object():
+    report = inspect_tool_contracts([_tool(schema={"type": "array"})])
+
+    assert report["passed"] is False
+    assert any(item["code"] == "invalid_schema_root_type" for item in report["errors"])
+
+
+def test_missing_description_is_a_warning_not_a_failure():
+    report = inspect_tool_contracts([_tool(description="  ")])
+
+    assert report["passed"] is True
+    assert report["errors"] == []
+    assert any(item["code"] == "missing_description" for item in report["warnings"])
+
+
+def test_non_json_values_are_rejected():
+    report = inspect_tool_contracts([
+        _tool(schema={"type": "object", "properties": {"value": {"default": float("nan")}}})
+    ])
+
+    assert report["passed"] is False
+    assert any(item["code"] == "non_json_schema" for item in report["errors"])
+
+
+def test_tool_without_input_schema_fails_closed():
+    report = inspect_tool_contracts([{"name": "solve", "description": "Run a solver"}])
+
+    assert report["passed"] is False
+    assert any(item["code"] == "invalid_input_schema" for item in report["errors"])