"""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")