From 303bd9aa21b42b9e5283e339a37114c9e299ef91 Mon Sep 17 00:00:00 2001 From: latte Date: Wed, 14 Oct 2026 17:22:45 +0200 Subject: [PATCH] test: sweep every tool through the MCP server against a seeded Gitea --- tests/integration/conftest.py | 13 +- tests/integration/harness.py | 4 + tests/integration/sweep_plan.py | 928 +++++++++++++++++++++++ tests/integration/sweep_seed.py | 445 +++++++++++ tests/integration/test_sweep.py | 54 ++ tests/integration/test_sweep_isolated.py | 43 ++ tests/test_sweep_plan.py | 273 +++++++ 7 files changed, 1759 insertions(+), 1 deletion(-) create mode 100644 tests/integration/sweep_plan.py create mode 100644 tests/integration/sweep_seed.py create mode 100644 tests/integration/test_sweep.py create mode 100644 tests/integration/test_sweep_isolated.py create mode 100644 tests/test_sweep_plan.py diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 9afa687..c17dfda 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -29,6 +29,7 @@ from tests.integration.harness import ( stop_gitea, unique, ) +from tests.integration.sweep_plan import LOG, write_report_if_enabled _HERE = Path(__file__).parent.resolve() ENABLE_ENV = "AEGIS_INTEGRATION" @@ -70,6 +71,14 @@ def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item namespace.cov_fail_under = 0 +def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None: + """Write ``docs/integration-coverage.md`` when ``AEGIS_IT_WRITE_REPORT=1`` and the sweep ran.""" + from aegis_gitea_mcp.catalog import load_catalog + + catalog = load_catalog() + write_report_if_enabled(catalog, catalog.spec_version) + + @pytest.fixture(scope="session") def gitea() -> Iterator[GiteaInstance]: """A throwaway Gitea in Docker, shared by the whole session.""" @@ -171,7 +180,9 @@ class McpSession: self.texts.append(text) payload = json.loads(text) assert isinstance(payload, dict) - return ToolResult(tool=tool, is_error=bool(raw.is_error), payload=payload, text=text) + result = ToolResult(tool=tool, is_error=bool(raw.is_error), payload=payload, text=text) + LOG.add(tool, result.status, result.is_error) + return result async def ok( self, tool: str, /, *, status: int | tuple[int, ...] = (200, 201, 204), **arguments: Any diff --git a/tests/integration/harness.py b/tests/integration/harness.py index f354571..74ddeff 100644 --- a/tests/integration/harness.py +++ b/tests/integration/harness.py @@ -272,6 +272,8 @@ def _gitea_env(root_url: str) -> dict[str, str]: "GITEA__security__INSTALL_LOCK": "true", "GITEA__security__SECRET_KEY": secrets.token_hex(32), "GITEA__security__INTERNAL_TOKEN": secrets.token_hex(32), + # Repository hook scripts are editable through the API (site admins only) when enabled. + "GITEA__security__DISABLE_GIT_HOOKS": "false", "GITEA__server__ROOT_URL": root_url, "GITEA__server__HTTP_PORT": str(GITEA_PORT), "GITEA__server__DISABLE_SSH": "true", @@ -285,6 +287,8 @@ def _gitea_env(root_url: str) -> dict[str, str]: "GITEA__repository__ENABLE_PUSH_CREATE_ORG": "false", "GITEA__mailer__ENABLED": "false", "GITEA__webhook__ALLOWED_HOST_LIST": "*", + # Lets repository migration and mirrors use this very instance as their remote. + "GITEA__migrations__ALLOW_LOCALNETWORKS": "true", "GITEA__packages__ENABLED": "true", "GITEA__api__MAX_RESPONSE_ITEMS": "50", } diff --git a/tests/integration/sweep_plan.py b/tests/integration/sweep_plan.py new file mode 100644 index 0000000..77513a3 --- /dev/null +++ b/tests/integration/sweep_plan.py @@ -0,0 +1,928 @@ +"""Plan for the all-tools sweep: arguments, ordering, expectations and the coverage report. + +The sweep calls every tool (538 with ``gitea_api_request``) once through the MCP server +against a seeded Gitea (:mod:`tests.integration.sweep_seed`). This module holds everything +that is data or pure logic, so it can be checked without Docker +(``tests/test_sweep_plan.py``): + +* :func:`build_arguments` turns a catalog operation plus the seed context into tool + arguments: path values from the context, required bodies generated from their schema, + and an override table (:data:`OVERRIDES`) for tools that need specific bodies or ids. +* :func:`ordered_operations` sorts the calls: read-only first, then creates and updates, + then deletes (leaf resources before their parents). +* :func:`classify` decides whether a result is acceptable: a 2xx, a status the swagger + documents for that operation, or an explicit entry of :data:`ALLOWANCES` with a reason. +* :func:`render_report` writes ``docs/integration-coverage.md``. +""" + +from __future__ import annotations + +import json +import os +import re +import secrets +from collections.abc import Callable, Iterable, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from aegis_gitea_mcp.catalog import Catalog, Operation +from tests.integration.sweep_seed import PNG_1X1, Ctx, b64, ssh_public_key + +REPO_ROOT = Path(__file__).resolve().parents[2] +SPEC_PATH = REPO_ROOT / "spec" / "gitea-swagger.v1.json" +REPORT_PATH = REPO_ROOT / "docs" / "integration-coverage.md" +REPORT_ENV = "AEGIS_IT_WRITE_REPORT" +RAW_TOOL = "gitea_api_request" + +#: Tools the generic sweep never calls, with the test that covers them in isolation. +EXCLUDED: dict[str, str] = { + "gitea_misc_delete_current_token": ( + "deletes the token the sweep itself runs with; covered by test_sweep_isolated.py " + "with a throwaway token" + ), +} + + +@dataclass(frozen=True) +class Allowance: + """A status that is accepted for a tool although the swagger does not document it.""" + + statuses: frozenset[int] + reason: str + + +def allow(reason: str, *statuses: int) -> Allowance: + """Build an :class:`Allowance`.""" + return Allowance(frozenset(statuses), reason) + + +#: Tools whose result may be a status the swagger does not list, each with its reason. +_NO_SIGNING_KEY = allow( + "the throwaway instance has no commit signing key; Gitea answers 404 although the " + "swagger lists no 404 for this operation", + 404, +) +ALLOWANCES: dict[str, Allowance] = { + "gitea_misc_get_signing_key": _NO_SIGNING_KEY, + "gitea_misc_get_signing_key_ssh": _NO_SIGNING_KEY, + "gitea_repo_signing_key": _NO_SIGNING_KEY, + "gitea_repo_signing_key_ssh": _NO_SIGNING_KEY, + "gitea_admin_add_user_badges": allow( + "Gitea answers 500 for an unknown badge slug and the API cannot create badges", 500 + ), + "gitea_admin_delete_unadopted_repository": allow( + "no unadopted repository exists on the instance; Gitea answers 404 although the swagger " + "lists only 204 and 403", + 404, + ), + "gitea_user_create_token": allow( + "Gitea only accepts basic auth here; the server always sends token authentication", 401 + ), + "gitea_user_get_tokens": allow( + "Gitea only accepts basic auth here; the server always sends token authentication", 401 + ), + "gitea_user_delete_access_token": allow( + "Gitea only accepts basic auth here; the server always sends token authentication", 401 + ), +} + + +# --------------------------------------------------------------------------- +# Spec helpers +# --------------------------------------------------------------------------- + + +def load_spec() -> dict[str, Any]: + """Load the pinned swagger.""" + data: dict[str, Any] = json.loads(SPEC_PATH.read_text(encoding="utf-8")) + return data + + +def documented_statuses(spec: dict[str, Any], operation: Operation) -> frozenset[int]: + """Status codes the swagger lists in ``responses`` of this operation.""" + item = spec["paths"][operation.path][operation.method.lower()] + return frozenset(int(code) for code in item["responses"] if code.isdigit()) + + +# --------------------------------------------------------------------------- +# Argument synthesis +# --------------------------------------------------------------------------- + +DEFAULT_PATH_VALUES: dict[str, str] = { + "owner": "owner", + "repo": "repo", + "org": "org", + "username": "user", + "collaborator": "user", + "assignee": "user", + "target": "user", + "user": "user", + "template_owner": "owner", + "template_repo": "template_repo", + "repo_name": "repo", + "index": "issue", + "sha": "sha", + "run": "run", + "job_id": "job_id", + "pageName": "page_name", + "team": "team_name", + "task": "cron_task", +} + +CONSTANT_PATH_VALUES: dict[str, Any] = { + "ref": "main", + "branch": "main", + "filepath": "README.md", + "archive": "main.zip", + "basehead": "main...feat", + "diffType": "diff", + "base": "main", + "head": "feat", + "type": "generic", + "version": "1.0.0", + "attempt": 1, + "artifact_id": 1, + "runner_id": 1, + "position": 1, + "workflow_id": "sweep.yml", + "topic": "extra", + "token": "none", +} + +# (regex over the path template, context key) for the ``{id}`` segment, first match wins. +ID_RULES: list[tuple[str, str]] = [ + (r"^/repos/\{owner\}/\{repo\}/pulls/comments/\{id\}", "review_comment"), + (r"^/repos/\{owner\}/\{repo\}/pulls/\{index\}/comments/\{id\}", "review_comment"), + (r"^/repos/\{owner\}/\{repo\}/pulls/\{index\}/reviews/\{id\}", "review"), + (r"^/repos/\{owner\}/\{repo\}/issues/comments/\{id\}", "comment"), + (r"^/repos/\{owner\}/\{repo\}/issues/\{index\}/comments/\{id\}", "comment"), + (r"^/repos/\{owner\}/\{repo\}/issues/\{index\}/labels/\{id\}", "label"), + (r"^/repos/\{owner\}/\{repo\}/issues/\{index\}/times/\{id\}", "time_id"), + (r"^/repos/\{owner\}/\{repo\}/labels/\{id\}", "label"), + (r"^/repos/\{owner\}/\{repo\}/milestones/\{id\}", "milestone"), + (r"^/repos/\{owner\}/\{repo\}/hooks/git/\{id\}", "git_hook"), + (r"^/repos/\{owner\}/\{repo\}/hooks/\{id\}", "repo_hook"), + (r"^/repos/\{owner\}/\{repo\}/keys/\{id\}", "deploy_key"), + (r"^/repos/\{owner\}/\{repo\}/releases/\{id\}", "release"), + (r"^/repos/\{owner\}/\{repo\}/tag_protections/\{id\}", "tag_protection"), + (r"^/repos/\{owner\}/\{repo\}/projects/\{id\}", "repo_project"), + (r"^/orgs/\{org\}/hooks/\{id\}", "org_hook"), + (r"^/orgs/\{org\}/labels/\{id\}", "org_label"), + (r"^/orgs/\{org\}/projects/\{id\}", "org_project"), + (r"^/user/hooks/\{id\}", "user_hook"), + (r"^/user/keys/\{id\}", "user_key"), + (r"^/user/projects/\{id\}", "user_project"), + (r"^/user/applications/oauth2/\{id\}", "oauth_app"), + (r"^/admin/hooks/\{id\}", "admin_hook"), + (r"^/admin/users/\{username\}/keys/\{id\}", "admin_user_key"), + (r"^/teams/\{id\}", "team_id"), + (r"^/repositories/\{id\}", "repo_id"), + (r"^/notifications/threads/\{id\}", "notification"), +] + + +def _path_value(operation: Operation, name: str, c: Ctx) -> Any: + """Value of the path parameter ``name`` of ``operation`` from the context.""" + path = operation.path + if name == "index": + return c["pr"] if "/pulls/" in path else c["issue"] + if name == "id": + for pattern, key in ID_RULES: + if re.match(pattern, path): + return c[key] + return 1 + if name == "attachment_id": + if "/releases/" in path: + return c["release_asset"] + return c["comment_attachment"] if "/issues/comments/" in path else c["issue_attachment"] + if name == "column_id": + if path.startswith("/orgs/"): + return c["org_column"] + return c["user_column"] if path.startswith("/user/") else c["repo_column"] + if name == "issue_id": + return c["org_issue_id"] if path.startswith("/orgs/") else c["issue1_id"] + if name == "secretname": + scope = path.split("/")[1] + return {"orgs": "ORG_SECRET", "user": c["user_secret"]}.get(scope, "REPO_SECRET") + if name == "variablename": + scope = path.split("/")[1] + return {"orgs": "ORG_VAR", "user": c["user_var"]}.get(scope, "REPO_VAR") + if name == "name": + if path.startswith("/gitignore/"): + return "Go" + if path.startswith("/label/"): + return "Advanced" + if path.startswith("/licenses/"): + return "MIT" + if path.startswith("/packages/"): + return c["pkg"] + if "branch_protections" in path: + return c["protection"] + return "none" + if name == "tag": + return c["release_tag"] if "/releases/" in path else "v0.0.1" + if name in DEFAULT_PATH_VALUES: + return c[DEFAULT_PATH_VALUES[name]] + if name in CONSTANT_PATH_VALUES: + return CONSTANT_PATH_VALUES[name] + return 1 + + +def _coerce(value: Any, schema: dict[str, Any]) -> Any: + """Convert ``value`` to the JSON type the argument schema asks for.""" + kind = schema.get("type") + if kind == "string": + return str(value) + if kind == "integer" and isinstance(value, str) and value.isdigit(): + return int(value) + return value + + +def sample(schema: dict[str, Any], name: str = "") -> Any: + """A minimal valid value for ``schema`` (required properties only).""" + if "enum" in schema: + return schema["enum"][0] + if "anyOf" in schema: + return sample(schema["anyOf"][0], name) + kind = schema.get("type") + if kind == "object" or "properties" in schema: + required = schema.get("required", []) + props = schema.get("properties", {}) + return {key: sample(props[key], key) for key in required if key in props} + if kind == "array": + return [] + if kind in ("integer", "number"): + return 1 + if kind == "boolean": + return False + lowered = name.lower() + if "email" in lowered: + return f"sweep-{secrets.token_hex(3)}@example.test" + if "url" in lowered or "addr" in lowered: + return "http://127.0.0.1:9/sweep" + if "color" in lowered: + return "#336699" + if "base64" in lowered or lowered == "content": + return b64("sweep") + return f"sweep-{secrets.token_hex(3)}" + + +def fresh(c: Ctx, prefix: str = "sw") -> str: + """A fresh name for one more resource.""" + return f"{prefix}-{c['sfx']}-{secrets.token_hex(2)}" + + +def hook_body(c: Ctx) -> dict[str, Any]: + """A webhook body of type gitea.""" + return { + "type": "gitea", + "config": {"url": "http://127.0.0.1:9/sweep", "content_type": "json"}, + "events": ["push"], + "active": True, + } + + +def local_url(c: Ctx) -> str: + """Base URL of the admin's repositories as seen from inside the Gitea container.""" + return f"http://127.0.0.1:3000/{c['admin']}" + + +def _project_overrides() -> dict[str, Override]: + """Column and issue handling of org, repo and user projects (all three work alike).""" + result: dict[str, Override] = {} + for scope, tool_scope in (("org", "org"), ("repo", "repo"), ("user", "user_current")): + result[f"gitea_{tool_scope}_move_project_columns"] = lambda c, s=scope: { + "body": {"column_ids": c[f"{s}_columns"]} + } + result[f"gitea_{tool_scope}_move_project_issue"] = lambda c, s=scope: { + "body": {"column_id": c[f"{s}_column2"]} + } + result[f"gitea_{tool_scope}_delete_project_column"] = lambda c, s=scope: { + "column_id": c[f"{s}_column2"] + } + result[f"gitea_{tool_scope}_remove_issue_from_project_column"] = lambda c, s=scope: { + "column_id": c[f"{s}_column2"] + } + return result + + +def _org_repo(c: Ctx) -> dict[str, Any]: + return {"owner": c["org"], "repo": c["org_repo"]} + + +def _org_repo_team(c: Ctx) -> dict[str, Any]: + return {"owner": c["org"], "repo": c["org_repo"], "team": c["team_name"]} + + +def _team_repo(c: Ctx) -> dict[str, Any]: + return {"org": c["org"], "repo": c["org_repo"]} + + +PATCH_TEXT = """\ +diff --git a/applied-{name}.txt b/applied-{name}.txt +new file mode 100644 +index 0000000..e69de29 +--- /dev/null ++++ b/applied-{name}.txt +@@ -0,0 +1 @@ ++applied +""" + +Override = Callable[[Ctx], dict[str, Any]] + +OVERRIDES: dict[str, Override] = { + # ------------------------------------------------------------------ reads + "gitea_repo_get_annotated_tag": lambda c: {"sha": c["tag_sha"]}, + "gitea_repo_get_push_mirror_by_remote_name": lambda c: {"name": c["push_mirror"]}, + "gitea_repo_delete_push_mirror": lambda c: {"name": c["push_mirror"]}, + "gitea_repo_add_push_mirror": lambda c: { + "body": { + "remote_address": f"{local_url(c)}/{c['push_target2']}.git", + "remote_username": c["admin"], + "remote_password": c["admin_password"], + "interval": "8h0m0s", + "sync_on_commit": False, + } + }, + "gitea_repo_mirror_sync": lambda c: {"repo": c["mirror_repo"]}, + "gitea_repo_merge_upstream": lambda c: { + "owner": c["org"], + "repo": c["fork_repo"], + "body": {"branch": "main"}, + }, + "gitea_repo_migrate": lambda c: { + "body": { + "clone_addr": f"{local_url(c)}/{c['template_repo']}.git", + "repo_name": fresh(c, "mig"), + "service": "git", + } + }, + "gitea_repo_add_team": lambda c: {**_org_repo(c), "team": c["team2_name"]}, + "gitea_repo_delete_team": lambda c: {**_org_repo(c), "team": c["team2_name"]}, + "gitea_repo_update_topics": lambda c: {"body": {"topics": ["sweep", "seed", "extra"]}}, + "gitea_issue_delete_stopwatch": lambda c: {"sudo": c["user"], "index": c["issue2"]}, + "gitea_org_block_user": lambda c: {"username": c["user3"]}, + "gitea_user_block_user": lambda c: {"username": c["user3"]}, + "gitea_repo_get_file_contents": lambda c: {"query_body": json.dumps({"files": ["README.md"]})}, + "gitea_repo_get_file_contents_post": lambda c: {"body": {"files": ["README.md"]}}, + "gitea_repo_list_git_refs": lambda c: {"ref": "heads/main"}, + "gitea_repo_get_blob": lambda c: {"sha": c["blob_sha"]}, + "gitea_repo_check_team": _org_repo_team, + "gitea_repo_list_teams": _org_repo, + "gitea_team_list_team_repo": _team_repo, + "gitea_team_add_team_repository": _team_repo, + "gitea_team_remove_team_repository": _team_repo, + "gitea_user_check_following": lambda c: {"username": c["admin"], "target": c["user"]}, + "gitea_misc_render_markdown": lambda c: {"body": {"Text": "# sweep", "Mode": "markdown"}}, + "gitea_misc_render_markup": lambda c: { + "body": {"Text": "# sweep", "Mode": "markdown", "Context": ""} + }, + "gitea_misc_render_markdown_raw": lambda c: {"body": "# sweep"}, + "gitea_notification_read_thread": lambda c: {"id": str(c["notification"])}, + "gitea_notification_get_thread": lambda c: {"id": str(c["notification"])}, + "gitea_user_get_tokens": lambda c: {"username": c["admin"]}, + # ---------------------------------------------------------------- actions + "gitea_actions_create_org_variable": lambda c: { + "variablename": "SW_NEW", + "body": {"value": "1"}, + }, + "gitea_actions_create_repo_variable": lambda c: { + "variablename": "SW_NEW", + "body": {"value": "1"}, + }, + "gitea_actions_create_user_variable": lambda c: { + "variablename": f"SW_NEW_{c['sfx'].upper()}", + "body": {"value": "1"}, + }, + "gitea_actions_dispatch_workflow": lambda c: {"body": {"ref": "main"}}, + "gitea_actions_update_org_secret": lambda c: {"body": {"data": "rotated-org-value"}}, + "gitea_actions_update_repo_secret": lambda c: {"body": {"data": "rotated-repo-value"}}, + "gitea_actions_update_user_secret": lambda c: {"body": {"data": "rotated-user-value"}}, + # ------------------------------------------------------------------ admin + "gitea_admin_add_user_badges": lambda c: {"body": {"badge_slugs": ["sweep"]}}, + "gitea_admin_delete_user_badges": lambda c: {"body": {"badge_slugs": ["sweep"]}}, + "gitea_admin_create_hook": lambda c: {"body": hook_body(c)}, + "gitea_admin_create_org": lambda c: {"body": {"username": fresh(c, "sw-aorg")}}, + "gitea_admin_create_public_key": lambda c: { + "body": {"title": fresh(c, "key"), "key": ssh_public_key()} + }, + "gitea_admin_create_repo": lambda c: {"body": {"name": fresh(c, "sw-arepo")}}, + "gitea_admin_create_user": lambda c: _new_user(c), + "gitea_admin_edit_user": lambda c: { + "username": c["user2"], + "body": {"source_id": 0, "login_name": c["user2"], "full_name": "Sweep Two"}, + }, + "gitea_admin_rename_user": lambda c: { + "username": c["rename_user"], + "body": {"new_username": fresh(c, "sw-renamed")}, + }, + "gitea_admin_delete_user": lambda c: {"username": c["del_user"]}, + # ------------------------------------------------------------------ hooks + "gitea_hook_org_create_hook": lambda c: {"body": hook_body(c)}, + "gitea_hook_repo_create_hook": lambda c: {"body": hook_body(c)}, + "gitea_hook_user_create_hook": lambda c: {"body": hook_body(c)}, + "gitea_hook_repo_edit_git_hook": lambda c: {"body": {"content": "#!/bin/sh\nexit 0\n"}}, + # ----------------------------------------------------------------- issues + "gitea_issue_create_issue": lambda c: {"body": {"title": fresh(c, "issue")}}, + "gitea_issue_create_comment": lambda c: {"body": {"body": "sweep comment"}}, + "gitea_issue_create_label": lambda c: {"body": {"name": fresh(c, "lbl"), "color": "#00aa00"}}, + "gitea_issue_create_milestone": lambda c: {"body": {"title": fresh(c, "ms")}}, + "gitea_issue_add_label": lambda c: {"body": {"labels": [c["label"]]}}, + "gitea_issue_replace_labels": lambda c: {"body": {"labels": [c["label"]]}}, + "gitea_issue_add_assignees": lambda c: {"body": {"assignees": [c["user"]]}}, + "gitea_issue_remove_assignees": lambda c: {"body": {"assignees": [c["admin"]]}}, + "gitea_issue_add_time": lambda c: {"body": {"time": 30}}, + "gitea_issue_edit_issue": lambda c: {"body": {"title": "Sweep main edited"}}, + "gitea_issue_edit_comment": lambda c: {"body": {"body": "edited"}}, + "gitea_issue_edit_comment_deprecated": lambda c: {"body": {"body": "edited again"}}, + "gitea_issue_edit_label": lambda c: {"body": {"name": "bug-renamed"}}, + "gitea_issue_edit_milestone": lambda c: {"body": {"title": "m1-renamed"}}, + "gitea_issue_edit_issue_attachment": lambda c: {"body": {"name": "renamed.txt"}}, + "gitea_issue_edit_issue_comment_attachment": lambda c: {"body": {"name": "renamed.txt"}}, + "gitea_issue_edit_issue_deadline": lambda c: {"body": {"due_date": "2030-01-01T00:00:00Z"}}, + "gitea_issue_lock_issue": lambda c: {"body": {"lock_reason": "too heated"}}, + "gitea_issue_move_issue_pin": lambda c: {"index": c["issue3"]}, + "gitea_issue_unpin_issue": lambda c: {"index": c["issue3"]}, + "gitea_issue_post_issue_reaction": lambda c: {"body": {"content": "heart"}}, + "gitea_issue_post_comment_reaction": lambda c: {"body": {"content": "heart"}}, + "gitea_issue_delete_issue_reaction": lambda c: {"body": {"content": "+1"}}, + "gitea_issue_delete_comment_reaction": lambda c: {"body": {"content": "+1"}}, + "gitea_issue_create_issue_blocking": lambda c: { + "body": {"index": c["issue5"], "owner": c["owner"], "repo": c["repo"]} + }, + "gitea_issue_remove_issue_blocking": lambda c: { + "body": {"index": c["issue3"], "owner": c["owner"], "repo": c["repo"]} + }, + "gitea_issue_create_issue_dependencies": lambda c: { + "body": {"index": c["issue4"], "owner": c["owner"], "repo": c["repo"]} + }, + "gitea_issue_remove_issue_dependencies": lambda c: { + "body": {"index": c["issue2"], "owner": c["owner"], "repo": c["repo"]} + }, + "gitea_issue_delete_comment_deprecated": lambda c: {"id": c["comment2"]}, + # ------------------------------------------------------------------- orgs + "gitea_org_create": lambda c: {"body": {"username": fresh(c, "sw-neworg")}}, + "gitea_org_create_label": lambda c: {"body": {"name": fresh(c, "lbl"), "color": "#00aa00"}}, + "gitea_org_create_org_repo": lambda c: {"body": {"name": fresh(c, "sw-orgrepo")}}, + "gitea_org_create_org_repo_deprecated": lambda c: {"body": {"name": fresh(c, "sw-orgrepo")}}, + "gitea_org_create_project": lambda c: {"body": {"title": fresh(c, "proj")}}, + "gitea_org_create_project_column": lambda c: { + "id": c["org_project"], + "body": {"title": fresh(c, "col")}, + }, + "gitea_org_edit": lambda c: {"body": {"description": "edited by the sweep"}}, + "gitea_org_rename": lambda c: { + "org": c["rename_org"], + "body": {"new_name": fresh(c, "sw-renorg")}, + }, + "gitea_org_delete": lambda c: {"org": c["del_org2"]}, + "gitea_org_delete_repos": lambda c: {"org": c["del_org"]}, + "gitea_org_update_avatar": lambda c: {"body": {"image": PNG_1X1}}, + "gitea_org_check_user_block": lambda c: {"username": c["user2"]}, + "gitea_org_unblock_user": lambda c: {"username": c["user2"]}, + # ---------------------------------------------------------------- package + "gitea_package_delete_package": lambda c: {"name": c["pkg2"]}, + # ------------------------------------------------------------------ pulls + "gitea_pull_create_pull_request": lambda c: { + "body": {"title": "Sweep PR three", "base": "main", "head": "feat3"} + }, + "gitea_pull_create_pull_review": lambda c: {"body": {"event": "COMMENT", "body": "sweep"}}, + "gitea_pull_submit_pull_review": lambda c: { + "sudo": c["user"], + "id": c["pending_review"], + "body": {"event": "COMMENT", "body": "submitted"}, + }, + "gitea_pull_delete_pull_review": lambda c: {"id": c["review2"]}, + "gitea_pull_dismiss_pull_review": lambda c: {"body": {"message": "stale", "priors": False}}, + "gitea_pull_create_pull_review_comment_reply": lambda c: {"body": {"body": "reply"}}, + "gitea_pull_create_pull_review_requests": lambda c: {"body": {"reviewers": [c["user2"]]}}, + "gitea_pull_delete_pull_review_requests": lambda c: {"body": {"reviewers": [c["user2"]]}}, + "gitea_pull_merge_pull_request": lambda c: {"index": c["pr2"], "body": {"do": "merge"}}, + "gitea_pull_edit_pull_request": lambda c: {"body": {"title": "Sweep PR edited"}}, + # --------------------------------------------------------------- releases + "gitea_release_create_release": lambda c: { + "body": {"tag_name": "v3.0.0", "target_commitish": "main"} + }, + "gitea_release_edit_release": lambda c: {"body": {"name": "renamed"}}, + "gitea_release_edit_release_attachment": lambda c: {"body": {"name": "renamed.bin"}}, + "gitea_release_delete_release_by_tag": lambda c: {"tag": c["release_tag2"]}, + # ------------------------------------------------------------------ repos + "gitea_repo_create_file": lambda c: { + "filepath": f"new-{fresh(c)}.txt", + "body": {"content": b64("new"), "message": "sweep new file"}, + }, + "gitea_repo_update_file": lambda c: { + "filepath": "to-update.txt", + "body": {"content": b64("after"), "sha": c["update_file_sha"], "message": "sweep update"}, + }, + "gitea_repo_delete_file": lambda c: { + "filepath": "to-delete.txt", + "body": {"sha": c["delete_file_sha"], "message": "sweep delete"}, + }, + "gitea_repo_change_files": lambda c: { + "body": { + "files": [{"operation": "create", "path": f"cf-{fresh(c)}.txt", "content": b64("c")}], + "message": "sweep change files", + } + }, + "gitea_repo_apply_diff_patch": lambda c: { + "body": {"content": PATCH_TEXT.format(name=fresh(c)), "message": "sweep patch"} + }, + "gitea_repo_create_branch": lambda c: { + "body": {"new_branch_name": fresh(c, "br"), "old_branch_name": "main"} + }, + "gitea_repo_create_branch_protection": lambda c: {"body": {"rule_name": "hotfix/*"}}, + "gitea_repo_create_tag": lambda c: {"body": {"tag_name": "v0.0.2", "target": "main"}}, + "gitea_repo_create_tag_protection": lambda c: { + "body": {"name_pattern": "w*", "whitelist_usernames": [c["admin"]]} + }, + "gitea_repo_create_fork": lambda c: { + "body": {"organization": c["del_org"], "name": fresh(c, "fork")} + }, + "gitea_repo_create_key": lambda c: { + "body": {"title": fresh(c, "key"), "key": ssh_public_key()} + }, + "gitea_repo_create_deploy_token": lambda c: {"body": {"title": fresh(c, "dt")}}, + "gitea_repo_create_status": lambda c: {"body": {"state": "success", "context": "ci/two"}}, + "gitea_repo_create_project": lambda c: {"body": {"title": fresh(c, "proj")}}, + "gitea_repo_create_project_column": lambda c: { + "id": c["repo_project"], + "body": {"title": fresh(c, "col")}, + }, + "gitea_repo_edit": lambda c: {"body": {"description": "edited by the sweep"}}, + "gitea_repo_edit_branch_protection": lambda c: {"body": {"required_approvals": 0}}, + "gitea_repo_edit_tag_protection": lambda c: { + "body": {"name_pattern": "v9*", "whitelist_usernames": [c["admin"]]} + }, + "gitea_repo_edit_project": lambda c: {"body": {"title": "renamed project"}}, + "gitea_repo_edit_project_column": lambda c: {"body": {"title": "renamed column"}}, + "gitea_repo_generate_repo": lambda c: { + "body": {"name": fresh(c, "gen"), "owner": c["admin"], "git_content": True} + }, + "gitea_repo_rename_branch": lambda c: {"branch": "rn", "body": {"name": "rn2"}}, + "gitea_repo_transfer": lambda c: {"repo": c["xfer_repo"], "body": {"new_owner": c["org"]}}, + "gitea_repo_update_avatar": lambda c: {"body": {"image": PNG_1X1}}, + "gitea_repo_update_branch": lambda c: { + "branch": "feat3", + "body": {"new_commit_id": c["sha"], "force": True}, + }, + "gitea_repo_update_branch_protection_priorities": lambda c: {"body": {"ids": []}}, + "gitea_repo_delete": lambda c: {"repo": c["del_repo"]}, + "gitea_repo_delete_branch": lambda c: {"branch": "del-branch"}, + "gitea_repo_delete_collaborator": lambda c: {"collaborator": c["user2"]}, + # ------------------------------------------------------------------- wiki + "gitea_wiki_create_wiki_page": lambda c: { + "body": {"title": "Page2", "content_base64": b64("two"), "message": "page two"} + }, + "gitea_wiki_edit_wiki_page": lambda c: { + "body": {"title": "Home", "content_base64": b64("# home v2\n"), "message": "edit"} + }, + # ------------------------------------------------------------------ teams + "gitea_team_create_team": lambda c: { + "body": {"name": fresh(c, "team"), "permission": "read", "units": ["repo.code"]} + }, + "gitea_team_edit_team": lambda c: { + "body": {"name": c["team_name"], "description": "edited by the sweep"} + }, + # ------------------------------------------------------------------ users + "gitea_user_add_email": lambda c: {"body": {"emails": [f"sw-{fresh(c)}@example.test"]}}, + "gitea_user_delete_email": lambda c: {"body": {"emails": [c["email"]]}}, + "gitea_user_check_user_block": lambda c: {"username": c["user2"]}, + "gitea_user_unblock_user": lambda c: {"username": c["user2"]}, + "gitea_user_create_oauth2_application": lambda c: { + "body": {"name": fresh(c, "app"), "redirect_uris": ["http://127.0.0.1:1/cb"]} + }, + "gitea_user_update_oauth2_application": lambda c: { + "body": {"name": fresh(c, "app2"), "redirect_uris": ["http://127.0.0.1:1/cb2"]} + }, + "gitea_user_current_create_repo": lambda c: {"body": {"name": fresh(c, "sw-urepo")}}, + "gitea_user_current_create_project": lambda c: {"body": {"title": fresh(c, "proj")}}, + "gitea_user_current_create_project_column": lambda c: { + "id": c["user_project"], + "body": {"title": fresh(c, "col")}, + }, + "gitea_user_current_post_key": lambda c: { + "body": {"title": fresh(c, "key"), "key": ssh_public_key()} + }, + "gitea_user_current_post_gpg_key": lambda c: {"body": {"armored_public_key": "not a key"}}, + "gitea_user_delete_access_token": lambda c: { + "username": c["admin"], + "token": c["throwaway_token"], + }, + "gitea_user_update_avatar": lambda c: {"body": {"image": PNG_1X1}}, + "gitea_user_update_settings": lambda c: {"body": {"description": "sweep"}}, + # -------------------------------------------------------------------- raw + RAW_TOOL: lambda c: {"method": "GET", "path": "/version"}, +} + + +OVERRIDES.update(_project_overrides()) + + +def _new_user(c: Ctx) -> dict[str, Any]: + name = fresh(c, "sw-new") + return { + "body": { + "username": name, + "email": f"{name}@example.test", + "password": secrets.token_urlsafe(18), + "must_change_password": False, + } + } + + +def build_arguments(operation: Operation, c: Ctx) -> dict[str, Any]: + """Synthesize the arguments of one catalog operation from the seed context.""" + args: dict[str, Any] = {} + for parameter in operation.parameters: + if parameter.location == "path": + value = _path_value(operation, parameter.name, c) + args[parameter.arg_name] = _coerce(value, parameter.schema) + elif parameter.is_file: + args[parameter.arg_name] = { + "filename": "sweep.txt", + "content_base64": b64("sweep"), + "content_type": "text/plain", + } + elif parameter.location == "body": + needs_members = bool(parameter.schema.get("required")) + args[parameter.arg_name] = ( + sample(parameter.schema) if parameter.required or needs_members else {} + ) + elif parameter.required: + args[parameter.arg_name] = sample(parameter.schema, parameter.name) + override = OVERRIDES.get(operation.tool_name) + if override is not None: + args.update(override(c)) + for name, value in list(args.items()): + parameter = next((p for p in operation.parameters if p.arg_name == name), None) + if parameter is not None and parameter.location == "path": + args[name] = _coerce(value, parameter.schema) + return args + + +# --------------------------------------------------------------------------- +# Ordering and classification +# --------------------------------------------------------------------------- + +PHASE_BY_METHOD = {"GET": 1, "HEAD": 1, "POST": 2, "PUT": 3, "PATCH": 4, "DELETE": 5} +#: Explicit priorities inside a phase (higher runs later; default 50). +PRIORITY: dict[str, int] = { + "gitea_pull_merge_pull_request": 90, + # Moving columns must list every column, so it runs before any column is created. + "gitea_org_move_project_columns": 10, + "gitea_repo_move_project_columns": 10, + "gitea_user_current_move_project_columns": 10, +} + + +def phase_of(operation: Operation) -> int: + """0 for read-only operations, then 2 (POST), 3 (PUT), 4 (PATCH) and 5 (DELETE).""" + if operation.read_only: + return 0 + return PHASE_BY_METHOD.get(operation.method, 2) + + +def sort_key(operation: Operation) -> tuple[int, int, int, str]: + """Sort key: phase, priority, leaf-first depth for deletes, name.""" + phase = phase_of(operation) + depth = -operation.path.count("/") if phase == 5 else 0 + return (phase, PRIORITY.get(operation.tool_name, 50), depth, operation.tool_name) + + +def ordered_operations(catalog: Catalog) -> list[Operation]: + """Every operation that the generic sweep calls, in execution order.""" + operations = [op for op in catalog.operations if op.tool_name not in EXCLUDED] + return sorted(operations, key=sort_key) + + +@dataclass +class SweepRecord: + """The outcome of one sweep call.""" + + tool: str + domain: str + method: str + path: str + status: int | None + is_error: bool + kind: str | None + message: str + outcome: str + reason: str = "" + + +def classify( + tool: str, + status: int | None, + is_error: bool, + documented: frozenset[int], +) -> tuple[str, str]: + """Return ``(outcome, reason)``: success, documented, allowance or unexpected.""" + if status is not None and 200 <= status < 300 and not is_error: + return "success", "" + if status is not None and status in documented: + return "documented", "" + allowance = ALLOWANCES.get(tool) + if allowance is not None and status in allowance.statuses: + return "allowance", allowance.reason + return "unexpected", "" + + +# --------------------------------------------------------------------------- +# Results and report +# --------------------------------------------------------------------------- + + +@dataclass +class CallLog: + """Every ``(tool, status, is_error)`` seen during the session, from every test.""" + + calls: list[tuple[str, int | None, bool]] = field(default_factory=list) + sweep: list[SweepRecord] = field(default_factory=list) + + def add(self, tool: str, status: int | None, is_error: bool) -> None: + """Record one call.""" + self.calls.append((tool, status, is_error)) + + +LOG = CallLog() + + +def _bucket(tool: str, log: CallLog) -> str: + """``success`` when any call of the session got a 2xx, else the sweep outcome.""" + for name, status, is_error in log.calls: + if name == tool and status is not None and 200 <= status < 300 and not is_error: + return "success" + for record in log.sweep: + if record.tool == tool: + return record.outcome + return "not called" + + +def render_report(catalog: Catalog, log: CallLog, gitea_version: str) -> str: + """Render the coverage report as Markdown (no em dashes, no bold).""" + tool_domain = {op.tool_name: op.domain for op in catalog.operations} + tool_domain[RAW_TOOL] = "misc" + for name in EXCLUDED: + tool_domain.setdefault(name, "misc") + by_domain: dict[str, list[str]] = {} + for tool, domain in sorted(tool_domain.items()): + by_domain.setdefault(domain, []).append(tool) + sweep = {record.tool: record for record in log.sweep} + totals = {"success": 0, "documented": 0, "allowance": 0, "other": 0} + sections: list[str] = [] + rows: list[str] = [] + for domain in sorted(by_domain): + tools = by_domain[domain] + buckets = {tool: _bucket(tool, log) for tool in tools} + counts = {key: sum(1 for b in buckets.values() if b == key) for key in totals} + counts["other"] = ( + len(tools) - counts["success"] - counts["documented"] - counts["allowance"] + ) + for key in totals: + totals[key] += counts[key] + rows.append( + f"| {domain} | {len(tools)} | {counts['success']} | {counts['documented']} | " + f"{counts['allowance']} | {counts['other']} |" + ) + lines = [f"## {domain}", ""] + ok = [t for t in tools if buckets[t] == "success"] + lines.append(f"Tools with a 2xx ({len(ok)}):") + lines.append("") + lines.append(", ".join(f"`{t}`" for t in ok) if ok else "none") + lines.append("") + documented = [t for t in tools if buckets[t] == "documented"] + lines.append(f"Tools with a documented non-2xx status ({len(documented)}):") + lines.append("") + if documented: + lines.extend(["| Tool | Status |", "| --- | --- |"]) + lines.extend(f"| `{t}` | {sweep[t].status} |" for t in documented) + else: + lines.append("none") + lines.append("") + allowed = [t for t in tools if buckets[t] == "allowance"] + lines.append(f"Allowances ({len(allowed)}):") + lines.append("") + if allowed: + lines.extend(["| Tool | Status | Reason |", "| --- | --- | --- |"]) + lines.extend(f"| `{t}` | {sweep[t].status} | {sweep[t].reason} |" for t in allowed) + else: + lines.append("none") + lines.append("") + other = [t for t in tools if buckets[t] not in ("success", "documented", "allowance")] + if other: + lines.append(f"Not covered by this run ({len(other)}):") + lines.append("") + lines.append(", ".join(f"`{t}`" for t in other)) + lines.append("") + sections.append("\n".join(lines)) + header = [ + "# Integration coverage", + "", + "Generated by the live-Gitea integration suite; do not edit by hand. Regenerate with:", + "", + "```bash", + f"{REPORT_ENV}=1 AEGIS_INTEGRATION=1 uv run pytest tests/integration -m integration --no-cov", + "```", + "", + f"Target: Gitea {gitea_version}. Every tool of the server ({len(tool_domain)}) is called " + "at least once through the MCP server against a seeded instance (see " + "`tests/integration/test_sweep.py`). A call counts as covered when it reaches Gitea and " + "answers 2xx, answers a status the swagger documents for that operation, or answers a " + "status listed as an explicit allowance below. Scenario tests " + "(`tests/integration/test_scenario_*.py`) add real state changes; a tool counts as 2xx " + "when any call in the run got a 2xx.", + "", + "| Domain | Tools | 2xx | Documented non-2xx | Allowances | Not covered |", + "| --- | --- | --- | --- | --- | --- |", + *rows, + f"| total | {len(tool_domain)} | {totals['success']} | {totals['documented']} | " + f"{totals['allowance']} | {totals['other']} |", + "", + ] + if EXCLUDED: + header.append("Excluded from the generic sweep and covered by a dedicated test:") + header.append("") + header.extend(f"- `{tool}`: {reason}" for tool, reason in sorted(EXCLUDED.items())) + header.append("") + return "\n".join(header) + "\n" + "\n".join(sections) + + +def write_report_if_enabled(catalog: Catalog, gitea_version: str) -> bool: + """Write the report when ``AEGIS_IT_WRITE_REPORT=1`` and the sweep ran.""" + if os.environ.get(REPORT_ENV) != "1" or not LOG.sweep: + return False + REPORT_PATH.write_text(render_report(catalog, LOG, gitea_version), encoding="utf-8") + return True + + +def failing(records: Iterable[SweepRecord]) -> list[SweepRecord]: + """The records whose outcome is ``unexpected``.""" + return [record for record in records if record.outcome == "unexpected"] + + +def summarize(records: Sequence[SweepRecord]) -> dict[str, int]: + """Counts per outcome.""" + result: dict[str, int] = {} + for record in records: + result[record.outcome] = result.get(record.outcome, 0) + 1 + return result + + +# --------------------------------------------------------------------------- +# The sweep itself +# --------------------------------------------------------------------------- + +CallTool = Callable[..., Any] + + +def _short(text: str, limit: int = 160) -> str: + return " ".join(text.split())[:limit] + + +async def run_sweep( + call: CallTool, catalog: Catalog, spec: dict[str, Any], c: Ctx +) -> list[SweepRecord]: + """Call every tool once (in plan order) and classify the results. + + ``call(tool, **arguments)`` must return an object with ``status``, ``is_error``, + ``error_kind`` and ``payload`` (a :class:`tests.integration.conftest.ToolResult`). + """ + records: list[SweepRecord] = [] + jobs: list[tuple[str, str, str, str, frozenset[int], dict[str, Any]]] = [] + for operation in ordered_operations(catalog): + jobs.append( + ( + operation.tool_name, + operation.domain, + operation.method, + operation.path, + documented_statuses(spec, operation), + build_arguments(operation, c), + ) + ) + jobs.append((RAW_TOOL, "misc", "GET", "/version", frozenset({200}), OVERRIDES[RAW_TOOL](c))) + for tool, domain, method, path, documented, arguments in jobs: + result = await call(tool, **arguments) + error = result.payload.get("error") if result.is_error else None + message = "" + if isinstance(error, dict): + message = _short(str(error.get("message", ""))) + outcome, reason = classify(tool, result.status, result.is_error, documented) + records.append( + SweepRecord( + tool=tool, + domain=domain, + method=method, + path=path, + status=result.status, + is_error=result.is_error, + kind=result.error_kind, + message=message, + outcome=outcome, + reason=reason, + ) + ) + return records diff --git a/tests/integration/sweep_seed.py b/tests/integration/sweep_seed.py new file mode 100644 index 0000000..d311204 --- /dev/null +++ b/tests/integration/sweep_seed.py @@ -0,0 +1,445 @@ +"""Seed data for the all-tools sweep. + +:func:`seed` creates the resources the sweep's arguments point at (through the Gitea API, +not through the tools under test) and returns the context dictionary. Everything created +here is fake data in a disposable instance. +""" + +from __future__ import annotations + +import base64 +import secrets +import time +from typing import Any + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from tests.integration.harness import GiteaInstance, HarnessError + +Ctx = dict[str, Any] + + +# --------------------------------------------------------------------------- +# Small helpers +# --------------------------------------------------------------------------- + + +def b64(text: str | bytes) -> str: + """Base64 text of ``text``.""" + raw = text.encode("utf-8") if isinstance(text, str) else text + return base64.b64encode(raw).decode("ascii") + + +def ssh_public_key() -> str: + """A fresh ed25519 public key in OpenSSH format (throwaway).""" + key = Ed25519PrivateKey.generate().public_key() + raw = key.public_bytes(serialization.Encoding.OpenSSH, serialization.PublicFormat.OpenSSH) + return raw.decode("ascii") + + +# A 1x1 transparent PNG, for the avatar endpoints. +PNG_1X1 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==" + + +# --------------------------------------------------------------------------- +# Seeding +# --------------------------------------------------------------------------- + + +class Seeder: + """Creates the sweep's seed data through the Gitea API (not through the tools).""" + + def __init__(self, gitea: GiteaInstance, user: str, user_token: str) -> None: + self.gitea = gitea + self.user = user + self.user_token = user_token + + def call( + self, + method: str, + path: str, + *, + token: str | None = None, + ok: tuple[int, ...] = (200, 201, 202, 204), + **kwargs: Any, + ) -> Any: + """Call the API, require a status in ``ok`` and return the JSON body (or None).""" + response = self.gitea.api(method, path, token=token, **kwargs) + if response.status_code not in ok: + raise HarnessError( + f"seed {method} {path} failed: HTTP {response.status_code}: {response.text[:200]}" + ) + if response.content and "json" in response.headers.get("content-type", ""): + return response.json() + return None + + def put_file(self, repo: str, path: str, text: str, branch: str = "main") -> dict[str, Any]: + """Create a file through the contents API.""" + owner = self.gitea.admin_login + body = {"content": b64(text), "message": f"add {path}", "branch": branch} + result: dict[str, Any] = self.call( + "POST", f"/repos/{owner}/{repo}/contents/{path}", json=body + ) + return result + + +WORKFLOW = """\ +name: sweep +on: [push, workflow_dispatch] +jobs: + hello: + runs-on: aegis-sweep-nonexistent + steps: + - run: echo hello +""" + + +def seed(gitea: GiteaInstance, user: str, user_token: str) -> Ctx: + """Create the sweep's seed data and return the context the arguments are built from.""" + s = Seeder(gitea, user, user_token) + admin = gitea.admin_login + sfx = secrets.token_hex(3) + c: Ctx = {"admin": admin, "user": user, "owner": admin, "sfx": sfx} + c["admin_password"] = gitea.admin_password + + # A second user plus throwaway users for the destructive admin tools. + c["user2"] = gitea.create_user(f"sw-u2-{sfx}")[0] + c["user3"] = gitea.create_user(f"sw-u3-{sfx}")[0] + c["del_user"] = gitea.create_user(f"sw-del-{sfx}")[0] + c["rename_user"] = gitea.create_user(f"sw-ren-{sfx}")[0] + + # Organization with a team and a repository. + c["org"] = org = f"sw-org-{sfx}" + s.call("POST", "/orgs", json={"username": org, "visibility": "public"}) + c["org_repo"] = f"sw-orgrepo-{sfx}" + s.call("POST", f"/orgs/{org}/repos", json={"name": c["org_repo"], "auto_init": True}) + team = s.call( + "POST", + f"/orgs/{org}/teams", + json={ + "name": "sweepers", + "permission": "write", + "includes_all_repositories": False, + "units": ["repo.code", "repo.issues", "repo.pulls"], + }, + ) + c["team_id"], c["team_name"] = team["id"], team["name"] + team2 = s.call( + "POST", + f"/orgs/{org}/teams", + json={"name": "sweepers2", "permission": "read", "units": ["repo.code"]}, + ) + c["team2_name"] = team2["name"] + s.call("PUT", f"/teams/{c['team_id']}/members/{user}") + s.call("PUT", f"/teams/{c['team_id']}/repos/{org}/{c['org_repo']}") + s.call("PUT", f"/orgs/{org}/public_members/{user}") + s.call("PUT", f"/orgs/{org}/blocks/{c['user2']}") + c["org_label"] = s.call( + "POST", f"/orgs/{org}/labels", json={"name": "org-bug", "color": "#ff0000"} + )["id"] + c["org_hook"] = s.call( + "POST", + f"/orgs/{org}/hooks", + json={"type": "gitea", "config": {"url": "http://127.0.0.1:9/h", "content_type": "json"}}, + )["id"] + s.call("PUT", f"/orgs/{org}/actions/secrets/ORG_SECRET", json={"data": "org-secret-value"}) + s.call("POST", f"/orgs/{org}/actions/variables/ORG_VAR", json={"value": "1"}) + c["org_issue_id"] = s.call( + "POST", f"/repos/{org}/{c['org_repo']}/issues", json={"title": "org issue"} + )["id"] + _project(s, c, "org", f"/orgs/{org}/projects") + + # Throwaway org and repository for the destructive tools. + c["del_org"] = f"sw-delorg-{sfx}" + s.call("POST", "/orgs", json={"username": c["del_org"]}) + c["del_org2"] = f"sw-delorg2-{sfx}" + s.call("POST", "/orgs", json={"username": c["del_org2"]}) + c["rename_org"] = f"sw-renorg-{sfx}" + s.call("POST", "/orgs", json={"username": c["rename_org"]}) + c["del_repo"] = f"sw-delrepo-{sfx}" + gitea.create_repo(admin, c["del_repo"]) + c["xfer_repo"] = f"sw-xfer-{sfx}" + gitea.create_repo(admin, c["xfer_repo"]) + c["template_repo"] = f"sw-tpl-{sfx}" + gitea.create_repo(admin, c["template_repo"]) + s.call("PATCH", f"/repos/{admin}/{c['template_repo']}", json={"template": True}) + + # The main seed repository. + c["repo"] = repo = f"sw-repo-{sfx}" + created = gitea.create_repo(admin, repo) + c["repo_id"] = created["id"] + base = f"/repos/{admin}/{repo}" + branch = s.call("GET", f"{base}/branches/main") + c["sha"] = branch["commit"]["id"] + s.put_file(repo, ".editorconfig", "root = true\n") + s.put_file(repo, "docs/a.txt", "alpha\n") + s.put_file(repo, ".gitea/workflows/sweep.yml", WORKFLOW) + c["sha"] = s.call("GET", f"{base}/branches/main")["commit"]["id"] + s.call("PUT", f"{base}/collaborators/{user}", json={"permission": "write"}) + s.call("PUT", f"{base}/collaborators/{c['user2']}", json={"permission": "write"}) + s.call("PUT", f"{base}/topics", json={"topics": ["sweep", "seed"]}) + s.call("PUT", f"{base}/topics/extra") + # A fork (for merge-upstream), a pull mirror and push-mirror targets, all from this instance. + local = f"http://127.0.0.1:3000/{admin}" + c["fork_repo"] = f"sw-fork-{sfx}" + s.call("POST", f"{base}/forks", json={"organization": org, "name": c["fork_repo"]}) + c["mirror_repo"] = f"sw-mirror-{sfx}" + s.call( + "POST", + "/repos/migrate", + json={ + "clone_addr": f"{local}/{c['template_repo']}.git", + "repo_name": c["mirror_repo"], + "service": "git", + "mirror": True, + "mirror_interval": "8h0m0s", + }, + ) + c["push_target"] = f"sw-ptarget-{sfx}" + gitea.create_repo(admin, c["push_target"], auto_init=False) + c["push_target2"] = f"sw-ptarget2-{sfx}" + gitea.create_repo(admin, c["push_target2"], auto_init=False) + c["push_mirror"] = s.call( + "POST", + f"{base}/push_mirrors", + json={ + "remote_address": f"{local}/{c['push_target']}.git", + "remote_username": admin, + "remote_password": gitea.admin_password, + "interval": "8h0m0s", + "sync_on_commit": False, + }, + )["remote_name"] + s.call( + "POST", + f"{base}/statuses/{c['sha']}", + json={"state": "success", "context": "ci/sweep"}, + ) + s.call("POST", f"{base}/tags", json={"tag_name": "v0.0.1", "target": "main"}) + annotated = s.call( + "POST", f"{base}/tags", json={"tag_name": "v0.0.3", "target": "main", "message": "note"} + ) + c["tag_sha"] = annotated["id"] + for name in ("feat", "feat2", "feat3", "rn", "del-branch"): + s.call( + "POST", f"{base}/branches", json={"new_branch_name": name, "old_branch_name": "main"} + ) + s.put_file(repo, "feat.txt", "feature\n", branch="feat") + s.put_file(repo, "feat2.txt", "feature two\n", branch="feat2") + s.put_file(repo, "feat3.txt", "feature three\n", branch="feat3") + s.call("POST", f"{base}/branch_protections", json={"rule_name": "release/*"}) + c["protection"] = "release/*" + s.call( + "POST", + f"{base}/tag_protections", + json={"name_pattern": "v9*", "whitelist_usernames": [admin]}, + ) + c["tag_protection"] = s.call("GET", f"{base}/tag_protections")[0]["id"] + s.put_file(repo, "to-delete.txt", "bye\n") + s.put_file(repo, "to-update.txt", "before\n") + c["delete_file_sha"] = s.call("GET", f"{base}/contents/to-delete.txt")["sha"] + c["update_file_sha"] = s.call("GET", f"{base}/contents/to-update.txt")["sha"] + c["blob_sha"] = s.call("GET", f"{base}/contents/README.md")["sha"] + c["page_name"] = "Home" + c["cron_task"] = "cleanup_hook_task_table" + + # Issues: 1 main, 2 dependency target, 3 pinned, 4/5 more targets. + for number, title in enumerate(("main", "dependency", "pinned", "target four", "target five")): + issue = s.call("POST", f"{base}/issues", json={"title": f"Sweep {title}", "body": "x"}) + assert issue["number"] == number + 1 + c["issue"], c["issue2"], c["issue3"], c["issue4"], c["issue5"] = 1, 2, 3, 4, 5 + c["issue1_id"] = s.call("GET", f"{base}/issues/1")["id"] + c["comment"] = s.call("POST", f"{base}/issues/1/comments", json={"body": "one"})["id"] + c["comment2"] = s.call("POST", f"{base}/issues/1/comments", json={"body": "two"})["id"] + c["label"] = s.call("POST", f"{base}/labels", json={"name": "bug", "color": "#d73a4a"})["id"] + s.call("POST", f"{base}/issues/1/labels", json={"labels": [c["label"]]}) + c["milestone"] = s.call("POST", f"{base}/milestones", json={"title": "m1"})["id"] + c["milestone_name"] = "m1" + s.call("POST", f"{base}/issues/1/assignees", json={"assignees": [admin]}) + s.call("PUT", f"{base}/issues/1/subscriptions/{user}") + s.call("POST", f"{base}/issues/1/reactions", json={"content": "+1"}) + s.call("POST", f"{base}/issues/comments/{c['comment']}/reactions", json={"content": "+1"}) + s.call("POST", f"{base}/issues/1/dependencies", json={"index": 2, "owner": admin, "repo": repo}) + s.call("POST", f"{base}/issues/1/blocks", json={"index": 3, "owner": admin, "repo": repo}) + # A comment by another user notifies the admin (author and assignee of issue 1). + s.call("POST", f"{base}/issues/1/comments", token=user_token, json={"body": "ping"}) + s.call("POST", f"{base}/issues/3/pin") + s.call("POST", f"{base}/issues/2/stopwatch/start", token=user_token) + time_entry = s.call("POST", f"{base}/issues/1/times", json={"time": 60}) + c["time_id"] = time_entry["id"] + attachment = s.call( + "POST", + f"{base}/issues/1/assets", + files={"attachment": ("issue.txt", b"issue attachment", "text/plain")}, + params={"name": "issue.txt"}, + ) + c["issue_attachment"] = attachment["id"] + attachment = s.call( + "POST", + f"{base}/issues/comments/{c['comment']}/assets", + files={"attachment": ("comment.txt", b"comment attachment", "text/plain")}, + params={"name": "comment.txt"}, + ) + c["comment_attachment"] = attachment["id"] + + # Pull requests: 6 is the main one, 7 is merged by the sweep. + pr = s.call("POST", f"{base}/pulls", json={"title": "Sweep PR", "base": "main", "head": "feat"}) + c["pr"] = pr["number"] + pr2 = s.call( + "POST", f"{base}/pulls", json={"title": "Sweep PR two", "base": "main", "head": "feat2"} + ) + c["pr2"] = pr2["number"] + pr_path = f"{base}/pulls/{c['pr']}" + review = s.call( + "POST", + f"{pr_path}/reviews", + token=user_token, + json={ + "event": "REQUEST_CHANGES", + "body": "please change", + "comments": [{"path": "feat.txt", "new_position": 1, "body": "line note"}], + }, + ) + c["review"] = review["id"] + c["review_comment"] = s.call("GET", f"{pr_path}/reviews/{review['id']}/comments")[0]["id"] + c["review2"] = s.call( + "POST", f"{pr_path}/reviews", token=user_token, json={"event": "COMMENT", "body": "fyi"} + )["id"] + c["pending_review"] = s.call( + "POST", f"{pr_path}/reviews", token=user_token, json={"body": "draft"} + )["id"] + s.call( + "POST", f"{pr_path}/requested_reviewers", json={"reviewers": [c["user2"]]}, ok=(201, 422) + ) + + # Release with an asset (and a second release removed by tag). + release = s.call( + "POST", f"{base}/releases", json={"tag_name": "v1.0.0", "target_commitish": "main"} + ) + c["release"], c["release_tag"] = release["id"], "v1.0.0" + asset = s.call( + "POST", + f"{base}/releases/{release['id']}/assets", + files={"attachment": ("rel.bin", b"release bytes", "application/octet-stream")}, + params={"name": "rel.bin"}, + ) + c["release_asset"] = asset["id"] + s.call("POST", f"{base}/releases", json={"tag_name": "v2.0.0", "target_commitish": "main"}) + c["release_tag2"] = "v2.0.0" + + # Hooks, keys, secrets, variables, wiki, projects. + hook_body = {"type": "gitea", "config": {"url": "http://127.0.0.1:9/h", "content_type": "json"}} + c["repo_hook"] = s.call("POST", f"{base}/hooks", json=hook_body)["id"] + s.call( + "PATCH", + f"{base}/hooks/git/pre-receive", + json={"content": "#!/bin/sh\nexit 0\n"}, + ok=(200,), + ) + c["git_hook"] = "pre-receive" + c["user_hook"] = s.call("POST", "/user/hooks", json=hook_body)["id"] + c["admin_hook"] = s.call("POST", "/admin/hooks", json=hook_body)["id"] + c["deploy_key"] = s.call( + "POST", f"{base}/keys", json={"title": f"deploy-{sfx}", "key": ssh_public_key()} + )["id"] + c["user_key"] = s.call( + "POST", "/user/keys", json={"title": f"mine-{sfx}", "key": ssh_public_key()} + )["id"] + c["admin_user_key"] = s.call( + "POST", f"/admin/users/{user}/keys", json={"title": f"k-{sfx}", "key": ssh_public_key()} + )["id"] + s.call("PUT", f"{base}/actions/secrets/REPO_SECRET", json={"data": "repo-secret-value"}) + c["user_secret"] = f"USER_SECRET_{sfx.upper()}" + c["user_var"] = f"USER_VAR_{sfx.upper()}" + s.call("PUT", f"/user/actions/secrets/{c['user_secret']}", json={"data": "user-secret-value"}) + s.call("POST", f"{base}/actions/variables/REPO_VAR", json={"value": "1"}) + s.call("POST", f"/user/actions/variables/{c['user_var']}", json={"value": "1"}) + s.call( + "POST", + f"{base}/wiki/new", + json={"title": "Home", "content_base64": b64("# home\n"), "message": "home"}, + ) + _project(s, c, "repo", f"{base}/projects") + _project(s, c, "user", "/user/projects") + c["oauth_app"] = s.call( + "POST", + "/user/applications/oauth2", + json={"name": f"sweep-{sfx}", "redirect_uris": ["http://127.0.0.1:1/cb"]}, + )["id"] + + # Current user: email, follow, star, watch, block. + c["email"] = f"sweep-{sfx}@example.test" + s.call("POST", "/user/emails", json={"emails": [c["email"]]}) + s.call("PUT", f"/user/following/{user}") + s.call("PUT", f"/user/starred/{admin}/{repo}") + s.call("PUT", f"{base}/subscription") + s.call("PUT", f"/user/blocks/{c['user2']}") + # A throwaway access token of the admin (basic auth), deleted by name. + token_name = f"sweep-{sfx}" + s.call( + "POST", + f"/users/{admin}/tokens", + auth=(admin, gitea.admin_password), + json={"name": token_name, "scopes": ["read:user"]}, + ) + c["throwaway_token"] = token_name + + # A generic package. + c["pkg"] = pkg = f"sweep-pkg-{sfx}" + upload = gitea.request( + "PUT", f"/api/packages/{admin}/generic/{pkg}/1.0.0/payload.bin", content=b"package" + ) + if upload.status_code != 201: + raise HarnessError(f"seed package upload failed: HTTP {upload.status_code}") + c["pkg_version"] = "1.0.0" + c["pkg2"] = f"sweep-pkg2-{sfx}" + upload = gitea.request( + "PUT", f"/api/packages/{admin}/generic/{c['pkg2']}/1.0.0/payload.bin", content=b"package" + ) + if upload.status_code != 201: + raise HarnessError(f"seed package upload failed: HTTP {upload.status_code}") + + # The workflow run created by the push of the workflow file (queued, no runner). + c["run"], c["job_id"] = _find_run(s, base) + c["notification"] = _find_notification(s, admin) + return c + + +def _project(s: Seeder, c: Ctx, scope: str, path: str) -> None: + """Create a project with two extra columns; store ids as ``_project/_column...``.""" + project = s.call("POST", path, json={"title": f"{scope} project"})["id"] + c[f"{scope}_project"] = project + ids: dict[str, int] = {} + for title in ("todo", "done"): + ids[title] = s.call("POST", f"{path}/{project}/columns", json={"title": title})["id"] + c[f"{scope}_column"], c[f"{scope}_column2"] = ids["todo"], ids["done"] + columns = s.call("GET", f"{path}/{project}/columns") + c[f"{scope}_columns"] = [column["id"] for column in columns] + + +def _find_run(s: Seeder, base: str) -> tuple[Any, Any]: + """Return ``(run id, job id)`` of the first workflow run, or ``(1, 1)`` if none appeared.""" + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + runs = s.call("GET", f"{base}/actions/runs") + entries = runs.get("workflow_runs") or [] + if entries: + run_id = entries[0]["id"] + jobs = s.call("GET", f"{base}/actions/runs/{run_id}/jobs") + job_entries = jobs.get("jobs") or [] + if job_entries: + return run_id, job_entries[0]["id"] + time.sleep(0.5) + return 1, 1 + + +def _find_notification(s: Seeder, admin: str) -> Any: + """Return the id of a notification of the admin if one exists, else 1.""" + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + listing = s.call("GET", "/notifications", params={"all": "true"}) + if listing: + return listing[0]["id"] + time.sleep(0.5) + return 1 diff --git a/tests/integration/test_sweep.py b/tests/integration/test_sweep.py new file mode 100644 index 0000000..966de47 --- /dev/null +++ b/tests/integration/test_sweep.py @@ -0,0 +1,54 @@ +"""The sweep: every tool is called at least once through the MCP server.""" + +from __future__ import annotations + +import pytest + +from aegis_gitea_mcp.catalog import load_catalog +from tests.integration.conftest import McpSession, RegularUser +from tests.integration.harness import GiteaInstance +from tests.integration.sweep_plan import ( + EXCLUDED, + LOG, + RAW_TOOL, + failing, + load_spec, + run_sweep, + summarize, +) +from tests.integration.sweep_seed import Ctx, seed + +#: At least this share of the tools must get a real 2xx, so a systematic argument problem +#: cannot hide behind documented 404 responses. +MIN_SUCCESS_SHARE = 0.55 + + +@pytest.fixture(scope="module") +def sweep_context(gitea: GiteaInstance, regular_user: RegularUser) -> Ctx: + """The seeded resources the sweep's arguments point at.""" + return seed(gitea, regular_user.login, regular_user.token) + + +async def test_every_tool_reaches_gitea(mcp: McpSession, sweep_context: Ctx) -> None: + catalog = load_catalog() + listed = {tool.name for tool in (await mcp.client.list_tools()).tools} + expected = {op.tool_name for op in catalog.operations} | {RAW_TOOL} + assert listed == expected + assert len(expected) == 538 + + records = await run_sweep(mcp.call, catalog, load_spec(), sweep_context) + LOG.sweep = records + + called = [record.tool for record in records] + assert len(called) == len(set(called)), "a tool was called twice" + assert set(called) == expected - set(EXCLUDED), "tools missing from the sweep" + + bad = failing(records) + summary = summarize(records) + report = "\n".join( + f" {r.tool}: status={r.status} kind={r.kind} {r.method} {r.path} :: {r.message}" + for r in bad + ) + assert not bad, f"{len(bad)} tool(s) with an unexpected result ({summary}):\n{report}" + success_share = summary.get("success", 0) / len(records) + assert success_share >= MIN_SUCCESS_SHARE, f"only {success_share:.0%} of tools got a 2xx" diff --git a/tests/integration/test_sweep_isolated.py b/tests/integration/test_sweep_isolated.py new file mode 100644 index 0000000..37df1d3 --- /dev/null +++ b/tests/integration/test_sweep_isolated.py @@ -0,0 +1,43 @@ +"""Tools the generic sweep must not call because they would break the run itself.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from mcp import Client + +from aegis_gitea_mcp.audit import AuditLogger +from aegis_gitea_mcp.server import build_server +from tests.integration.conftest import McpSession, integration_settings +from tests.integration.harness import GiteaInstance +from tests.integration.sweep_plan import EXCLUDED + + +async def test_delete_current_token_deletes_only_a_throwaway_token( + gitea: GiteaInstance, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + assert set(EXCLUDED) == {"gitea_misc_delete_current_token"} + monkeypatch.setenv("GITEA_URL", gitea.base_url) + throwaway = gitea.mint_token(gitea.admin_login, gitea.admin_password, ["all"]) + settings = integration_settings(gitea, tmp_path / "state", gitea_token=throwaway) + audit = AuditLogger(settings.resolved_audit_log_path) + try: + async with Client(build_server(settings, audit=audit)) as client: + session = McpSession(client, settings.resolved_audit_log_path) + current = await session.ok("gitea_misc_get_current_token") + assert "all" in current.data["scopes"] + assert gitea.api("GET", "/user", token=throwaway).status_code == 200 + + deleted = await session.ok("gitea_misc_delete_current_token") + assert deleted.data is None + assert throwaway not in session.audit_text() + + # The token is gone: Gitea now refuses it, while the harness admin token lives on. + assert gitea.api("GET", "/user", token=throwaway).status_code == 401 + assert gitea.api("GET", "/user").status_code == 200 + again = await session.call("gitea_misc_get_current_token") + assert again.is_error and again.status == 401 + assert throwaway not in again.text + finally: + audit.close() diff --git a/tests/test_sweep_plan.py b/tests/test_sweep_plan.py new file mode 100644 index 0000000..f8a9cc5 --- /dev/null +++ b/tests/test_sweep_plan.py @@ -0,0 +1,273 @@ +"""Checks of the integration sweep plan that need no Docker. + +The sweep itself runs against a live Gitea (``tests/integration/test_sweep.py``). These +tests make sure the plan stays consistent with the catalog: every tool gets arguments that +pass its schema, every tool is called exactly once in a sensible order, and every override, +allowance and exclusion names a real tool. +""" + +from __future__ import annotations + +from typing import Any + +import pytest +from jsonschema import Draft202012Validator + +from aegis_gitea_mcp.catalog import load_catalog +from aegis_gitea_mcp.server import RAW_INPUT_SCHEMA +from tests.integration import sweep_plan +from tests.integration.sweep_plan import ( + ALLOWANCES, + EXCLUDED, + OVERRIDES, + RAW_TOOL, + CallLog, + SweepRecord, + build_arguments, + classify, + documented_statuses, + load_spec, + ordered_operations, + phase_of, + render_report, + sort_key, +) + +CATALOG = load_catalog() +SPEC = load_spec() + +#: Context keys that hold strings; every other key is an integer id in the fake context. +STRING_SUFFIXES = ("_name", "_repo", "_user", "_sha", "_tag", "_token", "_task") +STRING_KEYS = { + "admin", + "user", + "user2", + "user3", + "owner", + "repo", + "org", + "sfx", + "email", + "protection", + "page_name", + "pkg", + "pkg2", + "pkg_version", + "sha", + "blob_sha", + "tag_sha", + "delete_file_sha", + "update_file_sha", + "milestone_name", + "release_tag", + "release_tag2", + "git_hook", + "cron_task", + "push_mirror", + "admin_password", + "throwaway_token", + "user_secret", + "user_var", + "del_org", + "del_org2", + "rename_org", + "org_repo", + "template_repo", + "xfer_repo", + "del_repo", + "fork_repo", + "mirror_repo", + "push_target", + "push_target2", + "del_user", + "rename_user", + "team_name", + "team2_name", +} + + +class FakeContext(dict[str, Any]): + """A seed context for tests: strings for the known string keys, 1 for everything else.""" + + def __missing__(self, key: str) -> Any: + if key in STRING_KEYS or key.endswith(STRING_SUFFIXES): + return "x1" + if key.endswith("_columns"): + return [1, 2, 3] + return 1 + + +def fake_context() -> FakeContext: + return FakeContext(sfx="abc123", admin="admin", user="alice", owner="admin") + + +def all_tool_names() -> set[str]: + return {op.tool_name for op in CATALOG.operations} | {RAW_TOOL} + + +def test_every_operation_gets_arguments_that_pass_its_schema() -> None: + context = fake_context() + problems: list[str] = [] + for operation in CATALOG.operations: + arguments = build_arguments(operation, context) + validator = Draft202012Validator(operation.input_schema) + errors = sorted(error.message[:80] for error in validator.iter_errors(arguments)) + if errors: + problems.append(f"{operation.tool_name}: {errors[0]}") + assert not problems, "\n".join(problems) + + +def test_raw_tool_arguments_pass_its_schema() -> None: + arguments = OVERRIDES[RAW_TOOL](fake_context()) + assert not list(Draft202012Validator(RAW_INPUT_SCHEMA).iter_errors(arguments)) + + +def test_plan_covers_every_tool_except_the_excluded_ones_exactly_once() -> None: + planned = [op.tool_name for op in ordered_operations(CATALOG)] + assert len(planned) == len(set(planned)) + assert set(planned) == {op.tool_name for op in CATALOG.operations} - set(EXCLUDED) + assert len(CATALOG.operations) + 1 == 538 + + +def test_overrides_allowances_and_exclusions_name_real_tools() -> None: + names = all_tool_names() + assert set(OVERRIDES) <= names + assert set(ALLOWANCES) <= names + assert set(EXCLUDED) <= names + assert all(allowance.reason for allowance in ALLOWANCES.values()) + assert all(reason for reason in EXCLUDED.values()) + + +def test_order_is_reads_then_creates_then_updates_then_deletes() -> None: + operations = ordered_operations(CATALOG) + phases = [phase_of(op) for op in operations] + assert phases == sorted(phases) + assert phase_of(next(op for op in operations if op.tool_name == "gitea_repo_get")) == 0 + first_delete = next(i for i, p in enumerate(phases) if p == 5) + assert all(op.method == "DELETE" for op in operations[first_delete:]) + assert all(op.read_only for op in operations if phase_of(op) == 0) + + +def test_deletes_run_leaf_resources_before_their_parents() -> None: + names = [op.tool_name for op in ordered_operations(CATALOG)] + leaf_first = [ + ("gitea_issue_delete_comment", "gitea_issue_delete"), + ("gitea_issue_delete_label", "gitea_issue_delete_milestone"), + ("gitea_release_delete_release_attachment", "gitea_release_delete_release"), + ("gitea_org_delete_repos", "gitea_org_delete"), + ] + for leaf, parent in leaf_first: + assert names.index(leaf) < names.index(parent), (leaf, parent) + assert names.index("gitea_pull_merge_pull_request") > names.index( + "gitea_pull_create_pull_review" + ) + + +def test_destructive_tools_target_throwaway_resources() -> None: + by_name = {op.tool_name: op for op in CATALOG.operations} + context = fake_context() + context.update( + repo="seed-repo", + org="seed-org", + user="seed-user", + del_repo="throwaway-repo", + del_org="throwaway-org", + del_org2="throwaway-org-two", + del_user="throwaway-user", + ) + assert build_arguments(by_name["gitea_repo_delete"], context)["repo"] == "throwaway-repo" + assert build_arguments(by_name["gitea_org_delete"], context)["org"] == "throwaway-org-two" + assert build_arguments(by_name["gitea_org_delete_repos"], context)["org"] == "throwaway-org" + assert ( + build_arguments(by_name["gitea_admin_delete_user"], context)["username"] == "throwaway-user" + ) + + +def test_documented_statuses_come_from_the_swagger() -> None: + operation = next(op for op in CATALOG.operations if op.tool_name == "gitea_repo_get") + assert documented_statuses(SPEC, operation) >= {200, 404} + for operation in CATALOG.operations: + assert documented_statuses(SPEC, operation), operation.tool_name + + +@pytest.mark.parametrize( + ("status", "is_error", "documented", "expected"), + [ + (200, False, frozenset(), "success"), + (204, False, frozenset(), "success"), + (404, True, frozenset({404}), "documented"), + (404, True, frozenset({403}), "unexpected"), + (None, True, frozenset({404}), "unexpected"), + (500, True, frozenset({404}), "unexpected"), + (301, True, frozenset({200}), "unexpected"), + ], +) +def test_classify( + status: int | None, is_error: bool, documented: frozenset[int], expected: str +) -> None: + assert classify("gitea_repo_get", status, is_error, documented)[0] == expected + + +def test_classify_accepts_listed_allowances_only_for_their_status() -> None: + tool = "gitea_user_create_token" + outcome, reason = classify(tool, 401, True, frozenset({201})) + assert outcome == "allowance" + assert "basic auth" in reason + assert classify(tool, 500, True, frozenset({201}))[0] == "unexpected" + assert classify("gitea_repo_get", 401, True, frozenset({200}))[0] == "unexpected" + + +def test_sort_key_orders_phase_before_name() -> None: + by_name = {op.tool_name: op for op in CATALOG.operations} + assert sort_key(by_name["gitea_repo_get"]) < sort_key(by_name["gitea_issue_create_issue"]) + assert sort_key(by_name["gitea_issue_create_issue"]) < sort_key(by_name["gitea_repo_edit"]) + + +def test_report_groups_by_domain_and_has_no_em_dash_or_bold() -> None: + log = CallLog() + log.add("gitea_repo_get", 200, False) + log.add("gitea_repo_delete", 204, False) + log.sweep = [ + SweepRecord("gitea_repo_get", "repo", "GET", "/x", 200, False, None, "", "success"), + SweepRecord( + "gitea_repo_list_forks", "repo", "GET", "/y", 404, True, "not_found", "", "documented" + ), + SweepRecord( + "gitea_misc_get_signing_key", + "misc", + "GET", + "/z", + 404, + True, + "not_found", + "", + "allowance", + "no signing key", + ), + ] + text = render_report(CATALOG, log, "28.1.0") + assert "# Integration coverage" in text + assert "## repo" in text + assert "`gitea_repo_get`" in text + assert "| `gitea_repo_list_forks` | 404 |" in text + assert "no signing key" in text + assert "—" not in text + assert "**" not in text + assert "Not covered by this run" in text + + +def test_report_is_only_written_when_enabled( + monkeypatch: pytest.MonkeyPatch, tmp_path: Any +) -> None: + target = tmp_path / "coverage.md" + monkeypatch.setattr(sweep_plan, "REPORT_PATH", target) + monkeypatch.setattr(sweep_plan, "LOG", CallLog()) + monkeypatch.delenv(sweep_plan.REPORT_ENV, raising=False) + assert sweep_plan.write_report_if_enabled(CATALOG, "28.1.0") is False + monkeypatch.setenv(sweep_plan.REPORT_ENV, "1") + assert sweep_plan.write_report_if_enabled(CATALOG, "28.1.0") is False # sweep did not run + sweep_plan.LOG.sweep = [ + SweepRecord("gitea_repo_get", "repo", "GET", "/x", 200, False, None, "", "success") + ] + assert sweep_plan.write_report_if_enabled(CATALOG, "28.1.0") is True + assert target.read_text(encoding="utf-8").startswith("# Integration coverage")