Files
AegisGitea-MCP/tests/test_catalog_build.py

627 lines
23 KiB
Python

"""Tests for schema generation, annotations and the catalog builder."""
from __future__ import annotations
import json
from typing import Any
import pytest
from jsonschema import Draft202012Validator
from aegis_gitea_mcp.catalog import Operation, load_catalog
from aegis_gitea_mcp.catalog.build import (
DESTRUCTIVE_POST_OPERATIONS,
READ_ONLY_POST_OPERATIONS,
build_catalog,
render_catalog,
spec_sha256,
)
from aegis_gitea_mcp.catalog.schema import (
MAX_DESCRIPTION_CHARS,
SUDO_SCHEMA,
resolve_schema,
trim_description,
)
FILE_SCHEMA_KEYS = {"filename", "content_base64", "content_type"}
def mini_spec(paths: dict[str, Any], definitions: dict[str, Any] | None = None) -> dict[str, Any]:
return {
"swagger": "2.0",
"info": {"title": "Mini", "version": "9.9.9"},
"basePath": "/api/v1",
"consumes": ["application/json"],
"produces": ["application/json"],
"paths": paths,
"definitions": definitions or {},
}
def build_ops(spec: dict[str, Any]) -> dict[str, Operation]:
result = build_catalog(json.dumps(spec).encode())
return {op["operation_id"]: Operation.from_dict(op) for op in result.document["operations"]}
def path_param(name: str, **extra: Any) -> dict[str, Any]:
return {"name": name, "in": "path", "required": True, "type": "string", **extra}
# ---------------------------------------------------------------- schema generation
def test_path_params_are_required_with_name_patterns() -> None:
spec = mini_spec(
{
"/repos/{owner}/{repo}/tags/{tag}": {
"get": {
"operationId": "repoGetTag",
"parameters": [
path_param("owner"),
path_param("repo"),
path_param("tag"),
{
"name": "id",
"in": "path",
"required": True,
"type": "integer",
"format": "int64",
},
],
}
}
}
)
schema = build_ops(spec)["repoGetTag"].input_schema
props = schema["properties"]
assert schema["required"] == ["owner", "repo", "tag", "id"]
assert props["owner"]["pattern"] == "^[A-Za-z0-9_.-]+$"
assert props["owner"]["maxLength"] == 255
assert props["owner"]["minLength"] == 1
assert "pattern" not in props["tag"]
assert props["tag"]["minLength"] == 1
assert props["id"] == {"type": "integer", "format": "int64"}
assert schema["additionalProperties"] is False
assert schema["type"] == "object"
def test_query_params_optional_enums_defaults_arrays() -> None:
spec = mini_spec(
{
"/things": {
"get": {
"operationId": "listThings",
"parameters": [
{
"name": "state",
"in": "query",
"type": "string",
"enum": ["a", "b"],
"default": "a",
},
{"name": "q", "in": "query", "type": "string", "required": True},
{
"name": "labels",
"in": "query",
"type": "array",
"items": {"type": "integer", "format": "int64"},
"collectionFormat": "multi",
},
{
"name": "csv",
"in": "query",
"type": "array",
"items": {"type": "string"},
"collectionFormat": "pipes",
},
],
}
}
}
)
operation = build_ops(spec)["listThings"]
props = operation.input_schema["properties"]
assert operation.input_schema["required"] == ["q"]
assert props["state"] == {"type": "string", "enum": ["a", "b"], "default": "a"}
assert props["labels"] == {"type": "array", "items": {"type": "integer", "format": "int64"}}
# A non-multi collection format is recorded on the Parameter only.
csv = next(p for p in operation.parameters if p.name == "csv")
assert csv.schema["x-collection-format"] == "pipes"
assert "x-collection-format" not in props["csv"]
def test_body_ref_is_inlined_and_vendor_keys_dropped() -> None:
definitions = {
"CreateThing": {
"type": "object",
"required": ["name"],
"properties": {
"name": {"type": "string", "x-go-name": "Name", "description": "x" * 500},
"nested": {"$ref": "#/definitions/Nested"},
"tags": {"type": "array", "items": {"type": "string"}},
},
"x-go-package": "code.gitea.io/gitea",
"example": {"name": "x"},
},
"Nested": {"type": "object", "properties": {"n": {"type": "integer"}}},
}
spec = mini_spec(
{
"/things": {
"post": {
"operationId": "createThing",
"parameters": [
{
"name": "body",
"in": "body",
"required": True,
"schema": {"$ref": "#/definitions/CreateThing"},
}
],
}
}
},
definitions,
)
operation = build_ops(spec)["createThing"]
body = operation.input_schema["properties"]["body"]
assert "$ref" not in json.dumps(operation.input_schema)
assert operation.input_schema["required"] == ["body"]
assert body["required"] == ["name"]
assert body["additionalProperties"] is False
assert body["properties"]["nested"]["properties"] == {"n": {"type": "integer"}}
assert body["properties"]["nested"]["additionalProperties"] is False
assert "x-go-package" not in body
assert "example" not in body
assert "x-go-name" not in body["properties"]["name"]
assert len(body["properties"]["name"]["description"]) <= MAX_DESCRIPTION_CHARS
parameter = operation.parameters[0]
assert (parameter.arg_name, parameter.location, parameter.required) == ("body", "body", True)
def test_optional_body_is_not_required() -> None:
spec = mini_spec(
{
"/things": {
"post": {
"operationId": "createThing",
"parameters": [{"name": "body", "in": "body", "schema": {"type": "string"}}],
}
}
}
)
operation = build_ops(spec)["createThing"]
assert operation.input_schema["required"] == []
assert operation.input_schema["properties"]["body"] == {"type": "string"}
def test_array_body_stays_array() -> None:
spec = mini_spec(
{
"/things": {
"put": {
"operationId": "replaceThings",
"parameters": [
{
"name": "body",
"in": "body",
"required": True,
"schema": {"type": "array", "items": {"type": "string"}},
}
],
}
}
}
)
body = build_ops(spec)["replaceThings"].input_schema["properties"]["body"]
assert body == {"type": "array", "items": {"type": "string"}}
def test_reference_cycle_collapses_to_plain_object() -> None:
definitions = {
"Node": {
"type": "object",
"properties": {
"name": {"type": "string"},
"parent": {"$ref": "#/definitions/Node"},
"children": {"type": "array", "items": {"$ref": "#/definitions/Node"}},
},
}
}
resolved = resolve_schema({"$ref": "#/definitions/Node"}, definitions)
assert resolved["properties"]["parent"] == {"type": "object"}
assert resolved["properties"]["children"]["items"] == {"type": "object"}
assert resolved["additionalProperties"] is False
def test_declared_additional_properties_are_kept() -> None:
schema = resolve_schema(
{
"type": "object",
"properties": {"a": {"type": "string"}},
"additionalProperties": {"type": "string"},
},
{},
)
assert schema["additionalProperties"] == {"type": "string"}
free_form = resolve_schema({"type": "object", "additionalProperties": {"type": "string"}}, {})
assert free_form == {"type": "object", "additionalProperties": {"type": "string"}}
def test_unknown_reference_is_an_error() -> None:
with pytest.raises(ValueError, match="unknown definition"):
resolve_schema({"$ref": "#/definitions/Missing"}, {})
with pytest.raises(ValueError, match="unsupported"):
resolve_schema({"$ref": "#/parameters/x"}, {})
def test_file_params_become_base64_objects() -> None:
spec = mini_spec(
{
"/uploads": {
"post": {
"operationId": "upload",
"consumes": ["multipart/form-data"],
"parameters": [
{"name": "name", "in": "formData", "type": "string"},
{
"name": "attachment",
"in": "formData",
"type": "file",
"required": True,
"description": "the file",
},
],
}
}
}
)
operation = build_ops(spec)["upload"]
attachment = operation.input_schema["properties"]["attachment"]
assert set(attachment["properties"]) == FILE_SCHEMA_KEYS
assert attachment["required"] == ["filename", "content_base64"]
assert attachment["additionalProperties"] is False
assert attachment["description"] == "the file"
file_param = next(p for p in operation.parameters if p.is_file)
assert file_param.location == "formData"
assert file_param.arg_name == "attachment"
plain = next(p for p in operation.parameters if p.name == "name")
assert plain.location == "formData"
assert not plain.is_file
assert operation.consumes == ("multipart/form-data",)
assert operation.input_schema["required"] == ["attachment"]
def test_every_tool_gets_optional_sudo() -> None:
for operation in load_catalog().operations:
props = operation.input_schema["properties"]
assert props["sudo"] == SUDO_SCHEMA
assert "sudo" not in operation.input_schema["required"]
assert all(parameter.name != "sudo" for parameter in operation.parameters)
def test_colliding_argument_names_are_prefixed() -> None:
spec = mini_spec(
{
"/things/{ref}": {
"post": {
"operationId": "mixed",
"parameters": [
path_param("ref"),
{"name": "ref", "in": "query", "type": "string"},
{"name": "sudo", "in": "query", "type": "string"},
{"name": "body", "in": "query", "type": "string"},
{"name": "body", "in": "body", "schema": {"type": "string"}},
],
}
}
}
)
operation = build_ops(spec)["mixed"]
arg_names = [parameter.arg_name for parameter in operation.parameters]
assert arg_names == ["ref", "query_ref", "query_sudo", "query_body", "body"]
assert len(set(arg_names)) == len(arg_names)
assert operation.input_schema["properties"]["sudo"] == SUDO_SCHEMA
def test_path_level_parameters_are_merged() -> None:
spec = mini_spec(
{
"/things/{name}": {
"parameters": [path_param("name", description="shared")],
"get": {
"operationId": "getThing",
"parameters": [{"name": "page", "in": "query", "type": "integer"}],
},
"delete": {
"operationId": "deleteThing",
"parameters": [path_param("name", description="own")],
},
}
}
)
ops = build_ops(spec)
assert [p.name for p in ops["getThing"].parameters] == ["name", "page"]
assert ops["getThing"].input_schema["properties"]["name"]["description"] == "shared"
assert ops["deleteThing"].input_schema["properties"]["name"]["description"] == "own"
assert [p.name for p in ops["deleteThing"].parameters] == ["name"]
def test_header_parameters_are_rejected() -> None:
spec = mini_spec(
{
"/things": {
"get": {
"operationId": "getThings",
"parameters": [{"name": "X-Thing", "in": "header", "type": "string"}],
}
}
}
)
with pytest.raises(ValueError, match="header"):
build_catalog(json.dumps(spec).encode())
def test_name_collisions_are_reported() -> None:
spec = mini_spec(
{
"/things": {"get": {"operationId": "listThings"}},
"/things2": {"get": {"operationId": "listThings2"}},
}
)
# Different operationIds that map to the same name: force one with a sibling path.
spec["paths"]["/other"] = {"get": {"operationId": "ListThings"}}
with pytest.raises(ValueError, match="collisions"):
build_catalog(json.dumps(spec).encode())
# ---------------------------------------------------------------- annotations and flags
@pytest.mark.parametrize(
("method", "read_only", "destructive", "idempotent"),
[
("get", True, False, True),
("post", False, False, False),
("put", False, True, True),
("patch", False, True, False),
("delete", False, True, True),
],
)
def test_method_annotations(
method: str, read_only: bool, destructive: bool, idempotent: bool
) -> None:
spec = mini_spec({"/things": {method: {"operationId": "doThing"}}})
operation = build_ops(spec)["doThing"]
assert operation.method == method.upper()
assert operation.read_only is read_only
assert operation.destructive is destructive
assert operation.idempotent is idempotent
def test_pagination_flag_needs_page_and_limit() -> None:
def query(name: str) -> dict[str, Any]:
return {"name": name, "in": "query", "type": "integer"}
spec = mini_spec(
{
"/both": {
"get": {"operationId": "both", "parameters": [query("page"), query("limit")]}
},
"/page": {"get": {"operationId": "pageOnly", "parameters": [query("page")]}},
"/none": {"get": {"operationId": "none"}},
}
)
ops = build_ops(spec)
assert ops["both"].paginated
assert not ops["pageOnly"].paginated
assert not ops["none"].paginated
def test_consumes_and_produces_fall_back_to_spec_defaults() -> None:
spec = mini_spec(
{
"/a": {"get": {"operationId": "a"}},
"/b": {
"post": {"operationId": "b", "consumes": ["text/plain"], "produces": ["text/html"]}
},
}
)
ops = build_ops(spec)
assert (ops["a"].consumes, ops["a"].produces) == (("application/json",), ("application/json",))
assert (ops["b"].consumes, ops["b"].produces) == (("text/plain",), ("text/html",))
def test_descriptions_are_short_and_admin_note_present() -> None:
spec = mini_spec(
{
"/admin/users": {
"post": {
"operationId": "adminCreateUser",
"summary": "Create a user",
"description": "d" * 900,
},
},
"/old": {"get": {"operationId": "oldThing", "deprecated": True}},
}
)
ops = build_ops(spec)
admin = ops["adminCreateUser"]
assert admin.summary == "Create a user"
assert len(admin.description) <= 500
created = build_ops(
mini_spec(
{
"/admin/users": {
"post": {"operationId": "adminCreateUser", "summary": "Create a user"}
}
}
)
)
assert "Requires a site-admin token." in created["adminCreateUser"].description
assert "Deprecated." in ops["oldThing"].description
assert ops["oldThing"].summary == "oldThing"
def test_trim_description_collapses_whitespace() -> None:
assert trim_description("a\n b") == "a b"
assert trim_description("x" * 400, 50).endswith("...")
assert len(trim_description("x" * 400, 50)) == 50
# ---------------------------------------------------------------- whole catalog
def test_build_is_deterministic_and_sorted() -> None:
spec = mini_spec(
{"/b": {"get": {"operationId": "bThing"}}, "/a": {"get": {"operationId": "aThing"}}}
)
raw = json.dumps(spec).encode()
first = build_catalog(raw)
second = build_catalog(raw)
assert render_catalog(first.document) == render_catalog(second.document)
names = [op["tool_name"] for op in first.document["operations"]]
assert names == sorted(names)
assert first.document["spec"] == {
"title": "Mini",
"version": "9.9.9",
"sha256": spec_sha256(raw),
}
assert sum(first.counts.values()) == 2
def test_rendered_catalog_is_valid_json_one_operation_per_line() -> None:
spec = mini_spec(
{"/a": {"get": {"operationId": "aThing"}}, "/b": {"get": {"operationId": "bThing"}}}
)
text = render_catalog(build_catalog(json.dumps(spec).encode()).document)
assert text.endswith("\n")
assert len(json.loads(text)["operations"]) == 2
assert text.count("\n") == 4
def test_every_real_input_schema_is_valid_draft_2020_12() -> None:
for operation in load_catalog().operations:
Draft202012Validator.check_schema(operation.input_schema)
assert operation.input_schema["type"] == "object"
assert operation.input_schema["additionalProperties"] is False
assert "$ref" not in json.dumps(operation.input_schema)
arg_names = [parameter.arg_name for parameter in operation.parameters]
assert len(set(arg_names)) == len(arg_names)
assert set(operation.input_schema["required"]) <= set(operation.input_schema["properties"])
assert set(arg_names) | {"sudo"} == set(operation.input_schema["properties"])
def test_real_schemas_reject_unknown_and_missing_arguments() -> None:
operation = load_catalog().get("gitea_repo_list_branches")
assert operation is not None
validator = Draft202012Validator(operation.input_schema)
assert not list(validator.iter_errors({"owner": "me", "repo": "r"}))
assert list(validator.iter_errors({"owner": "me"}))
assert list(validator.iter_errors({"owner": "me", "repo": "r", "extra": 1}))
assert list(validator.iter_errors({"owner": "../x", "repo": "r"}))
assert list(validator.iter_errors({"owner": "me", "repo": "r", "sudo": "a b"}))
def test_real_body_and_file_operations() -> None:
catalog = load_catalog()
create = catalog.get("gitea_issue_create_issue")
assert create is not None
body = create.input_schema["properties"]["body"]
assert "title" in body["properties"]
assert "title" in body["required"]
upload = catalog.get("gitea_issue_create_issue_attachment")
assert upload is not None
assert any(p.is_file and p.location == "formData" for p in upload.parameters)
assert catalog.get("gitea_repo_get_file_contents") is not None
query_body = catalog.get("gitea_repo_get_file_contents")
assert query_body is not None
assert any(p.name == "body" and p.arg_name == "query_body" for p in query_body.parameters)
def test_real_annotations() -> None:
catalog = load_catalog()
for operation in catalog.operations:
post_read_only = operation.operation_id in READ_ONLY_POST_OPERATIONS
post_destructive = operation.operation_id in DESTRUCTIVE_POST_OPERATIONS
assert operation.read_only == (operation.method == "GET" or post_read_only)
assert operation.destructive == (
operation.method in {"DELETE", "PUT", "PATCH"} or post_destructive
)
assert not (operation.read_only and operation.destructive)
if operation.paginated:
names = {p.name for p in operation.parameters}
assert {"page", "limit"} <= names
assert any(operation.paginated for operation in catalog.operations)
def test_shared_parameter_references_are_resolved() -> None:
spec = mini_spec(
{
"/things": {
"get": {"operationId": "listThings", "parameters": [{"$ref": "#/parameters/page"}]},
"post": {"operationId": "makeThing", "parameters": [{"$ref": "#/parameters/nope"}]},
}
}
)
spec["parameters"] = {"page": {"name": "page", "in": "query", "type": "integer"}}
with pytest.raises(ValueError, match="unknown parameter reference"):
build_catalog(json.dumps(spec).encode())
del spec["paths"]["/things"]["post"]
assert [p.name for p in build_ops(spec)["listThings"].parameters] == ["page"]
def test_unknown_parameter_location_is_rejected() -> None:
spec = mini_spec(
{
"/things": {
"get": {
"operationId": "a",
"parameters": [{"name": "x", "in": "cookie", "type": "string"}],
}
}
}
)
with pytest.raises(ValueError, match="unknown parameter location"):
build_catalog(json.dumps(spec).encode())
def test_post_annotation_overrides_in_the_real_catalog() -> None:
by_id = {op.operation_id: op for op in load_catalog().operations}
for operation_id in ("renderMarkdown", "renderMarkdownRaw", "renderMarkup"):
operation = by_id[operation_id]
assert operation.method == "POST"
assert operation.read_only and not operation.destructive and operation.idempotent
for operation_id in (
"repoMergePullRequest",
"repoTransfer",
"acceptRepoTransfer",
"rejectRepoTransfer",
"adminRenameUser",
"renameOrg",
"adminCronRun",
"repoMirrorSync",
"adminAdoptRepository",
"adminDeleteUnadoptedRepository",
"repoRenameBranch",
):
operation = by_id[operation_id]
assert operation.destructive and not operation.read_only, operation_id
# Ordinary creating POSTs stay non-destructive.
for operation_id in ("issueCreateIssue", "repoCreatePullRequest", "repoCreateBranch"):
assert not by_id[operation_id].destructive
assert not by_id[operation_id].read_only
def test_every_annotation_override_names_a_real_post_operation() -> None:
posts = {op.operation_id for op in load_catalog().operations if op.method == "POST"}
assert posts >= READ_ONLY_POST_OPERATIONS
assert posts >= DESTRUCTIVE_POST_OPERATIONS
assert not READ_ONLY_POST_OPERATIONS & DESTRUCTIVE_POST_OPERATIONS
def test_override_applies_to_a_mini_spec() -> None:
spec = mini_spec({"/markdown": {"post": {"operationId": "renderMarkdown"}}})
operation = build_ops(spec)["renderMarkdown"]
assert operation.read_only and not operation.destructive and operation.idempotent