SupplyAgent/tests/supply_infra/pipeline/test_preflight.py

from __future__ import annotations

from supply_infra.config import InfraSettings
from supply_infra.pipeline.preflight import (
    REQUIRED_PIPELINE_TABLES,
    evaluate_preflight,
)


def _settings(**overrides) -> InfraSettings:
    values = {
        "MYSQL_PASSWORD": "mysql-secret",
        "ODPS_ACCESS_ID": "odps-id",
        "ODPS_ACCESS_KEY": "odps-secret",
        "ODPS_PROJECT": "project",
        "SCHEDULER_ENABLED": True,
        "AIGC_API_TOKEN": "aigc-secret",
        "LOG_OSS_UPLOAD_ENABLED": False,
    }
    values.update(overrides)
    return InfraSettings(_env_file=None, **values)


def test_preflight_passes_for_complete_runtime_contract() -> None:
    result = evaluate_preflight(
        infra=_settings(AUTH_COOKIE_SECURE=True),
        openrouter_configured=True,
        table_names=set(REQUIRED_PIPELINE_TABLES),
        current_revision="head-1",
        expected_revision="head-1",
        require_scheduler=True,
        require_secure_cookie=True,
    )

    assert result.passed is True
    assert result.errors == []
    assert result.checks["pipeline_step_count"] == 15


def test_preflight_blocks_silent_partial_startup() -> None:
    result = evaluate_preflight(
        infra=_settings(
            SCHEDULER_ENABLED=False,
            AIGC_API_TOKEN="",
            CATEGORY_MATCH_API_URL="",
        ),
        openrouter_configured=False,
        table_names={"pipeline_run"},
        current_revision="old-head",
        expected_revision="new-head",
        require_scheduler=True,
        require_secure_cookie=True,
    )

    assert result.passed is False
    assert any("SCHEDULER_ENABLED" in item for item in result.errors)
    assert any("OPENROUTER_API_KEY" in item for item in result.errors)
    assert any("AIGC_API_TOKEN" in item for item in result.errors)
    assert any("CATEGORY_MATCH_API_URL" in item for item in result.errors)
    assert any("迁移版本不一致" in item for item in result.errors)
    assert any("缺少流水线依赖表" in item for item in result.errors)
    assert any("AUTH_COOKIE_SECURE" in item for item in result.errors)