bvandeusen/FabledSteward
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
dev
FabledSteward/tests/plugins/traefik/test_traefik_page.py
T
bvandeusenandClaude Opus 5.5 398312b8be fix(traefik): a renewed certificate is old, and list_certificates is bounded (#5864) 
A Traefik that hasn't restarted keeps reporting the certificate it
replaced, so codex.fabledsword.com's old one (21 days left) was listed
beside its renewal (81 days) and would have raised an expiry warning on
2026-10-22. A reported certificate whose every name a later-expiring
one covers is now old, on the page, the widget, the overview and MCP.

list_certificates takes `limit` (default 100) with total/truncated, and
says how many served certificates no HTTP router uses (`unrouted`).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 16:00:19 -04:00

436 lines
24 KiB
Python
Raw Permalink Blame History

"""The Traefik page puts failing and slow routers first and folds the healthy
ones into one table; certificates are graded by Settings → Thresholds (#5126)."""
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace as NS
import pytest
from plugins.traefik.routes import gone_routers, router_state, split_routers
from steward.core.settings import to_thresholds_cfg
from tests._templates import render_plugin_template
from tests.test_style_tokens import hex_colours
TH = to_thresholds_cfg({}) # latency crit 200 ms
def _latest(req=1.0, p95=20.0, e4=0.0, e5=0.0):
return NS(request_rate=req, response_bytes_rate=1000.0, latency_p95_ms=p95, latency_p99_ms=p95 * 1.5,
error_rate_4xx_pct=e4, error_rate_5xx_pct=e5,
scraped_at=datetime.now(timezone.utc) - timedelta(seconds=30))
@pytest.mark.parametrize("latest, grade", [
(_latest(), "ok"),
(_latest(e5=2.5), "crit"),
(_latest(p95=450.0), "warn"),
(_latest(e4=5.0), "warn"),
(_latest(p95=120.0, e4=0.5), "ok"), # slow-ish but under the critical cutoff
])
def test_router_state(latest, grade):
assert router_state(latest, TH)[0] == grade
def _router(name, latest):
state, why = router_state(latest, TH)
return {"name": name, "latest": latest, "state": state, "why": why,
**{f"sparkline_{k}": f"<svg data-line={k}></svg>" for k in ("req", "bytes", "p95", "4xx", "5xx")}}
def test_failing_routers_come_before_slow_ones_and_healthy_ones_split_off():
problems, healthy = split_routers([
_router("b-ok", _latest()), _router("slow", _latest(p95=450.0)),
_router("broken", _latest(e5=3.0)), _router("a-ok", _latest())])
assert [r["name"] for r in problems] == ["broken", "slow"]
assert [r["name"] for r in healthy] == ["a-ok", "b-ok"]
async def test_the_page_leads_with_problems_and_folds_the_healthy(app):
problems, healthy = split_routers([
_router("api", _latest(e5=3.0)), _router("web", _latest()), _router("docs", _latest())])
now = datetime.now(timezone.utc)
certs = [{"serial": "1", "cn": "soon.example", "sans": [], "routers": None,
"not_after": now + timedelta(days=2), "days_remaining": 2.0},
{"serial": "2", "cn": "fine.example", "sans": [], "routers": None,
"not_after": now + timedelta(days=60), "days_remaining": 60.0}]
async with app.test_request_context("/plugins/traefik/rows"):
page = await render_plugin_template(app, "traefik", "traefik/rows.html", problems=problems,
healthy=healthy, global_stat=None, certs=certs, range_key="24h",
error_4xx=1.0, error_5xx=0.1)
assert page.index(">api</h3>") < page.index('id="traefik-healthy"')
assert "3.0% of responses are 5xx" in page
assert ">2 healthy routers</summary>" in page
assert page.index(">docs</td>") < page.index(">web</td>")
assert 'class="tone-crit">2 days</td>' in page and 'class="tone-ok">60 days</td>' in page
assert "font-weight:600" not in page and not hex_colours(page), hex_colours(page)
# #5214: every figure that changes over time has its line, not only req/s, p95 and 5xx.
card = page[page.index(">api</h3>"):page.index('id="traefik-healthy"')]
assert [k for k in ("req", "bytes", "p95", "4xx", "5xx") if f"data-line={k}>" in card] == [
"req", "bytes", "p95", "4xx", "5xx"]
# The cards sit in a grid with its own bottom gap, not CSS columns (whose last margin is lost).
assert '<div class="node-grid" style="margin-bottom:1rem;">' in page and "column-width" not in page
def test_a_router_traefik_stopped_reporting_is_set_apart_not_failing():
live = _router("api", _latest())
removed = _router("old", _latest(e5=9.0))
removed["latest"].scraped_at = live["latest"].scraped_at - timedelta(minutes=10)
recent = _router("blip", _latest())
recent["latest"].scraped_at = live["latest"].scraped_at - timedelta(seconds=60)
current, gone = gone_routers([live, removed, recent], 30) # 3 scrapes = 90 s
assert [r["name"] for r in current] == ["api", "blip"]
assert [r["name"] for r in gone] == ["old"]
assert gone_routers([], 30) == ([], [])
async def test_gone_routers_are_listed_by_when_they_were_last_seen(app):
removed = _router("old", _latest(e5=9.0))
removed["latest"].scraped_at = datetime.now(timezone.utc) - timedelta(hours=2)
async with app.test_request_context("/plugins/traefik/rows"):
page = await render_plugin_template(app, "traefik", "traefik/rows.html", problems=[], healthy=[],
gone=[removed], global_stat=None, certs=[], range_key="24h",
error_4xx=1.0, error_5xx=0.1)
page = " ".join(page.split())
assert "1 router no longer reported</summary>" in page
assert "old</span> <span class=\"problem-detail\">last seen <time" in page and ">2h ago</time>" in page
assert "Failing" not in page
@pytest.mark.parametrize("days, says", [
(12.7, "12 days"), (1.2, "1 day"), (0.4, "expires today"),
(-0.3, "expired today"), (-1.5, "expired 1 day ago"), (-12.0, "expired 12 days ago"), (None, "—"),
])
def test_a_certificate_past_its_date_says_expired(days, says):
"""#5212: "expired 12 days ago", never "-12 days"."""
from steward.core.formatting import cert_left
assert cert_left(days) == says
@pytest.mark.parametrize("enabled, says, not_says", [
(False, "The access log is off", "No requests in the last"),
(True, "No requests in the last 24h", "The access log is off"),
])
async def test_the_traffic_empty_state_knows_whether_the_log_is_on(app, enabled, says, not_says):
async with app.test_request_context("/plugins/traefik/traffic"):
page = await render_plugin_template(app, "traefik", "traefik/traffic.html", summary=None, history=[],
range_key="24h", access_log_enabled=enabled,
access_log_path="/var/log/traefik/access.log")
assert says in page and not_says not in page
# ── Routers widget: current routers, certs on the router they serve (#5254) ──
@pytest.mark.parametrize("rule, hosts", [
("Host(`app.example.com`)", ["app.example.com"]),
("Host(`a.example.com`, `B.example.com`) && PathPrefix(`/api`)", ["a.example.com", "b.example.com"]),
("Host(`a.example.com`) || Host(`b.example.com`)", ["a.example.com", "b.example.com"]),
("HostSNI(`db.example.com`)", ["db.example.com"]),
("HostSNI(`*`)", []),
("PathPrefix(`/`)", []),
("", []),
])
def test_a_router_rule_names_its_hosts(rule, hosts):
from plugins.traefik.scraper import rule_hosts
assert rule_hosts(rule) == hosts
def test_a_certificate_covers_its_names_and_one_wildcard_label():
from plugins.traefik.scraper import cert_names, covers
names = cert_names("Example.com", "*.example.com, www.other.org")
assert names == {"example.com", "*.example.com", "www.other.org"}
assert covers(names, "example.com") and covers(names, "app.example.com")
assert covers(names, "www.other.org")
assert not covers(names, "a.b.example.com") and not covers(names, "other.org")
@pytest.mark.parametrize("metrics_url, api_url, base", [
("http://traefik:8080/metrics", "", "http://traefik:8080/api"),
("http://traefik:8080/metrics?x=1", "", "http://traefik:8080/api"),
("http://traefik:8080/metrics", "http://other:9000/api/", "http://other:9000/api"),
])
def test_the_api_sits_beside_the_metrics_unless_configured(metrics_url, api_url, base):
from plugins.traefik.scraper import api_base
assert api_base(metrics_url, api_url) == base
def test_a_metrics_router_finds_its_api_name_with_or_without_the_entrypoint():
from plugins.traefik.routes import router_hosts
domains = {"app@docker": ["app.example.com"]}
assert router_hosts("app@docker", domains) == ["app.example.com"]
assert router_hosts("websecure-app@docker", domains) == ["app.example.com"]
assert router_hosts("myapp@docker", domains) == []
def _cert(serial, cn, days, seen_ago=30, sans=None):
now = datetime.now(timezone.utc)
return NS(serial=serial, cn=cn, sans=sans, not_after=now + timedelta(days=days),
scraped_at=now - timedelta(seconds=seen_ago))
def test_a_certificate_traefik_stopped_reporting_is_old_not_current():
from plugins.traefik.routes import current_certs
live, renewed = _cert("1", "a.example", 80), _cert("2", "a.example", 5, seen_ago=3600)
served, old = current_certs([live, renewed], 60)
assert served == [live] and old == [renewed]
assert current_certs([], 60) == ([], [])
def test_a_certificate_a_later_one_renewed_is_old_though_traefik_still_reports_it():
"""codex.fabledsword.com on 2026-10-15: the replaced certificate (21 days
left) still reported beside its renewal (81 days), #5864."""
from plugins.traefik.routes import current_certs
old_one, renewal = _cert("1", "codex.example", 21), _cert("2", "codex.example", 81)
wider = _cert("3", "a.example", 90, sans="a.example,b.example")
narrower = _cert("4", "b.example", 30)
unrelated = _cert("5", "c.example", 10)
served, old = current_certs([old_one, renewal, wider, narrower, unrelated], 60)
assert {c.serial for c in served} == {"2", "3", "5"}
assert {c.serial for c in old} == {"1", "4"}
def test_each_router_gets_the_soonest_certificate_for_its_hosts():
from plugins.traefik.routes import certs_by_router
now = datetime.now(timezone.utc)
certs = [_cert("1", "*.example.com", 40), _cert("2", "app.example.com", 10), _cert("3", "x.org", 2)]
got = certs_by_router(["app@docker", "docs@docker", "plain@docker"],
{"app@docker": ["app.example.com"], "docs@docker": ["docs.example.com"]},
certs, now)
assert got["app@docker"]["cn"] == "app.example.com" and round(got["app@docker"]["days"]) == 10
assert got["docs@docker"]["cn"] == "*.example.com"
assert "plain@docker" not in got
async def test_the_widget_gives_a_verdict_then_routers_under_one_header(app):
slow = _router("api@docker", _latest(p95=450.0))
expiring = _router("shop@docker", _latest(req=3.0))
expiring.update(state="crit", why="certificate expires in 2 days",
cert={"cn": "shop.example", "days": 2.0, "grade": "crit"})
fine = _router("web@docker", _latest(req=9.0))
fine["cert"] = {"cn": "web.example", "days": 60.0, "grade": "ok"}
async with app.test_request_context("/plugins/traefik/widget"):
page = await render_plugin_template(
app, "traefik", "traefik/widget.html", problems=[expiring, slow], healthy=[fine],
healthy_more=4, total=7, loose_certs=[], domains_known=True)
page = " ".join(page.split())
assert "2 of 7 routers need a look." in page
assert page.count("<th>Router</th>") == 1 and "<th>TLS</th>" in page
# Problems first, each with why; the cert sits on its router, not above.
assert page.index(">shop") < page.index(">api") < page.index(">web")
assert "certificate expires in 2 days" in page and "p95 latency is 450 ms" in page
assert "60 days" in page and "Certificates expiring" not in page
assert '</span>shop<span class="w-sub' in page and ">shop@docker" not in page # provider off the label
assert "+4 more healthy" in page
async def test_a_router_in_trouble_says_since_when_on_the_widget(app):
"""#5810: the widget's problem rows carry the since the page shows."""
slow = _router("api@docker", _latest(p95=450.0))
slow.update(since=datetime.now(timezone.utc) - timedelta(hours=2), since_before=False)
failing = _router("pay@docker", _latest(e5=4.0))
failing.update(since=datetime.now(timezone.utc) - timedelta(days=1), since_before=True)
async with app.test_request_context("/plugins/traefik/widget"):
page = await render_plugin_template(
app, "traefik", "traefik/widget.html", problems=[failing, slow], healthy=[],
healthy_more=0, total=2, loose_certs=[], domains_known=True)
page = " ".join(page.split())
assert "all of the last 24 h" in page
assert page.count(", started ") == 1 and page.index("all of the last 24 h") < page.index(", started ")
async def test_without_router_domains_expiring_certs_follow_the_routers(app):
fine = _router("web@docker", _latest())
async with app.test_request_context("/plugins/traefik/widget"):
page = await render_plugin_template(
app, "traefik", "traefik/widget.html", problems=[], healthy=[fine], healthy_more=0, total=1,
loose_certs=[{"cn": "old.example", "days": 5.0}], domains_known=False)
page = " ".join(page.split())
assert "All 1 router healthy; 1 certificate expiring." in page
assert "<th>TLS</th>" not in page
assert page.index("<th>Router</th>") < page.index("Certificates expiring") < page.index("old.example")
async def test_the_page_sets_certificates_traefik_no_longer_serves_apart(app):
now = datetime.now(timezone.utc)
certs = [{"serial": "1", "cn": "app.example", "sans": ["www.app.example"], "routers": ["app"],
"not_after": now + timedelta(days=40), "days_remaining": 40.0},
{"serial": "2", "cn": "spare.example", "sans": [], "routers": [],
"not_after": now + timedelta(days=40), "days_remaining": 40.0}]
old = [{"serial": "3", "cn": "gone.example", "sans": [], "routers": [],
"not_after": now + timedelta(days=5), "days_remaining": 5.0}]
async with app.test_request_context("/plugins/traefik/rows"):
page = await render_plugin_template(app, "traefik", "traefik/rows.html", problems=[], healthy=[],
gone=[], global_stat=None, certs=certs, old_certs=old,
range_key="24h", error_4xx=1.0, error_5xx=0.1)
page = " ".join(page.split())
assert "also www.app.example" in page and ">app</td>" in page and ">not in use</td>" in page
assert page.index("spare.example") < page.index("No longer served (1)") < page.index("gone.example")
async def test_the_request_chart_says_its_figures_before_the_chart(app):
async with app.test_request_context("/plugins/traefik/widget/request_chart"):
page = await render_plugin_template(
app, "traefik", "traefik/widget_request_chart.html", labels_json="[1, 2]",
req_rate_json="[1.5, 4.25]", error_rate_json="[0, 0.5]", widget_id="7", span="6h",
days=False, now_req=4.25, peak_req=4.25, peak_err=0.5)
empty = await render_plugin_template(
app, "traefik", "traefik/widget_request_chart.html", labels_json="[]", span="6h")
page = " ".join(page.split())
assert "Now 4.25 req/s across all routers, peak 4.25. 5xx peaked at 0.50% of responses." in page
assert page.index("w-verdict") < page.index("<canvas")
assert "Requests/s (all routers)" in page and "max: 100" not in page
assert "couldn't be drawn" in page # a failure says why in place of the chart
assert "No Traefik requests recorded in the last 6h." in empty
def test_routers_are_graded_on_their_recent_window_not_one_scrape():
"""#5214: a 4xx share near the cutoff crossed it on one scrape and fell
back on the next, so problem cards came and went every poll."""
from plugins.traefik.routes import STATE_WINDOW_WORDS, grade_routers
spiky, quiet = _latest(e4=1.5), _latest(e4=1.5)
routers = [{"name": "spiky", "latest": spiky}, {"name": "quiet", "latest": quiet}]
# spiky's last 15 min average well under the cutoff; quiet has no window rows.
grade_routers(routers, {"spiky": NS(error_rate_4xx_pct=0.4, error_rate_5xx_pct=0.0, latency_p95_ms=20.0)}, TH)
assert routers[0]["state"] == "ok"
assert routers[1]["state"] == "warn" and not routers[1]["why"].endswith(STATE_WINDOW_WORDS)
held = [{"name": "held", "latest": _latest()}] # clean now, bad over the window
grade_routers(held, {"held": NS(error_rate_4xx_pct=3.0, error_rate_5xx_pct=0.0, latency_p95_ms=20.0)}, TH)
assert held[0]["state"] == "warn"
assert held[0]["why"] == "3.0% of responses are 4xx over the last 15 min"
def test_the_last_reload_gauge_is_a_moment_not_a_flag():
"""#5355: traefik_config_last_reload_success is a Unix time; read as a
0/1 flag it said "Reload failed" on every scrape."""
from plugins.traefik.scraper import compute_global_stats
at = compute_global_stats({"traefik_config_last_reload_success": {frozenset(): 1791460000.0},
"traefik_config_reloads_total": {frozenset(): 343.0}}, None)
assert at["config_last_reload_at"] == datetime.fromtimestamp(1791460000.0, timezone.utc)
assert at["config_reloads_total"] == 343.0
never = compute_global_stats({"traefik_config_last_reload_success": {frozenset(): 0.0}}, None)
assert never["config_last_reload_at"] is None
async def test_the_config_tile_says_when_it_last_reloaded(app):
stat = NS(config_last_reload_at=datetime.now(timezone.utc) - timedelta(minutes=3),
config_reloads_total=343.0, open_conns_total=None, process_memory_bytes=None)
async with app.test_request_context("/plugins/traefik/rows"):
page = await render_plugin_template(app, "traefik", "traefik/rows.html", problems=[], healthy=[],
gone=[], global_stat=stat, certs=[], old_certs=[], range_key="24h",
error_4xx=1.0, error_5xx=0.1)
assert "Config reloaded" in page and "3m ago" in page and "343 reloads" in page
assert "Reload failed" not in page
# ── Error cutoffs are settings, reachable from the page (#5380) ─────────────
def test_router_state_reads_the_error_cutoffs_from_settings():
strict = to_thresholds_cfg({"traefik.threshold.error_5xx_pct": 0.0, "traefik.threshold.error_4xx_pct": 0.2})
lenient = to_thresholds_cfg({"traefik.threshold.error_5xx_pct": 5.0, "traefik.threshold.error_4xx_pct": 10.0})
assert router_state(_latest(e5=0.05), strict)[0] == "crit"
assert router_state(_latest(e5=0.05), TH)[0] == "ok" # under the 0.1% default
assert router_state(_latest(e5=2.5), lenient)[0] == "ok"
assert router_state(_latest(e4=0.5), strict)[0] == "warn"
assert router_state(_latest(e4=5.0), lenient)[0] == "ok"
@pytest.mark.parametrize("role, linked", [("admin", True), ("viewer", False)])
async def test_the_page_links_an_admin_to_the_router_thresholds(app, role, linked):
from quart import session
async with app.test_request_context("/plugins/traefik/"):
session["user_role"] = role
page = await render_plugin_template(app, "traefik", "traefik/index.html", poll_interval=30,
access_log_enabled=False, current_range="1h")
assert ('href="/settings/thresholds/#traefik"' in page) is linked
# ── State first: the verdict and the request chart lead the page (#5710) ────
def _bucket(at, e5=0.0, e4=0.0, p95=20.0):
return NS(bucket=at, error_rate_5xx_pct=e5, error_rate_4xx_pct=e4, latency_p95_ms=p95)
def test_a_problem_dates_from_the_start_of_its_trailing_run_of_bad_buckets():
from plugins.traefik.routes import problem_since
t = [datetime(2026, 10, 14, h, tzinfo=timezone.utc) for h in range(8, 13)]
# Bad at 9, fine at 10, bad again from 11: the problem now began at 11.
history = [_bucket(t[0]), _bucket(t[1], e5=4.0), _bucket(t[2]), _bucket(t[3], e5=4.0),
_bucket(t[4], p95=900.0)]
assert problem_since(history, TH) == (t[3], False)
# Bad in every bucket: it began before the range shown.
assert problem_since([_bucket(x, e4=6.0) for x in t], TH) == (t[0], True)
# Fine in the newest bucket (a problem only the 15-minute window shows): no date.
assert problem_since(history[:3], TH) == (None, False)
assert problem_since([], TH) == (None, False)
async def _rows(app, routers, **ctx):
problems, healthy = split_routers(routers)
ctx = {"gone": [], "global_stat": None, "certs": [], "old_certs": [], "range_key": "24h",
"current_range": "24h", "error_4xx": 1.0, "error_5xx": 0.1, **ctx}
async with app.test_request_context("/plugins/traefik/rows"):
page = await render_plugin_template(app, "traefik", "traefik/rows.html", problems=problems,
healthy=healthy, **ctx)
return " ".join(page.split())
async def test_the_verdict_names_each_problem_and_since_when(app):
started = datetime.now(timezone.utc) - timedelta(hours=2)
api = _router("api", _latest(e5=3.0))
api.update(since=started, since_before=False)
slow = _router("slow", _latest(p95=450.0))
slow.update(since=started, since_before=True)
stat = NS(config_last_reload_at=None, config_reloads_total=None, open_conns_total=12.0,
process_memory_bytes=None)
page = await _rows(app, [api, slow, _router("web", _latest())], global_stat=stat,
gone=[_router("old", _latest())])
verdict = page[page.index('id="traefik-verdict"'):page.index("</p>")]
assert 'hx-swap-oob="true"' in verdict
assert '<strong class="tone-crit">2 routers need a look</strong>' in verdict
assert "api</span>, 3.0% of responses are 5xx, started <time" in verdict and "2h ago" in verdict
assert "slow</span>, p95 latency is 450 ms, all of the last 24h" in verdict
assert "1 healthy." in verdict and "1 no longer reported." in verdict
# The card says since when too; the proxy's figures follow the cards, not the verdict.
card = page[page.index(">api</h3>"):]
assert card.index("3.0% of responses are 5xx, started <time") < card.index("</p>")
assert page.index(">api</h3>") < page.index("Open connections") < page.index('id="traefik-healthy"')
assert "Needs attention" not in page and "Healthy routers" not in page # the verdict says them
async def test_the_verdict_when_all_is_well_or_nothing_is_reported(app):
well = await _rows(app, [_router("web", _latest()), _router("api", _latest())])
assert '<strong class="tone-ok">All 2 routers healthy</strong>' in well
assert "stat-strip" not in well # no figures, no empty strip
nothing = await _rows(app, [])
assert "No routers reported yet." in nothing
async def test_the_verdict_lists_three_problems_then_counts_the_rest(app):
page = await _rows(app, [_router(f"r{i}", _latest(e5=3.0)) for i in range(5)])
verdict = page[page.index('id="traefik-verdict"'):page.index("</p>")]
assert "r2</span>" in verdict and "r3</span>" not in verdict and "; 2 more below." in verdict
async def test_the_page_puts_the_verdict_then_the_request_chart_above_the_routers(app):
from quart import session
async with app.test_request_context("/plugins/traefik/"):
session["user_role"] = "viewer"
page = await render_plugin_template(app, "traefik", "traefik/index.html", poll_interval=30,
access_log_enabled=True, current_range="7d")
assert (page.index('id="traefik-verdict"') < page.index('id="traefik-chart"')
< page.index('id="traefik-rows"') < page.index('id="traefik-traffic"'))
chart = page[page.index('id="traefik-chart"'):page.index('id="traefik-rows"')]
assert 'hx-get="/plugins/traefik/widget/request_chart?wid=page"' in chart
assert 'hx-include="#time-range"' in chart and "rangeChange from:body" in chart
async def test_the_request_chart_over_days_labels_its_times_with_the_day(app):
kw = dict(labels_json="[1, 2]", req_rate_json="[1, 2]", error_rate_json="[0, 0]", widget_id="page",
span="7d", now_req=2, peak_req=2, peak_err=0)
async with app.test_request_context("/plugins/traefik/widget/request_chart"):
week = await render_plugin_template(app, "traefik", "traefik/widget_request_chart.html", days=True, **kw)
hours = await render_plugin_template(app, "traefik", "traefik/widget_request_chart.html", days=False, **kw)
assert '{ weekday: "short", day: "numeric", hour: "2-digit", minute: "2-digit", hourCycle: "h23" }' in week
assert '{ hour: "2-digit", minute: "2-digit", hourCycle: "h23" }' in hours
assert "last 7d, your local time" in week
Reference in New Issue View Git Blame Copy Permalink