"""Scenario: a real Actions run on an act_runner container, read back through the tools.""" from __future__ import annotations import asyncio import time from typing import Any from tests.integration.conftest import McpSession, RepoRef, b64, drop_org, runner_unavailable from tests.integration.harness import GiteaInstance, HarnessError, runner_logs, start_runner WORKFLOW_PATH = ".gitea/workflows/it.yml" WORKFLOW = """\ name: it on: [push, workflow_dispatch] jobs: hello: runs-on: aegis-host steps: - name: Say hello run: echo hello """ RUNNER_ONLINE_TIMEOUT = 90.0 RUN_TIMEOUT = 180.0 async def _wait_for(probe: Any, *, timeout: float, interval: float = 2.0) -> Any: """Await ``probe()`` until it returns something truthy; return it (or None on timeout).""" deadline = time.monotonic() + timeout while True: value = await probe() if value or time.monotonic() >= deadline: return value await asyncio.sleep(interval) async def test_workflow_run_on_act_runner( mcp: McpSession, gitea: GiteaInstance, fresh_repo: RepoRef ) -> None: repo = fresh_repo.args token = await mcp.ok("gitea_actions_repo_create_runner_registration_token", **repo) assert isinstance(token.data["token"], str) and token.data["token"] try: runner = start_runner(gitea, token.data["token"]) except HarnessError as exc: runner_unavailable(f"the act_runner container could not be started: {exc}") async def online() -> dict[str, Any] | None: runners = await mcp.ok("gitea_actions_get_repo_runners", **repo) entries = runners.data["runners"] if isinstance(runners.data, dict) else runners.data for entry in entries or []: if entry.get("status") in ("online", "idle", "active"): found: dict[str, Any] = entry return found return None registered = await _wait_for(online, timeout=RUNNER_ONLINE_TIMEOUT) if not registered: runner_unavailable( "the act_runner did not come online within " f"{RUNNER_ONLINE_TIMEOUT:.0f}s; runner log tail: {runner_logs(runner, 15)}" ) assert token.data["token"] not in runner_logs(runner, 200) # Commit the workflow through the contents tool, then dispatch it explicitly. await mcp.ok( "gitea_repo_create_file", status=201, filepath=WORKFLOW_PATH, body={"content": b64(WORKFLOW), "message": "add it workflow", "branch": "main"}, **repo, ) workflows = await mcp.ok("gitea_actions_list_repository_workflows", **repo) assert any(w["path"].endswith("it.yml") for w in workflows.data["workflows"]) await mcp.ok( "gitea_actions_dispatch_workflow", status=(200, 204), workflow_id="it.yml", body={"ref": "main"}, **repo, ) async def finished() -> list[dict[str, Any]] | None: runs = await mcp.ok("gitea_actions_get_workflow_runs", **repo) entries = runs.data["workflow_runs"] if len(entries) >= 2 and all(r["status"] == "completed" for r in entries): done: list[dict[str, Any]] = entries return done return None runs = await _wait_for(finished, timeout=RUN_TIMEOUT) assert runs, ( f"workflow runs did not complete within {RUN_TIMEOUT:.0f}s; " f"runner log tail: {runner_logs(runner, 20)}" ) assert {r["event"] for r in runs} == {"push", "workflow_dispatch"} assert [r["conclusion"] for r in runs] == ["success", "success"] run_id = runs[0]["id"] one = await mcp.ok("gitea_actions_get_workflow_run", run=run_id, **repo) assert one.data["status"] == "completed" jobs = await mcp.ok("gitea_actions_list_workflow_run_jobs", run=run_id, **repo) job = jobs.data["jobs"][0] assert job["conclusion"] == "success" job_id = str(job["id"]) # the job id is a string argument fetched = await mcp.ok("gitea_actions_get_workflow_job", job_id=job_id, **repo) assert fetched.data["id"] == job["id"] logs = await mcp.ok("gitea_actions_download_run_job_logs", job_id=job["id"], **repo) assert isinstance(logs.data, str) assert "hello" in logs.data artifacts = await mcp.ok("gitea_actions_get_artifacts_of_run", run=run_id, **repo) assert artifacts.data["artifacts"] == [] assert (await mcp.ok("gitea_actions_get_artifacts", **repo)).data["total_count"] == 0 tasks = await mcp.ok("gitea_actions_list_action_tasks", **repo) assert tasks.data["total_count"] >= 2 # The runner registered above, managed through the repository-scope runner tools. runner_id = registered["id"] got = await mcp.ok("gitea_actions_get_repo_runner", runner_id=str(runner_id), **repo) assert got.data["id"] == runner_id and got.data["name"] == registered["name"] off = await mcp.ok( "gitea_actions_update_repo_runner", runner_id=str(runner_id), body={"disabled": True}, **repo, ) assert off.data["disabled"] is True on = await mcp.ok( "gitea_actions_update_repo_runner", runner_id=str(runner_id), body={"disabled": False}, **repo, ) assert on.data["disabled"] is False # The completed run: logs archive, rerun of the run and of one job, then delete the run. archive = await mcp.ok("gitea_actions_get_workflow_run_logs", run=run_id, **repo) assert archive.data await mcp.ok("gitea_actions_rerun_workflow_run", status=201, run=run_id, **repo) await _wait_for_attempt(mcp, repo, run_id, attempt=2) rerun_jobs = await mcp.ok("gitea_actions_list_workflow_run_jobs", run=run_id, **repo) await mcp.ok( "gitea_actions_rerun_workflow_job", status=201, run=run_id, job_id=rerun_jobs.data["jobs"][0]["id"], **repo, ) await _wait_for_attempt(mcp, repo, run_id, attempt=3) await mcp.ok("gitea_actions_delete_action_run", run=run_id, **repo) await mcp.fail("gitea_actions_get_workflow_run", 404, run=run_id, **repo) await mcp.ok("gitea_actions_delete_repo_runner", runner_id=str(runner_id), **repo) await mcp.fail("gitea_actions_get_repo_runner", 404, runner_id=str(runner_id), **repo) async def _wait_for_attempt( mcp: McpSession, repo: dict[str, str], run_id: int, *, attempt: int ) -> None: """Wait until ``run_id`` reaches ``attempt`` and is completed again.""" async def done() -> bool: run = (await mcp.ok("gitea_actions_get_workflow_run", run=run_id, **repo)).data return bool(run["status"] == "completed" and run.get("run_attempt", 1) >= attempt) assert await _wait_for(done, timeout=RUN_TIMEOUT), ( f"run {run_id} did not finish attempt {attempt}" ) async def test_runner_management_at_every_scope(mcp: McpSession, gitea: GiteaInstance) -> None: """Register one runner per scope (org, user, admin) and manage it through the tools.""" org = gitea.unique("it-runner") await mcp.ok("gitea_org_create", status=201, body={"username": org}) try: scopes: list[tuple[str, str, dict[str, str], str]] = [ ("org", "gitea_actions_org_create_runner_registration_token", {"org": org}, "org"), ("user", "gitea_actions_user_create_runner_registration_token", {}, "user"), ("admin", "gitea_admin_create_runner_registration_token", {}, "admin"), ] for scope, token_tool, args, _ in scopes: token = await mcp.ok(token_tool, **args) suffix = f"{scope}-runner" try: container = start_runner( gitea, token.data["token"], suffix=suffix, label=f"aegis-{scope}" ) except HarnessError as exc: runner_unavailable(f"the {scope} runner could not be started: {exc}") return listing = ( "gitea_admin_get_runners" if scope == "admin" else (f"gitea_actions_get_{scope}_runners") ) getter = ( "gitea_admin_get_runner" if scope == "admin" else (f"gitea_actions_get_{scope}_runner") ) updater = ( "gitea_admin_update_runner" if scope == "admin" else (f"gitea_actions_update_{scope}_runner") ) deleter = ( "gitea_admin_delete_runner" if scope == "admin" else (f"gitea_actions_delete_{scope}_runner") ) async def registered( listing: str = listing, args: dict[str, str] = args, container: str = container ) -> dict[str, Any] | None: runners = await mcp.ok(listing, **args) entries = ( runners.data["runners"] if isinstance(runners.data, dict) else runners.data ) for entry in entries or []: if entry["name"] == container: found: dict[str, Any] = entry return found return None runner = await _wait_for(registered, timeout=RUNNER_ONLINE_TIMEOUT) if not runner: runner_unavailable( f"the {scope} runner did not register; log tail: {runner_logs(container, 15)}" ) return assert token.data["token"] not in runner_logs(container, 200) got = await mcp.ok(getter, runner_id=str(runner["id"]), **args) assert got.data["name"] == container changed = await mcp.ok( updater, runner_id=str(runner["id"]), body={"disabled": True}, **args ) assert changed.data["disabled"] is True await mcp.ok(deleter, runner_id=str(runner["id"]), **args) await mcp.fail(getter, 404, runner_id=str(runner["id"]), **args) finally: drop_org(gitea, org)