chongming-utils-task/tests/unit/test_timeouts.py

"""心跳与超时判定的测试。

全部基于 ``FakeClock``：不 sleep、不依赖真实时间，因此既快又不会有偶发失败。
"""

from __future__ import annotations

from datetime import UTC, datetime, timedelta

import pytest

from chongming_task.models import FailureKind, TimeoutScope
from chongming_task.policies import Timeout
from chongming_task.runtime.heartbeat import Heartbeat
from chongming_task.runtime.timeouts import (
    Deadline,
    is_expired,
    next_deadline,
    remaining_seconds,
)
from chongming_task.testing import FakeClock

_T0 = datetime(2026, 1, 1, tzinfo=UTC)


# ---------------------------------------------------------------------------
# 心跳
# ---------------------------------------------------------------------------


def test_heartbeat_starts_at_construction() -> None:
    clock = FakeClock(_T0)
    beat = Heartbeat(clock)
    assert beat.last_beat == _T0
    assert beat.idle_seconds() == 0.0


def test_heartbeat_advances_only_on_beat() -> None:
    """心跳由 checkpoint 驱动，不额外起线程；没 beat 就不该自己变新。"""
    clock = FakeClock(_T0)
    beat = Heartbeat(clock)
    clock.advance(30)
    assert beat.idle_seconds() == 30.0
    beat.beat()
    assert beat.last_beat == clock.now
    assert beat.idle_seconds() == 0.0


def test_heartbeat_staleness_boundary_is_inclusive() -> None:
    clock = FakeClock(_T0)
    beat = Heartbeat(clock)
    clock.advance(59.999)
    assert not beat.is_stale(60)
    clock.advance(0.001)
    assert beat.is_stale(60)


def test_heartbeat_rejects_non_positive_timeout() -> None:
    beat = Heartbeat(FakeClock(_T0))
    with pytest.raises(ValueError, match="必须为正数"):
        beat.is_stale(0)


# ---------------------------------------------------------------------------
# 超时判定
# ---------------------------------------------------------------------------


def test_no_timeout_configured_yields_none() -> None:
    assert next_deadline(Timeout()) is None


def test_start_scope_applies_only_before_started() -> None:
    timeout = Timeout(start=30)
    pending = next_deadline(timeout, submitted_at=_T0)
    assert pending is not None
    assert pending.scope is TimeoutScope.START
    assert pending.at == _T0 + timedelta(seconds=30)

    # 一旦开始执行，start 超时即失效。
    assert next_deadline(timeout, submitted_at=_T0, started_at=_T0) is None


def test_start_scope_needs_submitted_at() -> None:
    assert next_deadline(Timeout(start=30)) is None


def test_attempt_scope_measures_from_started_at() -> None:
    deadline = next_deadline(Timeout(per_attempt=600), started_at=_T0)
    assert deadline is not None
    assert deadline.scope is TimeoutScope.ATTEMPT
    assert deadline.at == _T0 + timedelta(seconds=600)


def test_idle_scope_follows_the_latest_heartbeat() -> None:
    """心跳推进后，idle 截止点必须随之顺延——否则长任务会被误杀。"""
    timeout = Timeout(idle=60)
    first = next_deadline(timeout, started_at=_T0, last_beat_at=_T0)
    assert first is not None
    assert first.at == _T0 + timedelta(seconds=60)

    later = _T0 + timedelta(seconds=45)
    second = next_deadline(timeout, started_at=_T0, last_beat_at=later)
    assert second is not None
    assert second.at == later + timedelta(seconds=60)


def test_idle_scope_requires_the_task_to_have_started() -> None:
    """任务还没跑起来就谈不上"卡住"，否则排队时长会被误判成静默超时。"""
    assert next_deadline(Timeout(idle=60), last_beat_at=_T0) is None


def test_before_start_only_the_start_scope_applies() -> None:
    timeout = Timeout(start=10, per_attempt=600, idle=60)
    deadline = next_deadline(timeout, submitted_at=_T0, last_beat_at=_T0)
    assert deadline is not None
    assert deadline.scope is TimeoutScope.START
    assert deadline.at == _T0 + timedelta(seconds=10)


def test_after_start_attempt_and_idle_compete_and_earliest_wins() -> None:
    """开始执行后只剩两档竞争：宽松的一档不能掩盖已经到期的一档。"""
    timeout = Timeout(per_attempt=600, idle=60)

    fresh_beat = next_deadline(timeout, started_at=_T0, last_beat_at=_T0)
    assert fresh_beat is not None
    assert fresh_beat.scope is TimeoutScope.IDLE

    # 心跳落后很久时，attempt 反而先到期。
    late_beat = _T0 + timedelta(seconds=590)
    stale_beat = next_deadline(timeout, started_at=_T0, last_beat_at=late_beat)
    assert stale_beat is not None
    assert stale_beat.scope is TimeoutScope.ATTEMPT
    assert stale_beat.at == _T0 + timedelta(seconds=600)


def test_deadline_maps_to_failure_kind() -> None:
    """超时是失败原因，不是生命周期状态；映射必须稳定。"""
    assert Deadline(TimeoutScope.START, _T0).kind is FailureKind.TIMEOUT_START
    assert Deadline(TimeoutScope.ATTEMPT, _T0).kind is FailureKind.TIMEOUT_ATTEMPT
    assert Deadline(TimeoutScope.IDLE, _T0).kind is FailureKind.TIMEOUT_IDLE


def test_is_expired_boundary_is_inclusive() -> None:
    deadline = Deadline(TimeoutScope.ATTEMPT, _T0)
    assert not is_expired(deadline, now=_T0 - timedelta(microseconds=1))
    assert is_expired(deadline, now=_T0)


def test_remaining_seconds_goes_negative_after_expiry() -> None:
    """调度器要把负数当成"立刻到期"，而不是交给 sleep——sleep(-1) 会抛异常。"""
    deadline = Deadline(TimeoutScope.ATTEMPT, _T0)
    assert remaining_seconds(deadline, now=_T0 - timedelta(seconds=5)) == 5.0
    assert remaining_seconds(deadline, now=_T0 + timedelta(seconds=5)) == -5.0