computer-playbook/tests/lint/repository/test_distros_spot.py

"""Lint: ``meta/distros.yml`` is the only distro list.

The Python layer reads the SPOT at runtime via :mod:`utils.distros`. Two
places cannot: ``default.env`` is parsed by ``scripts/meta/env/load.sh``
before any Python exists, and a GitHub ``env:`` block is static YAML. Both
therefore carry a materialised copy, and these rules pin that copy to the
SPOT so a distro added in one place cannot go missing in the other.

The image-reference rule guards the second half of the SPOT: the pkgmgr
and environment templates live in ``meta/distros.yml``; reintroducing an
interpolated ``pkgmgr-<distro>`` reference in code is the regression that
this file exists to catch.
"""

from __future__ import annotations

import json
import os
import re
import subprocess
import tempfile
import unittest
from pathlib import Path

from utils.cache.files import read_text
from utils.cache.yaml import load_yaml_any
from utils.distros import (
    FILE_META_DISTROS,
    IMAGE_PKGMGR,
    IMAGE_PKGMGR_VIRGIN,
    dev_runtime_images,
    distro_names,
    image_template,
)
from utils.env.parser import parse_static_env

from . import PROJECT_ROOT

WORKFLOWS_DIR = PROJECT_ROOT / ".github" / "workflows"
WORKSPACE_WORKFLOW = WORKFLOWS_DIR / "call-test-workspace.yml"
_MATRIX_REF_RE = re.compile(
    r"^\$\{\{\s*fromJson\(\s*needs\.([\w-]+)\.outputs\.([\w-]+)\s*\)\s*\}\}$",
    re.IGNORECASE,
)
_STEP_REF_RE = re.compile(r"^\$\{\{\s*steps\.([\w-]+)\.outputs\.([\w-]+)\s*\}\}$")
_GITHUB_SHELL = ("bash", "--noprofile", "--norc", "-eo", "pipefail", "-c")

_INTERPOLATED_PKGMGR_RE = re.compile(r"pkgmgr-[{$]")
_SHELL_LOOP_SLUG = "${d}"
_LIST_SOURCES = ("*.yml", "*.sh", "*.md", "Makefile", "default.env")


