test_scheduling.py

"""Cadence policy and due-work index (AD4, ADR-0005, ADR-0018, roadmap M4)."""
from __future__ import annotations

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

from openjobradar.scheduling.cadence import (
    TIER_COLD,
    TIER_HOT,
    TIER_WARM,
    TIERS,
    CadencePolicy,
    UnknownTierError,
    get_tier,
    next_interval_seconds,
)
from openjobradar.scheduling.due_work import (
    DueWorkIndex,
    InvalidWorkKeyError,
    UnknownWorkKindError,
    advance,
)
from openjobradar.tenancy import TenantContext

ALICE = TenantContext.from_sub("alice-sub-0000001")
BOB = TenantContext.from_sub("bob-sub-00000002")


# --- CadencePolicy -----------------------------------------------------------


def test_named_tiers_are_internally_consistent() -> None:
    for policy in TIERS.values():
        assert policy.min_interval_s <= policy.base_interval_s <= policy.max_interval_s
        assert policy.backoff_factor >= 1.0


@pytest.mark.parametrize(
    "kwargs",
    [
        dict(base_interval_s=100, min_interval_s=200, max_interval_s=300),  # min > base
        dict(base_interval_s=500, min_interval_s=100, max_interval_s=300),  # base > max
        dict(base_interval_s=100, min_interval_s=100, max_interval_s=100, backoff_factor=0.5),
    ],
)
def test_invalid_policy_rejected(kwargs: dict) -> None:
    with pytest.raises(ValueError):
        CadencePolicy("bad", **kwargs)


def test_get_tier_unknown_name_fails_closed() -> None:
    with pytest.raises(UnknownTierError):
        get_tier("scorching")


def test_interval_snaps_to_base_on_a_hit() -> None:
    assert next_interval_seconds(TIER_HOT, consecutive_empty_polls=0) == TIER_HOT.base_interval_s


def test_interval_backs_off_and_clamps_to_max() -> None:
    one_miss = next_interval_seconds(TIER_HOT, consecutive_empty_polls=1)
    assert one_miss == int(TIER_HOT.base_interval_s * TIER_HOT.backoff_factor)
    assert next_interval_seconds(TIER_HOT, consecutive_empty_polls=50) == TIER_HOT.max_interval_s


def test_negative_miss_count_rejected() -> None:
    with pytest.raises(ValueError):
        next_interval_seconds(TIER_WARM, consecutive_empty_polls=-1)


def test_digest_tier_has_no_adaptive_range() -> None:
    from openjobradar.scheduling.cadence import DIGEST_WEEKLY

    for n in (0, 1, 10):
        assert next_interval_seconds(DIGEST_WEEKLY, consecutive_empty_polls=n) == DIGEST_WEEKLY.base_interval_s


@composite
def policies(draw):
    min_i = draw(st.integers(1, 10_000))
    base_i = draw(st.integers(min_i, min_i + 10_000))
    max_i = draw(st.integers(base_i, base_i + 50_000))
    factor = draw(st.floats(1.0, 5.0, allow_nan=False, allow_infinity=False))
    return CadencePolicy("property", base_i, min_i, max_i, factor)


@given(policies(), st.integers(0, 30))
def test_interval_always_within_policy_bounds(policy: CadencePolicy, n: int) -> None:
    result = next_interval_seconds(policy, consecutive_empty_polls=n)
    assert policy.min_interval_s <= result <= policy.max_interval_s


@given(policies())
def test_interval_is_monotonic_nondecreasing_in_misses(policy: CadencePolicy) -> None:
    values = [next_interval_seconds(policy, consecutive_empty_polls=n) for n in range(6)]
    assert values == sorted(values)


# --- DueWorkIndex --------------------------------------------------------------


def test_schedule_requires_a_constructed_tenant_context() -> None:
    index = DueWorkIndex()
    with pytest.raises(TypeError):
        index.schedule("alice-sub-0000001", "poll_org", "acme", "2026-08-23T00:00:00Z")  # type: ignore[arg-type]


def test_schedule_rejects_unknown_work_kind() -> None:
    index = DueWorkIndex()
    with pytest.raises(UnknownWorkKindError):
        index.schedule(ALICE, "send_rocket", "acme", "2026-08-23T00:00:00Z")


@pytest.mark.parametrize("bad_key", ["", "-leading-dash", "has space", "trailing?question", "$$$"])
def test_schedule_rejects_malformed_work_key(bad_key: str) -> None:
    index = DueWorkIndex()
    with pytest.raises(InvalidWorkKeyError):
        index.schedule(ALICE, "poll_org", bad_key, "2026-08-23T00:00:00Z")


