cc-ci/runner/harness/meta.py

"""Single recipe-meta loader + declarative key registry (recipe-custom restructure P1; spec
docs/recipe-customization.md §8 R1).

THE one place `tests/<recipe>/recipe_meta.py` is `exec()`d. Every consumer (orchestrator, pytest
`meta` fixture, deploy env shaping, deps, warm-canonical enrollment, screenshot) reads the ONE
loaded `RecipeMeta` object instead of re-exec'ing the file and cherry-picking keys — that drift
(six divergent loaders, spec §4 L1–L6) is what made `SCREENSHOT` an unreachable knob (R2) and let
key typos silently disable coverage (R6).

Validation (locked decision, recipe-custom-restructure-full-plan.md):
- unknown ALL-CAPS top-level name → MetaError (hard error, fails fast at load; the all-recipes
  unit test catches it at PR time). Underscore-prefixed names (`_FOO`) are recipe-private and
  exempt; lowercase names (helper functions/imports) are ignored.
- type mismatch → MetaError. Callables are accepted ONLY for hook-typed keys.

The KEYS registry is the single source of truth for the key set: it drives validation, the
RecipeMeta dataclass fields, and the generated reference table in docs/recipe-customization.md §4
(scripts/gen-meta-docs.py; a unit test asserts the committed table matches).
"""

from __future__ import annotations

import copy
import dataclasses
import difflib
import os
from collections.abc import Callable

ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
TESTS_DIR = os.path.join(ROOT, "tests")


class MetaError(Exception):
    """A recipe_meta.py failed registry validation (unknown key / type mismatch / callable on a
    data key). Hard error by design: a typo'd key must fail the run at load, not silently reduce
    coverage (spec §8 R6 — the worst failure mode for a CI harness)."""


@dataclasses.dataclass(frozen=True)
class Key:
    """One registered recipe_meta key: name, type tag, default, one-line doc (rendered into the
    generated reference table), optional extra validator, and a deprecation marker (deprecated
    keys still load+validate but are scheduled for deletion)."""

    name: str
    type: str  # "int"|"str"|"tuple[int]"|"bool"|"dict_or_hook"|"hook"|"list[str]"|"dict"
    default: object
    doc: str
    validate: Callable[[object], None] | None = None
    deprecated: bool = False


KEYS: tuple[Key, ...] = (
    Key(
        "HEALTH_PATH",
        "str",
        "/",
        "Path probed for serving/health checks (deploy wait + generic `assert_serving`).",
    ),
    Key("HEALTH_OK", "tuple[int]", (200, 301, 302), "Acceptable HTTP status codes for health."),
    Key("DEPLOY_TIMEOUT", "int", 600, "Max seconds to wait for swarm convergence per deploy."),
    Key("HTTP_TIMEOUT", "int", 300, "Max seconds to wait for HTTP health after convergence."),
    Key(
        "BACKUP_CAPABLE",
        "bool",
        None,
        "Override the backup-tier capability auto-detect (compose `backupbot.backup` labels). `False` forces N/A; `True` forces the tier on; unset = auto-detect.",
    ),
    Key(
        "EXPECTED_NA",
        "dict",
        None,
        "Declare an N/A rung intentional: `{rung: reason}`. The cap stands either way; only the report wording changes.",
    ),
    Key(
        "READY_PROBE",
        "hook",
        None,
        "Callable returning extra readiness probes, run after install AND after upgrade: HTTP `{host, path, ok}` or TCP `{tcp_host, tcp_port, stable}`.",
    ),
    Key(
        "UPGRADE_BASE_VERSION",
        "str",
        None,
        "Exact published tag overriding the upgrade tier's base (default: `recipe_versions[-2]`).",
    ),
    Key(
        "BACKUP_VERIFY",
        "hook",
        None,
        "Callable `-> bool` post-backup data-capture check; `False` re-runs the backup (truncated-dump race guard), retried up to 3 attempts.",
    ),
    Key(
        "UPGRADE_EXTRA_ENV",
        "dict_or_hook",
        None,
        "Extra `.env` keys applied after the PR-head checkout, before the chaos redeploy (env that exists only at head).",
    ),
    Key(
        "EXTRA_ENV",
        "dict_or_hook",
        {},
        "Extra `.env` keys applied at EVERY deploy (base install AND upgrade old-app). Callable form derives values from the per-run domain.",
    ),
    Key(
        "DEPS",
        "list[str]",
        [],
        'Dep recipes deployed/provisioned alongside (e.g. `["keycloak"]`); creds land in `$CCCI_DEPS_FILE`.',
    ),
    Key(
        "WARM_CANONICAL",
        "bool",
        False,
        "Enroll the recipe in the warm/canonical app system (docs/warm.md): green cold runs on LATEST advance the canonical snapshot.",
    ),
    Key(
        "SCREENSHOT",
        "hook",
        None,
        "Callable driving Playwright to a safe, credential-free post-login view for the results-card screenshot (default: landing page).",
    ),
    # (CHAOS_BASE_DEPLOY, OIDC_AT_INSTALL and SKIP_GENERIC were deleted in restructure P2:
    # compose.ccci.yml is first-class + auto-chaos; install-time deps wiring is the only mode;
    # the generic floor is suppressible only via the dev-only CCCI_SKIP_GENERIC* env form.)
)

