Files
computer-playbook/tests/unit/python/utils/roles/test_meta_lookup.py
T
kevinveenbirkenbachandClaude Opus 5 2a7455bd90 feat(ci): spread the deploy matrix over amd64 and arm64
Both architectures were claimed to work and only one was ever exercised. The
deploy matrix now carries an architecture the way it carries mode, tor, distro
and filesystem, so every run proves both.

A job cannot ask for an architecture the way it asks for a distro, because the
architecture is the machine rather than something the job installs on it. The
matrix therefore carries a runner label and `runs-on` reads it, and an
architecture without a label in RUNNERS is a build error rather than a row that
lands on the wrong hardware and still reports green. assert_architecture.sh
checks the runner it actually got, because a label that stops resolving to
arm64 would otherwise go unnoticed forever.

Assignment is rotation, not chance: two architectures against five distros and
fifteen distro/filesystem positions are coprime enough that every combination
comes round. A role narrows the pool through `architectures` in its
meta/services.yml; without one it takes whatever the rotation hands it. An
empty intersection aborts instead of silently removing the role from the run.

Images are built per architecture and pushed under an architecture-suffixed
tag, then merged into the plain tag by manifest_all.sh; pushing the plain tag
from both builds would let the last one win and send half the rows to the wrong
platform. cleanup.sh now protects the `ci-<sha>-<architecture>` children of a
protected tag, because the plain tag is an index over them and deleting a child
leaves it unpullable on that platform.

label.py is split out of axes.py, which the 500-line limit no longer fits.
Writing a job title and reading one back belong together: a retrigger that
recovers an axis wrongly replays a different deploy than the one that failed,
and a drifting artifact name makes the failure reporter download nothing. The
leading `.*` in LABEL_RE is greedy on purpose so it anchors on the last mode
glyph; a reusable-workflow caller path carries a glyph of its own, and
test_selection.py guards the case.

Verified: 313 unit tests and 48 subtests across utils/github/variant,
utils/roles and cli/meta/ci. Locally, a deploy pinned to arm64 reaches ok=1444
and its image builds log aarch64-linux-gnu paths.

Not verified here: actionlint over the seven changed workflows, and a full
`make test`. The matrix doubles the row count, so runtime and runner
availability change for every CI run from here.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-19 07:02:44 +02:00

338 lines
10 KiB
Python

