praxis/tests/test_cohort_nightly.py

"""Nightly reconciliation + hook integration tests (TASK-07-06) — mocked PgStore.

Covers: scheduler timing (seconds until 03:00 CT), reconciliation recomputes
all windows, hook failure + nightly reconciliation = correct final state,
R-DASH-04 (nightly failure logs + retries next night).
"""

from __future__ import annotations

import datetime as _dt
from unittest.mock import AsyncMock, MagicMock

import pytest

from server.cohort.nightly import (
    CT,
    NightlyScheduler,
    seconds_until_next_03_ct,
)


# ── Scheduler timing ───────────────────────────────────────────────────────


def test_seconds_until_next_03_ct_future_today():
    # 01:00 CT → next 03:00 CT is in 2h
    now = _dt.datetime(2026, 8, 4, 1, 0, tzinfo=CT)
    secs = seconds_until_next_03_ct(now)
    assert 7190 <= secs <= 7200  # ~2h


def test_seconds_until_next_03_ct_past_today_wraps_tomorrow():
    # 04:00 CT → next 03:00 CT is tomorrow (23h)
    now = _dt.datetime(2026, 8, 4, 4, 0, tzinfo=CT)
    secs = seconds_until_next_03_ct(now)
    assert 82790 <= secs <= 82810  # ~23h


def test_seconds_until_next_03_ct_exactly_03_rolls_to_tomorrow():
    now = _dt.datetime(2026, 8, 4, 3, 0, 0, tzinfo=CT)
    secs = seconds_until_next_03_ct(now)
    # exactly 03:00:00 → next run is tomorrow (0 secs would mean "now", but
    # the scheduler sleeps then runs, so it must be ~24h)
    assert secs >= 86390  # ~24h


# ── TASK-12-04 (P1+ #6): zoneinfo DST-aware scheduler ───────────────────────


def test_nightly_scheduler_uses_zoneinfo_america_winnipeg():
    """TASK-12-04 (P1+ #6): CT is zoneinfo.ZoneInfo('America/Winnipeg') (DST-aware).

    The v0.4 fixed UTC-5 offset is replaced with ZoneInfo("America/Winnipeg")
    which correctly handles CST (UTC-6) in winter + CDT (UTC-5) in summer.
    """
    from zoneinfo import ZoneInfo

    assert isinstance(CT, ZoneInfo), f"CT should be a ZoneInfo, got {type(CT)}"
    assert str(CT) == "America/Winnipeg", f"CT should be America/Winnipeg, got {CT}"


def test_nightly_scheduler_dst_summer_cdt():
    """TASK-12-04 (P1+ #6): summer (August) → CDT (UTC-5).

    In August 2026, America/Winnipeg is on CDT (UTC-5). A 01:00 local time
    should be 06:00 UTC. The scheduler computes seconds until 03:00 local.
    """
    # 2026-08-04 is summer → CDT (UTC-5).
    now_local = _dt.datetime(2026, 8, 4, 1, 0, tzinfo=CT)
    # 01:00 CDT = 06:00 UTC.
    assert now_local.utcoffset() == _dt.timedelta(hours=-5), (
        f"August should be CDT (UTC-5), got offset {now_local.utcoffset()}"
    )
    secs = seconds_until_next_03_ct(now_local)
    # 01:00 → 03:00 = 2h = 7200s.
    assert 7190 <= secs <= 7200


def test_nightly_scheduler_dst_winter_cst():
    """TASK-12-04 (P1+ #6): winter (January) → CST (UTC-6).

    In January 2027, America/Winnipeg is on CST (UTC-6). A 01:00 local time
    should be 07:00 UTC. The v0.4 fixed UTC-5 offset would have been wrong
    by 1h in winter; the ZoneInfo correctly handles the DST transition.
    """
    # 2027-01-15 is winter → CST (UTC-6).
    now_local = _dt.datetime(2027, 1, 15, 1, 0, tzinfo=CT)
    assert now_local.utcoffset() == _dt.timedelta(hours=-6), (
        f"January should be CST (UTC-6), got offset {now_local.utcoffset()}"
    )
    secs = seconds_until_next_03_ct(now_local)
    # 01:00 → 03:00 = 2h = 7200s.
    assert 7190 <= secs <= 7200


