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 (refined operator-2026-05-28 SSO-dep plan §1):
    when a recipe declares `DEPS = [...]` in its `recipe_meta.py`, the orchestrator deploys each
    dep AFTER the generic tiers (between RESTORE and CUSTOM) and persists their per-run identity
    + SSO creds to `$CCCI_DEPS_FILE`. Tests access the dep's per-run domain via this fixture.
    For full SSO creds (realm/client/secret/admin) use the `deps_creds` fixture instead.

    Returns `{dep_recipe: domain}` (str→str). Empty when no deps declared OR deps-not-ready."""
    state = deps_mod.deps_as_dict(deps_mod.load_run_state())
    return {r: e["domain"] for r, e in state.items() if e.get("domain")}


@pytest.fixture(scope="session")
def deps_creds() -> dict[str, dict]:
    """Full SSO-creds dict for each declared dep (operator-2026-05-28 SSO-dep plan §1).
    `deps_creds["keycloak"]` returns the entry written by setup_custom_tests with keys
    domain/realm/client_id/client_secret/user/password/email/admin_user/admin_password/
    discovery_url/token_url/.... Use this in `@pytest.mark.requires_deps` tests that need to
    authenticate via OIDC."""
    return deps_mod.deps_as_dict(deps_mod.load_run_state())


def pytest_collection_modifyitems(config, items):
    """SSO-dep plan §4: tests marked `@pytest.mark.requires_deps` are skipped with reason
    `deps-not-ready: <captured-err>` when the orchestrator's setup_custom_tests step failed
    (orchestrator sets CCCI_DEPS_READY=0 in env). Non-deps custom tests are unaffected.

    This is failure-isolation per plan §1 — generic tiers cannot break the SSO-marked tests'
    skip status, and an SSO-setup failure cannot break the generic tiers (they run first)."""
    deps_ready_env = os.environ.get("CCCI_DEPS_READY", "1")
    if deps_ready_env == "1":
        return
    reason = os.environ.get("CCCI_DEPS_NOT_READY_REASON", "(no reason given)")
    skip_mark = pytest.mark.skip(reason=f"deps-not-ready: {reason}")
    skipped = 0
    for item in items:
        if "requires_deps" in item.keywords:
            item.add_marker(skip_mark)
            skipped += 1
    # F2-11: a skip-only pytest file exits 0, so without this the orchestrator can't tell
    # "SSO verified" from "SSO test silently skipped because deps weren't ready". Record the count
    # of requires_deps tests we skipped to a report file the orchestrator reads — it surfaces the
    # count in RUN SUMMARY and FAILS the recipe's SSO claim (a green exit must not mask an unrun
    # SSO test). Appended one line per pytest invocation (one per custom file); orchestrator sums.
    report = os.environ.get("CCCI_DEPS_SKIP_REPORT")
    if report and skipped:
        try:
            with open(report, "a") as f:
                f.write(f"{skipped}\n")
        except OSError:
            pass


def pytest_configure(config):
    """Register the `requires_deps` marker so pytest doesn't warn about it."""
    config.addinivalue_line(
        "markers",
        "requires_deps: test requires DEPS-declared services + setup_custom_tests success.",
    )


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)