AegisGitea-MCP/tests/test_executor_helpers.py

"""Shared builders for the executor tests (hand-made operations mirroring the swagger)."""

from __future__ import annotations

from typing import Any

from aegis_gitea_mcp.catalog.model import Operation, Parameter
from aegis_gitea_mcp.config import Settings
from aegis_gitea_mcp.identity import Identity

GITEA_TOKEN = "gitea-service-token-0123456789abcdef"
BEARER_TOKEN = "mcp-bearer-token-0123456789abcdefghijkl"
OAUTH_CLIENT_SECRET = "oauth-client-secret-0123456789abcdef"
USER_TOKEN = "user-oauth-token-0123456789abcdef"
API = "https://gitea.example.com/api/v1"


def make_settings(**overrides: Any) -> Settings:
    """Settings with every secret source populated."""
    values: dict[str, Any] = {
        "gitea_url": "https://gitea.example.com",
        "gitea_token": GITEA_TOKEN,
        "mcp_bearer_token": BEARER_TOKEN,
        "gitea_oauth_client_secret": OAUTH_CLIENT_SECRET,
    }
    values.update(overrides)
    return Settings(**values)


def admin_identity() -> Identity:
    return Identity(login="admin", mode="admin-token", token=GITEA_TOKEN)


def oauth_identity() -> Identity:
    return Identity(login="alice", mode="oauth", token=USER_TOKEN)


def param(
    name: str,
    location: str,
    schema: dict[str, Any] | None = None,
    *,
    required: bool = False,
    is_file: bool = False,
) -> Parameter:
    return Parameter(
        name=name,
        arg_name="body" if location == "body" else name,
        location=location,  # type: ignore[arg-type]
        required=required,
        schema=schema or {"type": "string"},
        is_file=is_file,
    )


def make_op(
    operation_id: str,
    method: str,
    path: str,
    parameters: list[Parameter],
    *,
    consumes: tuple[str, ...] = ("application/json",),
    produces: tuple[str, ...] = ("application/json",),
    paginated: bool = False,
) -> Operation:
    return Operation(
        tool_name=f"gitea_test_{operation_id.lower()}",
        domain="repo",
        operation_id=operation_id,
        method=method,
        path=path,
        summary="",
        description="",
        parameters=tuple(parameters),
        input_schema={},
        consumes=consumes,
        produces=produces,
        read_only=method == "GET",
        destructive=method == "DELETE",
        idempotent=method in {"GET", "PUT", "DELETE"},
        paginated=paginated,
    )


OWNER = param("owner", "path", required=True)
REPO = param("repo", "path", required=True)
PAGE = param("page", "query", {"type": "integer"})
LIMIT = param("limit", "query", {"type": "integer"})

LIST_BRANCHES = make_op(
    "repoListBranches",
    "GET",
    "/repos/{owner}/{repo}/branches",
    [OWNER, REPO, PAGE, LIMIT],
    paginated=True,
)
CREATE_ISSUE = make_op(
    "issueCreateIssue",
    "POST",
    "/repos/{owner}/{repo}/issues",
    [OWNER, REPO, param("body", "body", {"type": "object"})],
)
GET_RAW_FILE = make_op(
    "repoGetRawFile",
    "GET",
    "/repos/{owner}/{repo}/raw/{filepath}",
    [OWNER, REPO, param("filepath", "path", required=True), param("ref", "query")],
    produces=("application/octet-stream",),
)
GET_BRANCH = make_op(
    "repoGetBranch",
    "GET",
    "/repos/{owner}/{repo}/branches/{branch}",
    [OWNER, REPO, param("branch", "path", required=True)],
)
DELETE_BRANCH = make_op(
    "repoDeleteBranch",
    "DELETE",
    "/repos/{owner}/{repo}/branches/{branch}",
    [OWNER, REPO, param("branch", "path", required=True)],
)
UPLOAD_ATTACHMENT = make_op(
    "repoCreateReleaseAttachment",
    "POST",
    "/repos/{owner}/{repo}/releases/{id}/assets",
    [
        OWNER,
        REPO,
        param("id", "path", {"type": "integer"}, required=True),
        param("name", "query"),
        param("attachment", "formData", {"type": "object"}, is_file=True),
    ],
    consumes=("multipart/form-data", "application/octet-stream"),
)
PULL_DIFF = make_op(
    "repoDownloadPullDiffOrPatch",
    "GET",
    "/repos/{owner}/{repo}/pulls/{index}.{diffType}",
    [
        OWNER,
        REPO,
        param("index", "path", {"type": "integer"}, required=True),
        param("diffType", "path", required=True),
    ],
    produces=("text/plain",),
)
CREATE_USER = make_op(
    "adminCreateUser",
    "POST",
    "/admin/users",
    [param("body", "body", {"type": "object"})],
)
PUT_SECRET = make_op(
    "updateRepoSecret",
    "PUT",
    "/repos/{owner}/{repo}/actions/secrets/{secretname}",
    [OWNER, REPO, param("secretname", "path", required=True), param("body", "body")],
)
RENDER_RAW = make_op(
    "renderMarkdownRaw",
    "POST",
    "/markdown/raw",
    [param("body", "body", {"type": "string"})],
    consumes=("text/plain",),
    produces=("text/html",),
)
LIST_PULLS = make_op(
    "repoListPullRequests",
    "GET",
    "/repos/{owner}/{repo}/pulls",
    [
        OWNER,
        REPO,
        param("labels", "query", {"type": "array", "items": {"type": "integer"}}),
        param("state", "query"),
        param("draft", "query", {"type": "boolean"}),
        param("csvlist", "query", {"type": "array", "collectionFormat": "csv"}),
        PAGE,
        LIMIT,
    ],
    paginated=True,
)
CREATE_TOKEN = make_op(
    "userCreateToken",
    "POST",
    "/users/{username}/tokens",
    [param("username", "path", required=True), param("body", "body", {"type": "object"})],
)
REGISTRATION_TOKEN = make_op(
    "repoCreateRunnerRegistrationToken",
    "POST",
    "/repos/{owner}/{repo}/actions/runners/registration-token",
    [OWNER, REPO],
)
GET_VARIABLE_NOBODY = make_op(
    "repoListLabels",
    "GET",
    "/repos/{owner}/{repo}/labels",
    [OWNER, REPO],
)