"""Silence is alertable (#5062): a series' usual gap is learned from its
samples, an outage does not become its cadence, and a series counts as silent
after three usual gaps (at least five minutes; an hour while its cadence is
unknown)."""
import asyncio
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace as NS

import pytest

from steward.core.alerts import (
    NO_DATA_FLOOR_S, NO_DATA_UNKNOWN_S, _evaluate_rule, _track_samples, next_interval,
    no_data_after, silent_since,
)
from steward.models.alerts import AlertOperator, AlertStateEnum

NOW = datetime(2026, 10, 6, 12, 0, tzinfo=timezone.utc)


@pytest.mark.parametrize("interval,gap,want", [
    (None, 60, 60),             # the first gap is the cadence
    (0.0, 30, 30),
    (60, 60, 60),               # steady
    (60, 80, 65),               # a late sample barely moves it
    (60, 3600, 90),             # an outage counts as at most three gaps
    (60, -5, 45),               # never a negative gap
])
def test_next_interval(interval, gap, want):
    assert next_interval(interval, gap) == pytest.approx(want)


def test_a_new_cadence_is_learned_in_steps():
    interval, steps = 60.0, 0
    while no_data_after(interval) < 600:  # samples now arrive every 10 minutes
        interval, steps = next_interval(interval, 600), steps + 1
    assert steps == 3


@pytest.mark.parametrize("interval,want", [
    (None, NO_DATA_UNKNOWN_S), (0.0, NO_DATA_UNKNOWN_S),
    (30, NO_DATA_FLOOR_S), (100, 300), (600, 1800),
])
def test_no_data_after(interval, want):
    assert no_data_after(interval) == want


@pytest.mark.parametrize("last_ago,interval,watch_ago,silent_after", [
    (60, 60, 9999, None),            # reporting
    (300, 60, 9999, None),           # exactly at the limit is not yet silent
    (301, 60, 9999, 300),            # silent since last + 300s
    (2000, 600, 9999, 1800),         # slow series: three of its gaps
    (None, None, 3000, None),        # never sampled, watched under an hour
    (None, None, 4000, 3600),        # never sampled, watched over an hour
])
def test_silent_since(last_ago, interval, watch_ago, silent_after):
    last = NOW - timedelta(seconds=last_ago) if last_ago is not None else None
    watch = NOW - timedelta(seconds=watch_ago)
    got = silent_since(last, interval, watch, NOW)
    if silent_after is None:
        assert got is None
    else:
        assert got == (last or watch) + timedelta(seconds=silent_after)


def test_steward_s_own_outage_is_not_the_series_silence():
    """#5441: a series last sampled 18 minutes ago, while Steward has only been
    back for 2 minutes, is not silent yet; once Steward has watched it for
    three of its gaps with nothing arriving, it is, from then."""
    last = NOW - timedelta(minutes=18)
    watch = NOW - timedelta(days=1)
    assert silent_since(last, 60, watch, NOW, awake_since=NOW - timedelta(minutes=2)) is None
    awake = NOW - timedelta(minutes=6)
    assert silent_since(last, 60, watch, NOW, awake_since=awake) == awake + timedelta(seconds=300)
    # A sample since Steward came back is what counts, when there is one.
    assert silent_since(NOW - timedelta(minutes=1), 60, watch, NOW, awake_since=awake) is None


class _Session:
    def __init__(self):
        self.added = []

    def add(self, row):
        self.added.append(row)

    async def flush(self):
        pass


def _state(**kw):
    base = dict(state=AlertStateEnum.firing, last_sample_at=None, sample_interval_s=None,
                no_data_since=None, consecutive_failure_count=0, consecutive_ok_count=0)
    return NS(**{**base, **kw})


def _samples(*ago_values):
    return [NS(recorded_at=NOW - timedelta(seconds=a), value=v) for a, v in ago_values]


