SupplyAgent/tests/supply_infra/pipeline/test_health.py

from __future__ import annotations

from datetime import timedelta

from supply_infra.config import InfraSettings
from supply_infra.pipeline.dates import china_now
from supply_infra.pipeline.health import run_alert_level


def _settings() -> InfraSettings:
    return InfraSettings(_env_file=None)


def test_alert_level_escalates_against_next_schedule_deadline() -> None:
    settings = _settings()
    now = china_now()

    assert (
        run_alert_level(
            status="running",
            started_at=now - timedelta(hours=13),
            created_at=now - timedelta(hours=13),
            deadline_at=now + timedelta(hours=5),
            now=now,
            settings=settings,
        )
        == "warning"
    )
    assert (
        run_alert_level(
            status="running",
            started_at=now - timedelta(hours=1),
            created_at=now - timedelta(hours=1),
            deadline_at=now + timedelta(minutes=90),
            now=now,
            settings=settings,
        )
        == "critical"
    )
    assert (
        run_alert_level(
            status="blocked_previous_run",
            started_at=None,
            created_at=now - timedelta(hours=1),
            deadline_at=now + timedelta(minutes=20),
            now=now,
            settings=settings,
        )
        == "final_warning"
    )


def test_terminal_run_has_no_duration_alert() -> None:
    now = china_now()
    assert (
        run_alert_level(
            status="failed",
            started_at=now - timedelta(days=2),
            created_at=now - timedelta(days=2),
            deadline_at=now - timedelta(days=1),
            now=now,
            settings=_settings(),
        )
        is None
    )


def test_manual_run_without_deadline_only_uses_duration_warning() -> None:
    now = china_now()
    settings = _settings()

    assert (
        run_alert_level(
            status="running",
            started_at=now - timedelta(hours=1),
            created_at=now - timedelta(hours=1),
            deadline_at=None,
            now=now,
            settings=settings,
        )
        is None
    )
    assert (
        run_alert_level(
            status="running",
            started_at=now - timedelta(hours=13),
            created_at=now - timedelta(hours=13),
            deadline_at=None,
            now=now,
            settings=settings,
        )
        == "warning"
    )