feat(services): introduce canonical field for shared-role service entries

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 <noreply@anthropic.com>
This commit is contained in:
kevinveenbirkenbachandClaude Sonnet 4.6 committed 2026-03-30 15:58:54 +02:00
1 parent 563b0c3c8c
commit c74bd98969
4 files changed
+225 -6

No files matched your search

+7 -1
View File
@@ -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.<key>.type.
# canonical: <key>
# 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
+15 -5
View File
@@ -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(
@@ -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()
@@ -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()