def test_nightly_scheduler_dst_transition_spring_2027():
    """TASK-12-04 (P1+ #6): DST spring forward — 2027-03-14 02:00 → 03:00 CDT.

    On 2027-03-14, DST springs forward at 02:00 local (CST → CDT). The ZoneInfo
    correctly handles the transition (the 02:00 hour is skipped). The scheduler
    should still compute a valid seconds-until-03:00.
    """
    # 2027-03-14 01:00 CST (before spring forward) → 03:00 CDT is 1h later
    # (the 02:00 hour is skipped → 01:59 CST → 03:00 CDT).
    now_local = _dt.datetime(2027, 3, 14, 1, 0, tzinfo=CT)
    secs = seconds_until_next_03_ct(now_local)
    # 01:00 CST → 03:00 CDT is 1h (the 02:00 hour is skipped).
    # The exact value depends on the DST transition; assert it's ≤ 2h.
    assert 0 < secs <= 7200, f"spring-forward seconds should be <= 2h, got {secs}"


# ── Reconciliation recomputes all windows ──────────────────────────────────


class _FakeRecord(dict):
    """Mimics an asyncpg Record — dict(record) returns the dict."""
    pass


def _mock_pg_store_with_events(events):
    store = MagicMock()
    store.upsert_cohort_aggregate = AsyncMock()
    conn = MagicMock()
    rows = [_FakeRecord(e) for e in events]
    conn.fetch = AsyncMock(return_value=rows)
    cm = MagicMock()
    cm.__aenter__ = AsyncMock(return_value=conn)
    cm.__aexit__ = AsyncMock(return_value=None)
    store.pool = MagicMock()
    store.pool.acquire = MagicMock(return_value=cm)
    return store


@pytest.mark.asyncio
async def test_reconcile_recomputes_all_paths():
    events = [
        {"learner_ref": "l1", "path_id": "customer_service", "gate_outcome": "open",
         "rubric_scores_jsonb": '[{"criterion_id":"empathy","score":4.0}]',
         "recorded_at": _dt.datetime.now(_dt.timezone.utc)},
        {"learner_ref": "l2", "path_id": "customer_service", "gate_outcome": "open",
         "rubric_scores_jsonb": '[{"criterion_id":"empathy","score":3.0}]',
         "recorded_at": _dt.datetime.now(_dt.timezone.utc)},
        {"learner_ref": "l3", "path_id": "sales", "gate_outcome": "closed",
         "rubric_scores_jsonb": '[]',
         "recorded_at": _dt.datetime.now(_dt.timezone.utc)},
    ]
    store = _mock_pg_store_with_events(events)
    sched = NightlyScheduler()
    await sched.reconcile_now(store)
    # upserts should cover both paths × multiple metrics
    paths = {c.args[0] for c in store.upsert_cohort_aggregate.call_args_list}
    assert "customer_service" in paths
    assert "sales" in paths
    metrics = {c.args[1] for c in store.upsert_cohort_aggregate.call_args_list}
    assert "sessions_count" in metrics
    assert "active_learners_count" in metrics
    assert "gate_open_rate" in metrics


