laily
10d8800f07
Give ack, builder, and deployer an explicit init/check mode that reports missing project config instead of failing mid-work. Point builder at makefile.builder so its contract targets do not collide with an existing Makefile.
123 lines
5.2 KiB
Python
123 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
CHECK_PY = REPO_ROOT / "skills" / "deployer" / "scripts" / "deploy" / "check.py"
|
|
LIB_PY = REPO_ROOT / "skills" / "deployer" / "scripts" / "deploy" / "lib.py"
|
|
|
|
lib_spec = importlib.util.spec_from_file_location("deployer_lib", LIB_PY)
|
|
assert lib_spec is not None and lib_spec.loader is not None
|
|
deployer_lib = importlib.util.module_from_spec(lib_spec)
|
|
lib_spec.loader.exec_module(deployer_lib)
|
|
|
|
check_spec = importlib.util.spec_from_file_location("deployer_check", CHECK_PY)
|
|
assert check_spec is not None and check_spec.loader is not None
|
|
deployer_check = importlib.util.module_from_spec(check_spec)
|
|
check_spec.loader.exec_module(deployer_check)
|
|
|
|
|
|
def run_check(project: Path, hosts: set[str] | None = None) -> tuple[int, str]:
|
|
stdout = StringIO()
|
|
stderr = StringIO()
|
|
deployer_lib.PROJECT_ROOT = None
|
|
deployer_lib._SSH_HOSTS = None
|
|
deployer_check.lib.PROJECT_ROOT = None
|
|
deployer_check.lib._SSH_HOSTS = None
|
|
patched_hosts = hosts if hosts is not None else set()
|
|
with mock.patch("sys.stdout", stdout), mock.patch("sys.stderr", stderr):
|
|
with mock.patch.object(deployer_check.lib, "ssh_config_hosts", return_value=patched_hosts):
|
|
with mock.patch.dict(os.environ, {}, clear=False):
|
|
os.environ.pop("DEPLOYER_ROOT", None)
|
|
code = deployer_check.main(["--project", str(project)])
|
|
return code, stdout.getvalue() + stderr.getvalue()
|
|
|
|
|
|
class DeployerCheckTests(unittest.TestCase):
|
|
def tearDown(self) -> None:
|
|
deployer_lib.PROJECT_ROOT = None
|
|
deployer_lib._SSH_HOSTS = None
|
|
deployer_check.lib.PROJECT_ROOT = None
|
|
deployer_check.lib._SSH_HOSTS = None
|
|
|
|
def test_skill_documents_init_and_check_script(self) -> None:
|
|
skill = (REPO_ROOT / "skills" / "deployer" / "SKILL.md").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
self.assertIn("## 初始化", skill)
|
|
self.assertIn("scripts/deploy/check.py", skill)
|
|
self.assertIn("不要写假 node", skill)
|
|
|
|
def test_missing_layout_fails(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
project = Path(temp)
|
|
code, text = run_check(project)
|
|
self.assertEqual(code, 1, text)
|
|
self.assertIn("[FAIL] 1. 部署根存在", text)
|
|
self.assertIn("test/compose.yaml", text)
|
|
|
|
def test_compose_without_node_is_partial_failure(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
project = Path(temp)
|
|
env = project / ".pouch" / "deployer" / "test"
|
|
env.mkdir(parents=True)
|
|
(env / "compose.yaml").write_text("services:\n web:\n image: nginx\n", encoding="utf-8")
|
|
code, text = run_check(project, hosts={"my-vps"})
|
|
self.assertEqual(code, 1, text)
|
|
self.assertIn("[PASS] 1. 部署根存在", text)
|
|
self.assertIn("[PASS] 4. 至少有一个 compose.yaml", text)
|
|
self.assertIn("[FAIL] 5. 每个服务能解析 node", text)
|
|
self.assertIn("MISSING node", text)
|
|
|
|
def test_node_missing_from_ssh_config_fails(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
project = Path(temp)
|
|
root = project / ".pouch" / "deployer"
|
|
env = root / "test"
|
|
env.mkdir(parents=True)
|
|
(root / "_config.yaml").write_text("node: my-vps\n", encoding="utf-8")
|
|
(env / "compose.yaml").write_text("services:\n web:\n image: nginx\n", encoding="utf-8")
|
|
code, text = run_check(project, hosts=set())
|
|
self.assertEqual(code, 1, text)
|
|
self.assertIn("[PASS] 5. 每个服务能解析 node", text)
|
|
self.assertIn("[FAIL] 6. node 出现在 SSH config", text)
|
|
self.assertIn("my-vps NOT in ~/.ssh/config", text)
|
|
|
|
def test_ready_when_node_and_compose_present(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
project = Path(temp)
|
|
root = project / ".pouch" / "deployer"
|
|
env = root / "test"
|
|
env.mkdir(parents=True)
|
|
(root / "_config.yaml").write_text("node: my-vps\nbase_path: /opt/app\n", encoding="utf-8")
|
|
(env / "compose.yaml").write_text("services:\n web:\n image: nginx\n", encoding="utf-8")
|
|
code, text = run_check(project, hosts={"my-vps"})
|
|
self.assertEqual(code, 0, text)
|
|
self.assertIn("RESULT: PASSED", text)
|
|
self.assertIn("[PASS] 7. list 可发现服务", text)
|
|
|
|
def test_argocd_only_passes_compose_as_skip(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
project = Path(temp)
|
|
root = project / ".pouch" / "deployer"
|
|
root.mkdir(parents=True)
|
|
(root / "argocd.yaml").write_text(
|
|
"repo: git@git.example.com:org/infra-gitops.git\n",
|
|
encoding="utf-8",
|
|
)
|
|
code, text = run_check(project)
|
|
self.assertEqual(code, 0, text)
|
|
self.assertIn("[PASS] 3. Argo CD 指针", text)
|
|
self.assertIn("[SKIP] 4. 至少有一个 compose.yaml", text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|