bvandeusenandClaude Opus 5.5
72a3261582
- /ansible/playbooks lists every playbook across sources: what it says it does, its category and destructive flag, how it last ran; the whole row opens the playbook. A source with none is one quiet line. - /ansible/playbooks/<source>/<path>: the verdict (what it does, flagged destructive, last ran), the run form in place (dry run beside the button, options folded), the schedules that use it, its recent runs, the playbook itself folded. ?from=<run> fills the form from that run: Run again on a run's page. - /ansible/browse, its view page and the /run-form fragment are gone, not redirected (rule 122); every link points at the new pages. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
221 lines
12 KiB
Python
221 lines
12 KiB
Python
"""Ansible runs are named by what ran where, say how long ago and for how
|
|
long, and wear the one status badge every page uses (#5125)."""
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from quart import render_template, session
|
|
|
|
from steward.ansible.inventory_gen import run_title
|
|
from steward.ansible.origins import describe_origin
|
|
from steward.ansible.recap import host_recap, run_recap
|
|
from steward.ansible.routes import (
|
|
RUN_ORIGINS, RUN_STATUSES, hosts_worst_first, next_scheduled, run_duration, run_filters, run_view,
|
|
)
|
|
from steward.models.ansible import AnsibleRunStatus
|
|
from tests.test_style_tokens import hex_colours
|
|
|
|
|
|
@pytest.mark.parametrize("label, title", [
|
|
("target web-1", "update.yml on web-1"),
|
|
("group Docker Swarm", "update.yml on group Docker Swarm"),
|
|
("all inventory targets", "update.yml on all inventory targets"),
|
|
("homelab/hosts.ini", "update.yml on homelab/hosts.ini"),
|
|
("", "update.yml"),
|
|
])
|
|
def test_a_run_is_named_by_what_ran_where(label, title):
|
|
assert run_title("update.yml", label) == title
|
|
|
|
|
|
def _run(rid, status, started_min_ago, took_s=None, scope="steward:target:t1", check=False, results=None):
|
|
now = datetime.now(timezone.utc)
|
|
started = now - timedelta(minutes=started_min_ago)
|
|
return SimpleNamespace(
|
|
id=rid, playbook_path="update.yml", source_name="builtin", inventory_scope=scope,
|
|
status=status, started_at=started,
|
|
finished_at=started + timedelta(seconds=took_s) if took_s is not None else None,
|
|
params={"check": True} if check else None, results=results, output="")
|
|
|
|
|
|
def test_a_running_run_counts_its_duration_so_far():
|
|
now = datetime.now(timezone.utc)
|
|
run = _run("r", AnsibleRunStatus.running, 3)
|
|
assert run_duration(run, now) == pytest.approx(180, abs=2)
|
|
assert run_duration(_run("f", AnsibleRunStatus.success, 10, took_s=95), now) == 95
|
|
|
|
|
|
def _ctx(runs, verdict=None, f=None, more_url=None):
|
|
return {"runs": runs, "more_url": more_url, "statuses": RUN_STATUSES, "origins": RUN_ORIGINS,
|
|
"f": f or {"q": "", "status": "", "origin": ""}, "filtered": bool(f),
|
|
"verdict": verdict or {"days": 30, "total": 0, "failed": 0, "running": 0,
|
|
"latest_failed": None, "next": None}}
|
|
|
|
|
|
async def test_the_runs_list_names_each_run_with_one_badge(app):
|
|
now = datetime.now(timezone.utc)
|
|
labels = {"steward:target:t1": "target web-1", "steward:all": "all inventory targets"}
|
|
two = {"hosts": {"web-1": {"ok": 3, "changed": 1}, "web-2": {"ok": 3, "changed": 1}}}
|
|
runs = [run_view(r, labels, now) for r in (
|
|
_run("aaaa1111", AnsibleRunStatus.running, 2),
|
|
_run("bbbb2222", AnsibleRunStatus.failed, 60, took_s=125, scope="steward:all", check=True,
|
|
results={"hosts": {"db-1": {"ok": 1, "failed": 1}, "web-1": {"ok": 2}}}),
|
|
_run("cccc3333", AnsibleRunStatus.success, 120, took_s=40, results=two),
|
|
)]
|
|
more = "/ansible/?before=2026-10-07T01%3A00%3A00%2B00%3A00%7Cx"
|
|
async with app.test_request_context("/ansible/"):
|
|
session["user_role"] = "viewer"
|
|
page = await render_template("ansible/index.html", **_ctx(runs, more_url=more))
|
|
|
|
main = page[page.index("<main"):]
|
|
assert '<tr class="row-link">' in main # the whole row opens the run
|
|
assert ('<a class="row-target" href="/ansible/runs/aaaa1111" title="update.yml on web-1">'
|
|
'update.yml on web-1</a>') in main
|
|
assert "steward:target:t1" not in main # no raw scope, not even as a tooltip
|
|
assert ">update.yml on all inventory targets</a>" in main
|
|
assert '<span class="badge badge-info">Running</span>' in main
|
|
assert '<span class="badge badge-crit">Failed</span>' in main
|
|
assert '<span class="badge badge-ok">Succeeded</span>' in main
|
|
# Under each status, what the run did across its hosts (#5329).
|
|
assert '<span class="tone-crit">failed on db-1</span> · 2 hosts' in main
|
|
assert "2 hosts · 2 changed" in main
|
|
# The list continues itself as the reader scrolls (#5211): no cap of 50.
|
|
assert f'data-scroll-more hx-get="{more}"' in main
|
|
assert "2m 5s" in main and "40s" in main and " so far" in main
|
|
assert "dry run" in main and "ago" in main
|
|
assert "RUNNING" not in main and "FAILED" not in main # the old shouted status text is gone
|
|
assert not hex_colours(main), hex_colours(main)
|
|
|
|
|
|
async def test_ansible_pages_share_one_strip_of_tabs(app):
|
|
async with app.test_request_context("/ansible/"):
|
|
session["user_role"] = "viewer"
|
|
page = await render_template("ansible/index.html", **_ctx([]))
|
|
nav = page[page.index('<nav class="section-tabs" aria-label="Ansible">'):]
|
|
nav = nav[:nav.index("</nav>")]
|
|
for href in ("/ansible/", "/ansible/playbooks", "/ansible/schedules", "/ansible/inventory/targets"):
|
|
assert f'href="{href}"' in nav, href
|
|
assert nav.count('aria-current="page"') == 1 and '<a href="/ansible/" aria-current="page">Runs</a>' in nav
|
|
assert "No runs yet." in page and "No runs in the last 30 days" in page
|
|
|
|
|
|
async def test_the_runs_page_opens_on_its_verdict(app):
|
|
now = datetime.now(timezone.utc)
|
|
failed = run_view(_run("ffff0000", AnsibleRunStatus.failed, 90, took_s=5), {"steward:target:t1": "target web-1"}, now)
|
|
calm = {"days": 30, "total": 33, "failed": 0, "running": 0, "latest_failed": None,
|
|
"next": {"name": "Nightly prune", "at": now + timedelta(hours=2, minutes=5)}}
|
|
bad = {"days": 30, "total": 33, "failed": 2, "running": 1, "latest_failed": failed, "next": None}
|
|
async with app.test_request_context("/ansible/"):
|
|
session["user_role"] = "viewer"
|
|
ok = await render_template("ansible/index.html", **_ctx([], calm))
|
|
worse = await render_template("ansible/index.html", **_ctx([], bad, f={"q": "x", "status": "", "origin": ""}))
|
|
assert '<span class="tone-ok">33 runs in the last 30 days, all succeeded</span>' in ok
|
|
assert 'next: <a href="/ansible/schedules">Nightly prune</a> <time' in ok and ">in 2h" in ok
|
|
assert '<span class="tone-crit">2 of 33 runs failed</span> in the last 30 days' in worse
|
|
assert 'latest <a href="/ansible/runs/ffff0000">update.yml on web-1</a>' in worse
|
|
assert '<span class="tone-info">1 running now</span>' in worse
|
|
assert "No runs match these filters." in worse and 'value="x"' in worse
|
|
|
|
|
|
def test_the_run_filters_drop_what_they_do_not_know():
|
|
assert run_filters({"q": " prune ", "status": "failed", "origin": "mcp"}) == \
|
|
{"q": "prune", "status": "failed", "origin": "mcp"}
|
|
assert run_filters({"status": "exploded", "origin": "aliens"}) == {"q": "", "status": "", "origin": ""}
|
|
|
|
|
|
def test_the_next_schedule_is_the_soonest_enabled_one():
|
|
now = datetime.now(timezone.utc)
|
|
|
|
def sched(name, ran_min_ago, every_s, enabled=True):
|
|
return SimpleNamespace(name=name, enabled=enabled, interval_seconds=every_s,
|
|
last_run_at=now - timedelta(minutes=ran_min_ago) if ran_min_ago is not None else None)
|
|
daily, hourly = sched("daily", 60, 86400), sched("hourly", 30, 3600)
|
|
assert next_scheduled([daily, hourly], now)["name"] == "hourly"
|
|
assert next_scheduled([daily, sched("paused", 59, 3600, enabled=False)], now)["name"] == "daily"
|
|
assert next_scheduled([daily, sched("new", None, 3600)], now) == {"name": "new", "at": None} # due now
|
|
assert next_scheduled([], now) is None
|
|
|
|
|
|
def test_a_runs_recap_reads_in_words_worst_first():
|
|
hosts = {"web-1": {"ok": 3, "changed": 1}, "web-2": {"ok": 3, "changed": 0},
|
|
"db-1": {"ok": 1, "failed": 1}, "nas": {"unreachable": 1}}
|
|
assert run_recap({"hosts": hosts}) == [
|
|
("failed on db-1", "crit"), ("nas unreachable", "crit"), ("4 hosts", None), ("1 changed", None)]
|
|
assert run_recap({"hosts": {"a": {"ok": 2}}}) == [("1 host", None), ("nothing changed", "muted")]
|
|
assert run_recap({"hosts": {"a": {"changed": 1}}}, check=True) == [("1 host", None), ("1 would change", None)]
|
|
many = {h: {"failed": 1} for h in ("a", "b", "c", "d")}
|
|
assert run_recap({"hosts": many})[0] == ("failed on a, b and 2 more", "crit")
|
|
assert run_recap(None) == [] and run_recap({"hosts": {}}) == []
|
|
|
|
|
|
def test_a_hosts_recap_leaves_out_zeros():
|
|
assert host_recap({"ok": 3, "changed": 1, "unreachable": 0, "failed": 0, "skipped": 2}) == [
|
|
("3 ok", None), ("1 changed", None), ("2 skipped", "muted")]
|
|
assert host_recap({"ok": 1, "failed": 1, "unreachable": 1}) == [
|
|
("unreachable", "crit"), ("1 failed", "crit"), ("1 ok", None)]
|
|
assert host_recap({"changed": 2}, check=True) == [("2 would change", None)]
|
|
|
|
|
|
async def test_run_detail_titles_the_run_and_keeps_the_trail(app):
|
|
now = datetime.now(timezone.utc)
|
|
run = _run("dddd4444", AnsibleRunStatus.interrupted, 30, took_s=600)
|
|
hosts = hosts_worst_first({"hosts": {
|
|
"alpha": {"ok": 3, "changed": 0, "unreachable": 0, "failed": 0, "skipped": 0},
|
|
"zulu": {"ok": 1, "changed": 0, "unreachable": 0, "failed": 1, "skipped": 0}}})
|
|
async with app.test_request_context("/ansible/runs/dddd4444"):
|
|
session["user_role"] = "operator"
|
|
page = await render_template(
|
|
"ansible/run_detail.html", run=run, row=run_view(run, {"steward:target:t1": "target web-1"}, now),
|
|
origin={"text": "schedule nightly", "url": "/ansible/schedules", "kind": "schedule"},
|
|
hosts=[(h, host_recap(c)) for h, c in hosts])
|
|
|
|
assert "<title>update.yml on web-1 — Steward</title>" in page
|
|
assert 'href="/ansible/"' in page # breadcrumb back to the runs
|
|
# One verdict line (#5330): how it ended, how long, who started it.
|
|
verdict = page[page.index('<p class="page-verdict">'):]
|
|
verdict = verdict[:verdict.index("</p>")]
|
|
assert '<span class="badge badge-warn">Interrupted</span>' in verdict and " in 10m" in verdict
|
|
assert 'by <a href="/ansible/schedules">schedule nightly</a>' in verdict
|
|
assert "/cancel" not in page # an interrupted run has nothing to cancel
|
|
assert "dddd4444" not in page.replace("/ansible/runs/dddd4444", "")
|
|
# Each host's result in words, the failed host first.
|
|
assert page.index(">zulu<") < page.index(">alpha<")
|
|
assert '<span class="tone-crit">1 failed</span> · 1 ok' in page and ">3 ok<" in page
|
|
assert 'href="/ansible/playbooks/builtin/update.yml"' in page # the playbook it ran
|
|
assert 'href="/ansible/playbooks/builtin/update.yml?from=dddd4444#run-form">Run again</a>' in page
|
|
main = page[page.index("<main"):]
|
|
assert 'style="' not in main[:main.index('id="run-output"')] # no hand-set colours
|
|
assert not hex_colours(main), hex_colours(main)
|
|
|
|
|
|
async def test_run_started_is_on_the_palette(app):
|
|
run = _run("eeee5555", AnsibleRunStatus.running, 0)
|
|
async with app.test_request_context("/ansible/runs"):
|
|
page = await render_template("ansible/run_started.html", run=run, title="update.yml on web-1")
|
|
assert '<span class="badge badge-info">Running</span>' in page
|
|
assert "update.yml on web-1" in page and 'href="/ansible/runs/eeee5555"' in page
|
|
assert not hex_colours(page), hex_colours(page)
|
|
|
|
|
|
def _origin_of(triggered_by=None, origin=None, origin_id=None):
|
|
run = SimpleNamespace(triggered_by=triggered_by, origin=origin, origin_id=origin_id)
|
|
return describe_origin(run, {"u1": "alice"}, {"s1": "nightly"}, {"r1": "nas down"})
|
|
|
|
|
|
def test_a_run_says_who_or_what_started_it():
|
|
"""#5212: the runs list tells a schedule, an alert, an agent and a person apart."""
|
|
assert _origin_of("u1") == {"text": "alice", "url": None, "kind": "user"}
|
|
assert _origin_of(None, "schedule", "s1")["text"] == "schedule nightly"
|
|
assert _origin_of("u1", "schedule", "s1")["text"] == "schedule nightly (run now by alice)"
|
|
assert _origin_of(None, "alert", "r1") == {"text": "alert rule nas down",
|
|
"url": "/alerts/history?rule=r1", "kind": "alert"}
|
|
assert _origin_of("u1", "mcp")["text"] == "alice via MCP"
|
|
assert _origin_of(None, "schedule", "gone")["text"] == "a deleted schedule"
|
|
assert _origin_of("u-gone")["text"] == "a deleted user"
|
|
assert _origin_of()["text"] == "Steward" # automated, from before origins
|
|
|
|
|
|
def test_failed_hosts_come_first_then_changed_then_by_name():
|
|
hosts = {"c": {"changed": 2}, "b": {"failed": 1}, "a": {}, "d": {"unreachable": 1}}
|
|
assert [h for h, _ in hosts_worst_first({"hosts": hosts})] == ["b", "d", "c", "a"]
|
|
assert hosts_worst_first(None) == [] and hosts_worst_first({"hosts": None}) == []
|