From c74bd989690ebbb731d90bbd64e5abce84ecb5cc Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Mon, 30 Mar 2026 15:58:54 +0200 Subject: [PATCH] feat(services): introduce canonical field for shared-role service entries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds canonical: cdn to css/javascript entries in 20_services.yml so that role-based reverse lookup (web-svc-cdn → cdn) is deterministic. Updates _build_role_to_key to use the canonical field. Adds an integration test verifying bidirectional resolution and a lint test enforcing canonical consistency rules (exactly one primary, no chaining, matching roles). Co-Authored-By: Claude Sonnet 4.6 --- group_vars/all/20_services.yml | 8 +- plugins/lookup/service.py | 20 ++- tests/integration/test_services_canonical.py | 124 ++++++++++++++++++ tests/integration/test_services_resolvable.py | 79 +++++++++++ 4 files changed, 225 insertions(+), 6 deletions(-) create mode 100644 tests/integration/test_services_canonical.py create mode 100644 tests/integration/test_services_resolvable.py diff --git a/group_vars/all/20_services.yml b/group_vars/all/20_services.yml index 1a861796dd..405d87243d 100644 --- a/group_vars/all/20_services.yml +++ b/group_vars/all/20_services.yml @@ -5,9 +5,13 @@ # frontend: loaded by sys-front-inj-all for all deployed apps # backend: loaded by sys-stk-backend per application # -# Optional field: +# Optional fields: # role_template: "svc-db-{type}" # The placeholder {type} is substituted from compose.services..type. +# canonical: +# Required when multiple keys share the same role. Points to the primary +# key so that role-based reverse lookup always returns a deterministic id. +# The primary key must have no canonical field. Aliases must set canonical. # # See: docs/contributing/code/services.md @@ -23,10 +27,12 @@ services: css: role: web-svc-cdn type: frontend + canonical: cdn javascript: role: web-svc-cdn type: frontend + canonical: cdn logout: role: web-svc-logout diff --git a/plugins/lookup/service.py b/plugins/lookup/service.py index 2016ef6ebf..16f1b9f2ef 100644 --- a/plugins/lookup/service.py +++ b/plugins/lookup/service.py @@ -78,11 +78,21 @@ def _is_service_needed( def _build_role_to_key(services_map: Dict[str, Any]) -> Dict[str, str]: - return { - entry["role"]: key - for key, entry in services_map.items() - if isinstance(entry, dict) and "role" in entry - } + """Map each role name to its canonical service key. + + When multiple keys share the same role, the canonical key is determined by + the 'canonical' field on alias entries. The primary key (no 'canonical' + field) is used as the fallback so that role-based reverse lookup always + returns a single, deterministic id. + """ + result: Dict[str, str] = {} + for key, entry in services_map.items(): + if not isinstance(entry, dict) or "role" not in entry: + continue + role = entry["role"] + canonical_key = entry.get("canonical", key) + result[role] = canonical_key + return result def _resolve_term( diff --git a/tests/integration/test_services_canonical.py b/tests/integration/test_services_canonical.py new file mode 100644 index 0000000000..b270e1edff --- /dev/null +++ b/tests/integration/test_services_canonical.py @@ -0,0 +1,124 @@ +import os +import unittest +import yaml +from collections import defaultdict + + +class TestServicesCanonical(unittest.TestCase): + """Lint: canonical field in group_vars/all/20_services.yml must be consistent. + + Rules: + - If a role is shared by multiple keys, all alias keys MUST declare canonical. + - canonical must point to an existing key in the services map. + - The canonical target must share the same role as the alias. + - The canonical target must NOT itself have a canonical field (no chaining). + - Exactly one key per role must be the primary (no canonical field). + """ + + def setUp(self): + self.repo_root = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "..") + ) + services_file = os.path.join( + self.repo_root, "group_vars", "all", "20_services.yml" + ) + with open(services_file, encoding="utf-8") as f: + data = yaml.safe_load(f) or {} + self.services = data.get("services", {}) + self.assertGreater(len(self.services), 0, "services map must not be empty") + + # Build role → [keys] index + self.role_to_keys = defaultdict(list) + for key, entry in self.services.items(): + if isinstance(entry, dict) and "role" in entry: + self.role_to_keys[entry["role"]].append(key) + + def test_shared_roles_have_canonical(self): + """All keys sharing a role (except the primary) must declare canonical.""" + for role, keys in self.role_to_keys.items(): + if len(keys) < 2: + continue + primaries = [k for k in keys if "canonical" not in self.services[k]] + with self.subTest(role=role): + self.assertEqual( + len(primaries), + 1, + f"Role '{role}' is shared by {keys} but has " + f"{len(primaries)} primary key(s) (no canonical): {primaries}. " + f"Exactly one primary is required.", + ) + + def test_canonical_target_exists(self): + """canonical must reference an existing service key.""" + for key, entry in self.services.items(): + if not isinstance(entry, dict): + continue + canonical = entry.get("canonical") + if canonical is None: + continue + with self.subTest(key=key): + self.assertIn( + canonical, + self.services, + f"[{key}] canonical: '{canonical}' does not exist in services.", + ) + + def test_canonical_target_has_same_role(self): + """canonical target must map to the same role as the alias.""" + for key, entry in self.services.items(): + if not isinstance(entry, dict): + continue + canonical = entry.get("canonical") + if canonical is None: + continue + with self.subTest(key=key): + alias_role = entry.get("role") + target_entry = self.services.get(canonical, {}) + target_role = target_entry.get("role") if isinstance(target_entry, dict) else None + self.assertEqual( + alias_role, + target_role, + f"[{key}] canonical: '{canonical}' has role '{target_role}', " + f"but alias has role '{alias_role}'. Roles must match.", + ) + + def test_canonical_target_is_not_itself_an_alias(self): + """canonical must not point to another alias (no chaining).""" + for key, entry in self.services.items(): + if not isinstance(entry, dict): + continue + canonical = entry.get("canonical") + if canonical is None: + continue + with self.subTest(key=key): + target_entry = self.services.get(canonical, {}) + target_canonical = target_entry.get("canonical") if isinstance(target_entry, dict) else None + self.assertIsNone( + target_canonical, + f"[{key}] canonical: '{canonical}' is itself an alias " + f"(canonical: '{target_canonical}'). Chaining is not allowed.", + ) + + def test_unique_roles_must_not_have_canonical(self): + """A key that is the only key for its role must not declare canonical.""" + for key, entry in self.services.items(): + if not isinstance(entry, dict): + continue + canonical = entry.get("canonical") + if canonical is None: + continue + role = entry.get("role") + if role is None: + continue + with self.subTest(key=key): + siblings = self.role_to_keys.get(role, []) + self.assertGreater( + len(siblings), + 1, + f"[{key}] declares canonical: '{canonical}' but is the only " + f"key for role '{role}'. canonical is only needed for shared roles.", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/integration/test_services_resolvable.py b/tests/integration/test_services_resolvable.py new file mode 100644 index 0000000000..7d39a2fbbc --- /dev/null +++ b/tests/integration/test_services_resolvable.py @@ -0,0 +1,79 @@ +import os +import unittest +import yaml + +from plugins.lookup.service import LookupModule + + +class TestServicesResolvable(unittest.TestCase): + """Every entry in group_vars/all/20_services.yml must be resolvable + via the service lookup plugin both by service key and by role name.""" + + def setUp(self): + self.repo_root = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "..") + ) + services_file = os.path.join( + self.repo_root, "group_vars", "all", "20_services.yml" + ) + with open(services_file, encoding="utf-8") as f: + data = yaml.safe_load(f) or {} + self.services = data.get("services", {}) + self.assertGreater(len(self.services), 0, "services map must not be empty") + + def _run(self, term): + return LookupModule().run( + [term], + variables={ + "applications": {}, + "group_names": [], + "services": self.services, + }, + )[0] + + def _canonical_for(self, key): + entry = self.services[key] + return entry.get("canonical", key) + + def test_all_keys_resolvable(self): + """Every service key must resolve without error and return itself as id.""" + for key in self.services: + with self.subTest(key=key): + result = self._run(key) + self.assertEqual(result["id"], key) + + def test_all_roles_resolve_to_canonical(self): + """Role-based lookup must return the canonical key for that role.""" + seen_roles = set() + for key, entry in self.services.items(): + if not isinstance(entry, dict): + continue + role = entry.get("role") + if not role or role in seen_roles: + continue + seen_roles.add(role) + canonical = self._canonical_for(key) + with self.subTest(role=role, expected_canonical=canonical): + result = self._run(role) + self.assertEqual(result["role"], role) + self.assertEqual(result["id"], canonical) + + def test_key_and_role_produce_identical_result_for_primary_keys(self): + """For the canonical/primary key of a role, key and role lookups must agree.""" + for key, entry in self.services.items(): + if not isinstance(entry, dict): + continue + role = entry.get("role") + if not role: + continue + # Only test primary keys (no canonical field = this key IS canonical) + if "canonical" in entry: + continue + with self.subTest(key=key): + by_key = self._run(key) + by_role = self._run(role) + self.assertEqual(by_key, by_role) + + +if __name__ == "__main__": + unittest.main()