diff --git a/tests/test_port_alignment.py b/tests/test_port_alignment.py new file mode 100644 index 0000000..2b982bf --- /dev/null +++ b/tests/test_port_alignment.py @@ -0,0 +1,47 @@ +import sys +import tempfile +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools")) + +from port_alignment import build + + +class PortAlignmentTest(unittest.TestCase): + def test_reports_alignment(self): + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + (root / "mx_exporter").mkdir() + (root / "deployment" / "mx-exporter").mkdir(parents=True) + (root / "mx_exporter" / "__init__.py").write_text('parser.add_argument("-p", "--port", default=8000)\n', encoding="utf-8") + (root / "start_mxexporter.sh").write_text("HOST_PORT=8000\n", encoding="utf-8") + (root / "deployment" / "mx-exporter" / "mx-exporter-daemonset.yaml").write_text( + "port: 8000\ncontainerPort: 8000\n", encoding="utf-8" + ) + + report = build(root) + + self.assertTrue(report["aligned"]) + + def test_errors_when_yaml_ports_are_missing(self): + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + (root / "mx_exporter").mkdir() + (root / "deployment" / "mx-exporter").mkdir(parents=True) + (root / "mx_exporter" / "__init__.py").write_text( + 'parser.add_argument("-p", "--port", default=8000)\n', + encoding="utf-8", + ) + (root / "start_mxexporter.sh").write_text("HOST_PORT=8000\n", encoding="utf-8") + (root / "deployment" / "mx-exporter" / "mx-exporter-daemonset.yaml").write_text( + "kind: DaemonSet\n", + encoding="utf-8", + ) + + with self.assertRaises(ValueError): + build(root) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/port_alignment.py b/tools/port_alignment.py new file mode 100644 index 0000000..4166079 --- /dev/null +++ b/tools/port_alignment.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python3 +"""Audit mx-exporter port defaults across code and deployment files.""" + +from __future__ import annotations + +import argparse +import json +import re +from pathlib import Path + + +CLI_RE = re.compile(r'parser\.add_argument\("-p".*default=(\d+)') +SCRIPT_RE = re.compile(r"HOST_PORT=(\d+)") +YAML_PORT_RE = re.compile(r"(?:port|containerPort):\s*(\d+)") + + +def build(repo_root: Path) -> dict[str, object]: + cli_text = (repo_root / "mx_exporter" / "__init__.py").read_text(encoding="utf-8") + script_text = (repo_root / "start_mxexporter.sh").read_text(encoding="utf-8") + static_text = (repo_root / "deployment" / "mx-exporter" / "mx-exporter-daemonset.yaml").read_text(encoding="utf-8") + + cli_match = CLI_RE.search(cli_text) + script_match = SCRIPT_RE.search(script_text) + if not cli_match: + raise ValueError("Failed to find CLI default port in mx_exporter/__init__.py") + if not script_match: + raise ValueError("Failed to find default HOST_PORT in start_mxexporter.sh") + + cli_default = int(cli_match.group(1)) + script_default = int(script_match.group(1)) + static_ports = sorted({int(value) for value in YAML_PORT_RE.findall(static_text)}) + if not static_ports: + raise ValueError("Failed to find any ports in mx-exporter-daemonset.yaml") + aligned = script_default == cli_default and all(port == cli_default for port in static_ports) + return { + "cli_default_port": cli_default, + "script_default_port": script_default, + "static_ports": static_ports, + "aligned": aligned, + } + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repo-root", type=Path, default=Path(".")) + parser.add_argument("--output", type=Path) + args = parser.parse_args() + + text = json.dumps(build(args.repo_root), indent=2, ensure_ascii=False) + if args.output: + args.output.write_text(text + "\n", encoding="utf-8") + else: + print(text) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())