"""The Monitors page leads with what needs looking at (#5121): a count strip, problems first, each monitor on a page of its own, adding behind a button.""" from datetime import datetime, timedelta, timezone from types import SimpleNamespace import pytest from quart import render_template, session from steward.core.settings import to_thresholds_cfg from steward.monitors.service import problems_first, row_state from tests.test_style_tokens import hex_colours TH = to_thresholds_cfg({}) # latency warn 50 / crit 200 ms; uptime warn 99 / crit 95 % NOW = datetime.now(timezone.utc) def _mon(name, mtype="http", enabled=True, mid=None): return SimpleNamespace(id=mid or name, name=name, type=mtype, target=f"https://{name}/", enabled=enabled, host_id=None, check_interval_seconds=0, config={}) def _check(up=True, ms=20.0, ago_min=1): return SimpleNamespace(is_up=up, response_ms=ms, resolved_ip=None, status_code=200 if up else 503, error_msg=None if up else "503 Service Unavailable", tls_expires_at=None, checked_at=NOW - timedelta(minutes=ago_min)) @pytest.mark.parametrize("enabled, latest, uptime, state", [ (False, _check(False), 50.0, "paused"), # paused wins over everything (True, None, None, "pending"), (True, _check(False), 100.0, "down"), (True, _check(ms=20), 100.0, "up"), (True, _check(ms=120), 100.0, "up"), # slow but tolerable: coloured, not sorted (True, _check(ms=250), 100.0, "degraded"), # past the latency critical cutoff (True, _check(ms=20), 98.0, "degraded"), # uptime under its warning floor (True, _check(ms=20), None, "up"), # no uptime yet is not a problem ]) def test_row_state(enabled, latest, uptime, state): assert row_state(_mon("m", enabled=enabled), latest, uptime, TH) == state def test_problems_come_first_then_by_name(): rows = [{"monitor": _mon(n), "state": s} for n, s in [ ("alpha", "up"), ("Zed", "down"), ("beta", "paused"), ("gamma", "degraded"), ("delta", "pending"), ("able", "down")]] assert [r["monitor"].name for r in problems_first(rows)] == [ "able", "Zed", "gamma", "delta", "alpha", "beta"] def _row(m, state, latest, uptime=None, tls_days=None): return {"monitor": m, "state": state, "latest": latest, "uptime_pct": uptime, "tls_days": tls_days, "heartbeat": [{"state": "up" if latest.is_up else "down", "title": "t"}] if latest else []} async def test_the_list_counts_then_shows_problems_first(app): data = problems_first([ _row(_mon("web", mid="m1"), "up", _check(ms=30), 100.0, tls_days=2.0), _row(_mon("api", mid="m2"), "down", _check(False), 90.0), _row(_mon("nas", "icmp", mid="m3", enabled=False), "paused", None), ]) counts = {"up": 1, "degraded": 0, "down": 1, "pending": 0, "paused": 1} async with app.test_request_context("/monitors/rows"): session["user_role"] = "operator" page = await render_template("monitors/rows.html", monitor_data=data, counts=counts, range_key="24h") assert page.index('Down') < page.index("ping-row") assert page.index('href="/monitors/m2"') < page.index('href="/monitors/m1"') < page.index('href="/monitors/m3"') assert "HTTP · https://web/" in page and "Ping · https://nas/" in page assert '30 ms' in page assert '90.0%' in page and '100.0%' in page assert 'class="tone-crit" title="TLS certificate expires in 2 days">cert 2d' in page assert 'Down' in page and 'Paused' in page assert 'aria-label="Delete api"' in page and 'data-confirm="Delete monitor api?"' in page assert "onclick" not in page and not hex_colours(page), hex_colours(page) async def test_a_viewer_gets_no_row_actions_and_an_empty_board_says_so(app): async with app.test_request_context("/monitors/rows"): page = await render_template("monitors/rows.html", monitor_data=[_row(_mon("web"), "up", _check())], counts={"up": 1, "degraded": 0, "down": 0, "pending": 0, "paused": 0}, range_key="24h") empty = await render_template("monitors/rows.html", monitor_data=[], range_key="24h", counts=dict.fromkeys(("up", "degraded", "down", "pending", "paused"), 0)) assert "row-actions" not in page assert "No monitors yet" in empty and "/monitors/new" not in empty async def test_the_monitor_page_says_since_when_and_why(app): m = _mon("shop ", mid="m9") m.config = {"method": "GET", "expected_status": 200, "headers": {"Authorization": "s3cr3t-token"}, "verify_ssl": True} down = _check(False, ms=None) async with app.test_request_context("/monitors/m9"): session["user_role"] = "admin" page = await render_template( "monitors/detail.html", monitor=m, host=None, latest=down, failures=[down], state="down", since=datetime.now(timezone.utc) - timedelta(minutes=12), since_at_least=False, tls_days=None, uptime={"24h": 93.0, "7d": None, "30d": 99.5}, heartbeat=[{"state": "up", "title": "a"}, {"state": "down", "title": "b"}]) assert 'Monitors' in page and "shop <api>" in page assert "since 12m ago" in page assert "503 Service Unavailable" in page assert 'class="stat-tile stat-crit"' in page and "93.00%" in page assert "Authorization" in page and "s3cr3t-token" not in page # header names, never their values assert 'aria-label="Delete shop <api>"' in page assert not hex_colours(page), hex_colours(page) async def test_adding_is_its_own_page_and_keeps_what_was_typed(app): async with app.test_request_context("/monitors/new"): session["user_role"] = "operator" page = await render_template( "monitors/form.html", monitor=None, mtype="tcp", config={"port": "22"}, types=[], values={"name": "db", "target": "10.0.0.9", "host_id": "", "check_interval_seconds": "30"}, error="port must be between 1 and 65535, got 0.") assert 'Monitors' in page and 'action="/monitors/add"' in page assert '' in page assert 'value="db"' in page and 'value="10.0.0.9"' in page and 'value="30"' in page