@pytest.mark.asyncio
async def test_reconcile_suppresses_below_threshold():
    # 3 distinct learners → suppressed
    events = [
        {"learner_ref": f"l{i}", "path_id": "p", "gate_outcome": "open",
         "rubric_scores_jsonb": "[]",
         "recorded_at": _dt.datetime.now(_dt.timezone.utc)}
        for i in range(3)
    ]
    store = _mock_pg_store_with_events(events)
    sched = NightlyScheduler()
    await sched.reconcile_now(store)
    suppressed = [c for c in store.upsert_cohort_aggregate.call_args_list if c.args[6] is True]
    non_suppressed = [c for c in store.upsert_cohort_aggregate.call_args_list if c.args[6] is False]
    assert suppressed, "3 learners must be suppressed"
    assert not non_suppressed, "no cell should be non-suppressed with 3 learners"


@pytest.mark.asyncio
async def test_reconcile_no_events_no_op():
    store = _mock_pg_store_with_events([])
    sched = NightlyScheduler()
    await sched.reconcile_now(store)
    store.upsert_cohort_aggregate.assert_not_called()


# ── Hook failure → nightly reconciles ──────────────────────────────────────


@pytest.mark.asyncio
async def test_hook_failure_then_nightly_reconciles_correct_state():
    """A hook failure leaves no aggregate; the nightly job recomputes from
    mastery_gate_events and produces the correct final state."""
    events = [
        {"learner_ref": f"l{i}", "path_id": "p", "gate_outcome": "open",
         "rubric_scores_jsonb": "[]",
         "recorded_at": _dt.datetime.now(_dt.timezone.utc)}
        for i in range(10)
    ]
    store = _mock_pg_store_with_events(events)
    # Simulate hook failure: upsert raises first time, then nightly runs.
    # (In production the hook + nightly use the same store; here we just
    # verify the nightly path produces correct aggregates independently.)
    sched = NightlyScheduler()
    await sched.reconcile_now(store)
    non_suppressed = [c for c in store.upsert_cohort_aggregate.call_args_list if c.args[6] is False]
    assert non_suppressed, "nightly should produce non-suppressed cells for 10 learners"


# ── R-DASH-04: nightly failure logs + retries ──────────────────────────────


@pytest.mark.asyncio
async def test_r_dash_04_nightly_failure_does_not_crash_scheduler():
    """R-DASH-04: a reconciliation failure logs + the scheduler continues.

    The scheduler loop (_run_loop) catches exceptions from _reconcile and
    retries the next night. We simulate this by invoking the loop with a
    broken store and confirming the loop catches + continues.
    """
    store = MagicMock()
    store.upsert_cohort_aggregate = AsyncMock(side_effect=RuntimeError("db down"))
    store.pool = MagicMock()
    cm = MagicMock()
    cm.__aenter__ = AsyncMock(side_effect=RuntimeError("pool down"))
    cm.__aexit__ = AsyncMock(return_value=None)
    store.pool.acquire = MagicMock(return_value=cm)
    sched = NightlyScheduler()
    import server.cohort.nightly as nightly_mod
    orig = nightly_mod.seconds_until_next_03_ct
    calls = []
    def _fake_secs():
        calls.append(1)
        return 0.01
    nightly_mod.seconds_until_next_03_ct = _fake_secs
    try:
        task = await sched.start(store)
        await _sleep(0.1)
        await sched.stop()
        # The loop ran at least once despite the failure (R-DASH-04).
        assert len(calls) >= 1
    finally:
        nightly_mod.seconds_until_next_03_ct = orig


@pytest.mark.asyncio
async def test_scheduler_start_stop_lifecycle():
    store = _mock_pg_store_with_events([])
    sched = NightlyScheduler()
    # Patch seconds_until to be tiny so the loop is testable.
    import server.cohort.nightly as nightly_mod
    orig = nightly_mod.seconds_until_next_03_ct
    nightly_mod.seconds_until_next_03_ct = lambda: 0.01
    try:
        task = await sched.start(store)
        await _sleep(0.05)
        await sched.stop()
        assert task.cancelled() or task.done()
    finally:
        nightly_mod.seconds_until_next_03_ct = orig


async def _sleep(t: float) -> None:
    import asyncio
    await asyncio.sleep(t)