Added ports tests

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\.(?P<host>localhost|public)\.(?P<cat>[A-Za-z0-9_]+)\.(?P<svc>[A-Za-z0-9_]+)\b"),
+            # bracket notation: ports.host.cat['svc'] or ["svc"]
+            re.compile(r"ports\.(?P<host>localhost|public)\.(?P<cat>[A-Za-z0-9_]+)\[\s*['\"](?P<svc>[A-Za-z0-9_]+)['\"]\s*\]"),
+            # get(): ports.host.cat.get('svc')
+            re.compile(r"ports\.(?P<host>localhost|public)\.(?P<cat>[A-Za-z0-9_]+)\.get\(\s*['\"](?P<svc>[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.<host>.<category>.<service> 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()

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()