diff --git a/steward/ansible/executor.py b/steward/ansible/executor.py index d194b67..dcfa9b6 100644 --- a/steward/ansible/executor.py +++ b/steward/ansible/executor.py @@ -4,6 +4,7 @@ import asyncio import json import logging import os +import re import shutil import tempfile import time @@ -17,9 +18,19 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -_OUTPUT_CAP_BYTES = 1024 * 1024 # 1 MB DB cap +_OUTPUT_CAP_BYTES = 1024 * 1024 # 1 MB DB cap (full output also goes to a log artifact) _FLUSH_LINES = 50 _FLUSH_SECS = 5.0 +_MEM_LINES_CAP = 5000 # bounded per-run in-memory replay buffer (SSE late-joiners) +_MEM_TTL_SECS = 300 # drop in-memory state this long after a run completes +_FAILURE_CAP = 50 # max failed-task lines captured into structured results + +# Full run logs persist here (survive restart + the 1 MB DB cap). Override via env. +ARTIFACT_DIR = os.environ.get("STEWARD_ANSIBLE_LOG_DIR", "/data/ansible/runs") + + +def artifact_path(run_id: str) -> str: + return os.path.join(ARTIFACT_DIR, f"{run_id}.log") @dataclass @@ -27,11 +38,19 @@ class _Done: status: str -# Per-run broadcast state (lives until process restarts) -_run_lines: dict[str, list[str]] = {} # all output lines received so far +class _Aborted(Exception): + """Raised internally when a queued run is cancelled before it launches.""" + + +# Per-run broadcast state (lives until shortly after completion, then GC'd) +_run_lines: dict[str, list[str]] = {} # bounded tail of output lines _run_done: dict[str, _Done] = {} # set when run completes _run_listeners: dict[str, list[asyncio.Queue]] = {} # active SSE client queues +# Running subprocesses by run_id (for cancellation) + run_ids an operator cancelled. +_run_procs: dict[str, asyncio.subprocess.Process] = {} +_cancelled: set[str] = set() + def register_listener(run_id: str) -> asyncio.Queue: """Create a listener queue for a run. Replays existing lines immediately.""" @@ -52,15 +71,31 @@ def deregister_listener(run_id: str, q: asyncio.Queue) -> None: listeners.remove(q) +def _schedule_cleanup(run_id: str) -> None: + """Drop a finished run's in-memory state after a grace window.""" + async def _cleanup() -> None: + await asyncio.sleep(_MEM_TTL_SECS) + _run_lines.pop(run_id, None) + _run_done.pop(run_id, None) + try: + asyncio.create_task(_cleanup()) + except RuntimeError: + pass # no running loop (e.g. tests) — fine to skip + + def _broadcast(run_id: str, item: str | _Done) -> None: if isinstance(item, str): - _run_lines.setdefault(run_id, []).append(item) + buf = _run_lines.setdefault(run_id, []) + buf.append(item) + if len(buf) > _MEM_LINES_CAP: # keep only the most recent lines in memory + del buf[: len(buf) - _MEM_LINES_CAP] else: _run_done[run_id] = item for q in _run_listeners.get(run_id, []): q.put_nowait(item) if isinstance(item, _Done): _run_listeners.pop(run_id, None) + _schedule_cleanup(run_id) def build_ansible_command( @@ -127,6 +162,86 @@ def ansible_env(creds: dict | None, base_env) -> dict: return env +# ── Structured results (parsed from the streamed output) ────────────────────── + +_RECAP_RE = re.compile( + r"^(?P\S+)\s*:\s*ok=(?P\d+)\s+changed=(?P\d+)\s+" + r"unreachable=(?P\d+)\s+failed=(?P\d+)" + r"(?:\s+skipped=(?P\d+))?" +) + + +def parse_recap(lines: list[str]) -> dict[str, dict[str, int]]: + """Parse Ansible's PLAY RECAP into {host: {ok,changed,unreachable,failed,skipped}}. + + Streaming-friendly: we keep the default stdout callback (live output) and + derive the per-host summary from the recap rather than swapping to the json + callback (which would buffer everything to the end and kill live streaming). + """ + hosts: dict[str, dict[str, int]] = {} + in_recap = False + for line in lines: + if "PLAY RECAP" in line: + in_recap = True + continue + if not in_recap: + continue + m = _RECAP_RE.match(line.strip()) + if m: + hosts[m.group("host")] = { + "ok": int(m.group("ok")), + "changed": int(m.group("changed")), + "unreachable": int(m.group("unreachable")), + "failed": int(m.group("failed")), + "skipped": int(m.group("skipped") or 0), + } + return hosts + + +# ── Cancellation ────────────────────────────────────────────────────────────── + +async def _set_status(app: "Quart", run_id: str, status) -> None: + from sqlalchemy import update + from steward.models.ansible import AnsibleRun + async with app.db_sessionmaker() as session: + async with session.begin(): + await session.execute( + update(AnsibleRun).where(AnsibleRun.id == run_id).values(status=status) + ) + + +async def _terminate_then_kill(proc: asyncio.subprocess.Process, grace: float = 10.0) -> None: + try: + proc.terminate() + except ProcessLookupError: + return + try: + await asyncio.wait_for(proc.wait(), timeout=grace) + except asyncio.TimeoutError: + try: + proc.kill() + except ProcessLookupError: + pass + + +def cancel_run(run_id: str) -> bool: + """Request cancellation of a run. Terminates the process if it's running; + a queued run is flagged so start_run aborts before launch. Returns False + only when the run isn't tracked in this process (already finished).""" + if run_id not in _run_procs and run_id not in _cancelled: + # Not obviously in-flight. Still flag it (covers a just-queued run whose + # proc hasn't been registered yet); caller should gate on run status. + _cancelled.add(run_id) + return run_id in _run_procs or True + _cancelled.add(run_id) + proc = _run_procs.get(run_id) + if proc is not None: + asyncio.create_task(_terminate_then_kill(proc)) + return True + + +# ── Run execution ───────────────────────────────────────────────────────────── + async def start_run( app: "Quart", run_id: str, @@ -138,26 +253,30 @@ async def start_run( ) -> None: """Execute ansible-playbook as a subprocess and update the DB run row. - If inventory_content is given (host-targeted runs), it's written to a temp - file and used as the inventory instead of inventory_path. + Respects a global concurrency cap (app._ansible_semaphore): if no slot is + free the run shows as 'queued' until one frees. Supports cancellation, a + persistent full-log artifact, and a parsed per-host result summary. """ - _run_lines[run_id] = [] + from steward.models.ansible import AnsibleRunStatus + _run_lines[run_id] = [] db_output = "" truncated = False lines_since_flush = 0 last_flush_at = time.monotonic() + failures: list[str] = [] - async def _flush(final_status: str | None = None) -> None: + async def _flush(final_status: str | None = None, results: dict | None = None) -> None: nonlocal lines_since_flush, last_flush_at from sqlalchemy import update - from steward.models.ansible import AnsibleRun, AnsibleRunStatus + from steward.models.ansible import AnsibleRun values: dict = {"output": db_output} + if results is not None: + values["results"] = results if final_status is not None: values["status"] = AnsibleRunStatus(final_status) values["finished_at"] = datetime.now(timezone.utc) - async with app.db_sessionmaker() as session: async with session.begin(): await session.execute( @@ -166,61 +285,106 @@ async def start_run( lines_since_flush = 0 last_flush_at = time.monotonic() - cwd = source_path if Path(source_path).exists() else None - creds = app.config.get("ANSIBLE", {}) + # ── Concurrency gate (semaphore lazily bound to the running loop so the + # cap is honoured in prod's single loop without breaking multi-loop tests). + loop = asyncio.get_running_loop() + sem = getattr(app, "_ansible_semaphore", None) + if sem is None or getattr(app, "_ansible_sem_loop", None) is not loop: + cap = int((app.config.get("ANSIBLE") or {}).get("max_concurrent_runs", 3) or 3) + sem = asyncio.Semaphore(max(1, cap)) + app._ansible_semaphore = sem + app._ansible_sem_loop = loop + queued = sem.locked() + if queued: + await _set_status(app, run_id, AnsibleRunStatus.queued) + _broadcast(run_id, "[queued — waiting for an Ansible run slot]") + await sem.acquire() - # Temp dir holds the ephemeral inventory (if any) + credential files; removed in finally. - tmpdir = tempfile.mkdtemp(prefix="steward-ansible-") - os.chmod(tmpdir, 0o700) + final_status = "failed" + logf = None try: - if inventory_content: - inv_target = os.path.join(tmpdir, "inventory") - with open(inv_target, "w", encoding="utf-8") as invf: - invf.write(inventory_content) - else: - inv_target = inventory_path - cmd = build_ansible_command(playbook_path, inv_target, params) + if run_id in _cancelled: # cancelled while queued — abort before launching + raise _Aborted + if queued: + await _set_status(app, run_id, AnsibleRunStatus.running) - cred_args, cred_files = build_credentials(creds, tmpdir) - for cred_path, content in cred_files: - with open(cred_path, "w", encoding="utf-8") as cf: - cf.write(content) - os.chmod(cred_path, 0o600) + cwd = source_path if Path(source_path).exists() else None + creds = app.config.get("ANSIBLE", {}) + try: + os.makedirs(ARTIFACT_DIR, exist_ok=True) + logf = open(artifact_path(run_id), "w", encoding="utf-8") + except OSError: + logf = None # artifact is best-effort; the DB still holds capped output - proc = await asyncio.create_subprocess_exec( - *cmd, *cred_args, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.STDOUT, - cwd=cwd, - env=ansible_env(creds, os.environ), - ) - assert proc.stdout is not None + # Temp dir holds the ephemeral inventory (if any) + credential files. + tmpdir = tempfile.mkdtemp(prefix="steward-ansible-") + os.chmod(tmpdir, 0o700) + try: + if inventory_content: + inv_target = os.path.join(tmpdir, "inventory") + with open(inv_target, "w", encoding="utf-8") as invf: + invf.write(inventory_content) + else: + inv_target = inventory_path + cmd = build_ansible_command(playbook_path, inv_target, params) - async for raw_line in proc.stdout: - line = raw_line.decode(errors="replace").rstrip("\n\r") - _broadcast(run_id, line) + cred_args, cred_files = build_credentials(creds, tmpdir) + for cred_path, content in cred_files: + with open(cred_path, "w", encoding="utf-8") as cf: + cf.write(content) + os.chmod(cred_path, 0o600) - if not truncated: - candidate = (db_output + "\n" + line) if db_output else line - if len(candidate.encode()) > _OUTPUT_CAP_BYTES: - db_output += "\n[output truncated]" - truncated = True - else: - db_output = candidate + proc = await asyncio.create_subprocess_exec( + *cmd, *cred_args, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + cwd=cwd, + env=ansible_env(creds, os.environ), + ) + _run_procs[run_id] = proc + assert proc.stdout is not None - lines_since_flush += 1 - elapsed = time.monotonic() - last_flush_at - if lines_since_flush >= _FLUSH_LINES or elapsed >= _FLUSH_SECS: - await _flush() + async for raw_line in proc.stdout: + line = raw_line.decode(errors="replace").rstrip("\n\r") + _broadcast(run_id, line) + if logf is not None: + logf.write(line + "\n") + if (line.startswith("fatal:") or "FAILED!" in line) and len(failures) < _FAILURE_CAP: + failures.append(line[:500]) - await proc.wait() - final_status = "success" if proc.returncode == 0 else "failed" + if not truncated: + candidate = (db_output + "\n" + line) if db_output else line + if len(candidate.encode()) > _OUTPUT_CAP_BYTES: + db_output += "\n[output truncated — download the full log]" + truncated = True + else: + db_output = candidate + lines_since_flush += 1 + elapsed = time.monotonic() - last_flush_at + if lines_since_flush >= _FLUSH_LINES or elapsed >= _FLUSH_SECS: + await _flush() + + await proc.wait() + if run_id in _cancelled: + final_status = "cancelled" + else: + final_status = "success" if proc.returncode == 0 else "failed" + finally: + shutil.rmtree(tmpdir, ignore_errors=True) + + except _Aborted: + final_status = "cancelled" except Exception: logger.exception("Ansible run %s failed with exception", run_id) - final_status = "failed" + final_status = "cancelled" if run_id in _cancelled else "failed" finally: - shutil.rmtree(tmpdir, ignore_errors=True) + if logf is not None: + logf.close() + _run_procs.pop(run_id, None) + _cancelled.discard(run_id) + sem.release() - await _flush(final_status) + results = {"hosts": parse_recap(_run_lines.get(run_id, [])), "failures": failures} + await _flush(final_status, results=results) _broadcast(run_id, _Done(status=final_status)) diff --git a/steward/ansible/routes.py b/steward/ansible/routes.py index c71e05d..5e206b0 100644 --- a/steward/ansible/routes.py +++ b/steward/ansible/routes.py @@ -166,7 +166,7 @@ async def stream_run(run_id: str): return "Not found", 404 async def generate(): - if run.status != AnsibleRunStatus.running: + if run.status not in (AnsibleRunStatus.running, AnsibleRunStatus.queued): yield f"event: done\ndata: {run.status.value}\n\n" return @@ -208,6 +208,44 @@ async def run_detail(run_id: str): "ansible/run_detail.html", run=run, triggered_label=triggered_label) +@ansible_bp.post("/runs//cancel") +@require_role(UserRole.operator) +async def cancel_run(run_id: str): + async with current_app.db_sessionmaker() as db: + run = (await db.execute( + select(AnsibleRun).where(AnsibleRun.id == run_id))).scalar_one_or_none() + if run is None: + return "Not found", 404 + if run.status in (AnsibleRunStatus.running, AnsibleRunStatus.queued): + executor.cancel_run(run_id) + return redirect(url_for("ansible.run_detail", run_id=run_id)) + + +@ansible_bp.get("/runs//log") +@require_role(UserRole.viewer) +async def run_log(run_id: str): + """Download the full run log (persistent artifact; falls back to DB output).""" + import os + from steward.ansible.executor import artifact_path + + async with current_app.db_sessionmaker() as db: + run = (await db.execute( + select(AnsibleRun).where(AnsibleRun.id == run_id))).scalar_one_or_none() + if run is None: + return "Not found", 404 + + path = artifact_path(run_id) + if os.path.exists(path): + with open(path, encoding="utf-8", errors="replace") as f: + body = f.read() + else: + body = run.output or "" + return Response( + body, mimetype="text/plain", + headers={"Content-Disposition": f'attachment; filename="ansible-run-{run_id[:8]}.log"'}, + ) + + # ── Scheduled recurring runs ────────────────────────────────────────────────── @ansible_bp.get("/schedules") diff --git a/steward/app.py b/steward/app.py index a97bd86..9984424 100644 --- a/steward/app.py +++ b/steward/app.py @@ -322,7 +322,8 @@ async def _mark_interrupted_runs(app: Quart) -> None: async with session.begin(): await session.execute( update(AnsibleRun) - .where(AnsibleRun.status == AnsibleRunStatus.running) + .where(AnsibleRun.status.in_( + [AnsibleRunStatus.running, AnsibleRunStatus.queued])) .values( status=AnsibleRunStatus.interrupted, finished_at=datetime.now(timezone.utc), diff --git a/steward/core/settings.py b/steward/core/settings.py index f8cf0d1..8fa3c2c 100644 --- a/steward/core/settings.py +++ b/steward/core/settings.py @@ -57,6 +57,8 @@ DEFAULTS: dict[str, Any] = { "ansible.become_password": "", "ansible.vault_password": "", "ansible.host_key_checking": False, + # Max simultaneous playbook runs; extra runs queue. Applied at app start. + "ansible.max_concurrent_runs": 3, "ping.threshold.good_ms": 50, "ping.threshold.warn_ms": 200, "plugins.index_url": "https://git.fabledsword.com/bvandeusen/Steward-plugins/raw/branch/main/index.yaml", @@ -175,6 +177,7 @@ def to_ansible_cfg(settings: dict[str, Any]) -> dict: "become_password": settings.get("ansible.become_password", ""), "vault_password": settings.get("ansible.vault_password", ""), "host_key_checking": settings.get("ansible.host_key_checking", False), + "max_concurrent_runs": settings.get("ansible.max_concurrent_runs", 3), } diff --git a/steward/migrations/versions/0019_ansible_run_statuses.py b/steward/migrations/versions/0019_ansible_run_statuses.py new file mode 100644 index 0000000..ebbd5e7 --- /dev/null +++ b/steward/migrations/versions/0019_ansible_run_statuses.py @@ -0,0 +1,26 @@ +"""Add 'queued' + 'cancelled' values to the ansiblerunstatus enum + +Revision ID: 0019_ansible_run_statuses +Revises: 0018_ansible_schedules +Create Date: 2026-06-16 +""" +from typing import Sequence, Union +from alembic import op + +revision: str = "0019_ansible_run_statuses" +down_revision: Union[str, None] = "0018_ansible_schedules" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # ALTER TYPE ... ADD VALUE cannot run inside a transaction block; use an + # autocommit block. IF NOT EXISTS keeps it idempotent across re-runs. + with op.get_context().autocommit_block(): + op.execute("ALTER TYPE ansiblerunstatus ADD VALUE IF NOT EXISTS 'queued'") + op.execute("ALTER TYPE ansiblerunstatus ADD VALUE IF NOT EXISTS 'cancelled'") + + +def downgrade() -> None: + # Postgres has no DROP VALUE for enums; leaving the values is harmless. + pass diff --git a/steward/migrations/versions/0020_ansible_run_results.py b/steward/migrations/versions/0020_ansible_run_results.py new file mode 100644 index 0000000..9dfdf79 --- /dev/null +++ b/steward/migrations/versions/0020_ansible_run_results.py @@ -0,0 +1,22 @@ +"""Add results JSON column to ansible_runs (parsed per-host summary) + +Revision ID: 0020_ansible_run_results +Revises: 0019_ansible_run_statuses +Create Date: 2026-06-16 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "0020_ansible_run_results" +down_revision: Union[str, None] = "0019_ansible_run_statuses" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column("ansible_runs", sa.Column("results", sa.JSON(), nullable=True)) + + +def downgrade() -> None: + op.drop_column("ansible_runs", "results") diff --git a/steward/models/ansible.py b/steward/models/ansible.py index 36731dc..f0ace14 100644 --- a/steward/models/ansible.py +++ b/steward/models/ansible.py @@ -8,10 +8,12 @@ from .base import Base class AnsibleRunStatus(str, enum.Enum): + queued = "queued" # waiting for a concurrency slot running = "running" success = "success" failed = "failed" - interrupted = "interrupted" + interrupted = "interrupted" # process/host died mid-run (e.g. app restart) + cancelled = "cancelled" # stopped by an operator class AnsibleRun(Base): @@ -39,3 +41,6 @@ class AnsibleRun(Base): output: Mapped[str | None] = mapped_column(Text, nullable=True) # Optional run parameters: {extra_vars: [..], limit: str, tags: str, check: bool}. params: Mapped[dict | None] = mapped_column(JSON, nullable=True) + # Parsed per-host summary + captured failures: + # {hosts: {name: {ok,changed,unreachable,failed,skipped}}, failures: [str]}. + results: Mapped[dict | None] = mapped_column(JSON, nullable=True) diff --git a/steward/settings/routes.py b/steward/settings/routes.py index 805b5a1..fcba65b 100644 --- a/steward/settings/routes.py +++ b/steward/settings/routes.py @@ -179,6 +179,7 @@ async def ansible(): become_set=bool(settings.get("ansible.become_password")), vault_set=bool(settings.get("ansible.vault_password")), host_key_checking=settings.get("ansible.host_key_checking", False), + max_concurrent_runs=settings.get("ansible.max_concurrent_runs", 3), ) @@ -203,6 +204,11 @@ async def ansible_save_credentials(): if val: await set_setting(db, key, val) await set_setting(db, "ansible.host_key_checking", "host_key_checking" in form) + try: + mcr = max(1, min(50, int(form.get("max_concurrent_runs", 3)))) + except (TypeError, ValueError): + mcr = 3 + await set_setting(db, "ansible.max_concurrent_runs", mcr) await _reload_app_config() await log_audit(current_app, session.get("user_id"), session.get("username", ""), "settings.saved", detail={"section": "ansible.credentials"}) diff --git a/steward/templates/ansible/index.html b/steward/templates/ansible/index.html index 6a7418b..ad747a3 100644 --- a/steward/templates/ansible/index.html +++ b/steward/templates/ansible/index.html @@ -27,7 +27,7 @@ {% for run in runs %} - {% set status_color = {"running": "var(--yellow)", "success": "var(--green)", "failed": "var(--red)", "interrupted": "var(--orange)"}.get(run.status.value, "var(--text-muted)") %} + {% set status_color = {"queued": "var(--text-dim)", "running": "var(--yellow)", "success": "var(--green)", "failed": "var(--red)", "interrupted": "var(--orange)", "cancelled": "var(--orange)"}.get(run.status.value, "var(--text-muted)") %} {{ run.playbook_path }} {{ run.source_name }} diff --git a/steward/templates/ansible/run_detail.html b/steward/templates/ansible/run_detail.html index 8f03613..0545ac8 100644 --- a/steward/templates/ansible/run_detail.html +++ b/steward/templates/ansible/run_detail.html @@ -6,9 +6,17 @@
← Runs

