From 80ca12938b963f80e4aa32ec11e8428e660391ec Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Sat, 12 Jul 2025 16:27:23 +0200 Subject: [PATCH] Added ports tests --- .../test_ports_references_validity.py | 81 +++++++++++++++++++ tests/integration/test_ports_uniqueness.py | 76 +++++++++++++++++ 2 files changed, 157 insertions(+) create mode 100644 tests/integration/test_ports_references_validity.py create mode 100644 tests/integration/test_ports_uniqueness.py diff --git a/tests/integration/test_ports_references_validity.py b/tests/integration/test_ports_references_validity.py new file mode 100644 index 00000000..f184f826 --- /dev/null +++ b/tests/integration/test_ports_references_validity.py @@ -0,0 +1,81 @@ +import os +import unittest +import yaml +import re + +class TestPortReferencesValidity(unittest.TestCase): + @classmethod + def setUpClass(cls): + # locate and load the ports definition + base = os.path.dirname(__file__) + cls.ports_file = os.path.abspath( + os.path.join(base, '..', '..', 'group_vars', 'all', '09_ports.yml') + ) + if not os.path.isfile(cls.ports_file): + raise FileNotFoundError(f"{cls.ports_file} does not exist.") + + with open(cls.ports_file, 'r', encoding='utf-8') as f: + data = yaml.safe_load(f) or {} + + # collect all valid (host, category, service) triples + cls.valid = set() + for host, cats in data.get('ports', {}).items(): + if not isinstance(cats, dict): + continue + for cat, services in cats.items(): + if not isinstance(services, dict): + continue + for svc in services: + cls.valid.add((host, cat, svc)) + + # prepare regex patterns for the allowed reference forms + cls.patterns = [ + # dot notation: ports.host.cat.svc + re.compile(r"ports\.(?Plocalhost|public)\.(?P[A-Za-z0-9_]+)\.(?P[A-Za-z0-9_]+)\b"), + # bracket notation: ports.host.cat['svc'] or ["svc"] + re.compile(r"ports\.(?Plocalhost|public)\.(?P[A-Za-z0-9_]+)\[\s*['\"](?P[A-Za-z0-9_]+)['\"]\s*\]"), + # get(): ports.host.cat.get('svc') + re.compile(r"ports\.(?Plocalhost|public)\.(?P[A-Za-z0-9_]+)\.get\(\s*['\"](?P[A-Za-z0-9_]+)['\"]\s*\)"), + ] + + def test_port_references_point_to_defined_ports(self): + """ + Scan all .j2, .yml, .yaml files under roles/, group_vars/, host_vars/, tasks/, + templates/, and playbooks/ for any ports... references + (dot, [''], or .get('')) and verify each triple is defined in 09_ports.yml. + """ + project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')) + dirs_to_scan = ['roles', 'group_vars', 'host_vars', 'tasks', 'templates', 'playbooks'] + + errors = [] + + for dirname in dirs_to_scan: + dirpath = os.path.join(project_root, dirname) + if not os.path.isdir(dirpath): + continue + + for root, _, files in os.walk(dirpath): + for fname in files: + if not (fname.endswith(('.j2', '.yml', '.yaml'))): + continue + + path = os.path.join(root, fname) + with open(path, 'r', encoding='utf-8', errors='ignore') as fh: + for lineno, line in enumerate(fh, start=1): + for pat in self.patterns: + for m in pat.finditer(line): + triple = (m.group('host'), m.group('cat'), m.group('svc')) + if triple not in self.valid: + errors.append( + f"{path}:{lineno}: reference `{m.group(0)}` " + f"not found in {self.ports_file}" + ) + + if errors: + self.fail( + "Found invalid port references:\n" + + "\n".join(errors) + ) + +if __name__ == '__main__': + unittest.main() diff --git a/tests/integration/test_ports_uniqueness.py b/tests/integration/test_ports_uniqueness.py new file mode 100644 index 00000000..c4ccd622 --- /dev/null +++ b/tests/integration/test_ports_uniqueness.py @@ -0,0 +1,76 @@ +import os +import unittest +import yaml +from collections import defaultdict + +class TestPortsUniqueness(unittest.TestCase): + @classmethod + def setUpClass(cls): + base_dir = os.path.dirname(__file__) + cls.ports_file = os.path.abspath( + os.path.join(base_dir, '..', '..', 'group_vars', 'all', '09_ports.yml') + ) + # Try to load data; leave it as None if missing or invalid YAML + try: + with open(cls.ports_file, 'r', encoding='utf-8') as f: + cls.data = yaml.safe_load(f) or {} + except FileNotFoundError: + cls.data = None + except yaml.YAMLError as e: + raise RuntimeError(f"Failed to parse {cls.ports_file}: {e}") + + def test_ports_file_exists(self): + """Fail if the ports file is missing.""" + self.assertTrue( + os.path.isfile(self.ports_file), + f"{self.ports_file} does not exist." + ) + + def _collect_ports(self, section): + """ + Helper to collect a mapping from port -> list of 'category.service' identifiers + for a given section ('localhost' or 'public'). + """ + ports_section = self.data.get('ports', {}).get(section, {}) + port_map = defaultdict(list) + + for category, services in ports_section.items(): + if not isinstance(services, dict): + continue + for service, port in services.items(): + try: + port_num = int(port) + except (TypeError, ValueError): + self.fail(f"Invalid port value for {section}.{category}.{service}: {port!r}") + identifier = f"{category}.{service}" + port_map[port_num].append(identifier) + return port_map + + def _assert_unique(self, port_map, section): + """ + Assert no port number is assigned more than once in the given section. + """ + duplicates = {p: svcs for p, svcs in port_map.items() if len(svcs) > 1} + if duplicates: + msgs = [ + f"Port {p} is duplicated for services: {', '.join(svcs)}" + for p, svcs in duplicates.items() + ] + self.fail(f"Duplicate {section} ports found:\n" + "\n".join(msgs)) + + def test_unique_localhost_ports(self): + """All localhost‐exposed ports must be unique.""" + # Ensure the file was loaded + self.assertIsNotNone(self.data, f"{self.ports_file} does not exist or is unreadable.") + port_map = self._collect_ports('localhost') + self._assert_unique(port_map, 'localhost') + + def test_unique_public_ports(self): + """All public‐exposed ports must be unique.""" + # Ensure the file was loaded + self.assertIsNotNone(self.data, f"{self.ports_file} does not exist or is unreadable.") + port_map = self._collect_ports('public') + self._assert_unique(port_map, 'public') + +if __name__ == '__main__': + unittest.main()