company-ai-platform/tests/test_runtime_activation.py

import shutil
import subprocess
from datetime import timedelta
from pathlib import Path
from types import SimpleNamespace

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, text
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool

from app.application.scheduling import create_scheduler
from app.core.config import Settings, get_settings
from app.core.database import Base
from app.core.database.safety import validate_platform_migration_target
from app.core.utils.time import utc_now
from app.modules.events.constants import EventStatus
from app.modules.events.models import DomainEvent
from app.modules.feishu.constants import FeishuInboundStatus
from app.modules.feishu.models import FeishuEventReceipt
from app.modules.observability.constants import (
    HeartbeatComponent,
    ObservabilityKey,
    ObservabilityStatus,
)
from app.modules.observability.models import SystemHeartbeat
from app.modules.observability import runtime as observability_runtime
from app.modules.observability.service import ObservabilityService
from app.modules.workflows.constants import WorkflowStatus
from app.modules.workflows.models import WorkflowInstance
from app.tools import run_scheduler, runtime_preflight


def _production_settings(**overrides: object) -> Settings:
    values: dict[str, object] = {
        "app_env": "production",
        "database_url": (
            "postgresql+psycopg://app:runtime-database-password-2026@db/app"
        ),
        "api_key": "runtime-service-key-2026-primary",
        "audit_api_key": "runtime-audit-key-2026-independent",
        "cors_origins": ["https://internal.example.com"],
        "debug": False,
        "mask_sensitive_responses": True,
        "read_only_mode": True,
        "feishu_app_id": None,
        "feishu_app_secret": None,
        "feishu_event_transport": "disabled",
        "feishu_verification_token": None,
        "feishu_user_features_enabled": False,
    }
    values.update(overrides)
    return Settings(_env_file=None, **values)


def _stamp_alembic_head(engine: Engine) -> None:
    expected = runtime_preflight.expected_alembic_heads()
    assert len(expected) == 1
    with engine.begin() as connection:
        connection.execute(
            text("create table alembic_version (version_num varchar(32))")
        )
        connection.execute(
            text("insert into alembic_version (version_num) values (:revision)"),
            {"revision": expected[0]},
        )


def test_production_feishu_transport_fails_closed() -> None:
    with pytest.raises(ValueError, match="FEISHU_EVENT_TRANSPORT"):
        _production_settings(
            feishu_user_features_enabled=True,
            feishu_admin_identities=["tenant:open-id"],
        )

    with pytest.raises(ValueError, match="FEISHU_VERIFICATION_TOKEN"):
        _production_settings(
            feishu_event_transport="webhook",
            feishu_app_id="app-id",
            feishu_app_secret="feishu-app-secret-2026-secure-value",
        )

    with pytest.raises(ValueError, match="FEISHU_APP_ID"):
        _production_settings(feishu_event_transport="long_connection")

    settings = _production_settings(
        feishu_event_transport="long_connection",
        feishu_app_id="app-id",
        feishu_app_secret="feishu-app-secret-2026-secure-value",
        feishu_user_features_enabled=True,
        feishu_admin_identities=["tenant:open-id"],
    )
    assert settings.feishu_event_transport == "long_connection"


def test_production_database_target_fails_closed() -> None:
    with pytest.raises(ValueError, match="must use PostgreSQL"):
        _production_settings(
            database_url=(
                "mysql+pymysql://app:runtime-database-password-2026@legacy/business"
            ),
        )

    with pytest.raises(ValueError, match="must target different databases"):
        _production_settings(
            database_url=(
                "postgresql+psycopg://platform:platform-password-2026-secure@db/platform"
            ),
            legacy_database_url=(
                "postgresql://readonly:readonly-password-2026-secure@db:5432/platform"
            ),
        )

    settings = _production_settings(
        database_url=(
            "postgresql+psycopg://platform:platform-password-2026-secure@db/platform"
        ),
        legacy_database_url=(
            "mysql+pymysql://readonly:readonly-password-2026-secure@legacy/business"
        ),
    )
    assert settings.database_url.startswith("postgresql")


