scripts.check_env_fleet_agreement

Reference › scripts › check_env_fleet_agreement

The env and fleet agreement gate: the endpoints committed in the environment profiles must be the endpoints the infrastructure-as-code produces. tests/scripts/test_check_env_fleet_agreement.py imports it, and it reports through the tracer in tracer. See also check_env_subnet_scheme, which checks the profiles' address scheme.

Invocation: main accepts --tf-dir (the Terraform or OpenTofu directory) and passes every other argument on to FleetAgreementGate.execute. If the tofu state cannot be read it prints a FAIL line and returns 1.

FleetAgreementGate.from_tofu reads the outputs of the infrastructure state through a command runner, and check walks the gated fleet profiles. For each profile it derives the expected environment values from the Terraform outputs with an EnvDeriver, reads the committed file (_declared returns the declared keys and the keys owned by the generated block, _owned) and compares them (_compare, _compare_output). Each mismatch becomes an AgreementViolation that can render itself.

At a glance

Class diagram

Dependency diagram

API

Classes

AgreementViolation (dataclass)

Decorators: dataclass(frozen=True) · Line 62

One owned key whose committed state disagrees with the derivation.

FleetAgreementGate

Line 100

Compares the derived endpoints to every committed file that carries them.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/check_env_fleet_agreement.py

#!/usr/bin/env python3
"""Env/fleet agreement gate — the committed endpoints must be the IaC's endpoints.

``env/test.env``, ``env/dev.env`` and the ``TEST_*`` keys of ``.env.example`` are
DERIVED from ``tofu output -json`` (``infra/gen_env.py --profile test|dev``,
``--template``). This gate re-derives them and FAILS on:

* any owned key whose committed value differs from the derived one;
* any owned key that is commented out (``# KEY=...``) or absent — the 2026-09-21
  CHANGEME sweep commented ``TEST_MEMGRAPH_URI`` out of the host ``.env`` and the
  routing guard only noticed because a test happened to need it;
* a ``<profile>_fleet_endpoints`` terraform output (once applied) that disagrees
  with the derivation;
* a missing ``tofu`` binary or a missing/empty state (ADR-0077 item 4: missing is
  FAIL, never skip).

Measured 2026-10-08: ``.env.example`` carried ``TEST_MEMGRAPH_URI`` on the retired
prod-bridge address ``10.53.1.167`` for two months with nothing checking it.

The prod profile is not gated: ``env/prod.env`` deliberately routes through the
coordinators (``neo4j://...:7691``), not the data-node list ``gen_env.py`` prints.

Exit 0 = agreement, 1 = at least one violation or an unreadable state.
"""

from __future__ import annotations

import argparse
import re
import subprocess
import sys
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path

_REPO_ROOT = Path(__file__).resolve().parents[1]
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from infra.gen_env import (  # noqa: E402
    _TF_DIR,
    CommandRunner,
    EnvDeriver,
    FleetProfile,
    FleetStateError,
    SubprocessRunner,
    TerraformOutputs,
    tofu_binary,
)
from orchestrata_ultimate.core.tracing.tracer import Tracer  # noqa: E402

_ACTIVE = re.compile(r"^\s*([A-Z][A-Z0-9_]*)=(.*)$")
_COMMENTED = re.compile(r"^\s*#+\s*([A-Z][A-Z0-9_]*)=")
_TEMPLATE = ".env.example"
_GATED = (FleetProfile.test, FleetProfile.dev)
COMMENTED_OUT = "<commented out>"
ABSENT = "<absent>"
FILE_MISSING = "<file missing>"


@dataclass(frozen=True)
class AgreementViolation:
    """One owned key whose committed state disagrees with the derivation."""

    env_file: str
    key: str
    declared: str
    derived: str

    def render(self) -> str:
        return (
            f"{self.env_file}: {self.key}: declared {self.declared!r}, "
            f"derived {self.derived!r}"
        )


def _declared(path: Path) -> tuple[dict[str, str], set[str]]:
    """(active KEY->value, commented-out KEYs) of an env file; inline ``  # note`` stripped."""
    active: dict[str, str] = {}
    commented: set[str] = set()
    for line in path.read_text().splitlines():
        match = _ACTIVE.match(line)
        if match:
            active[match.group(1)] = re.split(r"\s+#", match.group(2), maxsplit=1)[
                0
            ].strip()
            continue
        hidden = _COMMENTED.match(line)
        if hidden:
            commented.add(hidden.group(1))
    return active, commented


