Hiddenden/AegisGitea-MCP
5
0 Fork 0
Code Issues 2 Pull Requests Actions Packages Projects 1 Releases 1 Wiki Activity

test: sweep every tool through the MCP server against a seeded Gitea

Browse Source
This commit is contained in:
Latte committed 2026-10-14 17:57:54 +02:00
1 parent 22c8157ca4
commit 303bd9aa21
7 files changed
+1759 -1
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/integration/conftest.py
+12 -1
View File
@@ -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
tests/integration/harness.py
+4
View File
@@ -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",
}
tests/integration/sweep_plan.py
+928
View File
@@ -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
tests/integration/sweep_seed.py
+445
View File
@@ -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 ``<scope>_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
tests/integration/test_sweep.py
+54
View File
@@ -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"
tests/integration/test_sweep_isolated.py
+43
View File
@@ -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()
tests/test_sweep_plan.py
+273
View File
@@ -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")