Octop/tests/unit/cron/test_cron_trigger.py

"""tests/unit/test_cron_trigger.py"""

from __future__ import annotations

import datetime as dt

import pytest
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.date import DateTrigger
from apscheduler.triggers.interval import IntervalTrigger

from octop.infra.cron.trigger import build_trigger, one_shot_disposition
from octop.infra.errors import ErrorCode, OctopError


def test_interval_seconds():
    trig = build_trigger("interval:60")
    assert isinstance(trig, IntervalTrigger)
    assert trig.interval == dt.timedelta(seconds=60)


def test_cron_expression():
    trig = build_trigger("cron:0 9 * * *")
    assert isinstance(trig, CronTrigger)


def _next_weekdays(trig: CronTrigger, *, count: int) -> list[int]:
    next_fire = trig.get_next_fire_time(None, dt.datetime(2026, 8, 8, 12, tzinfo=trig.timezone))
    actual: list[int] = []
    while next_fire is not None and len(actual) < count:
        actual.append(next_fire.weekday())
        next_fire = trig.get_next_fire_time(next_fire, next_fire)
    return actual


@pytest.mark.parametrize(
    ("weekday", "expected_weekday"),
    [
        ("0", 6),  # Sunday
        ("7", 6),  # Sunday (Unix alias)
        ("1", 0),  # Monday
        ("2", 1),  # Tuesday
        ("3", 2),  # Wednesday
        ("4", 3),  # Thursday
        ("5", 4),  # Friday
        ("6", 5),  # Saturday
        ("sun", 6),
        ("mon", 0),
        ("fri", 4),
        ("sat", 5),
        ("SUN", 6),
        ("Mon", 0),
    ],
)
def test_cron_weekday_uses_unix_semantics(weekday: str, expected_weekday: int):
    trig = build_trigger(f"cron:0 9 * * {weekday}")
    assert isinstance(trig, CronTrigger)
    assert _next_weekdays(trig, count=1) == [expected_weekday]


@pytest.mark.parametrize(
    ("weekday", "expected_weekdays"),
    [
        ("1-5", [0, 1, 2, 3, 4]),  # Mon-Fri
        ("mon-fri", [0, 1, 2, 3, 4]),
        ("0-6", [6, 0, 1, 2, 3, 4, 5]),  # every day via numeric range
        ("5-7", [6, 4, 5]),  # Fri-Sun; after Sat noon → Sun,Fri,Sat
        ("*/2", [6, 1, 3, 5]),  # Sun,Tue,Thu,Sat
        ("1-5/2", [0, 2, 4]),  # Mon,Wed,Fri
        ("1/2", [6, 0, 2, 4]),  # Mon,Wed,Fri,Sun; after Sat noon → Sun first
        ("0,6", [6, 5]),  # Sun,Sat
        ("1,3,5", [0, 2, 4]),  # Mon,Wed,Fri
        ("0,7", [6]),  # both Sunday aliases; deduped
        ("sun,sat", [6, 5]),
        ("0,sat", [6, 5]),  # mixed numeric + name
    ],
)
def test_cron_weekday_ranges_use_unix_semantics(weekday: str, expected_weekdays: list[int]):
    trig = build_trigger(f"cron:0 9 * * {weekday}")
    assert isinstance(trig, CronTrigger)
    assert _next_weekdays(trig, count=len(expected_weekdays)) == expected_weekdays


def test_date_iso():
    trig = build_trigger("date:2026-12-31T09:00:00")
    assert isinstance(trig, DateTrigger)


def test_unknown_kind_raises():
    with pytest.raises(OctopError) as ei:
        build_trigger("webhook:foo")
    assert ei.value.code is ErrorCode.CRON_TRIGGER_INVALID


def test_missing_colon_raises():
    with pytest.raises(OctopError):
        build_trigger("interval60")


def test_malformed_cron_raises():
    with pytest.raises(OctopError):
        build_trigger("cron:not a cron")


@pytest.mark.parametrize(
    "weekday",
    [
        "8",
        "5-1",
        "1/0",
        "*/0",
        "foo",
    ],
)
def test_cron_rejects_invalid_weekday(weekday: str):
    with pytest.raises(OctopError):
        build_trigger(f"cron:0 9 * * {weekday}")


