bvandeusenandClaude Opus 5.5
7c346354f2
/ansible/inventory/ shows every target (address, its linked Steward host, its groups) and every group (members, the schedules aimed at it) under a verdict line. The whole row opens the target or group. Add target and Add group open small dialogs; a failed add comes back with its dialog open and what was typed. The red x leaves the rows: Delete is on the target's or group's page, behind the confirm dialog, saying which schedules would be left with nothing to run on. The two list pages (/inventory/targets, /inventory/groups) are gone with no redirect (rule 122); their URLs keep only the Add POST. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
222 lines
12 KiB
Python
222 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/"):
|
|
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": {"id": "s1", "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/s1">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(id=f"id-{name}", 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) == {"id": "id-new", "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
|
|
# No run id as text: only in the links to the run and to run it again.
|
|
assert "dddd4444" not in page.replace("/ansible/runs/dddd4444", "").replace("?from=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" class="btn btn-sm btn-ghost">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}) == []
|