AegisGitea-MCP/tests/test_catalog_build.py

"""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 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:
        assert operation.read_only == (operation.method == "GET")
        assert operation.destructive == (operation.method in {"DELETE", "PUT", "PATCH"})
        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())