def test_production_placeholders_and_invalid_admin_identities_fail_closed() -> None:
    placeholder_key = "replace-with-a-random-service-key"
    with pytest.raises(ValueError, match="API_KEY/API_KEYS") as error:
        _production_settings(api_key=placeholder_key)
    assert placeholder_key not in str(error.value)
    with pytest.raises(ValueError, match="DATABASE_URL password"):
        _production_settings(
            database_url="postgresql+psycopg://app:change-me@db/app",
        )
    with pytest.raises(ValueError, match="tenant_key:open_id"):
        _production_settings(
            feishu_event_transport="long_connection",
            feishu_app_id="app-id",
            feishu_app_secret="feishu-app-secret-2026-secure-value",
            feishu_user_features_enabled=True,
            feishu_admin_identities=["not-a-valid-identity"],
        )


def test_platform_migration_target_never_accepts_mysql_or_legacy_database() -> None:
    with pytest.raises(RuntimeError, match="PostgreSQL or local SQLite"):
        validate_platform_migration_target(
            "mysql+pymysql://platform:secret@legacy/business",
            None,
        )
    with pytest.raises(RuntimeError, match="matches LEGACY_DATABASE_URL"):
        validate_platform_migration_target(
            "postgresql+psycopg://platform:one@db/platform",
            "postgresql://readonly:two@db:5432/platform",
        )


