feat: route badge reflects 7-day rolling average

The overall health badge on route cards now shows the rolling 7-day
average tier instead of the latest window-hours snapshot, so flapping
routes that are currently up still appear marginal/failing if the
week's mean warrants it. Same averaging drives the dashboard Route
Health widget's summary dot, the routes page summary strip counts,
and (already) the Route Trends chart line colors.

- compute_average_quality() in collector/routes.py (0/1/2 mean,
  thresholds 1.5/0.75, empty-history fallback) — kept in sync with
  the averageRouteTier JS helper in charts.js
- RouteRead / RouteDetail gain a 'quality_avg' field
- list/get/update handlers compute it per route; create skips (no
  meaningful history yet) and the frontend falls back to
  route_result.quality for brand-new routes
- diagnosis tooltip unchanged (still current-snapshot state text)
This commit is contained in:
Louis King committed 2026-07-19 17:23:54 +01:00
1 parent 37f8d7ae0b
commit fc60cd201a
8 files changed
+339 -33

No files matched your search

+108
View File
@@ -328,6 +328,114 @@ class TestGetRouteDetail:
assert second.json() == first_body
class TestRouteQualityAvg:
"""``quality_avg`` field — rolling 7-day average tier.
Backs the route card badge and summary strip counts on /routes so a
flapping route that's currently up still shows as marginal/failing if
the 7-day mean warrants it. See ``compute_average_quality`` in
``collector/routes.py`` for the algorithm.
"""
@staticmethod
def _make_route(session, *, enabled: bool = True, label: str = "Avg"):
# Use uuid-derived public_keys so concurrent tests using _sample_nodes
# (which always inserts "a"*64 / "b"*64) can't trip the unique
# constraint on Node.public_key during parallel xdist runs against
# the shared SQLite file.
suffix = uuid4().hex
keys = [(suffix[:32]).rjust(64, "0"), (suffix[32:64]).rjust(64, "0")]
nodes = [_make_node(session, k) for k in keys]
route = Route(from_label=label, to_label="Sink", enabled=enabled)
session.add(route)
session.flush()
for pos, n in enumerate(nodes):
session.add(
RouteNode(
route_id=route.id,
node_id=n.id,
position=pos,
expected_hash=n.public_key[:2].upper(),
)
)
session.commit()
return route
def test_present_on_list_for_enabled_routes(self, client_no_auth, api_db_session):
"""Each enabled route in the list response carries a computed tier.
With no traffic in the test DB every history bucket is no_coverage
which collapses to failing (mean 0); the point of the assertion is
that the field exists and is a valid tier, not the specific value.
"""
route = self._make_route(api_db_session, enabled=True, label="Enabled")
resp = client_no_auth.get("/api/v1/routes")
assert resp.status_code == 200
items = resp.json()["items"]
# Lookup by id — parallel tests in the same worker may leave
# other routes in the truncated-but-not-yet-reaped window.
matching = [i for i in items if i["id"] == str(route.id)]
assert len(matching) == 1
avg = matching[0]["quality_avg"]
assert avg in {"clear", "marginal", "failing"}
# No traffic in the test DB -> all no_coverage -> failing.
assert avg == "failing"
def test_none_for_disabled_routes(self, client_no_auth, api_db_session):
"""Disabled routes skip the computation; field is null.
Uses the per-route DETAIL endpoint rather than the list — the
list query races with other xdist workers' ``_truncate_all``
teardown against the shared SQLite file (the conftest only
isolates Postgres backends, not SQLite). The detail endpoint
scopes the read to a single row so a parallel truncate either
takes the row (404, not a false-pass) or leaves it.
"""
route = self._make_route(api_db_session, enabled=False, label="Disabled")
resp = client_no_auth.get(f"/api/v1/routes/{route.id}")
assert resp.status_code == 200
assert resp.json()["quality_avg"] is None
def test_present_on_detail(self, client_no_auth, api_db_session):
"""Detail endpoint also exposes the rolling average."""
route = self._make_route(api_db_session, enabled=True, label="Detail")
resp = client_no_auth.get(f"/api/v1/routes/{route.id}")
assert resp.status_code == 200
assert resp.json()["quality_avg"] == "failing"
def test_none_on_create_response(self, client_no_auth, api_db_session):
"""Create handler skips the rolling computation.
A brand-new route has no meaningful 7-day history; the frontend
falls back to ``route_result.quality`` via the
``q = route.quality_avg || route.route_result?.quality`` chain.
"""
nodes = _sample_nodes(api_db_session)
api_db_session.commit()
resp = client_no_auth.post(
"/api/v1/routes",
json={
"from_label": "Fresh",
"to_label": "Route",
"node_public_keys": [n.public_key for n in nodes],
},
headers={"X-User-Roles": "admin"},
)
assert resp.status_code == 201
assert resp.json()["quality_avg"] is None
def test_computed_on_update_response(self, client_no_auth, api_db_session):
"""Update handler recomputes the average (history pre-exists)."""
route = self._make_route(api_db_session, enabled=True, label="UpdateMe")
resp = client_no_auth.put(
f"/api/v1/routes/{route.id}",
json={"description": "now with description"},
headers={"X-User-Roles": "admin"},
)
assert resp.status_code == 200
assert resp.json()["quality_avg"] == "failing"
class TestUpdateRoute:
def test_update_from_to(self, client_no_auth, api_db_session):
nodes = _sample_nodes(api_db_session)
+87
View File
@@ -8,6 +8,7 @@ from sqlalchemy import select
from meshcore_hub.collector.routes import (
_has_any_hops_per_day,
_route_expected_hashes,
compute_average_quality,
derive_expected_hash,
derive_quality,
detect_observed_widths,
@@ -1085,6 +1086,92 @@ class TestEvaluateRouteHistory:
assert today_entry[1] == RouteQuality.UNKNOWN.value
class TestComputeAverageQuality:
"""Rolling-average tier over a history window (server-side badge source).
Mirrors the ``averageRouteTier`` JS helper in
``web/static/js/charts.js`` so the route card badge matches the chart
line color when both render the same window.
"""
@staticmethod
def _day(day_offset: int, quality: str, matched: int = 0):
return (
datetime(2024, 1, 1, tzinfo=timezone.utc).date()
+ timedelta(days=day_offset),
quality,
"healthy" if quality != "unknown" else "no_coverage",
matched,
)
def test_all_clear(self):
history = [self._day(i, RouteQuality.CLEAR.value) for i in range(7)]
assert compute_average_quality(history) == RouteQuality.CLEAR.value
def test_all_marginal(self):
history = [self._day(i, RouteQuality.MARGINAL.value) for i in range(7)]
assert compute_average_quality(history) == RouteQuality.MARGINAL.value
def test_all_failing(self):
history = [self._day(i, RouteQuality.FAILING.value) for i in range(7)]
assert compute_average_quality(history) == RouteQuality.FAILING.value
def test_mixed_clear_marginal_yields_clear(self):
# mean = (2+1+2+1+2+1+2)/7 ≈ 1.57 → clear
history = [
self._day(0, RouteQuality.CLEAR.value),
self._day(1, RouteQuality.MARGINAL.value),
self._day(2, RouteQuality.CLEAR.value),
self._day(3, RouteQuality.MARGINAL.value),
self._day(4, RouteQuality.CLEAR.value),
self._day(5, RouteQuality.MARGINAL.value),
self._day(6, RouteQuality.CLEAR.value),
]
assert compute_average_quality(history) == RouteQuality.CLEAR.value
def test_half_clear_half_failing_yields_marginal(self):
# mean = (2+0+2+0+2+0+2)/7 ≈ 1.14 → marginal
history = [
self._day(
i,
RouteQuality.CLEAR.value if i % 2 == 0 else RouteQuality.FAILING.value,
)
for i in range(7)
]
assert compute_average_quality(history) == RouteQuality.MARGINAL.value
def test_quarter_clear_three_quarters_failing_yields_failing(self):
# mean = (2+0+0+0+2+0+0)/7 ≈ 0.57 → failing
qualities = [
RouteQuality.CLEAR.value,
RouteQuality.FAILING.value,
RouteQuality.FAILING.value,
RouteQuality.FAILING.value,
RouteQuality.CLEAR.value,
RouteQuality.FAILING.value,
RouteQuality.FAILING.value,
]
history = [self._day(i, q) for i, q in enumerate(qualities)]
assert compute_average_quality(history) == RouteQuality.FAILING.value
def test_unknown_collapses_to_failing(self):
# unknown is in the failing tier (0) — a route with no coverage
# for the whole window averages to failing, not "unknown"
history = [self._day(i, RouteQuality.UNKNOWN.value) for i in range(7)]
assert compute_average_quality(history) == RouteQuality.FAILING.value
def test_empty_history_uses_fallback(self):
# Brand-new routes (no history yet) should not flash a misleading
# failing badge — fall back to the current snapshot.
assert (
compute_average_quality([], fallback=RouteQuality.MARGINAL.value)
== RouteQuality.MARGINAL.value
)
def test_empty_history_defaults_to_failing(self):
assert compute_average_quality([]) == RouteQuality.FAILING.value
# ---------------------------------------------------------------------------
# Branch-coverage tests (guards, exception handlers, optional-param paths)
# ---------------------------------------------------------------------------