bvandeusenandClaude Opus 5.5
c625528191
router_findings never passed `since`, so every failing router read as brand new and sorted last among the criticals. The overview now reads the failing routers' last day of history and dates each one with problem_since, as the dashboard widget does. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
83 lines
3.9 KiB
Python
83 lines
3.9 KiB
Python
"""The traefik plugin's MCP tools (#5838): which certificates and routers the
|
|
overview names, how a router is found by name, and what a row says."""
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace as NS
|
|
|
|
import pytest
|
|
|
|
import plugins.traefik as traefik_plugin
|
|
from plugins.traefik import mcp_tools as t
|
|
from steward.models.users import UserRole
|
|
|
|
NOW = datetime(2026, 10, 15, 12, 0, tzinfo=timezone.utc)
|
|
THRESHOLDS = {"tls_days": {"warn": 14, "crit": 3}}
|
|
|
|
|
|
def _cert(cn, days):
|
|
return NS(cn=cn, serial=f"s-{cn}", sans=None, not_after=NOW + timedelta(days=days), scraped_at=NOW)
|
|
|
|
|
|
def test_a_served_certificate_is_a_finding_inside_its_warning_days():
|
|
found = {f["subject"]: f for f in t.cert_findings(
|
|
[_cert("fine.example", 40), _cert("soon.example", 10), _cert("now.example", 2),
|
|
_cert("gone.example", -1)], NOW, THRESHOLDS)}
|
|
assert set(found) == {"soon.example", "now.example", "gone.example"}
|
|
assert found["soon.example"]["severity"] == "warning"
|
|
assert found["now.example"]["severity"] == "critical"
|
|
assert "expired" in found["gone.example"]["summary"]
|
|
|
|
|
|
def test_only_a_failing_router_is_an_overview_finding():
|
|
routers = [
|
|
{"name": "app@swarm", "state": "crit", "why": "12.0% of responses are 5xx over the last 15 min",
|
|
"meaning": {"meaning": "Mostly 502: Traefik couldn't reach the app."}},
|
|
{"name": "radarr@swarm", "state": "warn", "why": "7.2% of responses are 4xx"},
|
|
{"name": "ok@swarm", "state": "ok", "why": None},
|
|
]
|
|
[f] = t.router_findings(routers)
|
|
assert f["severity"] == "critical" and f["subject"] == "app@swarm"
|
|
assert "Mostly 502" in f["summary"] and f["next"] == 'get_router(router="app@swarm")'
|
|
assert f["since"] is None
|
|
|
|
|
|
def test_a_failing_router_says_since_when():
|
|
began = NOW - timedelta(hours=2)
|
|
[f] = t.router_findings([{"name": "app@swarm", "state": "crit", "why": "12.0% of responses are 5xx",
|
|
"since": began}])
|
|
assert f["since"] == "2026-10-15T10:00:00Z" and f["age_s"] > 0
|
|
|
|
|
|
def test_a_router_is_found_with_or_without_its_provider():
|
|
names = ["radarr@swarm", "sonarr@swarm", "api@internal", "web@docker", "web@file"]
|
|
assert t._match("radarr@swarm", names) == "radarr@swarm"
|
|
assert t._match("RADARR", names) == "radarr@swarm"
|
|
with pytest.raises(ValueError, match="matches web@docker, web@file"):
|
|
t._match("web", names)
|
|
with pytest.raises(ValueError, match="list_routers shows them"):
|
|
t._match("nope", names)
|
|
|
|
|
|
def test_a_row_names_codes_and_meaning_only_when_troubled():
|
|
latest = NS(request_rate=2.5, error_rate_4xx_pct=7.25, error_rate_5xx_pct=0.0, latency_p50_ms=12.0,
|
|
latency_p95_ms=40.0, latency_p99_ms=80.0, scraped_at=NOW)
|
|
troubled = t._router_row({"name": "radarr@swarm", "state": "warn", "why": "4xx", "latest": latest,
|
|
"codes": [("404", 30)], "meaning": {"action": "tune"}, "service": "arr_radarr",
|
|
"normed": True}, {"radarr@swarm": {"cn": "radarr.example", "days": 50.0}})
|
|
assert troubled["codes"] == [["404", 30]] and troubled["swarm_service"] == "arr_radarr"
|
|
assert troubled["graded_on_normal"] is True and troubled["certificate"]["days_left"] == 50.0
|
|
healthy = t._router_row({"name": "ok@swarm", "state": "ok", "why": None, "latest": latest}, {})
|
|
assert "codes" not in healthy and "meaning" not in healthy and "certificate" not in healthy
|
|
|
|
|
|
def test_the_plugin_registers_its_tools_for_viewers():
|
|
got = {}
|
|
traefik_plugin.register_mcp_tools(lambda fn, role, read_only: got.update({fn.__name__: (role, read_only)}))
|
|
assert got == {name: (UserRole.viewer, True)
|
|
for name in ("list_routers", "get_router", "list_certificates", "get_traffic")}
|
|
assert callable(traefik_plugin.mcp_overview)
|
|
|
|
|
|
async def test_list_routers_refuses_an_unknown_state():
|
|
with pytest.raises(ValueError, match="state must be one of"):
|
|
await t.list_routers(state="broken")
|