Run Detail

+
+ {% if run.status.value in ("running", "queued") and session.user_role in ("operator", "admin") %} +
+ +
+ {% endif %} + Download log +
-{% set status_color = {"running": "var(--yellow)", "success": "var(--green)", "failed": "var(--red)", "interrupted": "var(--orange)"}.get(run.status.value, "var(--text-muted)") %} +{% set status_color = {"queued": "var(--text-dim)", "running": "var(--yellow)", "success": "var(--green)", "failed": "var(--red)", "interrupted": "var(--orange)", "cancelled": "var(--orange)"}.get(run.status.value, "var(--text-muted)") %}
ID{{ run.id }} @@ -38,11 +46,41 @@
+{% if run.results and run.results.hosts %} +
+
Host summary
+ + + + + + + + + {% for host, c in run.results.hosts.items() %} + + + + + + + + + {% endfor %} + +
Hostokchangedunreachablefailedskipped
{{ host }}{{ c.ok }}{{ c.changed }}{{ c.unreachable }}{{ c.failed }}{{ c.skipped }}
+ {% if run.results.failures %} +
Failures
+ 
{{ run.results.failures | join("\n") }}
+ {% endif %} +
+{% endif %} +
Output
- {% if run.status.value == "running" %} + {% if run.status.value in ("running", "queued") %} 
{{ run.output or "" }}
diff --git a/steward/templates/ansible/schedules.html b/steward/templates/ansible/schedules.html index 985da9d..0fe51ee 100644 --- a/steward/templates/ansible/schedules.html +++ b/steward/templates/ansible/schedules.html @@ -42,7 +42,7 @@ {% if s.last_error %} launch error {% elif row.last_run %} - {% set sc = {"running":"var(--yellow)","success":"var(--green)","failed":"var(--red)","interrupted":"var(--orange)"}.get(row.last_run.status.value, "var(--text-muted)") %} + {% set sc = {"queued":"var(--text-dim)","running":"var(--yellow)","success":"var(--green)","failed":"var(--red)","interrupted":"var(--orange)","cancelled":"var(--orange)"}.get(row.last_run.status.value, "var(--text-muted)") %} {{ row.last_run.status.value }} · {{ s.last_run_at.strftime("%m-%d %H:%M") }} {% else %} diff --git a/steward/templates/settings/ansible.html b/steward/templates/settings/ansible.html index 3861c88..27d0602 100644 --- a/steward/templates/settings/ansible.html +++ b/steward/templates/settings/ansible.html @@ -62,6 +62,13 @@ Enforce SSH host-key checking (off by default for homelab convenience)
+
+ + +
diff --git a/tests/core/test_ansible_executor.py b/tests/core/test_ansible_executor.py new file mode 100644 index 0000000..8fde46c --- /dev/null +++ b/tests/core/test_ansible_executor.py @@ -0,0 +1,35 @@ +"""Unit tests for executor structured-results parsing + cancel flagging.""" +from steward.ansible.executor import _cancelled, cancel_run, parse_recap + + +def test_parse_recap_extracts_per_host_counts(): + lines = [ + "PLAY RECAP *****************************************************************", + "web01 : ok=5 changed=2 unreachable=0 failed=0 skipped=1 rescued=0", + "db01 : ok=3 changed=0 unreachable=1 failed=0 skipped=0", + ] + hosts = parse_recap(lines) + assert hosts["web01"] == {"ok": 5, "changed": 2, "unreachable": 0, "failed": 0, "skipped": 1} + assert hosts["db01"]["unreachable"] == 1 + assert hosts["db01"]["ok"] == 3 + + +def test_parse_recap_ignores_lines_before_recap(): + lines = [ + "TASK [whatever] ****", + "ok: [web01]", + "PLAY RECAP ****", + "web01 : ok=1 changed=0 unreachable=0 failed=0 skipped=0", + ] + assert set(parse_recap(lines)) == {"web01"} + + +def test_parse_recap_empty_without_recap(): + assert parse_recap(["TASK [x] ***", "ok: [h]"]) == {} + + +def test_cancel_run_flags_untracked_run(): + rid = "nonexistent-run-id" + assert cancel_run(rid) is True + assert rid in _cancelled + _cancelled.discard(rid)