"""Metric anomalies (#5740): what counts as a departure, and how one is shown
on Alerts → Anomalies, the Overview and the weekly report. The detection,
lapsing and learning passes run against Postgres in
tests/integration/test_anomalies.py."""
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from zoneinfo import ZoneInfo

from quart import render_template, session

from steward.core import mail
from steward.core.anomalies import DEPART, WINDOW, AnomalyView, _gate_open, departs
from steward.core.norms import Norm
from tests.test_style_tokens import hex_colours

NOW = datetime.now(timezone.utc)
NORM = Norm(kind="day", slot=14, usual_hold=10.0, usual_peak=20.0, days_seen=7)


def test_a_departure_is_above_the_usual_hold_by_the_ratio_and_the_floor():
    assert departs(50.0, NORM, floor=5)
    assert not departs(10.0 * DEPART, NORM, floor=0)       # at the line is not over it
    assert not departs(14.0, NORM, floor=5)                # over ×1.25, but by less than the floor
    assert departs(15.5, NORM, floor=5)
    assert not departs(0.3, Norm("day", 1, 0.1, 0.2, 7), floor=2)   # 0.1% → 0.3% is never news


def _anomaly(**kw):
    a = dict(id="a1", source_module="host_agent", resource_name="nas", metric_name="cpu_pct",
             started_at=NOW - timedelta(hours=3), last_started_at=NOW - timedelta(hours=1),
             ended_at=None, occurrences=1, peak_level=82.0, peak_at=NOW, usual_hold=30.0,
             usual_peak=45.0, review="to_review", reviewed_by=None, reviewed_at=None, note=None)
    a.update(kw)
    return SimpleNamespace(**a)


def _view(href="/hosts/h1", **kw):
    return AnomalyView(_anomaly(**kw), None, href)


def test_a_rows_link_opens_its_page_over_a_range_that_reaches_it():
    assert _view().link == "/hosts/h1?range=24h"
    assert _view(last_started_at=NOW - timedelta(days=3)).link == "/hosts/h1?range=7d"
    assert _view(last_started_at=NOW - timedelta(days=12)).link == "/hosts/h1?range=30d"
    assert _view(href=None).link is None


async def _page(app, role, to_review, past):
    async with app.test_request_context("/alerts/anomalies"):
        session["user_id"], session["user_role"] = "u1", role
        return await render_template("alerts/anomalies.html", review_days=14,
                                     to_review=[{"v": v, "opens_group": i == 0, "group": "Today"}
                                                for i, v in enumerate(to_review)],
                                     past=[{"v": v, "opens_group": i == 0, "group": "Older"}
                                           for i, v in enumerate(past)])


async def test_the_page_lists_what_waits_for_review_first(app):
    waiting = _view(resource_name="waiting-box")
    done = _view(id="a2", resource_name="done-box", review="normal", reviewed_by="ann",
                 reviewed_at=NOW, ended_at=NOW - timedelta(minutes=5),
                 last_started_at=NOW - timedelta(minutes=30))
    page = await _page(app, "operator", [waiting], [done])
    assert "1 to review" in page
    assert page.index("waiting-box") < page.index("done-box")
    assert "Metric anomaly" in page and "Marked normal by ann" in page
    assert "Ongoing since" in page and "usual for that hour" in page
    assert 'action="/alerts/anomalies/a1/review"' in page              # to review: buttons
    assert 'action="/alerts/anomalies/a2/review"' not in page          # reviewed: none
    assert 'href="/hosts/h1?range=24h"' in page
    assert ">Today<" in page and ">Older<" in page
    assert not hex_colours(page), hex_colours(page)


async def test_a_viewer_sees_anomalies_but_cannot_mark_them(app):
    page = await _page(app, "viewer", [_view()], [])
    assert "nas" in page and "This is normal" not in page and "Real problem" not in page


async def test_a_repeat_says_how_often_and_when_the_latest_was(app):
    v = _view(occurrences=5, ended_at=NOW - timedelta(minutes=10))
    page = await _page(app, "operator", [v], [])
    assert "5 times since" in page and ", last" in page


async def test_a_quiet_page_says_nothing_waits(app):
    page = await _page(app, "admin", [], [])
    assert "Nothing to review" in page and "within 14 days" in page


