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,
    runtime_components_ready,
)


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"
    )


def test_runtime_readiness_requires_all_workers_and_reconciler() -> None:
    assert runtime_components_ready(
        scheduler_enabled=True,
        scheduler_running=True,
        active_worker_count=4,
        expected_worker_count=4,
        reconciler_running=True,
    )
    assert not runtime_components_ready(
        scheduler_enabled=True,
        scheduler_running=True,
        active_worker_count=3,
        expected_worker_count=4,
        reconciler_running=True,
    )
    assert not runtime_components_ready(
        scheduler_enabled=True,
        scheduler_running=True,
        active_worker_count=4,
        expected_worker_count=4,
        reconciler_running=False,
    )


def test_runtime_readiness_allows_intentionally_disabled_scheduler() -> None:
    assert runtime_components_ready(
        scheduler_enabled=False,
        scheduler_running=False,
        active_worker_count=1,
        expected_worker_count=1,
        reconciler_running=True,
    )