AegisGitea-MCP/tests/test_sweep_plan.py

"""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 "\u2014" 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")


def _record(tool: str, status: int | None, is_error: bool = False) -> SweepRecord:
    return SweepRecord(tool, "repo", "GET", "/x", status, is_error, None, "", "success")


def test_outcome_class_is_2xx_or_the_exact_status() -> None:
    assert sweep_plan.outcome_class(_record("t", 200)) == "2xx"
    assert sweep_plan.outcome_class(_record("t", 204)) == "2xx"
    assert sweep_plan.outcome_class(_record("t", 404, True)) == 404
    assert sweep_plan.outcome_class(_record("t", 200, True)) == 200  # an error flag wins
    assert sweep_plan.outcome_class(_record("t", None, True)) == "no-status"


def test_baseline_reports_every_regression_at_once_and_notes_upgrades() -> None:
    baseline: dict[str, Any] = {"a": "2xx", "b": 404, "c": 404, "d": "2xx", "gone": "2xx"}
    records = [
        _record("a", 404, True),  # 2xx -> 404: regression
        _record("b", 200),  # 404 -> 2xx: upgrade
        _record("c", 403, True),  # 404 -> 403: a different status is a regression
        _record("d", 200),  # unchanged
        _record("new", 200),  # not in the baseline
    ]

    diff = sweep_plan.compare_baseline(records, baseline)

    assert diff.upgrades == ["b: baseline 404, now 2xx"]
    assert len(diff.regressions) == 4
    assert diff.regressions[0].startswith("a: baseline 2xx, got 404")
    assert diff.regressions[1].startswith("c: baseline 404, got 403")
    assert "new: no baseline entry" in diff.regressions[2]
    assert diff.regressions[3] == "gone: in the baseline but not called by the sweep"


def test_baseline_round_trips_and_covers_the_whole_sweep(tmp_path: Any) -> None:
    target = tmp_path / "baseline.json"
    records = [_record("b", 404, True), _record("a", 200)]
    sweep_plan.write_baseline(records, "28.1.0", target)
    assert sweep_plan.load_baseline(target) == {"a": "2xx", "b": 404}
    assert list(sweep_plan.load_baseline(target)) == ["a", "b"]

    # The committed baseline names exactly the tools the sweep calls.
    committed = sweep_plan.load_baseline()
    swept = {op.tool_name for op in ordered_operations(CATALOG)} | {RAW_TOOL}
    assert set(committed) == swept
    assert all(value == "2xx" or isinstance(value, int) for value in committed.values())