def test_environment_overrides_dotenv(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    base = tmp_path / ".env"
    base.write_text("SCHEDULER_ENABLED=true\n", encoding="utf-8")
    monkeypatch.delenv("SCHEDULER_ENABLED", raising=False)

    assert Settings(_env_file=base).scheduler_enabled is True

    monkeypatch.setenv("SCHEDULER_ENABLED", "false")
    assert Settings(_env_file=base).scheduler_enabled is False


def test_scheduler_process_refuses_disabled_flag(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        run_scheduler,
        "get_settings",
        lambda: SimpleNamespace(scheduler_enabled=False),
    )
    monkeypatch.setattr(
        run_scheduler,
        "create_scheduler",
        lambda: pytest.fail("disabled scheduler must not be created"),
    )

    with pytest.raises(RuntimeError, match="SCHEDULER_ENABLED"):
        run_scheduler.main()


def test_runtime_preflight_requires_exact_alembic_head(tmp_path: Path) -> None:
    database_path = tmp_path / "runtime-preflight.db"
    database_url = f"sqlite:///{database_path.as_posix()}"
    settings = Settings(
        _env_file=None,
        database_url=database_url,
        feishu_event_transport="disabled",
    )
    expected = runtime_preflight.expected_alembic_heads()
    assert len(expected) == 1
    engine = create_engine(database_url)
    try:
        with engine.begin() as connection:
            connection.execute(
                text("create table alembic_version (version_num varchar(32))")
            )
            connection.execute(
                text("insert into alembic_version (version_num) values ('old')")
            )
        with pytest.raises(
            runtime_preflight.RuntimePreflightError,
            match="current Alembic head",
        ):
            runtime_preflight.run_preflight(settings)

        with engine.begin() as connection:
            connection.execute(text("delete from alembic_version"))
            connection.execute(
                text(
                    "insert into alembic_version (version_num) values (:revision)"
                ),
                {"revision": expected[0]},
            )
        result = runtime_preflight.run_preflight(settings)
        assert result.revisions == expected
    finally:
        engine.dispose()


def test_scheduler_registers_immediate_runtime_heartbeats(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("TASK_QUEUE_ENABLED", "true")
    monkeypatch.setenv("FEISHU_USER_FEATURES_ENABLED", "false")
    get_settings.cache_clear()
    try:
        scheduler = create_scheduler()

        assert scheduler.get_job("scheduler_heartbeat").next_run_time is not None
        assert scheduler.get_job("worker_heartbeat").next_run_time is not None
        assert scheduler.get_job("event_dispatch").next_run_time is not None
        assert scheduler.get_job("feishu_inbound_event_cycle") is not None
    finally:
        get_settings.cache_clear()


def test_required_component_heartbeats_control_readiness(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("FEISHU_USER_FEATURES_ENABLED", "true")
    monkeypatch.setenv("FEISHU_ADMIN_IDENTITIES", "")
    monkeypatch.setenv("FEISHU_EVENT_TRANSPORT", "long_connection")
    monkeypatch.setenv("FEISHU_APP_ID", "app-id")
    monkeypatch.setenv("FEISHU_APP_SECRET", "app-secret")
    monkeypatch.setenv("TASK_QUEUE_ENABLED", "false")
    get_settings.cache_clear()
    engine = create_engine("sqlite://")
    Base.metadata.create_all(engine)
    _stamp_alembic_head(engine)
    try:
        with Session(engine) as db:
            service = ObservabilityService(db)
            missing = service.ready()
            assert missing[ObservabilityKey.STATUS] == ObservabilityStatus.DEGRADED
            assert missing[ObservabilityKey.CHECKS][ObservabilityKey.SCHEDULER][
                ObservabilityKey.STATUS
            ] == ObservabilityStatus.DEGRADED
            assert missing[ObservabilityKey.CHECKS][ObservabilityKey.FEISHU_EVENTS][
                ObservabilityKey.STATUS
            ] == ObservabilityStatus.DEGRADED

            service.record_heartbeat(
                HeartbeatComponent.SCHEDULER,
                "scheduler-test",
            )
            service.record_heartbeat(
                HeartbeatComponent.FEISHU_EVENTS,
                "feishu-events-test",
            )
            ready = service.ready()
            assert ready[ObservabilityKey.STATUS] == ObservabilityStatus.OK

            heartbeat = db.query(SystemHeartbeat).filter_by(
                component=HeartbeatComponent.SCHEDULER,
            ).one()
            heartbeat.last_seen_at = utc_now() - timedelta(minutes=10)
            db.commit()
            stale = service.ready()
            assert stale[ObservabilityKey.CHECKS][ObservabilityKey.SCHEDULER][
                ObservabilityKey.STATUS
            ] == ObservabilityStatus.DEGRADED
    finally:
        get_settings.cache_clear()
        engine.dispose()


def test_api_lifecycle_records_immediate_heartbeat(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(engine)
    factory = sessionmaker(bind=engine, expire_on_commit=False)
    monkeypatch.setattr(observability_runtime, "SessionLocal", factory)
    monkeypatch.setattr(
        observability_runtime,
        "get_settings",
        lambda: SimpleNamespace(heartbeat_interval_seconds=3600),
    )
    app = FastAPI()
    observability_runtime.attach_api_heartbeat(app)

    try:
        with TestClient(app):
            thread = app.state.api_heartbeat_thread
            assert thread.is_alive()
            with Session(engine) as db:
                heartbeat = db.query(SystemHeartbeat).filter_by(
                    component=HeartbeatComponent.API,
                    instance_id=app.state.api_heartbeat_instance_id,
                ).one()
                assert heartbeat.status == ObservabilityStatus.OK

        assert not thread.is_alive()
    finally:
        engine.dispose()


def test_production_readiness_requires_api_heartbeat(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    settings = SimpleNamespace(
        app_env="production",
        scheduler_enabled=False,
        task_queue_enabled=False,
        feishu_user_features_enabled=False,
        feishu_event_transport="disabled",
        event_dispatch_enabled=True,
        heartbeat_interval_seconds=60,
        heartbeat_retention_seconds=86400,
    )
    monkeypatch.setattr(
        "app.modules.observability.service.get_settings",
        lambda: settings,
    )
    engine = create_engine("sqlite://")
    Base.metadata.create_all(engine)
    _stamp_alembic_head(engine)
    try:
        with Session(engine) as db:
            service = ObservabilityService(db)
            service.record_heartbeat(
                HeartbeatComponent.SCHEDULER,
                "scheduler-test",
            )

            missing = service.ready()
            assert missing[ObservabilityKey.CHECKS][ObservabilityKey.API][
                ObservabilityKey.STATUS
            ] == ObservabilityStatus.DEGRADED
            assert missing[ObservabilityKey.CHECKS][ObservabilityKey.API][
                "reason"
            ] == "heartbeat_missing"

            service.record_heartbeat(
                HeartbeatComponent.API,
                "api-test",
            )
            ready = service.ready()
            assert ready[ObservabilityKey.STATUS] == ObservabilityStatus.OK
            assert ready[ObservabilityKey.CHECKS][ObservabilityKey.API][
                ObservabilityKey.STATUS
            ] == ObservabilityStatus.OK
    finally:
        engine.dispose()


def test_production_schema_readiness_requires_alembic_head(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        "app.modules.observability.service.get_settings",
        lambda: SimpleNamespace(app_env="production"),
    )
    engine = create_engine("sqlite://")
    try:
        with Session(engine) as db:
            db.execute(text("create table alembic_version (version_num varchar(32))"))
            db.execute(
                text("insert into alembic_version (version_num) values ('old-revision')")
            )
            db.commit()
            service = ObservabilityService(db)
            mismatch = service._schema_check()
            assert mismatch[ObservabilityKey.STATUS] == ObservabilityStatus.DEGRADED
            expected = mismatch["expected"]
            assert len(expected) == 1

            db.execute(text("delete from alembic_version"))
            db.execute(
                text("insert into alembic_version (version_num) values (:revision)"),
                {"revision": expected[0]},
            )
            db.commit()
            assert service._schema_check()[ObservabilityKey.STATUS] == ObservabilityStatus.OK
    finally:
        engine.dispose()


def test_historical_terminal_failures_do_not_block_readiness() -> None:
    engine = create_engine("sqlite://")
    Base.metadata.create_all(engine)
    try:
        with Session(engine) as db:
            db.add(
                DomainEvent(
                    event_id="terminal-event",
                    event_type="test.terminal",
                    aggregate_type="test",
                    status=EventStatus.FAILED,
                    max_attempts=1,
                )
            )
            db.add(
                WorkflowInstance(
                    code="WF-TERMINAL",
                    workflow_type="test",
                    aggregate_type="test",
                    status=WorkflowStatus.FAILED,
                )
            )
            db.commit()

            service = ObservabilityService(db)
            events = service._events_check()
            workflows = service._workflows_check()
            assert events[ObservabilityKey.STATUS] == ObservabilityStatus.OK
            assert events["failed"] == 1
            assert workflows[ObservabilityKey.STATUS] == ObservabilityStatus.OK
            assert workflows["failed"] == 1
    finally:
        engine.dispose()


def test_feishu_inbound_metrics_expose_counts_without_payloads() -> None:
    engine = create_engine("sqlite://")
    Base.metadata.create_all(engine)
    try:
        with Session(engine) as db:
            for index, status_value in enumerate(
                (
                    FeishuInboundStatus.PENDING,
                    FeishuInboundStatus.RETRY,
                    FeishuInboundStatus.PROCESSING,
                    FeishuInboundStatus.FAILED,
                )
            ):
                db.add(
                    FeishuEventReceipt(
                        event_key=f"metric-event-{index}",
                        source="webhook",
                        status=status_value,
                        payload={"text": "must not appear in metrics"},
                    )
                )
            db.commit()

            metrics = ObservabilityService(db)._feishu_inbound_metrics()
            assert metrics == {
                "pending": 1,
                "retry": 1,
                "processing": 1,
                "failed": 1,
                "reply_pending": 0,
                "reply_retry": 0,
                "reply_processing": 0,
                "reply_failed": 0,
            }
            assert "payload" not in metrics
            assert "must not appear" not in str(metrics)
    finally:
        engine.dispose()


def test_runtime_deployment_uses_single_dotenv_file() -> None:
    compose = Path("docker-compose.yml").read_text(encoding="utf-8")
    external_compose = Path("docker-compose.external-db.yml").read_text(
        encoding="utf-8"
    )
    gitignore = Path(".gitignore").read_text(encoding="utf-8")
    start_script = Path("scripts/start_runtime.ps1").read_text(encoding="utf-8")

    assert "feishu-events:" in compose
    assert 'profiles: ["long-connection"]' in compose
    assert "/api/v1/health/ready" in compose
    assert "/api/v1/health/live" not in compose
    assert "${COMPOSE_DATABASE_URL:-postgresql+psycopg://" in compose
    assert "@db:5432/" in compose
    assert "${DATABASE_URL:" not in compose
    assert compose.count("APP_ENV: production") == 5
    assert 'profiles: ["internal-database"]' in external_compose
    assert external_compose.count("depends_on: !override") == 5
    assert external_compose.count(
        "${COMPOSE_DATABASE_URL:?COMPOSE_DATABASE_URL is required "
        "for external PostgreSQL mode}"
    ) == 5
    assert "${POSTGRES_PASSWORD" not in external_compose
    assert compose.count("restart: unless-stopped") >= 6
    assert compose.count("path: .env") == 1
    assert compose.count("required: true") == 1
    assert ".env.runtime" not in compose
    assert ".env" in gitignore.splitlines()
    assert ".env.runtime" not in gitignore.splitlines()
    assert "-WindowStyle Hidden" in start_script
    assert "TASK_QUEUE_ENABLED = \"false\"" in start_script
    assert "app.tools.runtime_preflight" in start_script
    assert "/api/v1/health/ready" in start_script


def test_default_and_external_compose_configs_validate_without_project_dotenv(
    tmp_path: Path,
) -> None:
    docker = shutil.which("docker")
    if docker is None:
        pytest.skip("Docker CLI is not installed")

    base_compose = tmp_path / "docker-compose.yml"
    external_compose = tmp_path / "docker-compose.external-db.yml"
    shutil.copyfile("docker-compose.yml", base_compose)
    shutil.copyfile("docker-compose.external-db.yml", external_compose)
    (tmp_path / ".env").write_text(
        "\n".join(
            (
                "API_KEY=config-test-service-key",
                "AUDIT_API_KEY=config-test-audit-key",
                "CORS_ORIGINS=[]",
            )
        )
        + "\n",
        encoding="utf-8",
    )

    default_interpolation = tmp_path / "default.compose.env"
    default_interpolation.write_text(
        "POSTGRES_PASSWORD=config-test-database-password\n",
        encoding="utf-8",
    )
    default_command = [
        docker,
        "compose",
        "--project-directory",
        str(tmp_path),
        "--env-file",
        str(default_interpolation),
        "-f",
        str(base_compose),
    ]
    _run_compose_config([*default_command, "config", "--quiet"], tmp_path)
    default_services = _run_compose_config(
        [*default_command, "config", "--services"],
        tmp_path,
    )
    assert "db" in default_services.splitlines()

    external_interpolation = tmp_path / "external.compose.env"
    external_interpolation.write_text(
        (
            "COMPOSE_DATABASE_URL="
            "postgresql+psycopg://config-user:config-password@"
            "external.invalid:5432/platform\n"
        ),
        encoding="utf-8",
    )
    external_command = [
        docker,
        "compose",
        "--project-directory",
        str(tmp_path),
        "--env-file",
        str(external_interpolation),
        "-f",
        str(base_compose),
        "-f",
        str(external_compose),
    ]
    _run_compose_config([*external_command, "config", "--quiet"], tmp_path)
    external_services = _run_compose_config(
        [*external_command, "config", "--services"],
        tmp_path,
    )
    assert "db" not in external_services.splitlines()


def _run_compose_config(command: list[str], cwd: Path) -> str:
    result = subprocess.run(
        command,
        cwd=cwd,
        check=False,
        capture_output=True,
        text=True,
        timeout=30,
    )
    assert result.returncode == 0, result.stderr
    return result.stdout