97 lines
3.4 KiB
Python
97 lines
3.4 KiB
Python
"""The sweep: every tool is called at least once through the MCP server."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from aegis_gitea_mcp.audit import validate_audit_log_integrity
|
|
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 (
|
|
BASELINE_ENV,
|
|
EXCLUDED,
|
|
LOG,
|
|
RAW_TOOL,
|
|
compare_baseline,
|
|
failing,
|
|
load_baseline,
|
|
load_spec,
|
|
run_sweep,
|
|
summarize,
|
|
write_baseline,
|
|
)
|
|
from tests.integration.sweep_seed import Ctx, seed
|
|
|
|
|
|
@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, gitea: GiteaInstance, regular_user: RegularUser
|
|
) -> 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}"
|
|
|
|
# The pinned outcome per tool: an argument or path regression cannot hide behind a
|
|
# documented 404, because a 2xx that turns into any other status fails here.
|
|
if os.environ.get(BASELINE_ENV) == "1":
|
|
write_baseline(records, load_catalog().spec_version)
|
|
else:
|
|
diff = compare_baseline(records, load_baseline())
|
|
if diff.upgrades:
|
|
print(
|
|
f"\n{len(diff.upgrades)} tool(s) improved on the baseline; refresh it with "
|
|
f"{BASELINE_ENV}=1:\n " + "\n ".join(diff.upgrades)
|
|
)
|
|
assert not diff.regressions, (
|
|
f"{len(diff.regressions)} tool(s) differ from sweep_baseline.json "
|
|
f"(regenerate with {BASELINE_ENV}=1 only when the change is intended):\n "
|
|
+ "\n ".join(diff.regressions)
|
|
)
|
|
|
|
# Whatever the 538 calls did, no credential or secret value came back in a result.
|
|
hidden = [
|
|
gitea.admin_token,
|
|
gitea.admin_password,
|
|
regular_user.token,
|
|
regular_user.password,
|
|
"rotated-org-value",
|
|
"rotated-repo-value",
|
|
"rotated-user-value",
|
|
"org-secret-value",
|
|
"repo-secret-value",
|
|
"user-secret-value",
|
|
]
|
|
for value in hidden:
|
|
assert all(value not in text for text in mcp.texts), "a secret leaked into a result"
|
|
assert value not in mcp.audit_text(), "a secret leaked into the audit log"
|
|
|
|
# One audit entry per call, and the hash chain over all of them is intact.
|
|
assert len(mcp.audit_entries()) == len(records)
|
|
valid, errors = validate_audit_log_integrity(mcp.audit_path)
|
|
assert valid, errors[:3]
|