"""Unit tests for utils.roles.meta_lookup.
Covers:
* service role with run_after / lifecycle on its primary entity;
* multi-entity role where the primary entity is a metadata-only holder;
* absent run_after (returns []);
* absent lifecycle (returns None);
* malformed meta/services.yml (raises a clear error).
"""
from __future__ import annotations
import textwrap
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
from utils.roles.mapping import ROLE_FILE_META_SERVICES
from utils.roles.meta_lookup import (
MetaServicesShapeError,
get_role_architectures,
get_role_lifecycle,
get_role_placement,
get_role_run_after,
get_role_skip,
iter_roles_with_placement,
)
class _RoleFixtures:
def __init__(self, root: Path):
self.root = root
def write(self, role_name: str, body: str) -> Path:
role_dir = self.root / role_name
(role_dir / "meta").mkdir(parents=True, exist_ok=True)
path = role_dir / ROLE_FILE_META_SERVICES
path.write_text(textwrap.dedent(body).lstrip("\n"), encoding="utf-8")
return role_dir
class TestMetaLookup(unittest.TestCase):
def setUp(self) -> None:
self.tmp = TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.fx = _RoleFixtures(Path(self.tmp.name))
def test_service_role_with_run_after_and_lifecycle(self) -> None:
role_dir = self.fx.write(
"web-app-gitea",
"""
gitea:
image: gitea/gitea
run_after:
- svc-db-postgres
- web-app-keycloak
lifecycle: stable
""",
)
self.assertEqual(
get_role_run_after(role_dir, role_name="web-app-gitea"),
["svc-db-postgres", "web-app-keycloak"],
)
self.assertEqual(
get_role_lifecycle(role_dir, role_name="web-app-gitea"),
"stable",
)
def test_multi_entity_metadata_only_primary_holder(self) -> None:
role_dir = self.fx.write(
"web-app-bluesky",
"""
bluesky:
run_after:
- web-app-keycloak
lifecycle: alpha
api:
ports:
local:
http: 8030
web:
ports:
local:
http: 8031
""",
)
self.assertEqual(
get_role_run_after(role_dir, role_name="web-app-bluesky"),
["web-app-keycloak"],
)
self.assertEqual(
get_role_lifecycle(role_dir, role_name="web-app-bluesky"),
"alpha",
)
def test_returns_empty_list_when_run_after_absent(self) -> None:
role_dir = self.fx.write(
"web-app-yourls",
"""
yourls:
lifecycle: beta
""",
)
self.assertEqual(
get_role_run_after(role_dir, role_name="web-app-yourls"),
[],
)
def test_returns_none_when_lifecycle_absent(self) -> None:
role_dir = self.fx.write(
"web-app-yourls",
"""
yourls:
run_after:
- svc-db-postgres
""",
)
self.assertIsNone(get_role_lifecycle(role_dir, role_name="web-app-yourls"))
def test_returns_empty_when_services_file_missing(self) -> None:
role_dir = self.fx.root / "desk-something"
role_dir.mkdir(parents=True)
self.assertEqual(
get_role_run_after(role_dir, role_name="desk-something"),
[],
)
self.assertIsNone(get_role_lifecycle(role_dir, role_name="desk-something"))
def test_malformed_yaml_raises_clear_error(self) -> None:
role_dir = self.fx.root / "web-app-broken"
(role_dir / "meta").mkdir(parents=True)
(role_dir / ROLE_FILE_META_SERVICES).write_text(
"key: value\n:foo: ]invalid[\n - not\n", encoding="utf-8"
)
with self.assertRaises(MetaServicesShapeError):
get_role_run_after(role_dir, role_name="web-app-broken")
def test_non_dict_root_raises_clear_error(self) -> None:
role_dir = self.fx.root / "web-app-listroot"
(role_dir / "meta").mkdir(parents=True)
(role_dir / ROLE_FILE_META_SERVICES).write_text(
"- this\n- is\n- a list\n", encoding="utf-8"
)
with self.assertRaises(MetaServicesShapeError):
get_role_lifecycle(role_dir, role_name="web-app-listroot")
def test_placement_manager_is_returned(self) -> None:
role_dir = self.fx.write(
"svc-registry-cache",
"""
cache:
placement: manager
""",
)
self.assertEqual(
get_role_placement(role_dir, role_name="svc-registry-cache"),
"manager",
)
def test_placement_absent_returns_none(self) -> None:
role_dir = self.fx.write(
"web-app-yourls",
"""
yourls:
lifecycle: beta
""",
)
self.assertIsNone(get_role_placement(role_dir, role_name="web-app-yourls"))
def test_iter_roles_with_placement_filters_correctly(self) -> None:
self.fx.write(
"svc-registry-cache",
"cache:\n placement: manager\n",
)
self.fx.write(
"svc-registry-docker",
"docker:\n placement: manager\n",
)
self.fx.write(
"svc-prx-openresty",
"openresty:\n placement: manager\n",
)
self.fx.write(
"web-app-yourls",
"yourls:\n lifecycle: beta\n",
)
self.fx.write(
"web-app-only-worker",
"only-worker:\n placement: worker\n",
)
managers = iter_roles_with_placement("manager", roles_dir=self.fx.root)
self.assertEqual(
managers,
["svc-prx-openresty", "svc-registry-cache", "svc-registry-docker"],
)
def test_iter_roles_with_placement_empty_placement_returns_empty(
self,
) -> None:
self.fx.write(
"svc-registry-cache",
"cache:\n placement: manager\n",
)
self.assertEqual(
iter_roles_with_placement("", roles_dir=self.fx.root),
[],
)
def test_iter_roles_with_placement_missing_roles_dir(self) -> None:
self.assertEqual(
iter_roles_with_placement(
"manager", roles_dir=self.fx.root / "does-not-exist"
),
[],
)
def test_skip_returns_disabled_modes(self) -> None:
role_dir = self.fx.write(
"svc-storage-nfs-client",
"""
nfs-client:
modes:
compose:
enabled: false
swarm:
enabled: false
""",
)
self.assertEqual(
get_role_skip(role_dir, role_name="svc-storage-nfs-client"),
["compose", "swarm"],
)
def test_skip_absent_modes_returns_empty(self) -> None:
role_dir = self.fx.write(
"web-app-yourls",
"""
yourls:
lifecycle: beta
""",
)
self.assertEqual(get_role_skip(role_dir, role_name="web-app-yourls"), [])
def test_partial_modes_default_enabled(self) -> None:
role_dir = self.fx.write(
"web-app-yourls",
"""
yourls:
modes:
compose:
enabled: true
""",
)
self.assertEqual(get_role_skip(role_dir, role_name="web-app-yourls"), [])
def test_modes_enabled_non_bool_raises(self) -> None:
role_dir = self.fx.write(
"web-app-broken",
"""
broken:
modes:
compose:
enabled: maybe
""",
)
with self.assertRaises(MetaServicesShapeError):
get_role_skip(role_dir, role_name="web-app-broken")
class TestRoleArchitectures(unittest.TestCase):
"""``meta/services.yml`` narrows which hardware a role can deploy on."""
def setUp(self) -> None:
self._tmp = TemporaryDirectory()
self.fx = _RoleFixtures(Path(self._tmp.name))
self.addCleanup(self._tmp.cleanup)
def test_a_declared_subset_is_returned(self) -> None:
role_dir = self.fx.write(
"web-app-amd-only",
"""
amd-only:
architectures:
- amd64
""",
)
self.assertEqual(
get_role_architectures(role_dir, role_name="web-app-amd-only"),
["amd64"],
)
def test_no_declaration_means_the_role_takes_whatever_the_run_assigns(
self,
) -> None:
role_dir = self.fx.write(
"web-app-open",
"""
open:
lifecycle: beta
""",
)
self.assertEqual(get_role_architectures(role_dir, role_name="web-app-open"), [])
def test_a_missing_services_file_means_no_declaration(self) -> None:
role_dir = Path(self._tmp.name) / "web-app-bare"
(role_dir / "meta").mkdir(parents=True, exist_ok=True)
self.assertEqual(get_role_architectures(role_dir, role_name="web-app-bare"), [])
def test_an_unknown_architecture_raises(self) -> None:
role_dir = self.fx.write(
"web-app-typo",
"""
typo:
architectures:
- amd65
""",
)
with self.assertRaises(MetaServicesShapeError):
get_role_architectures(role_dir, role_name="web-app-typo")
def test_an_empty_list_raises_instead_of_dropping_every_row(self) -> None:
role_dir = self.fx.write(
"web-app-empty",
"""
empty:
architectures: []
""",
)
with self.assertRaises(MetaServicesShapeError):
get_role_architectures(role_dir, role_name="web-app-empty")
if __name__ == "__main__":
unittest.main()