MediaCrawler/tests/test_schedule.py

# -*- coding: utf-8 -*-
"""Tests for monitor task schedule arithmetic.

Everything here is timezone-local, matching the implementation: the container is
pinned to the operator's zone via TZ, so the tests build their expectations from
naive local datetimes too and stay correct wherever they run.
"""

from datetime import datetime, time, timedelta

import pytest

from api.monitor import schedule


def _ms(moment: datetime) -> int:
    return int(moment.timestamp() * 1000)


def test_interval_is_now_plus_the_interval():
    after = _ms(datetime(2026, 10, 7, 9, 0))

    nxt = schedule.next_occurrence(
        mode=schedule.MODE_INTERVAL,
        interval_minutes=120,
        hours=[],
        days=[],
        minute=0,
        after_ms=after,
    )

    assert nxt == after + 120 * 60_000


def test_daily_takes_the_soonest_remaining_time_today():
    after = _ms(datetime(2026, 10, 7, 8, 0))

    nxt = schedule.next_occurrence(
        mode=schedule.MODE_DAILY,
        interval_minutes=60,
        hours=[18, 9],  # deliberately unsorted
        days=[],
        minute=30,
        after_ms=after,
    )

    assert nxt == _ms(datetime(2026, 10, 7, 9, 30))


def test_daily_rolls_over_to_tomorrow_once_every_time_has_passed():
    after = _ms(datetime(2026, 10, 7, 20, 0))

    nxt = schedule.next_occurrence(
        mode=schedule.MODE_DAILY,
        interval_minutes=60,
        hours=[9, 18],
        days=[],
        minute=30,
        after_ms=after,
    )

    assert nxt == _ms(datetime(2026, 10, 8, 9, 30))


def test_a_slot_exactly_now_belongs_to_the_next_day():
    """Strictly-after, so the run that just fired does not fire again.

    The scheduler advances with ``after_ms`` set to the moment the run started,
    which is at or just past the slot -- if the comparison were inclusive it would
    pick the same slot back up and loop.
    """
    after = _ms(datetime(2026, 10, 7, 9, 30))

    nxt = schedule.next_occurrence(
        mode=schedule.MODE_DAILY,
        interval_minutes=60,
        hours=[9],
        days=[],
        minute=30,
        after_ms=after,
    )

    assert nxt == _ms(datetime(2026, 10, 8, 9, 30))


def test_weekly_jumps_to_the_next_selected_weekday():
    after_dt = datetime(2026, 10, 7, 8, 0)
    target = (after_dt.weekday() + 2) % 7

    nxt = schedule.next_occurrence(
        mode=schedule.MODE_WEEKLY,
        interval_minutes=60,
        hours=[10],
        days=[target],
        minute=0,
        after_ms=_ms(after_dt),
    )

    expected = datetime.combine((after_dt + timedelta(days=2)).date(), time(10, 0))
    assert nxt == _ms(expected)


def test_weekly_can_fire_later_the_same_day():
    after_dt = datetime(2026, 10, 7, 8, 0)

    nxt = schedule.next_occurrence(
        mode=schedule.MODE_WEEKLY,
        interval_minutes=60,
        hours=[21],
        days=[after_dt.weekday()],
        minute=15,
        after_ms=_ms(after_dt),
    )

    assert nxt == _ms(datetime.combine(after_dt.date(), time(21, 15)))


def test_weekly_without_weekdays_means_every_day():
    """Otherwise an empty day selection would match nothing and never fire."""
    after = _ms(datetime(2026, 10, 7, 8, 0))

    nxt = schedule.next_occurrence(
        mode=schedule.MODE_WEEKLY,
        interval_minutes=60,
        hours=[9],
        days=[],
        minute=0,
        after_ms=after,
    )

    assert nxt == _ms(datetime(2026, 10, 7, 9, 0))


def test_a_clock_schedule_with_no_times_can_never_fire():
    """Returned as None so the caller can park the task instead of leaving it due."""
    nxt = schedule.next_occurrence(
        mode=schedule.MODE_DAILY,
        interval_minutes=60,
        hours=[],
        days=[],
        minute=0,
        after_ms=_ms(datetime(2026, 10, 7, 8, 0)),
    )

    assert nxt is None


@pytest.mark.parametrize(
    "raw, expected",
    [
        ("9,18", [9, 18]),
        ("18,9", [9, 18]),  # stored order is not guaranteed
        ("9,9,9", [9]),
        ("", []),
        (None, []),
        ("9, 18 ", [9, 18]),
        ("9,99,-1,abc,", [9]),  # junk is dropped, never raised
    ],
)
def test_parse_hours_is_forgiving(raw, expected):
    assert schedule.parse_hours(raw) == expected


def test_parse_days_accepts_the_whole_week():
    assert schedule.parse_days("0,1,2,3,4,5,6") == [0, 1, 2, 3, 4, 5, 6]
    assert schedule.parse_days("7,-1") == []


@pytest.mark.parametrize(
    "mode, interval, hours, days, minute, expected",
    [
        (schedule.MODE_INTERVAL, 360, [], [], 0, "每 6 小时"),
        (schedule.MODE_INTERVAL, 1440, [], [], 0, "每 1 天"),
        (schedule.MODE_INTERVAL, 45, [], [], 0, "每 45 分钟"),
        (schedule.MODE_DAILY, 60, [9, 18], [], 30, "每天 09:30、18:30"),
        (schedule.MODE_WEEKLY, 60, [10], [0, 1, 2, 3, 4], 0, "周一、周二、周三、周四、周五 10:00"),
        (schedule.MODE_WEEKLY, 60, [10], [], 0, "每天 10:00"),
        (schedule.MODE_DAILY, 60, [], [], 0, "未设置时间"),
    ],
)
def test_describe(mode, interval, hours, days, minute, expected):
    assert (
        schedule.describe(
            mode=mode, interval_minutes=interval, hours=hours, days=days, minute=minute
        )
        == expected
    )


def test_format_round_trips_through_parse():
    hours = [9, 12, 18]
    days = [0, 4]
    assert schedule.parse_hours(schedule.format_hours(hours)) == hours
    assert schedule.parse_days(schedule.format_days(days)) == days