140 lines
7.6 KiB
Python
140 lines
7.6 KiB
Python
"""Audit entries name the thing they are about, linked to its page, and keep
|
|
the recorded name of something deleted since (#5126)."""
|
|
from datetime import datetime, timezone
|
|
from types import SimpleNamespace as NS
|
|
|
|
from quart import render_template
|
|
|
|
from steward.audit.service import entity_labels
|
|
|
|
|
|
class _Empty:
|
|
"""A session whose every query finds nothing: every entity was deleted."""
|
|
async def execute(self, _stmt):
|
|
return NS(all=lambda: [], scalars=lambda: NS(all=lambda: []))
|
|
|
|
|
|
def _ev(etype, eid, action="x.y"):
|
|
return NS(entity_type=etype, entity_id=eid, action=action, username="admin",
|
|
timestamp=datetime(2026, 10, 6, 12, 0, tzinfo=timezone.utc))
|
|
|
|
|
|
async def test_a_deleted_entity_keeps_the_name_its_entry_recorded():
|
|
items = [{"event": _ev("host", "h-123456789"), "detail": {"name": "web-1"}},
|
|
{"event": _ev("monitor", "m-abcdef012"), "detail": {}},
|
|
{"event": _ev("plugin", "docker"), "detail": {}},
|
|
{"event": _ev(None, None), "detail": {}}]
|
|
labels = await entity_labels(_Empty(), items)
|
|
assert labels[("host", "h-123456789")] == {"label": "web-1", "url": None, "gone": True}
|
|
assert labels[("monitor", "m-abcdef012")] == {"label": "m-abcdef", "url": None, "gone": True}
|
|
assert labels[("plugin", "docker")] == {"label": "docker", "url": "/settings/plugins/", "gone": False}
|
|
assert len(labels) == 3
|
|
|
|
|
|
async def test_the_page_names_and_links_the_entity(app):
|
|
events = [
|
|
{"event": _ev("host", "h1", "host.renamed"), "detail": {"old": "a", "new": "b"},
|
|
"entity": {"label": "web-1", "url": "/hosts/h1", "gone": False}},
|
|
{"event": _ev("monitor", "m9", "monitor.deleted"), "detail": {},
|
|
"entity": {"label": "api check", "url": None, "gone": True}},
|
|
]
|
|
async with app.test_request_context("/audit/"):
|
|
page = await render_template("audit/list.html", events=events, after=None, action_filter="")
|
|
assert '<a href="/hosts/h1">web-1</a>' in page
|
|
assert "api check" in page and "(deleted)" in page
|
|
assert "h1</td>" not in page # no bare id where a name is known
|
|
|
|
|
|
|
|
async def test_the_log_continues_itself_instead_of_a_load_more_link(app):
|
|
events = [{"event": _ev("host", "h1", "host.renamed"), "detail": {}, "entity": None}]
|
|
async with app.test_request_context("/audit/"):
|
|
more = await render_template("audit/list.html", events=events, last_group="Today",
|
|
after="2026-10-07T03:00:00+00:00|0190abc", action_filter="host.")
|
|
last = await render_template("audit/_rows.html", events=events, after=None, action_filter="")
|
|
assert ('data-scroll-more hx-get="/audit/?before=2026-10-07T03%3A00%3A00%2B00%3A00%7C0190abc'
|
|
'&action=host.&group=Today"') in more
|
|
assert "Load more" not in more
|
|
assert "data-scroll-more" not in last and "End of the list." in last
|
|
|
|
|
|
def test_an_entry_reads_as_a_sentence_with_its_detail_in_words():
|
|
"""#5212: no action codes, Python reprs or raw scopes on the page."""
|
|
from steward.audit.service import action_text, detail_facts
|
|
assert action_text("monitor.disabled") == "paused monitor"
|
|
assert action_text("some.new_thing") == "some.new_thing" # unknown: its code
|
|
facts = detail_facts({"rule_id": "r1", "check": False, "vars": ["a", "b"], "secret_vars": [],
|
|
"scope": "steward:group:g1", "interval_seconds": 7200,
|
|
"section": "ansible.credentials", "run_id": "abc", "limit": None,
|
|
"carried": {"monitor_results": 12}},
|
|
{"steward:group:g1": "group Docker Swarm"})
|
|
said = {f["label"]: f["text"] for f in facts}
|
|
assert said == {"dry run": "no", "variables": "a, b", "secret variables": "none",
|
|
"scope": "group Docker Swarm", "every": "2h", "section": "Ansible: credentials",
|
|
"run": "open", "history carried over": "monitor results 12"}
|
|
assert [f["url"] for f in facts if f["label"] == "run"] == ["/ansible/runs/abc"]
|
|
|
|
|
|
async def test_the_row_says_what_was_done_not_its_code(app):
|
|
events = [{"event": _ev("monitor", "m1", "monitor.disabled"), "detail": {},
|
|
"entity": {"label": "api check", "url": "/monitors/m1", "gone": False},
|
|
"action": "paused monitor", "facts": [{"label": "dry run", "text": "no", "url": None}]}]
|
|
async with app.test_request_context("/audit/"):
|
|
from steward.audit.service import AREAS
|
|
page = await render_template("audit/list.html", events=events, after=None,
|
|
action_filter="monitor.", areas=AREAS)
|
|
assert 'title="monitor.disabled">paused monitor</td>' in page
|
|
assert "dry run</span>" in page and "False" not in page
|
|
assert '<option value="monitor." selected>Monitors</option>' in page
|
|
|
|
|
|
async def test_each_entry_shows_its_exact_moment_in_the_readers_zone(app):
|
|
"""Milestone 490: a record is read for when, not how long ago."""
|
|
from quart import session
|
|
from steward.core import formatting as f
|
|
events = [{"event": _ev("host", "h1", "host.renamed"), "detail": {}, "entity": None,
|
|
"facts": [{"label": "was", "text": "web-0", "url": None},
|
|
{"label": "AI token", "text": "#1", "url": None}]}]
|
|
async with app.test_request_context("/audit/"):
|
|
session["tz"] = "America/New_York"
|
|
page = await render_template("audit/list.html", events=events, after=None, action_filter="")
|
|
ny = f.zone("America/New_York")
|
|
assert f"<th>When ({f.tz_label(ny)})</th>" in page
|
|
# 12:00 UTC is 08:00 in New York, and the hover names the zone.
|
|
assert f'<td class="nowrap">{f.when(events[0]["event"].timestamp, ny)}</td>' in page
|
|
assert "08:00:00 EDT" in page and " ago<" not in page
|
|
# One fact a line, labels quiet, and the table on classes.
|
|
assert '<div><span class="tone-dim">was</span> web-0</div>' in page
|
|
table = page[page.index('<table class="table">'):]
|
|
assert "style=" not in table
|
|
|
|
|
|
def test_dividers_open_each_day_group_once_even_across_pages():
|
|
"""#5214: the operator wants Today, Yesterday, the last week and older as
|
|
dividers, with every row's own date beside it."""
|
|
from steward.audit.service import mark_day_groups
|
|
now = datetime(2026, 10, 8, 18, 0, tzinfo=timezone.utc)
|
|
|
|
def at(d, h=12):
|
|
return NS(timestamp=datetime(2026, 10, d, h, 0, tzinfo=timezone.utc))
|
|
|
|
page1 = [{"event": at(8)}, {"event": at(8, 9)}, {"event": at(7)}, {"event": at(5)}]
|
|
assert mark_day_groups(page1, timezone.utc, "", now) == "Last 7 days"
|
|
assert [(i["group"], i["opens_group"]) for i in page1] == [
|
|
("Today", True), ("Today", False), ("Yesterday", True), ("Last 7 days", True)]
|
|
# The next page carries the group the last ended in, so it doesn't reopen it.
|
|
page2 = [{"event": at(3)}, {"event": at(1)}]
|
|
assert mark_day_groups(page2, timezone.utc, "Last 7 days", now) == "Older"
|
|
assert [(i["group"], i["opens_group"]) for i in page2] == [("Last 7 days", False), ("Older", True)]
|
|
|
|
|
|
async def test_a_divider_row_heads_each_group(app):
|
|
events = [{"event": _ev("host", "h1", "host.renamed"), "detail": {}, "entity": None, "facts": [],
|
|
"group": "Today", "opens_group": True},
|
|
{"event": _ev("host", "h1", "host.renamed"), "detail": {}, "entity": None, "facts": [],
|
|
"group": "Today", "opens_group": False}]
|
|
async with app.test_request_context("/audit/"):
|
|
page = await render_template("audit/_rows.html", events=events, after=None, action_filter="")
|
|
assert page.count('<tr class="table-divider"><th colspan="5" scope="rowgroup">Today</th></tr>') == 1
|
|
assert "Today 1" not in page and ">Tue 6 Oct " in page # the row keeps its date
|