test_notify_decision.py

"""Tenant-aware notify decision: thresholds, quiet hours, channels (roadmap M4, ADR-0019)."""
from __future__ import annotations

import pytest
from hypothesis import given
from hypothesis import strategies as st

from openjobradar.notify import QuietHours, decide, enabled_channels

# --- QuietHours --------------------------------------------------------------


@pytest.mark.parametrize("bad_hour", [-1, 24, 100, "5", 5.0, True])
def test_quiet_hours_rejects_out_of_range_or_non_int_hours(bad_hour) -> None:
    with pytest.raises(ValueError):
        QuietHours(bad_hour, 6)


def test_quiet_hours_same_start_and_end_means_no_restriction() -> None:
    window = QuietHours(9, 9)
    assert all(not window.contains(h) for h in range(24))


def test_quiet_hours_non_wrapping_window() -> None:
    window = QuietHours(1, 6)
    assert window.contains(1) and window.contains(5)
    assert not window.contains(0) and not window.contains(6)


def test_quiet_hours_wrapping_window_past_midnight() -> None:
    window = QuietHours(22, 6)
    assert window.contains(23) and window.contains(0) and window.contains(5)
    assert not window.contains(6) and not window.contains(21)


def test_next_open_hour_skips_the_whole_window() -> None:
    window = QuietHours(22, 6)
    assert window.next_open_hour(22) == 6
    assert window.next_open_hour(23) == 6
    assert window.next_open_hour(5) == 6


def test_next_open_hour_wraps_around_midnight_when_needed() -> None:
    window = QuietHours(20, 2)  # quiet 20:00 - 01:59
    assert window.next_open_hour(23) == 2


@given(st.integers(0, 22), st.integers(0, 23), st.integers(0, 23))
def test_next_open_hour_is_always_outside_the_window(start, width_offset, at_hour) -> None:
    end = (start + 1 + width_offset) % 24
    if end == start:
        return  # QuietHours forbids a zero-width window meaning "always quiet"; skip that draw
    window = QuietHours(start, end)
    if not window.contains(at_hour):
        return  # only meaningful when we start inside the window
    assert not window.contains(window.next_open_hour(at_hour))


# --- channels ------------------------------------------------------------------


def test_email_enabled_by_default_matching_platform_defaults() -> None:
    assert enabled_channels(None) == ("email",)
    assert enabled_channels({}) == ("email",)


def test_email_can_be_explicitly_disabled() -> None:
    assert enabled_channels({"email": {"enabled": False}}) == ()


def test_webhook_is_opt_in() -> None:
    assert enabled_channels({"webhook": {"url": "https://example.com/hook"}}) == ("email",)
    assert enabled_channels({"webhook": {"enabled": True, "url": "https://example.com/hook"}}) == (
        "email",
        "webhook",
    )


def test_webhook_none_is_treated_as_disabled() -> None:
    assert enabled_channels({"webhook": None}) == ("email",)


# --- decide ----------------------------------------------------------------------


def test_score_below_digest_threshold_is_suppressed() -> None:
    result = decide(40, alert_threshold=80, digest_threshold=60, channels_config={}, dry_run=False)
    assert result.verdict == "suppressed"
    assert result.channels == ()
    assert result.deferred_until_hour_utc is None


def test_score_between_thresholds_queues_digest() -> None:
    result = decide(65, alert_threshold=80, digest_threshold=60, channels_config={}, dry_run=False)
    assert result.verdict == "digest"
    assert result.channels == ("email",)


def test_score_at_or_above_alert_threshold_alerts() -> None:
    result = decide(80, alert_threshold=80, digest_threshold=60, channels_config={}, dry_run=False)
    assert result.verdict == "alert"
    assert result.deferred_until_hour_utc is None


def test_alert_deferred_when_it_lands_in_quiet_hours() -> None:
    result = decide(
        90,
        alert_threshold=80,
        digest_threshold=60,
        channels_config={},
        dry_run=False,
        quiet_hours=QuietHours(22, 6),
        at_hour_utc=23,
    )
    assert result.verdict == "alert"
    assert result.deferred_until_hour_utc == 6
    assert "deferred" in result.reason


def test_alert_outside_quiet_hours_is_not_deferred() -> None:
    result = decide(
        90,
        alert_threshold=80,
        digest_threshold=60,
        channels_config={},
        dry_run=False,
        quiet_hours=QuietHours(22, 6),
        at_hour_utc=12,
    )
    assert result.deferred_until_hour_utc is None


def test_digest_verdict_ignores_quiet_hours_entirely() -> None:
    result = decide(
        65,
        alert_threshold=80,
        digest_threshold=60,
        channels_config={},
        dry_run=False,
        quiet_hours=QuietHours(0, 23),  # quiet almost all day
        at_hour_utc=12,
    )
    assert result.verdict == "digest"
    assert result.deferred_until_hour_utc is None


def test_dry_run_passes_through_regardless_of_verdict() -> None:
    for score in (10, 65, 90):
        result = decide(score, alert_threshold=80, digest_threshold=60, channels_config={}, dry_run=True)
        assert result.is_dry_run is True
    # never-silent-death: a high-scoring dry-run posting is still an "alert", just a preview
    alert = decide(95, alert_threshold=80, digest_threshold=60, channels_config={}, dry_run=True)
    assert alert.verdict == "alert" and alert.is_dry_run is True


def test_quiet_hours_without_at_hour_utc_rejected() -> None:
    with pytest.raises(ValueError):
        decide(
            90,
            alert_threshold=80,
            digest_threshold=60,
            channels_config={},
            dry_run=False,
            quiet_hours=QuietHours(22, 6),
        )


@pytest.mark.parametrize("bad_score", [-1, 101, "80", 50.5])
def test_invalid_score_rejected(bad_score) -> None:
    with pytest.raises(ValueError):
        decide(bad_score, alert_threshold=80, digest_threshold=60, channels_config={}, dry_run=False)


def test_inverted_thresholds_rejected() -> None:
    with pytest.raises(ValueError):
        decide(90, alert_threshold=50, digest_threshold=60, channels_config={}, dry_run=False)


@given(
    st.integers(0, 100),
    st.integers(0, 100),
    st.integers(0, 100),
)
def test_verdict_is_monotonic_nondecreasing_in_score(digest_threshold, gap, score_offset) -> None:
    alert_threshold = min(100, digest_threshold + gap)
    rank = {"suppressed": 0, "digest": 1, "alert": 2}
    scores = sorted({0, digest_threshold, alert_threshold, 100, min(100, score_offset)})
    verdicts = [
        decide(
            s,
            alert_threshold=alert_threshold,
            digest_threshold=digest_threshold,
            channels_config={},
            dry_run=False,
        ).verdict
        for s in scores
    ]
    ranks = [rank[v] for v in verdicts]
    assert ranks == sorted(ranks)