def test_schedule_accepts_the_legal_work_key_charset() -> None:
    index = DueWorkIndex()
    entry = index.schedule(ALICE, "poll_org", "acme-co.jobs:us-remote@v2+1", "2026-08-23T00:00:00Z")
    assert entry.work_key == "acme-co.jobs:us-remote@v2+1"


@pytest.mark.parametrize("bad_ts", ["2026-08-23", "not-a-date", "2026-08-23 00:00:00", ""])
def test_schedule_rejects_malformed_timestamp(bad_ts: str) -> None:
    index = DueWorkIndex()
    with pytest.raises(ValueError):
        index.schedule(ALICE, "poll_org", "acme", bad_ts)


def test_get_and_unschedule_roundtrip() -> None:
    index = DueWorkIndex()
    entry = index.schedule(ALICE, "poll_org", "acme", "2026-08-23T00:00:00Z")
    assert index.get(ALICE, "poll_org", "acme") == entry
    assert index.unschedule(ALICE, "poll_org", "acme") is True
    assert index.get(ALICE, "poll_org", "acme") is None
    assert index.unschedule(ALICE, "poll_org", "acme") is False


def test_due_before_scans_across_tenants_ordered_oldest_first() -> None:
    index = DueWorkIndex()
    index.schedule(ALICE, "poll_org", "acme", "2026-08-23T06:00:00Z")
    index.schedule(BOB, "poll_org", "globex", "2026-08-23T01:00:00Z")
    index.schedule(ALICE, "digest", "weekly", "2026-08-24T00:00:00Z")

    due = index.due_before("2026-08-23T12:00:00Z")
    assert [e.user_id for e in due] == [BOB.user_id, ALICE.user_id]  # bob's earlier due time first


def test_due_before_excludes_not_yet_due() -> None:
    index = DueWorkIndex()
    index.schedule(ALICE, "poll_org", "acme", "2026-08-30T00:00:00Z")
    assert index.due_before("2026-08-23T00:00:00Z") == []


def test_schedule_rejects_negative_consecutive_empty() -> None:
    index = DueWorkIndex()
    with pytest.raises(ValueError):
        index.schedule(ALICE, "poll_org", "acme", "2026-08-23T00:00:00Z", consecutive_empty=-1)


def test_unschedule_and_get_require_a_constructed_tenant_context() -> None:
    index = DueWorkIndex()
    with pytest.raises(TypeError):
        index.unschedule("alice-sub-0000001", "poll_org", "acme")  # type: ignore[arg-type]
    with pytest.raises(TypeError):
        index.get("alice-sub-0000001", "poll_org", "acme")  # type: ignore[arg-type]


def test_len_reports_total_entry_count_across_tenants() -> None:
    index = DueWorkIndex()
    assert len(index) == 0
    index.schedule(ALICE, "poll_org", "acme", "2026-08-23T00:00:00Z")
    index.schedule(BOB, "digest", "weekly", "2026-08-24T00:00:00Z")
    assert len(index) == 2


def test_advance_resets_miss_counter_and_cadence_on_a_hit() -> None:
    index = DueWorkIndex()
    entry = index.schedule(ALICE, "poll_org", "acme", "2026-08-23T00:00:00Z", consecutive_empty=3)
    updated = advance(entry, found_new=True, policy=TIER_HOT, now="2026-08-23T06:00:00Z")
    assert updated.consecutive_empty == 0
    assert updated.next_due_at == "2026-08-23T06:15:00Z"  # +900s base interval
    assert entry.consecutive_empty == 3  # original is untouched (frozen)


def test_advance_backs_off_cadence_on_a_miss() -> None:
    index = DueWorkIndex()
    entry = index.schedule(ALICE, "poll_org", "acme", "2026-08-23T00:00:00Z")
    updated = advance(entry, found_new=False, policy=TIER_HOT, now="2026-08-23T06:00:00Z")
    assert updated.consecutive_empty == 1
    expected_seconds = int(TIER_HOT.base_interval_s * TIER_HOT.backoff_factor)
    assert updated.next_due_at == f"2026-08-23T06:{expected_seconds // 60:02d}:{expected_seconds % 60:02d}Z"


def test_advance_rejects_malformed_now() -> None:
    entry = DueWorkIndex().schedule(ALICE, "poll_org", "acme", "2026-08-23T00:00:00Z")
    with pytest.raises(ValueError):
        advance(entry, found_new=True, policy=TIER_COLD, now="tomorrow")