kite/backend/tests/test_compose_migrations.py

"""
Static regression tests for the Compose migration gate.

These tests parse docker-compose.yml (via PyYAML) and assert the structural
invariants that ensure Alembic runs before any application process starts.
No Docker daemon is required — they are pure-Python configuration checks.

Requirements covered:
  CONN-04 — Cloud connections require an up-to-date schema before use
  CLOUD-01 — cloud_items/cloud_folder_states tables must exist at runtime
"""
from __future__ import annotations

import os
import pathlib

import pytest
import yaml


REPO_ROOT = pathlib.Path(__file__).resolve().parents[2]
COMPOSE_PATH = REPO_ROOT / "docker-compose.yml"

# When running inside the Docker container the repo root is not mounted;
# skip the whole module rather than erroring with FileNotFoundError.
if not COMPOSE_PATH.exists():
    pytest.skip(
        f"docker-compose.yml not found at {COMPOSE_PATH} — "
        "these static analysis tests must run from the host repo root, not inside a container.",
        allow_module_level=True,
    )

APPLICATION_SERVICES = {"backend", "celery-worker", "celery-beat"}


@pytest.fixture(scope="module")
def compose() -> dict:
    """Load docker-compose.yml once for all tests in this module."""
    with COMPOSE_PATH.open() as fh:
        return yaml.safe_load(fh)


# ---------------------------------------------------------------------------
# 1. Exactly one service runs "alembic upgrade head"
# ---------------------------------------------------------------------------

def test_exactly_one_alembic_upgrade_service(compose):
    """Exactly one service must run 'alembic upgrade head' — the migrate service."""
    alembic_services = [
        name
        for name, svc in compose["services"].items()
        if "alembic upgrade head" in " ".join(
            svc.get("command", [])
            if isinstance(svc.get("command"), list)
            else [str(svc.get("command", ""))]
        )
    ]
    assert len(alembic_services) == 1, (
        f"Expected exactly 1 service running 'alembic upgrade head', "
        f"found: {alembic_services}"
    )
    assert alembic_services[0] == "migrate", (
        f"Migration service must be named 'migrate', got: {alembic_services[0]}"
    )


# ---------------------------------------------------------------------------
# 2. The migrate service uses DATABASE_MIGRATE_URL, not DATABASE_URL
# ---------------------------------------------------------------------------

def test_migrate_service_uses_migrate_url(compose):
    """The migrate service must reference DATABASE_MIGRATE_URL."""
    svc = compose["services"]["migrate"]
    env = svc.get("environment", [])
    # environment may be a list of "KEY=VALUE" or "KEY" strings, or a dict
    if isinstance(env, dict):
        keys = set(env.keys())
    else:
        keys = {e.split("=")[0] for e in env}

    assert "DATABASE_MIGRATE_URL" in keys, (
        "migrate service must declare DATABASE_MIGRATE_URL in its environment"
    )
    assert "DATABASE_URL" not in keys, (
        "migrate service must NOT have DATABASE_URL — it uses the DDL-capable migrate role only"
    )


# ---------------------------------------------------------------------------
# 3. Application services must NOT run alembic upgrade head in their command
# ---------------------------------------------------------------------------

def test_application_services_do_not_run_alembic(compose):
    """backend/celery-worker/celery-beat must not run 'alembic upgrade' themselves."""
    for name in APPLICATION_SERVICES:
        svc = compose["services"][name]
        cmd = svc.get("command", "")
        if isinstance(cmd, list):
            cmd = " ".join(cmd)
        assert "alembic upgrade" not in cmd, (
            f"Service '{name}' must not run 'alembic upgrade' — only the migrate service should"
        )


# ---------------------------------------------------------------------------
# 4. All application services depend on migrate with service_completed_successfully
# ---------------------------------------------------------------------------

def test_application_services_depend_on_migrate(compose):
    """backend, celery-worker, and celery-beat must all wait for migrate to complete."""
    for name in APPLICATION_SERVICES:
        svc = compose["services"][name]
        depends = svc.get("depends_on", {})
        assert "migrate" in depends, (
            f"Service '{name}' is missing 'migrate' in depends_on"
        )
        condition = depends["migrate"].get("condition")
        assert condition == "service_completed_successfully", (
            f"Service '{name}' depends_on.migrate.condition must be "
            f"'service_completed_successfully', got: {condition!r}"
        )


# ---------------------------------------------------------------------------
# 5. The migrate service preserves container hardening
# ---------------------------------------------------------------------------

def test_migrate_service_has_container_hardening(compose):
    """migrate service must have read_only, cap_drop, and no-new-privileges."""
    svc = compose["services"]["migrate"]

    assert svc.get("read_only") is True, (
        "migrate service must set read_only: true"
    )

    cap_drop = svc.get("cap_drop", [])
    assert "ALL" in cap_drop, (
        "migrate service must cap_drop ALL"
    )

    sec_opts = svc.get("security_opt", [])
    assert "no-new-privileges:true" in sec_opts, (
        "migrate service must set no-new-privileges:true"
    )


# ---------------------------------------------------------------------------
# 6. The migrate service has restart: "no" (it is one-shot, not a daemon)
# ---------------------------------------------------------------------------

def test_migrate_service_is_one_shot(compose):
    """migrate service must have restart: 'no' — it exits after running migrations."""
    svc = compose["services"]["migrate"]
    restart = svc.get("restart", "")
    assert restart == "no", (
        f"migrate service restart policy must be 'no' (one-shot), got: {restart!r}"
    )


# ---------------------------------------------------------------------------
# 7. The migrate service only depends on postgres (not minio/redis)
# ---------------------------------------------------------------------------

def test_migrate_service_minimal_dependencies(compose):
    """migrate only needs postgres to be healthy — minio/redis are runtime concerns."""
    svc = compose["services"]["migrate"]
    depends = svc.get("depends_on", {})
    assert "postgres" in depends, "migrate must depend on postgres"
    assert depends["postgres"]["condition"] == "service_healthy", (
        "migrate must wait for postgres to be healthy"
    )
    for extra in ("minio", "redis", "minio-init"):
        assert extra not in depends, (
            f"migrate should NOT depend on '{extra}' — only postgres is needed for schema migrations"
        )