"""Which deploy row replays the role's README instructions, and in what mode. The instructions test used to be a job of its own that picked one random role per run. Riding along on a deploy row instead costs no extra runner and covers every role the sweep touches. The row it rides on is the role's smallest variant -- the one enabling the fewest services -- because that is the shape the README's Production block describes: the bare role, with the optional providers switched off. The replay is a second deploy on the same runner, so the row it picks must be one the guide can actually deploy: the README has to carry a Production block, and the guide's own mode (``compose`` for a role that ships a stack, ``host`` for one installed onto the machine) has to be a mode the role runs and tests. """ from __future__ import annotations from functools import cache from typing import TYPE_CHECKING from utils.cache.files import PROJECT_ROOT, read_text from utils.roles.deploy import role_has_stack from utils.roles.meta_lookup import get_role_mode_enabled, get_role_test_skips if TYPE_CHECKING: from collections.abc import Collection, Mapping, Sequence from typing import Any PRODUCTION_HEADING = "### Production" """The README section :file:`scripts/github/guide/compose_deploy.sh` replays. A role without it has no instructions to test.""" def _enabled_services(variant: Mapping[str, Any]) -> int: services = variant.get("services") or {} return sum( 1 for config in services.values() if isinstance(config, dict) and config.get("enabled") ) def smallest_variant( variants: Sequence[Mapping[str, Any]], allowed: Collection[str] | None = None ) -> str: """The variant that enables the fewest services, as the matrix writes it; the lowest index wins a tie, and ``''`` means none qualifies. Args: variants: the role's rendered variant configs, in index order. allowed: the variants this sweep actually deploys. A variant the sweep cuts as redundant coverage never reaches a runner, so hanging the replay on it would silently drop the role from the instructions test; the next-smallest deployed variant carries it instead. """ candidates = [ index for index in range(len(variants)) if allowed is None or str(index) in allowed ] if not candidates: return "" return str( min(candidates, key=lambda index: (_enabled_services(variants[index]), index)) ) @cache def guide_deployable(app: str) -> str: """The mode the guide would replay *app* in, or ``''`` when it cannot. Independent of the variant, and cached: the answer is a property of the role's files, and :func:`utils.github.variant.axes.assign` asks it once per row of every chunk. """ role_dir = PROJECT_ROOT / "roles" / app readme = role_dir / "README.md" if not readme.is_file() or PRODUCTION_HEADING not in read_text(str(readme)): return "" mode = "compose" if role_has_stack(role_dir) else "host" if mode in get_role_test_skips(role_dir, role_name=app): return "" if not get_role_mode_enabled(role_dir, mode=mode, role_name=app): return "" return mode def guide_variant( app: str, variants_per_app: Mapping[str, Sequence[Mapping[str, Any]]] | None, allowed: Collection[str], ) -> tuple[str, str]: """The variant of *app* that carries the replay, and the mode it runs in. ``('', '')`` when the role has no instructions to replay or this sweep deploys no variant that could carry them. Args: app: the role. variants_per_app: rendered variant configs per app; without them no variant can be told to be the smallest one. allowed: the variants this sweep deploys, as the matrix writes them. """ variants = (variants_per_app or {}).get(app) if not variants: return "", "" mode = guide_deployable(app) if not mode: return "", "" variant = smallest_variant(variants, allowed) return (variant, mode) if variant else ("", "")