增加监控导出端口一致性audit

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

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