feat(ansible): one strip of tabs, a run list that opens on its verdict, a run page in one line (#5328, #5329, #5330)
CI / lint (push) Successful in 3s
CI / integration (push) Successful in 1m25s
CI / unit (push) Successful in 1m46s
CI / publish (push) Successful in 1m9s

- section_tabs macro; Runs · Playbooks · Schedules · Inventory on every
  Ansible page, replacing each page's own set of header buttons.
- Runs: a verdict line (last 30 days, failures first with the latest named,
  running now, the schedule due next); under each status what the run did
  across its hosts (ansible/recap.py); a filter bar (playbook, status, who
  started it) that swaps the table and carries into the scroll paging.
- Run page: status, duration, start and who started it in one line; each
  host's result in words, worst first; the playbook it ran links to it;
  the log and failures on classes rather than inline colours.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-07 23:45:04 -04:00
co-authored by Claude Opus 5.5
parent 7499c38037
commit a02f224f22
18 changed files with 519 additions and 111 deletions
+107 -15
View File
@@ -8,7 +8,10 @@ from quart import render_template, session
from steward.ansible.inventory_gen import run_title
from steward.ansible.origins import describe_origin
from steward.ansible.routes import hosts_worst_first, run_duration, run_view
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
@@ -24,14 +27,14 @@ 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):
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=None, output="")
params={"check": True} if check else None, results=results, output="")
def test_a_running_run_counts_its_duration_so_far():
@@ -41,56 +44,145 @@ def test_a_running_run_counts_its_duration_so_far():
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),
_run("cccc3333", AnsibleRunStatus.success, 120, took_s=40),
_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", runs=runs, after="2026-10-07T01:00:00+00:00|x")
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">update.yml on web-1</a>' in main
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 'data-scroll-more hx-get="/ansible/?before=2026-10-07T01%3A00%3A00%2B00%3A00%7Cx"' in main
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/browse", "/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=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}}}))
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
assert '<span class="badge badge-warn">Interrupted</span>' in page
assert "<dd>10m</dd>" in page
# 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
# #5212: who started it, by name; no run id in the page; the failed host first.
assert '<dd><a href="/ansible/schedules">schedule nightly</a></dd>' in page
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/browse/builtin/update.yml"' in page # the playbook it ran
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)
@@ -0,0 +1,85 @@
"""Integration: the Ansible Runs tab (#5329) against real Postgres. Its
verdict counts the last 30 days; the filter bar narrows by playbook, status
and origin; an HTMX request from the bar gets the table alone, and the next
page as the reader scrolls keeps the filters. Requires STEWARD_DATABASE_URL."""
from __future__ import annotations
import asyncio
import os
import time
import uuid
from datetime import datetime, timedelta, timezone
import pytest
pytestmark = pytest.mark.integration
@pytest.fixture
def app():
if not os.environ.get("STEWARD_DATABASE_URL"):
pytest.skip("needs Postgres")
from steward.app import create_app
return create_app(testing=False)
def test_the_runs_tab_counts_and_filters(app):
from steward.models.ansible import AnsibleRun, AnsibleRunStatus
from steward.models.users import User, UserRole
async def _go():
tag = uuid.uuid4().hex[:8]
uid = str(uuid.uuid4())
now = datetime.now(timezone.utc)
def run(name, status, mins, origin=None, by=None, results=None):
return AnsibleRun(playbook_path=f"it{tag}/{name}.yml", source_name="steward-builtin",
inventory_scope="steward:all", status=status, origin=origin,
triggered_by=by, started_at=now - timedelta(minutes=mins),
finished_at=now - timedelta(minutes=mins - 1), results=results)
async with app.db_sessionmaker() as s:
async with s.begin():
s.add(User(id=uid, username=f"runs-{tag}", email=f"runs-{tag}@test",
password_hash="x", role=UserRole.viewer, is_active=True))
async with s.begin():
s.add_all([
run("prune", AnsibleRunStatus.success, 10, origin="schedule",
results={"hosts": {"a": {"ok": 2, "changed": 1}, "b": {"ok": 2, "changed": 1}}}),
run("update", AnsibleRunStatus.failed, 20, by=uid,
results={"hosts": {f"db-{tag}": {"ok": 1, "failed": 1}}}),
])
client = app.test_client()
async with client.session_transaction() as sess:
sess.update(user_id=uid, user_role="viewer", username=f"runs-{tag}", login_at=time.time())
async def get(url, **kw):
resp = await client.get(url, **kw)
return resp.status_code, await resp.get_data(as_text=True)
out = {"page": await get(f"/ansible/?q=it{tag}"),
"failed": await get(f"/ansible/?q=it{tag}&status=failed", headers={"HX-Request": "true"}),
"sched": await get(f"/ansible/?q=it{tag}&origin=schedule", headers={"HX-Request": "true"}),
"person": await get(f"/ansible/?q=it{tag}&origin=manual", headers={"HX-Request": "true"})}
async with app.db_sessionmaker() as s:
async with s.begin():
from sqlalchemy import delete
await s.execute(delete(AnsibleRun).where(AnsibleRun.playbook_path.like(f"it{tag}/%")))
await s.delete(await s.get(User, uid))
return tag, out
tag, out = asyncio.run(_go())
status, page = out["page"]
assert status == 200
assert '<p class="page-verdict">' in page and "in the last 30 days" in page # the verdict queries ran
assert f"it{tag}/prune.yml" in page and f"it{tag}/update.yml" in page
assert "2 hosts · 2 changed" in page and f'<span class="tone-crit">failed on db-{tag}</span>' in page
assert '<nav class="section-tabs" aria-label="Ansible">' in page
failed = out["failed"][1]
assert "<html" not in failed and '<p class="page-verdict">' not in failed # the table alone
assert f"it{tag}/update.yml" in failed and f"it{tag}/prune.yml" not in failed
sched = out["sched"][1]
assert f"it{tag}/prune.yml" in sched and f"it{tag}/update.yml" not in sched
person = out["person"][1]
assert f"it{tag}/update.yml" in person and f"it{tag}/prune.yml" not in person