class TestDistrosSpot(unittest.TestCase):
    def setUp(self) -> None:
        self.expected = " ".join(distro_names())

    def test_default_env_declares_no_distro_list(self) -> None:
        self.assertNotIn(
            "INFINITO_DISTROS",
            parse_static_env(PROJECT_ROOT / "default.env"),
            "INFINITO_DISTROS is generated by utils/env/handlers/infinito/"
            f"distros.py from {FILE_META_DISTROS}; a static default in "
            "default.env would be a second source of truth.",
        )

    def test_every_literal_distro_list_matches_spot(self) -> None:
        """A workflow ``env:``/``default:`` value, a shell array and a docs
        table cannot call the SPOT, so they carry the list verbatim. Any run
        of three or more declared distro names in a value position must
        therefore spell the SPOT in full: a stale copy that lost a distro
        fails here. Comment lines and YAML ``description:`` keys are prose
        that names a subset as an example, so they are not value positions.
        """
        run = re.compile(
            r"(?:\b(?:{names})\b[\s,]+){{2,}}\b(?:{names})\b".format(
                names="|".join(distro_names())
            )
        )
        this_file = Path(__file__).resolve()
        offenders: list[str] = []
        for pattern in _LIST_SOURCES:
            for path in sorted(PROJECT_ROOT.glob(f"**/{pattern}")):
                if {".git", ".claude", "node_modules"} & set(path.parts):
                    continue
                if not path.is_file():
                    continue
                if path.resolve() == this_file:
                    continue
                rel = path.relative_to(PROJECT_ROOT).as_posix()
                for idx, line in enumerate(read_text(str(path)).splitlines(), 1):
                    stripped = line.strip()
                    if stripped.startswith(("#", "description:")):
                        continue
                    for found in run.findall(line):
                        listed = re.split(r"[\s,]+", found.strip())
                        if listed != list(distro_names()):
                            offenders.append(f"{rel}:{idx}: {' '.join(listed)}")

        if offenders:
            body = "\n".join(f"  - {o}" for o in offenders)
            self.fail(
                f"Literal distro list(s) drifted from {FILE_META_DISTROS} "
                f"(expected {self.expected!r}):\n{body}"
            )

    def test_workspace_matrix_matches_spot(self) -> None:
        """The matrix is an expression, so the pin walks the wiring from the
        axis to the resolver step, runs that step the way the runner does and
        compares what it writes to ``GITHUB_OUTPUT`` against the SPOT."""
        jobs = load_yaml_any(str(WORKSPACE_WORKFLOW), default_if_missing={})["jobs"]
        consumer = jobs["test-workspace"]
        axis = consumer["strategy"]["matrix"]["include"]
        ref = _MATRIX_REF_RE.match(str(axis))
        self.assertIsNotNone(
            ref,
            f"call-test-workspace.yml must derive its matrix from {FILE_META_DISTROS} "
            f"via fromJson(needs.<job>.outputs.<name>), got {axis!r}.",
        )
        resolver_id, output_name = ref.groups()
        self.assertIn(
            resolver_id,
            consumer.get("needs", ()),
            f"job 'test-workspace' consumes {resolver_id!r} without needing it.",
        )
        resolver = jobs[resolver_id]
        published = resolver.get("outputs", {}).get(output_name)
        wiring = _STEP_REF_RE.match(str(published))
        self.assertIsNotNone(
            wiring,
            f"job {resolver_id!r} must publish {output_name!r} from a step "
            f"output, got {published!r}.",
        )
        step_id, step_key = wiring.groups()
        steps = [s for s in resolver["steps"] if s.get("id") == step_id]
        self.assertEqual(
            len(steps), 1, f"job {resolver_id!r} declares no step id {step_id!r}."
        )
        with tempfile.TemporaryDirectory() as tmp:
            output = Path(tmp) / "github_output"
            output.touch()
            proc = subprocess.run(
                [*_GITHUB_SHELL, steps[0]["run"]],
                cwd=PROJECT_ROOT,
                env={**os.environ, "GITHUB_OUTPUT": str(output)},
                capture_output=True,
                text=True,
                check=False,
            )
            self.assertEqual(proc.returncode, 0, proc.stderr)
            key, _, value = read_text(str(output)).strip().partition("=")
        self.assertEqual(
            key,
            step_key,
            f"the resolver writes {key!r}, {resolver_id!r} publishes {step_key!r}.",
        )
        entries = json.loads(value)
        self.assertEqual(
            tuple(entry["image"] for entry in entries),
            dev_runtime_images(),
            f"the {resolver_id!r} job's resolver drifted from {FILE_META_DISTROS}.",
        )
        self.assertTrue(
            all(entry["track"] for entry in entries),
            f"every {resolver_id!r} entry must carry a workspace track: {entries}",
        )

    def _sources(self, pattern: str):
        for path in sorted(PROJECT_ROOT.glob(f"**/{pattern}")):
            if ".git" in path.parts or "node_modules" in path.parts:
                continue
            for idx, line in enumerate(read_text(str(path)).splitlines(), 1):
                if _INTERPOLATED_PKGMGR_RE.search(line):
                    yield path.relative_to(PROJECT_ROOT).as_posix(), idx, line.strip()

    def test_python_and_make_compose_no_pkgmgr_reference(self) -> None:
        offenders = [
            f"{rel}:{idx}: {line}"
            for pattern in ("*.py", "Makefile")
            for rel, idx, line in self._sources(pattern)
        ]
        if offenders:
            body = "\n".join(f"  - {o}" for o in offenders)
            self.fail(
                f"pkgmgr image reference composed outside {FILE_META_DISTROS}:\n"
                f"{body}\n\nRender it via utils.distros.pkgmgr_image() or read "
                "INFINITO_PARENT_IMAGE from the env layer."
            )

    def test_shell_pkgmgr_references_match_the_spot_template(self) -> None:
        """Shell loops render ``{slug}`` from their own variable, so the SPOT
        cannot be called per distro. The literal must still be the SPOT
        template rendered with the declared owner and tag."""
        static = parse_static_env(PROJECT_ROOT / "default.env")
        allowed = {
            image_template(kind).format(
                owner=static["INFINITO_PARENT_IMAGE_OWNER"],
                slug=_SHELL_LOOP_SLUG,
                tag=static["INFINITO_PARENT_IMAGE_TAG"],
            )
            for kind in (IMAGE_PKGMGR, IMAGE_PKGMGR_VIRGIN)
        }

        offenders = [
            f"{rel}:{idx}: {line}"
            for rel, idx, line in self._sources("*.sh")
            if not any(ref in line for ref in allowed)
        ]
        if offenders:
            body = "\n".join(f"  - {o}" for o in offenders)
            self.fail(
                f"Shell pkgmgr reference does not match a {FILE_META_DISTROS} "
                f"template:\n{body}\n\nExpected one of: {sorted(allowed)}"
            )


if __name__ == "__main__":
    unittest.main()