async def _widget(app, logged_in, anomalies, to_review):
    async with app.test_request_context("/status/widget"):
        if logged_in:
            session["user_id"] = "u1"
        return await render_template("status/widget.html", entries=[], up=0, down=0, pending=0,
                                     total_hosts=1, alerts=[], anomalies=anomalies,
                                     to_review=to_review, widget_id="1")


async def test_the_overview_warns_of_an_open_anomaly_and_counts_those_to_review(app):
    page = await _widget(app, True, [_view()], 3)
    assert '<span class="badge badge-warn">Metric anomaly</span>' in page
    assert "nas · CPU" in page
    assert '<a href="/alerts/anomalies">3 metric anomalies to review</a>' in page
    assert "No monitors configured" not in page
    shared = await _widget(app, False, [_view()], 1)
    assert "1 metric anomaly to review" in shared and 'href="/alerts/anomalies"' not in shared


async def test_a_quiet_overview_has_no_review_line(app):
    page = await _widget(app, True, [], 0)
    assert "to review" not in page


REPORT = {"generated_at": datetime(2026, 10, 11, 12, 0, tzinfo=timezone.utc),
          "uptime_summary": [], "active_alerts": [], "alert_events": {"fired": 0, "resolved": 0},
          "top_routers": [], "hosts_down": [],
          "anomalies": {
              "seen": [{"name": "nas — CPU", "review": "to_review", "occurrences": 2,
                        "started_at": datetime(2026, 10, 6, 7, 0, tzinfo=timezone.utc),
                        "last_started_at": datetime(2026, 10, 9, 7, 10, tzinfo=timezone.utc),
                        "ended_at": datetime(2026, 10, 9, 7, 40, tzinfo=timezone.utc),
                        "level": "held 82%, usual 30%"}],
              "lapsed": [{"name": "web01 — Load", "review": "lapsed", "occurrences": 1,
                          "started_at": datetime(2026, 9, 20, 3, 0, tzinfo=timezone.utc),
                          "last_started_at": datetime(2026, 9, 20, 3, 0, tzinfo=timezone.utc),
                          "ended_at": datetime(2026, 9, 20, 4, 0, tzinfo=timezone.utc),
                          "level": "held 9.1, usual 2.0"}],
              "to_review": 1}}


def test_the_weekly_report_lists_the_weeks_anomalies_in_the_readers_zone():
    to = mail.Recipient("ann@example.com", ZoneInfo("America/New_York"), False, "ann")
    _, text, html = mail.report_email({"base_url": "https://steward.example"}, REPORT, to)
    for part in (text, html):
        assert "nas — CPU" in part and "2 times since Tue 6 Oct 2026 03:00 EDT" in part
        assert "Fri 9 Oct 2026 03:10 EDT – Fri 9 Oct 2026 03:40 EDT" in part
        assert "web01 — Load" in part and "lapsed unreviewed" in part
        assert "1 anomaly waits to be marked normal or real" in part
        assert "https://steward.example/alerts/anomalies" in part
    assert "METRIC ANOMALIES THIS WEEK" in text


def test_a_week_without_anomalies_leaves_the_section_out():
    to = mail.Recipient("ann@example.com", timezone.utc, False, "ann")
    quiet = {**REPORT, "anomalies": {"seen": [], "lapsed": [], "to_review": 0}}
    _, text, html = mail.report_email({}, quiet, to)
    assert "anomal" not in (text + html).lower()


def test_a_share_gated_on_a_rate_is_judged_only_once_it_stands_for_enough_events():
    """#6105: three 404s on a quiet router read 7%. With `fewest`, the share
    is judged only once rate × window × share reaches it."""
    learned = SimpleNamespace(gate=("req", lambda: 30 / WINDOW.total_seconds()), fewest=lambda: 20)
    key = ("traefik", "r", "e4")
    per_s = 100 / WINDOW.total_seconds()                               # 100 requests in the window
    assert not _gate_open(learned, key, {key: 7.5, ("traefik", "r", "req"): per_s})          # 7.5 errors
    assert _gate_open(learned, key, {key: 25.0, ("traefik", "r", "req"): per_s})             # 25
    assert not _gate_open(learned, key, {key: 90.0, ("traefik", "r", "req"): 10 / WINDOW.total_seconds()})
    assert _gate_open(SimpleNamespace(gate=("req", 0.0), fewest=None), key, {key: 1.0})