_REGISTRY: dict[str, Key] = {k.name: k for k in KEYS}

# The one validated, attribute-access view of a recipe's customization. Generated from KEYS so the
# field set can never drift from the registry (frozen: consumers share one immutable object).
RecipeMeta = dataclasses.make_dataclass(
    "RecipeMeta",
    [(k.name, object, dataclasses.field(default=None)) for k in KEYS],
    frozen=True,
)
RecipeMeta.__doc__ = (
    "Validated per-recipe customization (one field per registered key; attribute access). "
    "Built ONLY by meta.load()."
)


def meta_path(recipe: str, tests_dir: str | None = None) -> str:
    """Canonical path of a recipe's meta file (pure)."""
    return os.path.join(tests_dir or TESTS_DIR, recipe, "recipe_meta.py")


def _coerce(key: Key, value: object, path: str) -> object:
    """Validate `value` against `key`'s declared type; normalize containers (tuple[int]/list[str]).
    Raises MetaError on mismatch — including a callable supplied for a data-typed key."""
    t = key.type
    if callable(value) and t not in ("hook", "dict_or_hook"):
        raise MetaError(
            f"{path}: {key.name} is a data key (type {t}) — callables are accepted only for "
            f"hook-typed keys"
        )
    if t == "int":
        if isinstance(value, int) and not isinstance(value, bool):
            return value
    elif t == "str":
        if isinstance(value, str):
            return value
    elif t == "bool":
        if isinstance(value, bool):
            return value
    elif t == "tuple[int]":
        if isinstance(value, tuple | list) and all(
            isinstance(x, int) and not isinstance(x, bool) for x in value
        ):
            return tuple(value)
    elif t == "list[str]":
        if isinstance(value, tuple | list) and all(isinstance(x, str) for x in value):
            return list(value)
    elif t == "dict":
        if isinstance(value, dict):
            return value
    elif (
        t == "hook"
        and callable(value)
        or t == "dict_or_hook"
        and (isinstance(value, dict) or callable(value))
    ):
        return value
    raise MetaError(f"{path}: {key.name} must be {t}, got {type(value).__name__} ({value!r})")


def load(recipe: str, tests_dir: str | None = None):
    """Load + validate a recipe's customization -> RecipeMeta. THE only exec() of recipe_meta.py.

    Missing file -> all registry defaults (the zero-config baseline, spec §2). Unknown
    non-underscore ALL-CAPS top-level name or type mismatch -> MetaError (hard error).
    `tests_dir` overrides the recipe-meta root (unit tests / fixtures)."""
    path = meta_path(recipe, tests_dir)
    values = {k.name: copy.copy(k.default) for k in KEYS}
    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 name in sorted(ns):
            if name.startswith("_") or not name.isupper():
                continue  # _FOO = recipe-private (exempt); lowercase = helpers/imports (ignored)
            key = _REGISTRY.get(name)
            if key is None:
                near = difflib.get_close_matches(name, _REGISTRY, n=1)
                hint = f" — did you mean {near[0]!r}?" if near else ""
                raise MetaError(
                    f"{path}: unknown recipe_meta key {name!r}{hint}. Registered keys: "
                    f"{', '.join(sorted(_REGISTRY))}. Recipe-private constants must be "
                    f"underscore-prefixed (e.g. _{name})."
                )
            values[name] = _coerce(key, ns[name], path)
            if key.validate:
                key.validate(values[name])
    return RecipeMeta(**values)


def as_dict(meta) -> dict:
    """RecipeMeta -> {key: value} (every registered key, defaults included)."""
    return dataclasses.asdict(meta)


def non_default(meta) -> dict:
    """The keys a recipe explicitly customized: {key: value} where value differs from the registry
    default. Hooks compare by identity-vs-None (a set hook is always non-default). Feeds the run's
    customization manifest (P5)."""
    out = {}
    for k in KEYS:
        v = getattr(meta, k.name)
        if v != k.default:
            out[k.name] = v
    return out


def _env_map(value, domain: str) -> dict[str, str]:
    if callable(value):
        value = value(domain)
    return {str(k): str(v) for k, v in (value or {}).items()}


def extra_env(meta, domain: str) -> dict[str, str]:
    """Resolve EXTRA_ENV (dict or callable(domain)->dict) to the concrete per-run env map."""
    return _env_map(meta.EXTRA_ENV, domain)


def upgrade_extra_env(meta, domain: str) -> dict[str, str]:
    """Resolve UPGRADE_EXTRA_ENV (dict or callable(domain)->dict) to the concrete env map."""
    return _env_map(meta.UPGRADE_EXTRA_ENV, domain)