@pytest.mark.parametrize("spec", ["interval:0", "interval:-5", "interval:-1"])
def test_interval_rejects_non_positive_seconds(spec: str) -> None:
    """A zero or negative interval is an unbounded hot loop, not a schedule."""
    with pytest.raises(OctopError, match="positive integer"):
        build_trigger(spec)


def test_cron_trigger_uses_requested_timezone() -> None:
    """A built cron trigger must run on the requested zone, not the host OS zone.

    APScheduler only injects the scheduler timezone for string trigger specs; a pre-built
    ``CronTrigger`` keeps whatever zone it was constructed with, and falls back to the host
    local zone when none is given.
    """
    trig = build_trigger("cron:0 9 * * *", timezone="America/New_York")
    assert isinstance(trig, CronTrigger)
    assert str(trig.timezone) == "America/New_York"

    fire = trig.get_next_fire_time(None, dt.datetime(2026, 1, 1, tzinfo=dt.UTC))
    # 09:00 in New York (EST, UTC-5) is 14:00 UTC.
    assert fire == dt.datetime(2026, 1, 1, 14, 0, tzinfo=dt.UTC)


def test_date_trigger_uses_requested_timezone() -> None:
    """Naive ``date:`` ISO times are wall-clock in the requested zone, not the host OS zone."""
    trig = build_trigger("date:2026-12-31T09:00:00", timezone="Asia/Shanghai")
    assert isinstance(trig, DateTrigger)
    assert (trig.run_date.hour, trig.run_date.utcoffset()) == (9, dt.timedelta(hours=8))


def test_interval_one_second_is_accepted() -> None:
    trig = build_trigger("interval:1")
    assert isinstance(trig, IntervalTrigger)
    assert trig.interval == dt.timedelta(seconds=1)


def test_one_shot_future_is_scheduled() -> None:
    run_at = dt.datetime(2026, 10, 18, 9, tzinfo=dt.UTC)
    now = dt.datetime(2026, 10, 17, 9, tzinfo=dt.UTC)
    assert one_shot_disposition(run_at, now=now, last_run_at=None) == "schedule"


def test_one_shot_within_grace_and_never_run_is_scheduled() -> None:
    run_at = dt.datetime(2026, 10, 17, 8, tzinfo=dt.UTC)
    now = dt.datetime(2026, 10, 17, 9, tzinfo=dt.UTC)
    assert one_shot_disposition(run_at, now=now, last_run_at=None) == "schedule"


def test_one_shot_past_grace_and_never_run_is_missed() -> None:
    run_at = dt.datetime(2026, 10, 1, 9, tzinfo=dt.UTC)
    now = dt.datetime(2026, 10, 17, 9, tzinfo=dt.UTC)
    assert one_shot_disposition(run_at, now=now, last_run_at=None) == "miss"


def test_one_shot_already_ran_for_slot_is_closed() -> None:
    run_at = dt.datetime(2026, 10, 17, 8, tzinfo=dt.UTC)
    now = dt.datetime(2026, 10, 17, 9, tzinfo=dt.UTC)
    assert one_shot_disposition(run_at, now=now, last_run_at=int(run_at.timestamp())) == "close"


def test_one_shot_early_run_closes_after_the_slot_passes() -> None:
    run_at = dt.datetime(2026, 10, 17, 8, tzinfo=dt.UTC)
    now = dt.datetime(2026, 10, 17, 9, tzinfo=dt.UTC)
    early = int(dt.datetime(2026, 10, 17, 7, tzinfo=dt.UTC).timestamp())
    assert one_shot_disposition(run_at, now=now, last_run_at=early) == "close"


def test_one_shot_early_run_can_be_rearmed_while_the_slot_is_still_future() -> None:
    run_at = dt.datetime(2026, 10, 18, 9, tzinfo=dt.UTC)
    now = dt.datetime(2026, 10, 17, 9, tzinfo=dt.UTC)
    assert one_shot_disposition(run_at, now=now, last_run_at=int(now.timestamp())) == "schedule"