diff --git a/roles/sys-ctl-hlth-disc-space/tasks/01_core.yml b/roles/sys-ctl-hlth-disc-space/tasks/01_core.yml index 6fe8fb97..d7779a68 100644 --- a/roles/sys-ctl-hlth-disc-space/tasks/01_core.yml +++ b/roles/sys-ctl-hlth-disc-space/tasks/01_core.yml @@ -4,7 +4,7 @@ when: run_once_sys_ctl_alm_compose is not defined - include_role: - name: sys-service + name: sys-service vars: system_service_on_calendar: "{{ SYS_SCHEDULE_HEALTH_DISC_SPACE }}" system_service_timer_enabled: true diff --git a/tests/integration/test_sys_service_requires_system_service_id.py b/tests/integration/test_sys_service_requires_system_service_id.py new file mode 100644 index 00000000..a7bc99f0 --- /dev/null +++ b/tests/integration/test_sys_service_requires_system_service_id.py @@ -0,0 +1,85 @@ +#!/usr/bin/env python3 +import os +import glob +import unittest +import yaml + + +def _safe_yaml_load(path): + try: + with open(path, "r", encoding="utf-8") as f: + doc = yaml.safe_load(f) + # A tasks file can be a list (usual) or a dict (blocks, etc.) + return doc + except Exception as e: + raise AssertionError(f"Failed to parse YAML: {path}\n{e}") from e + + +class TestSysServiceRequiresSystemServiceId(unittest.TestCase): + def setUp(self): + # Repo root = three levels up from this file: tests/integration/.py + self.repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) + self.roles_dir = os.path.join(self.repo_root, "roles") + self.assertTrue(os.path.isdir(self.roles_dir), f"'roles' directory not found at: {self.roles_dir}") + + def _iter_task_files(self, role_dir): + tasks_dir = os.path.join(role_dir, "tasks") + if not os.path.isdir(tasks_dir): + return + patterns = ["*.yml", "*.yaml"] + for pattern in patterns: + for path in glob.glob(os.path.join(tasks_dir, pattern)): + yield path + + # also scan nested includes like tasks/**/*.yml + for pattern in patterns: + for path in glob.glob(os.path.join(tasks_dir, "**", pattern), recursive=True): + yield path + + def _role_includes_sys_service(self, tasks_doc) -> bool: + """ + Return True if any task includes: + - include_role: + name: sys-service + """ + def check_task(task): + if not isinstance(task, dict): + return False + if "include_role" not in task: + return False + inc = task.get("include_role") + if isinstance(inc, dict): + name = inc.get("name") + return name == "sys-service" + # Rare shorthand: include_role: sys-service (not common, but just in case) + if isinstance(inc, str): + return inc.strip() == "sys-service" + return False + + # tasks_doc can be a list, dict (with 'block'), or None + if isinstance(tasks_doc, list): + for t in tasks_doc: + if check_task(t): + return True + # handle blocks within list items + if isinstance(t, dict) and "block" in t and isinstance(t["block"], list): + for bt in t["block"]: + if check_task(bt): + return True + elif isinstance(tasks_doc, dict): + # top-level block file (rare) + if "block" in tasks_doc and isinstance(tasks_doc["block"], list): + for bt in tasks_doc["block"]: + if check_task(bt): + return True + return False + + def _vars_has_system_service_id(self, role_dir): + vars_dir = os.path.join(role_dir, "vars") + if not os.path.isdir(vars_dir): + return (False, "vars/ directory not found") + candidates = [] + candidates.extend(glob.glob(os.path.join(vars_dir, "main.yml"))) + candidates.extend(glob.glob(os.path.join(vars_dir, "main.yaml"))) + if not candidates: + return diff --git a/tests/integration/test_system_service_id_matches_role.py b/tests/integration/test_system_service_id_matches_role.py new file mode 100644 index 00000000..935d1227 --- /dev/null +++ b/tests/integration/test_system_service_id_matches_role.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python3 +import os +import glob +import unittest +import yaml + + +class TestSystemServiceIdMatchesRole(unittest.TestCase): + def setUp(self): + # Repo root = three levels up from this file: tests/integration/.py + self.repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) + self.roles_dir = os.path.join(self.repo_root, "roles") + self.assertTrue(os.path.isdir(self.roles_dir), f"'roles' directory not found at: {self.roles_dir}") + + def _load_yaml(self, path: str): + with open(path, "r", encoding="utf-8") as f: + return yaml.safe_load(f) or {} + + def test_system_service_id_equals_role_name(self): + role_dirs = [d for d in os.listdir(self.roles_dir) + if os.path.isdir(os.path.join(self.roles_dir, d))] + + self.assertGreater(len(role_dirs), 0, f"No role directories found in {self.roles_dir}") + + for role in sorted(role_dirs): + with self.subTest(role=role): + vars_dir = os.path.join(self.roles_dir, role, "vars") + if not os.path.isdir(vars_dir): + continue + + candidates = [] + candidates.extend(glob.glob(os.path.join(vars_dir, "main.yml"))) + candidates.extend(glob.glob(os.path.join(vars_dir, "main.yaml"))) + if not candidates: + continue + + vars_file = sorted(candidates, key=lambda p: (not p.endswith("main.yml"), p))[0] + data = self._load_yaml(vars_file) + + if "system_service_id" not in (data or {}): + continue + + value = str(data.get("system_service_id")).strip() + allowed = {role, role + "@", "{{ application_id }}"} + + self.assertIn( + value, + allowed, + ( + f"[{role}] system_service_id mismatch in {vars_file}.\n" + f" Allowed: {sorted(allowed)}\n" + f" Actual: {value}" + ) + ) + + +if __name__ == "__main__": + unittest.main()