cc-ci/tests/conftest.py

"""Shared pytest fixtures for recipe CI (plan §4.3).

A run is parameterized by env: RECIPE, REF (PR head sha), PR, SRC (head repo). The harness
computes a unique app domain per run so concurrent runs never collide, and GUARANTEES teardown
(undeploy + volume + secret removal) via a finalizer, even on failure.
"""

from __future__ import annotations

import os
import sys

import pytest

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "runner"))
from harness import deps as deps_mod, lifecycle, naming  # noqa: E402


def _short(s: str, n: int = 8) -> str:
    return "".join(c for c in s if c.isalnum())[:n] or "local"


def _recipe_meta(recipe: str) -> dict:
    """Optional per-recipe config so enrolling a recipe needs NO shared-harness change (D5).
    A recipe may ship tests/<recipe>/recipe_meta.py with any of: HEALTH_PATH (str),
    HEALTH_OK (tuple of status codes), DEPLOY_TIMEOUT (int), HTTP_TIMEOUT (int)."""
    path = os.path.join(os.path.dirname(__file__), recipe, "recipe_meta.py")
    meta = {
        "HEALTH_PATH": "/",
        "HEALTH_OK": (200, 301, 302),
        "DEPLOY_TIMEOUT": 600,
        "HTTP_TIMEOUT": 300,
    }
    if os.path.exists(path):
        ns: dict = {}
        with open(path) as fh:
            exec(compile(fh.read(), path, "exec"), ns)  # noqa: S102 (trusted, in-repo)
        for k in meta:
            if k in ns:
                meta[k] = ns[k]
    return meta


@pytest.fixture(scope="session")
def recipe() -> str:
    return os.environ.get("RECIPE", "custom-html")


@pytest.fixture(scope="session")
def app_domain(recipe) -> str:
    # Docker swarm config/secret names = <stackname>_<res>_<ver> must be <= 64 chars, and
    # stackname is the sanitized domain. ".ci.commoninternet.net" alone is 22 chars, so the
    # subdomain label must stay short. Use <recipe[:4]>-<6hex(recipe|pr|ref)> — unique per run,
    # collision-safe across recipes (full recipe in the hash), readable context lives in the
    # Drone build params + PR comment. (Deviation from plan §4.0 long name; see DECISIONS.md.)
    return naming.app_domain(recipe, os.environ.get("PR", "0"), os.environ.get("REF"))


@pytest.fixture(scope="session")
def meta(recipe) -> dict:
    return _recipe_meta(recipe)


@pytest.fixture(scope="session")
def live_app() -> str:
    """Phase 1d shared-deployment contract: the orchestrator deploys ONCE and runs each tier
    (generic or overlay) as its own pytest invocation against that single live deployment, passing
    its domain in CCCI_APP_DOMAIN. Tiers are assertion-only (and lifecycle ops mutate in place) —
    they NEVER deploy or tear down. This guarantees one deploy + one teardown per run (DG4.1)."""
    domain = os.environ.get("CCCI_APP_DOMAIN")
    assert domain, "CCCI_APP_DOMAIN not set — a tier must run under the deploy-once orchestrator"
    return domain


@pytest.fixture(scope="session")
def deps_apps() -> dict[str, str]:
    """Phase 2 Q2.3 dependency-resolver contract: when a recipe declares `DEPS = [...]` in its
    `recipe_meta.py`, the orchestrator deploys each dep BEFORE the recipe under test, persists
    their per-run identity to `$CCCI_DEPS_FILE`, and tears them down LAST in finally. Tests access
    the dep's per-run domain via this fixture: `deps_apps["keycloak"]` returns the dep's domain
    or raises KeyError if the dep wasn't declared. Returns {} when the recipe declared no deps."""
    state = deps_mod.load_run_state()
    return {entry["recipe"]: entry["domain"] for entry in state if entry.get("domain")}


def _wait_healthy(domain, meta):
    lifecycle.wait_healthy(
        domain,
        ok_codes=tuple(meta["HEALTH_OK"]),
        path=meta["HEALTH_PATH"],
        deploy_timeout=meta["DEPLOY_TIMEOUT"],
        http_timeout=meta["HTTP_TIMEOUT"],
    )


@pytest.fixture
def deployed(recipe, app_domain, meta, request):
    """Function-scoped: deploy the current/$REF version healthy, guaranteed teardown after.
    Used by stages that start from current (install/backup)."""
    version = os.environ.get("VERSION") or None
    lifecycle.janitor()
    request.addfinalizer(lambda: lifecycle.teardown_app(app_domain))
    lifecycle.deploy_app(recipe, app_domain, version=version)
    _wait_healthy(app_domain, meta)
    return app_domain


@pytest.fixture(scope="session")
def deployed_app(recipe, app_domain, meta):
    """Install stage: deploy the recipe and wait until healthy; tear down at session end."""
    version = os.environ.get("VERSION") or None
    lifecycle.janitor()  # sweep orphans from crashed runs first
    try:
        lifecycle.deploy_app(recipe, app_domain, version=version, secrets=True)
        _wait_healthy(app_domain, meta)
        yield app_domain
    finally:
        lifecycle.teardown_app(app_domain)