def test_tracking_learns_the_gap_and_skips_late_samples():
    st = _state()
    assert _track_samples(st, _samples((180, 1), (120, 1), (60, 1)), NOW) is None
    assert (st.last_sample_at, st.sample_interval_s) == (NOW - timedelta(seconds=60), 60)
    _track_samples(st, _samples((90, 1)), NOW)  # older than the last
    assert (st.last_sample_at, st.sample_interval_s) == (NOW - timedelta(seconds=60), 60)


def test_a_sample_after_silence_clears_it_and_says_so():
    st = _state(last_sample_at=NOW - timedelta(minutes=20), sample_interval_s=60,
                no_data_since=NOW - timedelta(minutes=15))
    note = _track_samples(st, _samples((0, 42.0)), NOW)
    assert st.no_data_since is None and st.state == AlertStateEnum.firing  # still firing
    assert st.sample_interval_s == 90       # the outage counted as three gaps, at most
    assert note == "data resumed after 15m silent"


_RULE = NS(id="r", operator=AlertOperator.gt, threshold=90.0,
           consecutive_failures_required=1, consecutive_ok_required=1)


@pytest.mark.parametrize("before,value,want_event,want_notice", [
    # Back, and breaching: one event, silent to firing (#5442).
    (AlertStateEnum.inactive, 95.0, ("no_data", "firing"), "firing"),
    # Back, and fine: one event, silent back to how it was.
    (AlertStateEnum.inactive, 10.0, ("no_data", "inactive"), "data_resumed"),
    # A firing alert that went silent comes back fine: it resolves, from silent.
    (AlertStateEnum.firing, 10.0, ("no_data", "resolved"), "resolved"),
    # ... or still breaching: it stays firing, and only the resume is news.
    (AlertStateEnum.firing, 95.0, ("no_data", "firing"), "data_resumed"),
])
def test_a_resumed_series_gets_one_event_for_what_its_first_sample_did(
        before, value, want_event, want_notice):
    session, st = _Session(), _state(state=before)
    said = "data resumed after 15m silent"
    out = asyncio.run(_evaluate_rule(session, _RULE, st, value, NOW, resumed=said))
    [event] = session.added
    assert (event.from_state, event.to_state) == want_event
    assert event.note == said and event.transitioned_at == NOW
    assert [(n, e) for n, e, _ in out] == [(want_notice, event.id)]
    # Without a resume, a sample that changes nothing records nothing.
    session = _Session()
    assert asyncio.run(_evaluate_rule(session, _RULE, _state(state=st.state), value, NOW)) == []
    assert session.added == []


def test_muted_by_names_the_window_that_covers_the_rule():
    from steward.core.alerts import _muted_by
    rule = NS(source_module="host_agent", resource_name="nas")
    windows = [NS(name="other", scope_source="docker", scope_resource=None),
               NS(name="nas work", scope_source="host_agent", scope_resource="nas")]
    assert _muted_by(windows, rule) == "notification muted: maintenance window nas work"
    assert _muted_by(windows[:1], rule) is None
    assert _muted_by([NS(name="all", scope_source=None, scope_resource=None)], rule) == (
        "notification muted: maintenance window all")


def test_a_flushed_backlog_counts_as_heard_when_it_arrives():
    """#5720: an agent's backlog, flushed a couple of samples a post, is hours
    old. Judged by its newest sample, the series read silent again between
    posts and resumed twice. It counts as heard when Steward heard it, and the
    rest of the backlog reads as late."""
    st = _state(last_sample_at=NOW - timedelta(hours=3), sample_interval_s=60,
                no_data_since=NOW - timedelta(hours=2, minutes=55))
    note = _track_samples(st, _samples((2 * 3600 + 60, 5.0), (2 * 3600, 5.0)), NOW)
    assert note == "data resumed after 2h 55m silent"
    assert st.last_sample_at == NOW
    assert silent_since(st.last_sample_at, st.sample_interval_s, NOW, NOW) is None
    # The next post of the same backlog is late: no gap, no second resume.
    assert _track_samples(st, _samples((2 * 3600 - 60, 5.0)), NOW) is None
    assert st.last_sample_at == NOW