"""Integration: a metric out of its normal is an anomaly to review (#5740).

The operator's rulings (System 182): an anomaly stays "to review" until
someone marks it, even after it ends; a repeat joins the row still waiting;
an ended one nobody reviews lapses after the review window, an ongoing one
never; and only an anomaly marked normal is learned.
Requires STEWARD_DATABASE_URL.
"""
from __future__ import annotations

import asyncio
import os
import uuid
from datetime import datetime, timedelta, timezone

import pytest

pytestmark = pytest.mark.integration

METRIC = "cpu_pct"
NOW = datetime(2026, 10, 12, 10, 0, tzinfo=timezone.utc)


@pytest.fixture
def app():
    if not os.environ.get("STEWARD_DATABASE_URL"):
        pytest.skip("needs Postgres")
    from steward.app import create_app
    return create_app(testing=False)


@pytest.fixture
def src():
    """A source of the test's own, learned for its run. A detection pass judges
    every learned key, so another test's keys must not be learned here; within
    a test, resources are reported at times far enough apart not to overlap."""
    from steward.core import norms
    name = f"anom{uuid.uuid4().hex[:8]}"
    norms.learn_normal(name, METRIC, 5)
    yield name
    norms.LEARNED.pop((name, METRIC), None)


def _norms(src: str, res: str, hold: float = 10.0, peak: float = 20.0):
    """A learned day normal for every hour: holds `hold`, peaks at `peak`."""
    from steward.models.norms import MetricNorm
    return [MetricNorm(source_module=src, resource_name=res, metric_name=METRIC, slot=h,
                       kind="day", usual_hold=hold, usual_peak=peak, days_seen=7, updated_at=NOW)
            for h in range(24)]