def _owned(lines: Sequence[str]) -> dict[str, str]:
    return dict(
        ln.split("=", 1) for ln in lines if "=" in ln and not ln.startswith("#")
    )


class FleetAgreementGate:
    """Compares the derived endpoints to every committed file that carries them."""

    def __init__(
        self,
        outputs: TerraformOutputs,
        deriver: EnvDeriver | None = None,
        tracer: Tracer | None = None,
    ) -> None:
        self._outputs = outputs
        self._tracer = tracer if tracer is not None else Tracer.default()
        self._deriver = (
            deriver if deriver is not None else EnvDeriver(outputs, self._tracer)
        )

    @classmethod
    def from_tofu(cls, tf_dir: Path, runner: CommandRunner) -> FleetAgreementGate:
        return cls(TerraformOutputs.from_tofu(tf_dir, runner))

    def check(
        self, repo_root: Path, profiles: Sequence[FleetProfile] = _GATED
    ) -> list[AgreementViolation]:
        """Every disagreement between derived and committed (empty = agreement)."""
        with self._tracer.span("FleetAgreementGate.check"):
            found: list[AgreementViolation] = []
            for profile in profiles:
                derived = _owned(self._deriver.env_lines(profile))
                found += self._compare(profile.env_file(repo_root), repo_root, derived)
                found += self._compare_output(profile, derived)
                if profile is FleetProfile.test:
                    template = _owned(self._deriver.template_lines(profile))
                    found += self._compare(repo_root / _TEMPLATE, repo_root, template)
            return found

    def _compare(
        self, path: Path, repo_root: Path, derived: Mapping[str, str]
    ) -> list[AgreementViolation]:
        name = str(path.relative_to(repo_root))
        if not path.is_file():
            return [
                AgreementViolation(name, k, FILE_MISSING, v) for k, v in derived.items()
            ]
        active, commented = _declared(path)
        out: list[AgreementViolation] = []
        for key, want in derived.items():
            if key in active:
                if active[key] != want:
                    out.append(AgreementViolation(name, key, active[key], want))
            elif key in commented:
                out.append(AgreementViolation(name, key, COMMENTED_OUT, want))
            else:
                out.append(AgreementViolation(name, key, ABSENT, want))
        return out

    def _compare_output(
        self, profile: FleetProfile, derived: Mapping[str, str]
    ) -> list[AgreementViolation]:
        declared = self._outputs.declared_endpoints(profile)
        if declared is None:
            return []
        name = f"infra/terraform output {profile.value}_fleet_endpoints"
        return [
            AgreementViolation(name, key, str(declared.get(key, ABSENT)), want)
            for key, want in derived.items()
            if declared.get(key) != want
        ]

    def execute(self, argv: Sequence[str]) -> int:
        """CLI body: report violations, return the exit code."""
        parser = argparse.ArgumentParser(description=__doc__)
        parser.add_argument("--repo-root", type=Path, default=_REPO_ROOT)
        parser.add_argument(
            "--profile",
            action="append",
            choices=[p.value for p in _GATED],
            help="Profile to check (repeatable; default: test and dev).",
        )
        args = parser.parse_args(list(argv))
        profiles = [FleetProfile(p) for p in args.profile] if args.profile else _GATED
        try:
            violations = self.check(args.repo_root, profiles)
        except FleetStateError as exc:
            print(f"FAIL: {exc}", file=sys.stderr)
            return 1
        for violation in violations:
            print(f"FAIL: {violation.render()}")
        if violations:
            return 1
        print(
            "env/fleet agreement OK: "
            + ", ".join(p.value for p in profiles)
            + f" derived from terraform match their committed files and {_TEMPLATE}"
        )
        return 0


def main(argv: Sequence[str] | None = None, runner: CommandRunner | None = None) -> int:
    parser = argparse.ArgumentParser(add_help=False)
    parser.add_argument("--tf-dir", type=Path, default=_TF_DIR)
    known, rest = parser.parse_known_args(list(sys.argv[1:] if argv is None else argv))
    if "-h" in rest or "--help" in rest:
        rest = ["--help"]
    try:
        gate = FleetAgreementGate.from_tofu(
            known.tf_dir, runner if runner is not None else SubprocessRunner()
        )
    except (subprocess.CalledProcessError, FileNotFoundError) as exc:
        print(f"FAIL: {tofu_binary()} could not read the state: {exc}", file=sys.stderr)
        return 1
    return gate.execute(rest)


if __name__ == "__main__":
    sys.exit(main())