feat: add ops center and node onboarding flow

domain-api/tests/test_drive_ops_center_docs.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+import os
+import re
+import subprocess
+import unittest
+from pathlib import Path
+
+
+class DriveOpsCenterDocsTests(unittest.TestCase):
+    def test_authoritative_docs_only_reference_supported_commands(self) -> None:
+        repo_root = Path(__file__).resolve().parents[2]
+        script_path = repo_root / "domain-api" / "deploy" / "multi-region" / "drive_ops_center.sh"
+        doc_paths = [
+            repo_root / "domain-api" / "deploy" / "multi-region" / "README.md",
+            repo_root / "docs" / "25_domainCheck_海外单脑控制面上线收口总表.md",
+            repo_root / "docs" / "26_domainCheck_发布前运行验证与交付模板.md",
+        ]
+
+        help_output = subprocess.run(
+            ["bash", str(script_path), "help"],
+            cwd=repo_root,
+            check=True,
+            capture_output=True,
+            text=True,
+        ).stdout
+
+        supported_commands = self._extract_supported_commands(help_output)
+        documented_commands = self._extract_documented_commands(doc_paths)
+
+        missing_commands = sorted(documented_commands - supported_commands)
+        self.assertEqual(
+            [],
+            missing_commands,
+            msg=(
+                "These drive_ops_center.sh commands are referenced by authoritative docs "
+                "but missing from the CLI help output: "
+                + ", ".join(missing_commands)
+            ),
+        )
+        self.assertIn("stack-diagnosis", documented_commands)
+        self.assertIn("stack-diagnosis", supported_commands)
+
+    def test_stack_diagnosis_summary_argument_normalization_does_not_treat_summary_as_host(self) -> None:
+        repo_root = Path(__file__).resolve().parents[2]
+        script_path = repo_root / "domain-api" / "deploy" / "multi-region" / "drive_ops_center.sh"
+        env = dict(os.environ)
+        env["DOMAINCHECK_OPS_CENTER_ENV"] = "/tmp/domaincheck-ops-center-does-not-exist"
+        env["OPS_MAINLAND_API_BASE_URL"] = "http://127.0.0.1:9"
+
+        result = subprocess.run(
+            ["bash", str(script_path), "stack-diagnosis", "summary"],
+            cwd=repo_root,
+            check=True,
+            capture_output=True,
+            text=True,
+            env=env,
+        )
+
+        combined_output = f"{result.stdout}\n{result.stderr}"
+        self.assertNotIn("Could not resolve host: summary", combined_output)
+        self.assertNotIn("NameError", combined_output)
+        self.assertIn("[9/9] condensed stack summary", combined_output)
+
+    def test_env_drift_recommends_stack_diagnosis_instead_of_stack_check(self) -> None:
+        repo_root = Path(__file__).resolve().parents[2]
+        script_path = repo_root / "domain-api" / "deploy" / "multi-region" / "check_env_drift.sh"
+        script_text = script_path.read_text(encoding="utf-8")
+
+        self.assertIn(
+            'recommended_actions.append("bash domain-api/deploy/multi-region/drive_ops_center.sh stack-diagnosis http://127.0.0.1:8100 summary")',
+            script_text,
+        )
+        self.assertNotIn(
+            'recommended_actions.append("bash domain-api/deploy/multi-region/drive_ops_center.sh stack-check http://127.0.0.1:8100 summary")',
+            script_text,
+        )
+
+    @staticmethod
+    def _extract_supported_commands(help_output: str) -> set[str]:
+        commands: set[str] = set()
+        in_commands_section = False
+        for raw_line in help_output.splitlines():
+            line = raw_line.rstrip()
+            if line.strip() == "commands:":
+                in_commands_section = True
+                continue
+            if line.strip() == "examples:":
+                break
+            if not in_commands_section:
+                continue
+            stripped = line.strip()
+            if not stripped:
+                continue
+            command = stripped.split()[0]
+            if re.fullmatch(r"[a-z0-9-]+", command):
+                commands.add(command)
+        return commands
+
+    @staticmethod
+    def _extract_documented_commands(doc_paths: list[Path]) -> set[str]:
+        pattern = re.compile(r"drive_ops_center\.sh\s+([a-z0-9-]+)")
+        commands: set[str] = set()
+        for doc_path in doc_paths:
+            text = doc_path.read_text(encoding="utf-8")
+            commands.update(match.group(1) for match in pattern.finditer(text))
+        return commands
+
+
+if __name__ == "__main__":
+    unittest.main()