mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 10:59:59 +00:00
A module that dies on an unhandled exception carries no msg: the AnsiballZ wrapper returns only failed and exception, whose event holds the sole account of what went wrong. The aggregation defaulted to the literal string package install failed and discarded it, so an AUR fetch that threw in run 30713666732 reported nothing at all and took a full investigation to place. Walk msg, exception.event.msg, exception, module_stderr, stderr instead, and tag every result with the module that produced it. This applies to apt, dnf, pacman and every other module the plugin drives, not only to the AUR. The aur module fetches the RPC record and the snapshot tarball through bare open_url calls with no error handling, so one blip kills a deploy; distribution repositories need none of this because apt, dnf and pacman retry inside their own transport. Give the calls that reach an endpoint of their own a single shared policy. Only the AUR path is declared today; copr, ppa and build carry it against the day they are. Split the result along its responsibilities so no file passes 250 lines: calls.py holds the vocabulary, sources.py one builder per acquisition path, plan.py only the dispatcher, outcome.py the reporting and facts.py the fact shapes. The action plugin is left with executing modules. Tests follow the code. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
124 lines
4.5 KiB
Python
124 lines
4.5 KiB
Python
"""Lint: every declaration produces an executable install plan.
|
|
|
|
A declaration can pass the shape and coverage lints and still blow up at
|
|
deploy time, because :func:`~utils.packages.plan.build_plan` reads keys a
|
|
repository definition may omit. This walks the real registry - every id,
|
|
every default distribution, both states - and requires a plan that names
|
|
only modules a deploy can actually resolve.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from utils.packages.calls import STATE_ABSENT, STATE_PRESENT
|
|
from utils.packages.plan import build_plan
|
|
from utils.packages.registry import build_registry, resolve
|
|
from utils.packages.schema import DISTRO_FAMILY
|
|
|
|
from . import PROJECT_ROOT
|
|
|
|
ALLOWED_MODULES: frozenset[str] = frozenset(
|
|
{
|
|
"ansible.builtin.command",
|
|
"ansible.builtin.dnf",
|
|
"ansible.builtin.lineinfile",
|
|
"ansible.builtin.package",
|
|
"ansible.builtin.replace",
|
|
"ansible.builtin.user",
|
|
"ansible.builtin.yum_repository",
|
|
"community.general.copr",
|
|
"community.general.pacman",
|
|
"kewlfft.aur.aur",
|
|
}
|
|
)
|
|
"""Modules a package install may invoke. Widening this set means a deploy
|
|
gains a new dependency, which is a decision, not a side effect: add the
|
|
collection to requirements/requirements.galaxy.yml in the same change."""
|
|
|
|
|
|
class TestPackagePlans(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.registry = build_registry(PROJECT_ROOT)
|
|
|
|
def _plans(self):
|
|
for package_id, declaration in sorted(self.registry.items()):
|
|
for distro in sorted(DISTRO_FAMILY):
|
|
spec = resolve(declaration, distro)
|
|
if spec is None:
|
|
continue
|
|
for state in (STATE_PRESENT, STATE_ABSENT):
|
|
yield package_id, distro, state, spec
|
|
|
|
def _buildable(self):
|
|
"""Plans that build, so only one test reports an unbuildable one."""
|
|
for package_id, distro, state, spec in self._plans():
|
|
try:
|
|
plan = build_plan(spec, state)
|
|
except Exception:
|
|
continue
|
|
yield package_id, distro, state, spec, plan
|
|
|
|
def test_every_declaration_builds_a_plan(self) -> None:
|
|
offenders: list[str] = []
|
|
checked = 0
|
|
for package_id, distro, state, spec in self._plans():
|
|
checked += 1
|
|
try:
|
|
build_plan(spec, state)
|
|
except Exception as exc:
|
|
offenders.append(
|
|
f"{package_id} / {distro} / {state}: {type(exc).__name__}: {exc}"
|
|
)
|
|
|
|
self.assertGreater(checked, 0, "the registry resolved nothing to plan")
|
|
if offenders:
|
|
self.fail(
|
|
f"{len(offenders)} of {checked} declaration(s) raise while their "
|
|
"install plan is built, so the deploy would fail on them:\n"
|
|
+ "\n".join(f" - {o}" for o in offenders)
|
|
)
|
|
|
|
def test_plans_name_only_allowed_modules(self) -> None:
|
|
offenders: list[str] = []
|
|
for package_id, distro, state, _spec, plan in self._buildable():
|
|
offenders.extend(
|
|
f"{package_id} / {distro} / {state}: '{call.module}'"
|
|
for call in plan
|
|
if call.module not in ALLOWED_MODULES
|
|
)
|
|
|
|
if offenders:
|
|
self.fail(
|
|
f"{len(offenders)} plan step(s) name a module outside the "
|
|
"allowed set. Declare the collection in "
|
|
"requirements/requirements.galaxy.yml and add the module to "
|
|
"ALLOWED_MODULES in the same change:\n"
|
|
+ "\n".join(f" - {o}" for o in offenders)
|
|
)
|
|
|
|
def test_a_present_plan_installs_the_declared_names(self) -> None:
|
|
offenders: list[str] = []
|
|
for package_id, distro, state, spec, plan in self._buildable():
|
|
if state != STATE_PRESENT or not spec.names:
|
|
continue
|
|
installed: set[str] = set()
|
|
for call in plan:
|
|
name = call.args.get("name")
|
|
installed.update(name if isinstance(name, list) else [])
|
|
missing = sorted(set(spec.names) - installed)
|
|
if missing:
|
|
offenders.append(
|
|
f"{package_id} / {distro}: plan never installs {', '.join(missing)}"
|
|
)
|
|
|
|
if offenders:
|
|
self.fail(
|
|
f"{len(offenders)} plan(s) drop a declared name on the way to "
|
|
"the install call:\n" + "\n".join(f" - {o}" for o in offenders)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|