async def _report(s, src: str, res: str, until: datetime, value: float):
    """A sample a minute over the window before `until`, at `value`."""
    from steward.core.anomalies import WINDOW
    from steward.models.metrics import PluginMetric, PluginMetricLatest
    n = int(WINDOW.total_seconds() // 60)
    s.add_all([PluginMetric(source_module=src, resource_name=res, metric_name=METRIC,
                            value=value, recorded_at=until - timedelta(minutes=m))
               for m in range(n)])
    await s.merge(PluginMetricLatest(source_module=src, resource_name=res, metric_name=METRIC,
                                     value=value, recorded_at=until))


async def _steps(app, src: str, res: str, steps, *, seed: bool = True):
    """Seed the normal (unless `seed` is false), then for each (at, value)
    report the window before `at` and run one detection pass at `at`. Returns
    the key's anomalies, oldest first, with their spans."""
    from sqlalchemy import select
    from steward.core.anomalies import detect
    from steward.models.norms import MetricAnomaly, MetricAnomalySpan

    async with app.db_sessionmaker() as s:
        if seed:
            async with s.begin():
                s.add_all(_norms(src, res))
        for at, value in steps:
            async with s.begin():
                await _report(s, src, res, at, value)
            async with s.begin():
                await detect(s, at)
        rows = list((await s.execute(
            select(MetricAnomaly).where(MetricAnomaly.resource_name == res)
            .order_by(MetricAnomaly.started_at))).scalars())
        spans = {a.id: list((await s.execute(
            select(MetricAnomalySpan).where(MetricAnomalySpan.anomaly_id == a.id)
            .order_by(MetricAnomalySpan.started_at))).scalars()) for a in rows}
        return rows, spans


def _run(app, src: str, res: str, steps):
    """_steps for a test that needs nothing else. A test doing more runs it
    all in one asyncio.run: the app's pooled connections belong to the loop
    that opened them, so a second loop can't use them."""
    return asyncio.run(_steps(app, src, res, steps))


def _res(tag: str) -> str:
    return f"{tag}-{uuid.uuid4().hex[:8]}"


def test_a_level_within_its_normal_is_no_anomaly(app, src):
    # Usual hold 10: 12 is under ×1.25, and 14 is over it but not by the floor of 5.
    rows, _ = _run(app, src, _res("calm"), [(NOW, 12.0), (NOW + timedelta(minutes=30), 14.0)])
    assert rows == []


def test_a_step_change_stays_open_and_to_review(app, src):
    from steward.core.anomalies import TO_REVIEW
    res = _res("step")
    rows, spans = _run(app, src, res, [(NOW + timedelta(minutes=m), 50.0) for m in (0, 30, 60, 90)])
    (a,) = rows
    assert a.review == TO_REVIEW and a.ended_at is None and a.occurrences == 1
    assert a.peak_level == 50.0 and a.usual_hold == 10.0 and a.usual_peak == 20.0
    assert a.started_at == NOW - timedelta(minutes=15)
    assert len(spans[a.id]) == 1 and spans[a.id][0].ended_at is None


def test_one_that_ended_unwatched_is_still_to_review_and_a_repeat_joins_it(app, src):
    from steward.core.anomalies import TO_REVIEW
    res = _res("repeat")
    back = NOW + timedelta(minutes=30)
    rows, spans = _run(app, src, res, [
        (NOW, 60.0), (back, 10.0),                                    # out, then back
        (NOW + timedelta(hours=3), 40.0),                             # again, later
        (NOW + timedelta(hours=3, minutes=30), 10.0),
        (NOW + timedelta(hours=3, minutes=40), 45.0),                 # back out within a window
    ])
    (a,) = rows
    assert a.review == TO_REVIEW
    # Two occurrences: the third departure came back within one window of
    # the second ending, so it continues the second.
    assert a.occurrences == 2 and len(spans[a.id]) == 2
    first, second = spans[a.id]
    assert first.ended_at == back and first.peak_level == 60.0
    # The continuation's window still held some of the 10s, so the
    # occurrence's peak stays its first level.
    assert second.ended_at is None and second.peak_level == 40.0
    assert a.peak_level == 60.0 and a.last_started_at == second.started_at


def test_an_ended_anomaly_lapses_after_the_window_and_an_ongoing_one_never(app, src):
    from sqlalchemy import select
    from steward.core.anomalies import LAPSED, TO_REVIEW, detect, lapse
    from steward.models.norms import MetricAnomaly
    ended, going = _res("ended"), _res("going")
    end = NOW + timedelta(minutes=30)

    async def _go():
        await _steps(app, src, ended, [(NOW, 60.0), (end, 10.0)])
        await _steps(app, src, going, [(NOW + timedelta(days=2), 60.0)])
        async with app.db_sessionmaker() as s:
            async with s.begin():
                # The ongoing one keeps reporting high, so it stays open.
                await _report(s, src, going, end + timedelta(days=30), 60.0)
                await detect(s, end + timedelta(days=30))
            async with s.begin():
                await lapse(s, end + timedelta(days=13), 14)
                mid = {a.resource_name: a.review for a in (await s.execute(
                    select(MetricAnomaly).where(MetricAnomaly.resource_name.in_((ended, going)))
                )).scalars()}
            async with s.begin():
                await lapse(s, end + timedelta(days=30), 14)
            s.expire_all()
            late = {a.resource_name: (a.review, a.ended_at) for a in (await s.execute(
                select(MetricAnomaly).where(MetricAnomaly.resource_name.in_((ended, going)))
            )).scalars()}
            return mid, late

    mid, late = asyncio.run(_go())
    assert mid == {ended: TO_REVIEW, going: TO_REVIEW}
    assert late[ended][0] == LAPSED
    assert late[going] == (TO_REVIEW, None)


def test_a_repeat_after_review_is_a_new_row(app, src):
    from steward.core.anomalies import REAL, TO_REVIEW, mark
    res = _res("again")

    async def _go():
        rows, _ = await _steps(app, src, res, [(NOW, 60.0), (NOW + timedelta(minutes=30), 10.0)])
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await mark(s, rows[0].id, REAL, "tester", "disk filled")
        return await _steps(app, src, res, [(NOW + timedelta(hours=5), 60.0)], seed=False)

    rows, _ = asyncio.run(_go())
    assert [(a.review, a.occurrences) for a in rows] == [(REAL, 1), (TO_REVIEW, 1)]
    assert rows[0].note == "disk filled" and rows[0].reviewed_by == "tester"


def test_to_review_is_listed_first(app, src):
    from steward.core.anomalies import NORMAL, TO_REVIEW, list_anomalies, mark
    older, newer = _res("older"), _res("newer")
    seen = {}

    async def _go():
        old_rows, _ = await _steps(app, src, older, [(NOW, 60.0), (NOW + timedelta(minutes=30), 10.0)])
        new_rows, _ = await _steps(app, src, newer, [(NOW + timedelta(hours=2), 60.0),
                                                     (NOW + timedelta(hours=2, minutes=30), 10.0)])
        seen["older"] = old_rows[0].review
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await mark(s, new_rows[0].id, NORMAL, "tester")
            got = await list_anomalies(s, limit=1000)
            return [a.resource_name for a in got if a.resource_name in (older, newer)]

    assert asyncio.run(_go()) == [older, newer]
    assert seen["older"] == TO_REVIEW


@pytest.mark.parametrize("review,learned_hold", [
    ("normal", 90.0), ("real", 10.0), ("lapsed", 10.0), ("to_review", 10.0)])
def test_only_an_anomaly_marked_normal_is_learned(app, src, review, learned_hold):
    """Seven days at 10, but 14:00–15:00 held 90 on three of them, each under
    one anomaly. Only when it is marked normal does 14:00 learn 90."""
    from steward.core.norms import norms_at, rebuild_norms
    from steward.models.metrics import PluginMetricHourly
    from steward.models.norms import MetricAnomaly, MetricAnomalySpan
    res = _res(f"learn-{review}")
    hot_days = (1, 2, 3)
    rows = []
    for d in range(1, 8):
        for h in range(24):
            bucket = NOW.replace(hour=0) - timedelta(days=d) + timedelta(hours=h)
            v = 90.0 if (d in hot_days and h == 14) else 10.0
            rows.append(PluginMetricHourly(source_module=src, resource_name=res, metric_name=METRIC,
                                           bucket=bucket, value_avg=v, value_max=v, value_p95=v,
                                           sample_count=60))
    first = NOW.replace(hour=14) - timedelta(days=max(hot_days))

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                s.add_all(rows)
                a = MetricAnomaly(source_module=src, resource_name=res, metric_name=METRIC,
                                  started_at=first, last_started_at=first, occurrences=3,
                                  ended_at=first + timedelta(hours=1), peak_level=90.0,
                                  peak_at=first, usual_hold=10.0, usual_peak=10.0, review=review)
                s.add(a)
                await s.flush()
                s.add_all([MetricAnomalySpan(
                    anomaly_id=a.id, started_at=NOW.replace(hour=14) - timedelta(days=d),
                    ended_at=NOW.replace(hour=15) - timedelta(days=d), peak_level=90.0)
                    for d in hot_days])
            async with s.begin():
                await rebuild_norms(s, NOW)
            got = await norms_at(s, [(src, res, METRIC)], NOW.replace(hour=14))
            return got[(src, res, METRIC)]

    norm = asyncio.run(_go())
    assert norm.usual_hold == learned_hold


def test_cleanup_prunes_reviewed_anomalies_and_never_one_to_review(app, src):
    from sqlalchemy import select
    from steward.core.anomalies import REAL, mark, prune
    from steward.models.norms import MetricAnomaly
    kept, gone = _res("kept"), _res("gone")
    later = NOW + timedelta(days=1)

    async def _go():
        await _steps(app, src, kept, [(NOW, 60.0), (NOW + timedelta(minutes=30), 10.0)])
        rows, _ = await _steps(app, src, gone, [(later, 60.0), (later + timedelta(minutes=30), 10.0)])
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await mark(s, rows[0].id, REAL, "tester")
            async with s.begin():
                await prune(s, NOW + timedelta(days=2))
            return {a.resource_name for a in (await s.execute(
                select(MetricAnomaly).where(MetricAnomaly.resource_name.in_((kept, gone)))
            )).scalars()}

    assert asyncio.run(_go()) == {kept}


def test_a_gated_metric_is_judged_only_while_its_gate_is_open(app, src):
    """A router's 4xx share means nothing over a handful of requests (#5741):
    learned with gate=("req", 1.0), it is judged only while the resource's
    own "req" averages at least 1 over the window."""
    from steward.core import norms
    from steward.core.anomalies import WINDOW
    from steward.models.metrics import PluginMetric, PluginMetricLatest
    norms.learn_normal(src, METRIC, 5, gate=("req", 1.0))
    res = _res("gated")
    quiet, busy = NOW, NOW + timedelta(hours=2)

    async def _req(s, until, value):
        n = int(WINDOW.total_seconds() // 60)
        s.add_all([PluginMetric(source_module=src, resource_name=res, metric_name="req",
                                value=value, recorded_at=until - timedelta(minutes=m)) for m in range(n)])
        await s.merge(PluginMetricLatest(source_module=src, resource_name=res, metric_name="req",
                                         value=value, recorded_at=until))

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await _req(s, quiet, 0.2)
                await _req(s, busy, 5.0)
        return await _steps(app, src, res, [(quiet, 60.0), (busy, 60.0)])

    rows, spans = asyncio.run(_go())
    (a,) = rows                                   # only the busy window opened one
    assert spans[a.id][0].started_at == busy - WINDOW


def test_a_metric_learns_only_from_its_since(app, src):
    """A metric whose meaning changed (Traefik latency, #5755) learns only
    from history after `since`: seven days held 90, but only the last three
    are after it, under the five a normal needs."""
    from steward.core import norms
    from steward.core.norms import norms_at, rebuild_norms
    from steward.models.metrics import PluginMetricHourly
    res = _res("since")
    rows = [PluginMetricHourly(source_module=src, resource_name=res, metric_name=METRIC,
                               bucket=NOW.replace(hour=0) - timedelta(days=d) + timedelta(hours=h),
                               value_avg=90.0, value_max=90.0, value_p95=90.0, sample_count=60)
            for d in range(1, 8) for h in range(24)]

    async def _go(since):
        norms.learn_normal(src, METRIC, 5, since=since)
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await rebuild_norms(s, NOW)
            got = await norms_at(s, [(src, res, METRIC)], NOW.replace(hour=14))
            return got.get((src, res, METRIC))

    async def _both():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                s.add_all(rows)
        return await _go(None), await _go(NOW.replace(hour=0) - timedelta(days=3))

    learned, held_back = asyncio.run(_both())
    assert learned is not None and learned.usual_hold == 90.0
    assert held_back is None


def test_a_metric_learned_without_its_parts_ignores_them(app, src):
    """A host's CPU learns a normal and each of its cores doesn't (#5742):
    with parts=False, "<host>:core0" is neither learned nor judged, and the
    floor may be a function read at each pass."""
    from sqlalchemy import select
    from steward.core import norms
    from steward.core.anomalies import detect, levels_now
    from steward.core.norms import norms_at, rebuild_norms
    from steward.models.metrics import PluginMetricHourly
    from steward.models.norms import MetricAnomaly
    host = _res("whole")
    part = f"{host}:core0"
    rows = [PluginMetricHourly(source_module=src, resource_name=r, metric_name=METRIC,
                               bucket=NOW.replace(hour=0) - timedelta(days=d) + timedelta(hours=h),
                               value_avg=10.0, value_max=10.0, value_p95=10.0, sample_count=60)
            for r in (host, part) for d in range(1, 8) for h in range(24)]
    at = NOW + timedelta(days=3)

    async def _go():
        norms.learn_normal(src, METRIC, lambda: 5.0, parts=False)
        async with app.db_sessionmaker() as s:
            async with s.begin():
                s.add_all(rows)
            async with s.begin():
                await rebuild_norms(s, NOW)
            async with s.begin():
                learned = await norms_at(s, [(src, host, METRIC), (src, part, METRIC)],
                                         NOW.replace(hour=14))
            async with s.begin():
                await _report(s, src, host, at, 90.0)
                await _report(s, src, part, at, 90.0)
            async with s.begin():
                levels = await levels_now(s, at)
            async with s.begin():
                await detect(s, at)
            async with s.begin():
                opened = list((await s.execute(select(MetricAnomaly.resource_name).where(
                    MetricAnomaly.source_module == src))).scalars())
            return learned, levels, opened

    learned, levels, opened = asyncio.run(_go())
    assert set(learned) == {(src, host, METRIC)}
    assert (src, host, METRIC) in levels and (src, part, METRIC) not in levels
    assert opened == [host]      # 90 against a usual 10: the host departs, its core isn't judged