242 lines
9.8 KiB
Python
242 lines
9.8 KiB
Python
"""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)
|