diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index f9d39d2..9afa687 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -6,14 +6,22 @@ Tests in this directory are marked ``integration`` and skipped unless from __future__ import annotations +import asyncio +import base64 +import json import os import re -from collections.abc import Iterator +from collections.abc import AsyncIterator, Iterator from dataclasses import dataclass, field from pathlib import Path +from typing import Any import pytest +from mcp import Client +from aegis_gitea_mcp.audit import AuditLogger +from aegis_gitea_mcp.config import Settings +from aegis_gitea_mcp.server import build_server from tests.integration.harness import ( GiteaInstance, HarnessError, @@ -97,3 +105,231 @@ def regular_user(gitea: GiteaInstance) -> RegularUser: login, password = gitea.create_user(gitea.unique("it-user")) token = gitea.mint_token(login, password, ["all"]) return RegularUser(login=login, password=password, token=token) + + +# --------------------------------------------------------------------------- +# Calling tools through the MCP server +# --------------------------------------------------------------------------- + + +@dataclass +class ToolResult: + """One tool call result: the parsed envelope plus the raw text.""" + + tool: str + is_error: bool + payload: dict[str, Any] + text: str + + @property + def status(self) -> int | None: + """HTTP status Gitea answered with (None when the call never reached Gitea).""" + error = self.payload.get("error") + if isinstance(error, dict): + status = error.get("status") + return status if isinstance(status, int) else None + status = self.payload.get("status") + return status if isinstance(status, int) else None + + @property + def data(self) -> Any: + """The ``data`` member of a success envelope.""" + return self.payload.get("data") + + @property + def error_kind(self) -> str | None: + """The ``error.kind`` of a failed call.""" + error = self.payload.get("error") + return error.get("kind") if isinstance(error, dict) else None + + @property + def identity(self) -> dict[str, Any]: + """The ``identity`` object of the envelope.""" + identity = self.payload.get("identity") + return identity if isinstance(identity, dict) else {} + + +class McpSession: + """A connected in-memory MCP client plus bookkeeping for secret-leak assertions. + + Every result text is kept in :attr:`texts` so a test can assert that a secret value + never came back, whatever tool returned it. + """ + + def __init__(self, client: Client, audit_path: Path) -> None: + self.client = client + self.audit_path = audit_path + self.texts: list[str] = [] + + async def call(self, tool: str, /, **arguments: Any) -> ToolResult: + """Call ``tool`` with ``arguments`` and return the parsed result (never raises).""" + raw = await self.client.call_tool(tool, arguments) + assert raw.content, f"{tool}: empty result" + block = raw.content[0] + text = getattr(block, "text", None) + assert isinstance(text, str), f"{tool}: result is not text" + 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) + + async def ok( + self, tool: str, /, *, status: int | tuple[int, ...] = (200, 201, 204), **arguments: Any + ) -> ToolResult: + """Call ``tool`` and assert it succeeded with one of the expected statuses.""" + result = await self.call(tool, **arguments) + allowed = (status,) if isinstance(status, int) else status + assert not result.is_error and result.status in allowed, f"{tool} failed: {_brief(result)}" + return result + + async def fail(self, tool: str, /, status: int, **arguments: Any) -> ToolResult: + """Call ``tool`` and assert Gitea refused it with ``status``.""" + result = await self.call(tool, **arguments) + assert result.is_error and result.status == status, ( + f"{tool} expected HTTP {status}: {_brief(result)}" + ) + return result + + def audit_entries(self) -> list[dict[str, Any]]: + """Payloads of the ``tool_invocation`` entries written to the audit log so far.""" + lines = self.audit_path.read_text(encoding="utf-8").splitlines() + entries = [json.loads(line) for line in lines if line.strip()] + return [e["payload"] for e in entries if e.get("event_type") == "tool_invocation"] + + def audit_text(self) -> str: + """The whole audit log file as text.""" + return self.audit_path.read_text(encoding="utf-8") + + +def _brief(result: ToolResult) -> str: + """A short, secret-free description of a result for assertion messages.""" + error = result.payload.get("error") + detail = error if isinstance(error, dict) else {"status": result.status} + return json.dumps(detail, ensure_ascii=False)[:400] + + +def integration_settings(gitea: GiteaInstance, data_dir: Path, **overrides: Any) -> Settings: + """Server settings for the live instance: admin token, state under ``data_dir``.""" + values: dict[str, Any] = { + "gitea_url": gitea.base_url, + "gitea_token": gitea.admin_token, + "data_dir": data_dir, + "audit_log_path": data_dir / "audit.log", + "request_timeout_seconds": 60, + } + values.update(overrides) + return Settings(**values) + + +class BackgroundContext: + """Enter an async context manager in its own task and leave it from that same task. + + The MCP client uses anyio task groups, which must be exited by the task that entered + them. Async pytest fixtures set up and tear down in different tasks, so the context + lives in a helper task instead. + """ + + def __init__(self, manager: Any) -> None: + self._manager = manager + self._ready = asyncio.Event() + self._stop = asyncio.Event() + self._value: Any = None + self._error: BaseException | None = None + self._task: asyncio.Task[None] | None = None + + async def _run(self) -> None: + try: + async with self._manager as value: + self._value = value + self._ready.set() + await self._stop.wait() + except BaseException as exc: + self._error = exc + raise + finally: + self._ready.set() + + async def open(self) -> Any: + """Enter the context and return its value (raises what entering raised).""" + self._task = asyncio.create_task(self._run()) + await self._ready.wait() + if self._error is not None: + raise self._error + return self._value + + async def close(self) -> None: + """Leave the context from the task that entered it.""" + self._stop.set() + if self._task is not None: + await self._task + + +@pytest.fixture +async def mcp( + gitea: GiteaInstance, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> AsyncIterator[McpSession]: + """An in-memory MCP client on a server running as the admin of the live instance.""" + monkeypatch.setenv("GITEA_URL", gitea.base_url) + settings = integration_settings(gitea, tmp_path / "mcp-state") + audit = AuditLogger(settings.resolved_audit_log_path) + server = build_server(settings, audit=audit) + holder = BackgroundContext(Client(server)) + try: + client = await holder.open() + yield McpSession(client, settings.resolved_audit_log_path) + finally: + await holder.close() + audit.close() + + +# --------------------------------------------------------------------------- +# Scenario helpers +# --------------------------------------------------------------------------- + + +def b64(text: str | bytes) -> str: + """Base64 text of ``text`` (the contents API and file arguments take base64).""" + raw = text.encode("utf-8") if isinstance(text, str) else text + return base64.b64encode(raw).decode("ascii") + + +@dataclass(frozen=True) +class RepoRef: + """A repository created for one test.""" + + owner: str + name: str + + @property + def args(self) -> dict[str, str]: + """The ``owner`` and ``repo`` tool arguments.""" + return {"owner": self.owner, "repo": self.name} + + @property + def full(self) -> str: + """``owner/name``.""" + return f"{self.owner}/{self.name}" + + +@pytest.fixture +def fresh_repo(gitea: GiteaInstance, unique_name: str) -> Iterator[RepoRef]: + """A public repository with a README on ``main`` owned by the admin, deleted afterwards.""" + gitea.create_repo(gitea.admin_login, unique_name) + try: + yield RepoRef(gitea.admin_login, unique_name) + finally: + gitea.api("DELETE", f"/repos/{gitea.admin_login}/{unique_name}") + + +def drop_org(gitea: GiteaInstance, org: str) -> None: + """Best effort: delete an organization together with its repositories.""" + listing = gitea.api("GET", f"/orgs/{org}/repos", params={"limit": 50}) + if listing.status_code == 200: + for repo in listing.json(): + gitea.api("DELETE", f"/repos/{org}/{repo['name']}") + gitea.api("DELETE", f"/orgs/{org}") + + +def drop_user(gitea: GiteaInstance, login: str) -> None: + """Best effort: purge a user created by a test.""" + gitea.api("DELETE", f"/admin/users/{login}", params={"purge": "true"}) diff --git a/tests/integration/harness.py b/tests/integration/harness.py index 1ef14e7..f354571 100644 --- a/tests/integration/harness.py +++ b/tests/integration/harness.py @@ -34,6 +34,13 @@ import httpx GITEA_IMAGE = ( "gitea/gitea:28.1.0@sha256:f505a0d3a41323b786e05942fc7a4075e07ae7c169fd9485649aa55ee543a21a" ) +# Actions runner image (digest-pinned, tag 0.6.1 at the time of writing). Jobs run in +# host mode inside this very container, so no Docker socket and no bind mounts are needed. +ACT_RUNNER_IMAGE = ( + "gitea/act_runner:0.6.1@sha256:b5c35d6bdbb9bb25e531230bfc7cc663cb751406cbec90a2a891b85fea54de86" +) +RUNNER_LABEL = "aegis-host" +RUNNER_REGISTRATION_ENV = "GITEA_RUNNER_REGISTRATION_TOKEN" NAME_PREFIX = "aegis-it-" GITEA_PORT = 3000 ADMIN_LOGIN = "aegis-admin" @@ -176,6 +183,7 @@ class RunResources: container: str network: str attached_container: str | None = None # job container joined in CI mode + extra_containers: list[str] = field(default_factory=list) # e.g. the act_runner cleaned: bool = False @@ -194,6 +202,8 @@ def cleanup_resources(res: RunResources) -> None: _ACTIVE.pop(res.run_id, None) if res.attached_container: run_docker("network", "disconnect", "-f", res.network, res.attached_container, check=False) + for extra in res.extra_containers: + run_docker("rm", "-f", "-v", extra, check=False) run_docker("rm", "-f", "-v", res.container, check=False) run_docker("network", "rm", res.network, check=False) @@ -311,11 +321,27 @@ class GiteaInstance: ``token`` defaults to the admin token; pass ``anonymous=True`` for an unauthenticated call, or ``auth=(login, password)`` for basic auth. """ + return self.request( + method, f"/api/v1/{path.lstrip('/')}", token, anonymous=anonymous, **kwargs + ) + + def request( + self, + method: str, + path: str, + token: str | None = None, + *, + anonymous: bool = False, + **kwargs: Any, + ) -> httpx.Response: + """Call any absolute ``path`` of the instance (for example ``/api/packages/...``). + + Authentication works like :meth:`api`. + """ headers: dict[str, str] = dict(kwargs.pop("headers", None) or {}) if not anonymous and "auth" not in kwargs: headers["Authorization"] = f"token {token or self.admin_token}" - url = f"{self.base_url}/api/v1/{path.lstrip('/')}" - return self._client.request(method, url, headers=headers, **kwargs) + return self._client.request(method, f"{self.base_url}{path}", headers=headers, **kwargs) def unique(self, prefix: str) -> str: """Return ``-<8 hex>``, valid as a Gitea user, org or repo name.""" @@ -529,3 +555,48 @@ def stop_gitea(instance: GiteaInstance) -> None: res = _ACTIVE.get(instance.run_id) if res is not None: cleanup_resources(res) + + +def start_runner(instance: GiteaInstance, registration_token: str) -> str: + """Start an ``act_runner`` container registered with ``registration_token``. + + The runner joins the run's private network, reaches Gitea by its container name + (so it works in local and in container network mode) and runs jobs labelled + ``aegis-host`` in host mode inside its own container. The token travels in the + environment of the docker CLI and is forwarded with ``-e NAME`` (never in argv). + The container is registered for cleanup before it is created. Returns its name. + """ + with _LOCK: + res = _ACTIVE.get(instance.run_id) + if res is None: + raise HarnessError("the Gitea instance is not tracked (already stopped?)") + ensure_image(ACT_RUNNER_IMAGE) + name = f"{NAME_PREFIX}{instance.run_id}-runner" + res.extra_containers.append(name) + run_docker( + "run", + "-d", + "--name", + name, + "--network", + instance.network, + "--label", + "aegis-it=1", + "-e", + RUNNER_REGISTRATION_ENV, + "-e", + f"GITEA_INSTANCE_URL=http://{instance.container}:{GITEA_PORT}", + "-e", + f"GITEA_RUNNER_NAME={name}", + "-e", + f"GITEA_RUNNER_LABELS={RUNNER_LABEL}:host", + ACT_RUNNER_IMAGE, + env={RUNNER_REGISTRATION_ENV: registration_token}, + timeout=300, + ) + return name + + +def runner_logs(name: str, lines: int = 60) -> str: + """Return the last log lines of a runner container (for failure diagnostics).""" + return _container_logs_tail(name, lines) diff --git a/tests/integration/test_scenario_org_repo.py b/tests/integration/test_scenario_org_repo.py new file mode 100644 index 0000000..723e8f7 --- /dev/null +++ b/tests/integration/test_scenario_org_repo.py @@ -0,0 +1,231 @@ +"""Scenarios: organizations, teams and repositories, driven through the MCP server.""" + +from __future__ import annotations + +import base64 +import time +from typing import Any + +from tests.integration.conftest import McpSession, RegularUser, RepoRef, b64, drop_org +from tests.integration.harness import GiteaInstance + + +def names(items: Any, key: str = "name") -> list[str]: + """The ``key`` member of every dict in a list result.""" + return [item[key] for item in items] + + +async def test_org_and_team_lifecycle( + mcp: McpSession, gitea: GiteaInstance, regular_user: RegularUser +) -> None: + org = gitea.unique("it-org") + repo = gitea.unique("team-repo") + try: + created = await mcp.ok( + "gitea_org_create", status=201, body={"username": org, "visibility": "public"} + ) + assert created.data["name"] == org + + fetched = await mcp.ok("gitea_org_get", org=org) + assert fetched.data["name"] == org + + team = await mcp.ok( + "gitea_team_create_team", + status=201, + org=org, + body={ + "name": "devs", + "permission": "write", + "includes_all_repositories": False, + "units": ["repo.code", "repo.issues", "repo.pulls"], + }, + ) + team_id = team.data["id"] + assert team.data["name"] == "devs" + + listed = await mcp.ok("gitea_team_list_teams", org=org) + assert {"devs", "Owners"} <= set(names(listed.data)) + found = await mcp.ok("gitea_team_search", org=org, q="dev") + assert "devs" in names(found.data["data"]) + + await mcp.ok("gitea_team_add_team_member", id=team_id, username=regular_user.login) + members = await mcp.ok("gitea_team_list_team_members", id=team_id) + assert regular_user.login in names(members.data, "login") + await mcp.ok("gitea_team_list_team_member", id=team_id, username=regular_user.login) + assert regular_user.login in names( + (await mcp.ok("gitea_org_list_members", org=org)).data, "login" + ) + + await mcp.ok("gitea_org_create_org_repo", status=201, org=org, body={"name": repo}) + await mcp.ok("gitea_team_add_team_repository", id=team_id, org=org, repo=repo) + repos = await mcp.ok("gitea_team_list_team_repos", id=team_id) + assert repo in names(repos.data) + await mcp.ok("gitea_team_list_team_repo", id=team_id, org=org, repo=repo) + assert (await mcp.ok("gitea_org_list_repos", org=org)).data + + edited = await mcp.ok( + "gitea_team_edit_team", id=team_id, body={"name": "devs", "description": "the devs"} + ) + assert edited.data["description"] == "the devs" + + await mcp.ok("gitea_team_remove_team_repository", id=team_id, org=org, repo=repo) + await mcp.fail("gitea_team_list_team_repo", 404, id=team_id, org=org, repo=repo) + await mcp.ok("gitea_team_remove_team_member", id=team_id, username=regular_user.login) + await mcp.fail("gitea_team_list_team_member", 404, id=team_id, username=regular_user.login) + + await mcp.ok("gitea_team_delete_team", id=team_id) + await mcp.fail("gitea_team_get_team", 404, id=team_id) + + await mcp.ok("gitea_repo_delete", owner=org, repo=repo) + await mcp.ok("gitea_org_delete", org=org) + await mcp.fail("gitea_org_get", 404, org=org) + finally: + drop_org(gitea, org) + + +async def test_repo_files_branches_tags_and_topics( + mcp: McpSession, gitea: GiteaInstance, fresh_repo: RepoRef +) -> None: + repo = fresh_repo.args + info = await mcp.ok("gitea_repo_get", **repo) + assert info.data["full_name"] == fresh_repo.full + assert info.data["default_branch"] == "main" + + # Files through the contents API: create, read, update, delete. + path = "docs/hello.txt" + created = await mcp.ok( + "gitea_repo_create_file", + status=201, + filepath=path, + body={"content": b64("hello"), "message": "add hello", "branch": "main"}, + **repo, + ) + assert created.data["content"]["path"] == path + got = await mcp.ok("gitea_repo_get_contents", filepath=path, **repo) + assert base64.b64decode(got.data["content"]) == b"hello" + sha = got.data["sha"] + updated = await mcp.ok( + "gitea_repo_update_file", + filepath=path, + body={"content": b64("hello again"), "sha": sha, "message": "update hello"}, + **repo, + ) + new_sha = updated.data["content"]["sha"] + assert new_sha != sha + raw = await mcp.ok("gitea_repo_get_raw_file", filepath=path, **repo) + body = raw.data + if isinstance(body, dict): # served as a binary type: base64 envelope + body = base64.b64decode(body["content_base64"]).decode("utf-8") + assert body == "hello again" + listing = await mcp.ok("gitea_repo_get_contents_list", **repo) + assert "README.md" in names(listing.data) + deleted = await mcp.ok( + "gitea_repo_delete_file", + filepath=path, + body={"sha": new_sha, "message": "remove hello"}, + **repo, + ) + assert deleted.data["commit"]["message"].startswith("remove hello") + await mcp.fail("gitea_repo_get_contents", 404, filepath=path, **repo) + + # Branches. + await mcp.ok( + "gitea_repo_create_branch", + status=201, + body={"new_branch_name": "feature", "old_branch_name": "main"}, + **repo, + ) + assert "feature" in names((await mcp.ok("gitea_repo_list_branches", **repo)).data) + assert (await mcp.ok("gitea_repo_get_branch", branch="feature", **repo)).data["name"] == ( + "feature" + ) + await mcp.ok("gitea_repo_delete_branch", branch="feature", **repo) + await mcp.fail("gitea_repo_get_branch", 404, branch="feature", **repo) + + # Tags. + tag = await mcp.ok( + "gitea_repo_create_tag", + status=201, + body={"tag_name": "v0.1.0", "target": "main", "message": "first"}, + **repo, + ) + assert tag.data["name"] == "v0.1.0" + assert "v0.1.0" in names((await mcp.ok("gitea_repo_list_tags", **repo)).data) + await mcp.ok("gitea_repo_get_tag", tag="v0.1.0", **repo) + await mcp.ok("gitea_repo_delete_tag", tag="v0.1.0", **repo) + await mcp.fail("gitea_repo_get_tag", 404, tag="v0.1.0", **repo) + + # Topics. + await mcp.ok("gitea_repo_update_topics", body={"topics": ["aegis", "integration"]}, **repo) + assert set((await mcp.ok("gitea_repo_list_topics", **repo)).data["topics"]) == { + "aegis", + "integration", + } + await mcp.ok("gitea_repo_add_topic", topic="extra", **repo) + await mcp.ok("gitea_repo_delete_topic", topic="aegis", **repo) + assert set((await mcp.ok("gitea_repo_list_topics", **repo)).data["topics"]) == { + "integration", + "extra", + } + + # Edit and read back. + edited = await mcp.ok("gitea_repo_edit", body={"description": "edited by it"}, **repo) + assert edited.data["description"] == "edited by it" + + +async def test_repo_collaborators_fork_transfer_archive_search( + mcp: McpSession, gitea: GiteaInstance, regular_user: RegularUser, fresh_repo: RepoRef +) -> None: + repo = fresh_repo.args + me = regular_user.login + + # Collaborators. + await mcp.ok( + "gitea_repo_add_collaborator", collaborator=me, body={"permission": "write"}, **repo + ) + await mcp.ok("gitea_repo_check_collaborator", collaborator=me, **repo) + assert me in names((await mcp.ok("gitea_repo_list_collaborators", **repo)).data, "login") + permission = await mcp.ok("gitea_repo_get_repo_permissions", collaborator=me, **repo) + assert permission.data["permission"] == "write" + await mcp.ok("gitea_repo_delete_collaborator", collaborator=me, **repo) + await mcp.fail("gitea_repo_check_collaborator", 404, collaborator=me, **repo) + + # Archive: a binary envelope (base64, content type, size). + archive = await mcp.ok("gitea_repo_get_archive", archive="main.zip", **repo) + blob = base64.b64decode(archive.data["content_base64"]) + assert blob.startswith(b"PK") + assert archive.data["size"] == len(blob) + assert archive.data["content_type"].startswith("application/") + + # Search finds the repository by name. + deadline = time.monotonic() + 15 + found: list[str] = [] + while time.monotonic() < deadline and fresh_repo.name not in found: + search = await mcp.ok("gitea_repo_search", q=fresh_repo.name) + found = names(search.data["data"]) + if fresh_repo.name not in found: + time.sleep(0.5) + assert fresh_repo.name in found + + # Fork as another user (admin sudo), then list forks. + fork = await mcp.ok("gitea_repo_create_fork", status=(201, 202), sudo=me, body={}, **repo) + assert fork.identity["sudo"] == me + assert fork.data["owner"]["login"] == me + forks = await mcp.ok("gitea_repo_list_forks", **repo) + assert me in [item["owner"]["login"] for item in forks.data] + await mcp.ok("gitea_repo_delete", owner=me, repo=fresh_repo.name) + + # Transfer to an organization created for the purpose. + org = gitea.unique("it-xfer") + try: + await mcp.ok("gitea_org_create", status=201, body={"username": org}) + await mcp.ok("gitea_repo_transfer", status=(201, 202), body={"new_owner": org}, **repo) + moved = await mcp.ok("gitea_repo_get", owner=org, repo=fresh_repo.name) + assert moved.data["owner"]["login"] == org + # Gitea redirects the old path to the new owner; redirects are never followed. + redirected = await mcp.fail("gitea_repo_get", 301, **repo) + assert redirected.error_kind == "redirect" + await mcp.ok("gitea_repo_edit", owner=org, repo=fresh_repo.name, body={"archived": True}) + assert (await mcp.ok("gitea_repo_get", owner=org, repo=fresh_repo.name)).data["archived"] + finally: + drop_org(gitea, org)