399 lines
17 KiB
Python
399 lines
17 KiB
Python
"""覆盖率 78%->85% 第 2 步：deploy.py 的剩余缺口真实行为测试。
|
||
|
||
只测本地逻辑与参数分发：真实临时 git 仓库建包、真实 tarball
|
||
读清单；凡是触网/触主机的步骤一律用 subprocess/ssh_cmd/transfer
|
||
替身在调用边界拦住（替身只决定命令返回值，不替换被测函数
|
||
本身），绝不连真实主机、不跑真实部署。"""
|
||
import io
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from contextlib import redirect_stderr, redirect_stdout
|
||
from types import SimpleNamespace
|
||
from unittest import mock
|
||
|
||
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(
|
||
os.path.abspath(__file__))))
|
||
sys.path.insert(0, os.path.join(REPO_ROOT, "deploy"))
|
||
import deploy as deploy_mod # noqa: E402
|
||
|
||
|
||
def _make_git_repo(path):
|
||
def git(*args):
|
||
subprocess.run(["git", "-C", path] + list(args),
|
||
check=True, capture_output=True, text=True)
|
||
|
||
subprocess.run(["git", "init", "-q", path], check=True,
|
||
capture_output=True)
|
||
git("config", "user.email", "test@test.local")
|
||
git("config", "user.name", "Test")
|
||
with open(os.path.join(path, "hello.txt"), "w") as f:
|
||
f.write("hello\n")
|
||
git("add", ".")
|
||
git("commit", "-qm", "init")
|
||
head = subprocess.run(["git", "-C", path, "rev-parse", "HEAD"],
|
||
check=True, capture_output=True,
|
||
text=True).stdout.strip()
|
||
git("update-ref", "refs/remotes/origin/main", head)
|
||
return head
|
||
|
||
|
||
def _cp(args=(), returncode=0, **kwargs):
|
||
return subprocess.CompletedProcess(args=list(args),
|
||
returncode=returncode,
|
||
stdout="", stderr="")
|
||
|
||
|
||
class DeployCleanTreeTest(unittest.TestCase):
|
||
def test_clean_and_dirty_and_allow_dirty(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
_make_git_repo(tmp)
|
||
old_root = deploy_mod.REPO_ROOT
|
||
deploy_mod.REPO_ROOT = tmp
|
||
try:
|
||
deploy_mod.check_clean_tree(False) # 干净：不退出
|
||
with open(os.path.join(tmp, "hello.txt"), "a") as f:
|
||
f.write("dirty\n")
|
||
with self.assertRaises(SystemExit) as ctx:
|
||
deploy_mod.check_clean_tree(False)
|
||
self.assertEqual(ctx.exception.code, 2)
|
||
out = io.StringIO()
|
||
with redirect_stdout(out):
|
||
deploy_mod.check_clean_tree(True) # 只告警不退出
|
||
self.assertIn("警告", out.getvalue())
|
||
subprocess.run(["git", "-C", tmp, "add", "."],
|
||
check=True, capture_output=True)
|
||
with self.assertRaises(SystemExit):
|
||
deploy_mod.check_clean_tree(False) # 暂存区脏也拦
|
||
finally:
|
||
deploy_mod.REPO_ROOT = old_root
|
||
|
||
def test_run_failure_exits(self):
|
||
with self.assertRaises(SystemExit) as ctx:
|
||
deploy_mod.run(["false"], check=True)
|
||
self.assertEqual(ctx.exception.code, 1)
|
||
r = deploy_mod.run(["false"], check=False)
|
||
self.assertNotEqual(r.returncode, 0)
|
||
|
||
|
||
class DeployRemoteHelpersTest(unittest.TestCase):
|
||
def test_remote_sha_none_on_rc(self):
|
||
with mock.patch.object(deploy_mod.subprocess, "run",
|
||
return_value=_cp(returncode=1)):
|
||
self.assertIsNone(deploy_mod.remote_sha("h", "u", "/tmp/x"))
|
||
|
||
def test_remote_rm_issues_rm_command(self):
|
||
seen = []
|
||
|
||
def fake_run(cmd, **kwargs):
|
||
seen.append(cmd)
|
||
return _cp(cmd)
|
||
|
||
with mock.patch.object(
|
||
deploy_mod, "ssh_cmd",
|
||
side_effect=lambda h, u, c: ["ssh", "dummy", c]):
|
||
with mock.patch.object(deploy_mod.subprocess, "run",
|
||
side_effect=fake_run):
|
||
deploy_mod.remote_rm("h", "u", "/tmp/x.tar.gz")
|
||
self.assertTrue(any(c[-1] == "rm -f /tmp/x.tar.gz" for c in seen))
|
||
|
||
def test_build_release_archive_failure_exits(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
_make_git_repo(tmp)
|
||
old_root = deploy_mod.REPO_ROOT
|
||
deploy_mod.REPO_ROOT = tmp
|
||
try:
|
||
# rev-parse 能过、archive 必败的 ref：一个 blob 的 sha
|
||
#（rev-parse 成功，git archive 拒绝非提交对象）。
|
||
blob = subprocess.run(
|
||
["git", "-C", tmp, "hash-object", "-w", "--stdin"],
|
||
input="blob-content\n", check=True, capture_output=True,
|
||
text=True).stdout.strip()
|
||
self.assertEqual(len(blob), 40)
|
||
with self.assertRaises(SystemExit) as ctx:
|
||
deploy_mod.build_release(blob, tmp)
|
||
self.assertEqual(ctx.exception.code, 1)
|
||
finally:
|
||
deploy_mod.REPO_ROOT = old_root
|
||
|
||
|
||
class DeployTransferTest(unittest.TestCase):
|
||
def _helpers(self, key=True, plain=True):
|
||
def fake_helper(name):
|
||
if name == "chunk_send_key_explicit.py" and key:
|
||
return "/helpers/chunk_send_key_explicit.py"
|
||
if name == "chunk_send.py" and plain:
|
||
return "/helpers/chunk_send.py"
|
||
return None
|
||
|
||
return fake_helper
|
||
|
||
def test_all_attempts_fail_exits(self):
|
||
def fake_run(cmd, **kwargs):
|
||
if cmd and cmd[0] == "scp":
|
||
return subprocess.CompletedProcess(
|
||
args=cmd, returncode=1, stdout="", stderr="reset")
|
||
return subprocess.CompletedProcess(
|
||
args=cmd, returncode=1, stdout="", stderr="chunk fail")
|
||
|
||
with mock.patch.object(deploy_mod, "helper_path",
|
||
side_effect=self._helpers()):
|
||
with mock.patch.object(deploy_mod.subprocess, "run",
|
||
side_effect=fake_run):
|
||
with mock.patch.dict(os.environ,
|
||
{"SSH_PASSWORD": "x"}):
|
||
with self.assertRaises(SystemExit) as ctx:
|
||
deploy_mod.transfer("h", "u", "/tmp/a", "/tmp/b",
|
||
"0" * 64)
|
||
self.assertEqual(ctx.exception.code, 1)
|
||
|
||
def test_chunk_decode_failure_continues_then_exits(self):
|
||
def fake_run(cmd, **kwargs):
|
||
if cmd and cmd[0] == "scp":
|
||
return subprocess.CompletedProcess(
|
||
args=cmd, returncode=1, stdout="", stderr="reset")
|
||
if cmd and cmd[0] == "python3":
|
||
return _cp(cmd) # 分片脚本本身成功
|
||
return subprocess.CompletedProcess(
|
||
args=cmd, returncode=1, stdout="", stderr="decode fail")
|
||
|
||
with mock.patch.object(deploy_mod, "helper_path",
|
||
side_effect=self._helpers()):
|
||
with mock.patch.object(deploy_mod, "ssh_cmd",
|
||
return_value=["ssh", "dummy"]):
|
||
with mock.patch.object(deploy_mod.subprocess, "run",
|
||
side_effect=fake_run):
|
||
with mock.patch.dict(os.environ,
|
||
{"SSH_PASSWORD": "x"}):
|
||
with self.assertRaises(SystemExit):
|
||
deploy_mod.transfer("h", "u", "/tmp/a",
|
||
"/tmp/b", "0" * 64)
|
||
|
||
def test_sha_mismatch_removes_and_exits(self):
|
||
removed = []
|
||
|
||
def fake_run(cmd, **kwargs):
|
||
return _cp(cmd) # scp 成功
|
||
|
||
with mock.patch.object(deploy_mod, "helper_path",
|
||
side_effect=self._helpers(key=False,
|
||
plain=False)):
|
||
with mock.patch.object(deploy_mod.subprocess, "run",
|
||
side_effect=fake_run):
|
||
with mock.patch.object(
|
||
deploy_mod, "remote_sha",
|
||
return_value="1" * 64):
|
||
with mock.patch.object(
|
||
deploy_mod, "remote_rm",
|
||
side_effect=lambda h, u, p: removed.append(p)):
|
||
with self.assertRaises(SystemExit):
|
||
deploy_mod.transfer("h", "u", "/tmp/a",
|
||
"/tmp/b", "0" * 64)
|
||
self.assertEqual(removed, ["/tmp/b"])
|
||
|
||
|
||
class DeployViaGitTest(unittest.TestCase):
|
||
def _args(self, **over):
|
||
base = dict(tarball=None, build_only=False, host="h.example",
|
||
user="root", dry_run=False,
|
||
repo="git@github.com:openqj/codex-app-server.git",
|
||
git_key="/root/.ssh/id_codex_github",
|
||
allow_dirty=True, skip_tests=True, ref="HEAD")
|
||
base.update(over)
|
||
return SimpleNamespace(**base)
|
||
|
||
def test_conflicts_and_guards_exit_2(self):
|
||
for over, in_msg in [
|
||
(dict(tarball="/tmp/x.tgz"), "tarball"),
|
||
(dict(build_only=True), "build-only"),
|
||
(dict(host=None, dry_run=False), "host"),
|
||
(dict(repo="https://github.com/x/y.git"), "SSH")]:
|
||
with self.subTest(over=over):
|
||
err = io.StringIO()
|
||
with redirect_stderr(err):
|
||
with self.assertRaises(SystemExit) as ctx:
|
||
deploy_mod.deploy_via_git(self._args(**over))
|
||
self.assertEqual(ctx.exception.code, 2)
|
||
|
||
def test_dry_run_success(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
head = _make_git_repo(tmp)
|
||
old_root = deploy_mod.REPO_ROOT
|
||
deploy_mod.REPO_ROOT = tmp
|
||
try:
|
||
out = io.StringIO()
|
||
with redirect_stdout(out):
|
||
rc = deploy_mod.deploy_via_git(
|
||
self._args(dry_run=True))
|
||
self.assertEqual(rc, 0)
|
||
self.assertIn(head, out.getvalue())
|
||
finally:
|
||
deploy_mod.REPO_ROOT = old_root
|
||
|
||
def _ssh_fake(self, head, script_rc=0, script_out=None, live=None):
|
||
real_run = subprocess.run
|
||
|
||
def fake_run(cmd, **kwargs):
|
||
if cmd and cmd[0] == "git":
|
||
return real_run(cmd, **kwargs)
|
||
last = cmd[-1] if cmd else ""
|
||
if "cat /opt/codex-app-server/DEPLOYED_COMMIT" in last:
|
||
return subprocess.CompletedProcess(
|
||
args=cmd, returncode=0,
|
||
stdout=(live if live is not None else head) + "\n",
|
||
stderr="")
|
||
if last.startswith("sh "):
|
||
return subprocess.CompletedProcess(
|
||
args=cmd, returncode=script_rc,
|
||
stdout=(script_out if script_out is not None
|
||
else "RELEASE_OK commit=%s" % head),
|
||
stderr="remote-err" if script_rc != 0 else "")
|
||
return _cp(cmd)
|
||
|
||
return fake_run
|
||
|
||
def test_non_dry_run_success_and_failures(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
head = _make_git_repo(tmp)
|
||
old_root = deploy_mod.REPO_ROOT
|
||
deploy_mod.REPO_ROOT = tmp
|
||
try:
|
||
transfers = []
|
||
with mock.patch.object(
|
||
deploy_mod, "ssh_cmd",
|
||
side_effect=lambda h, u, c: ["ssh", "dummy", c]):
|
||
with mock.patch.object(
|
||
deploy_mod, "transfer",
|
||
side_effect=lambda *a: transfers.append(a)):
|
||
with mock.patch.object(
|
||
deploy_mod.subprocess, "run",
|
||
side_effect=self._ssh_fake(head)):
|
||
rc = deploy_mod.deploy_via_git(self._args())
|
||
self.assertEqual(rc, 0)
|
||
self.assertEqual(len(transfers), 2)
|
||
# 主机脚本失败 -> exit 1
|
||
with mock.patch.object(
|
||
deploy_mod, "ssh_cmd",
|
||
side_effect=lambda h, u, c: ["ssh", "dummy", c]):
|
||
with mock.patch.object(deploy_mod, "transfer"):
|
||
with mock.patch.object(
|
||
deploy_mod.subprocess, "run",
|
||
side_effect=self._ssh_fake(
|
||
head, script_rc=1, script_out="boom")):
|
||
with self.assertRaises(SystemExit):
|
||
deploy_mod.deploy_via_git(self._args())
|
||
# live 二次核对不符 -> exit 1
|
||
with mock.patch.object(
|
||
deploy_mod, "ssh_cmd",
|
||
side_effect=lambda h, u, c: ["ssh", "dummy", c]):
|
||
with mock.patch.object(deploy_mod, "transfer"):
|
||
with mock.patch.object(
|
||
deploy_mod.subprocess, "run",
|
||
side_effect=self._ssh_fake(
|
||
head, live="0" * 40)):
|
||
with self.assertRaises(SystemExit):
|
||
deploy_mod.deploy_via_git(self._args())
|
||
finally:
|
||
deploy_mod.REPO_ROOT = old_root
|
||
|
||
|
||
class DeployMainTest(unittest.TestCase):
|
||
def _tarball(self, tmp):
|
||
head = _make_git_repo(tmp)
|
||
old_root = deploy_mod.REPO_ROOT
|
||
deploy_mod.REPO_ROOT = tmp
|
||
try:
|
||
tarball, sha, commit = deploy_mod.build_release("HEAD", tmp)
|
||
finally:
|
||
deploy_mod.REPO_ROOT = old_root
|
||
self.assertEqual(commit, head)
|
||
return tarball, head
|
||
|
||
def test_no_host_errors(self):
|
||
with self.assertRaises(SystemExit) as ctx:
|
||
deploy_mod.main([])
|
||
self.assertEqual(ctx.exception.code, 2)
|
||
|
||
def test_tarball_mismatch_and_dry_run(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
tarball, head = self._tarball(tmp)
|
||
with self.assertRaises(SystemExit) as ctx:
|
||
deploy_mod.main(["h.example", "--tarball", tarball,
|
||
"--commit", "0" * 40])
|
||
self.assertEqual(ctx.exception.code, 2)
|
||
rc = deploy_mod.main(["h.example", "--tarball", tarball,
|
||
"--dry-run"])
|
||
self.assertEqual(rc, 0)
|
||
|
||
def test_build_only(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
out = io.StringIO()
|
||
with redirect_stdout(out):
|
||
rc = deploy_mod.main(["--build-only", "--allow-dirty",
|
||
"--skip-tests", "--out-dir", tmp])
|
||
self.assertEqual(rc, 0)
|
||
self.assertIn("仅构建", out.getvalue())
|
||
self.assertTrue(any(n.endswith(".tar.gz")
|
||
for n in os.listdir(tmp)))
|
||
|
||
def test_git_dry_run_via_main(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
_make_git_repo(tmp)
|
||
old_root = deploy_mod.REPO_ROOT
|
||
deploy_mod.REPO_ROOT = tmp
|
||
try:
|
||
rc = deploy_mod.main(["--source", "git", "--dry-run",
|
||
"--skip-tests", "--allow-dirty"])
|
||
self.assertEqual(rc, 0)
|
||
finally:
|
||
deploy_mod.REPO_ROOT = old_root
|
||
|
||
def _ssh_fake_main(self, head, script_rc=0, live=None):
|
||
def fake_run(cmd, **kwargs):
|
||
last = cmd[-1] if cmd else ""
|
||
if "cat /opt/codex-app-server/DEPLOYED_COMMIT" in last:
|
||
return subprocess.CompletedProcess(
|
||
args=cmd, returncode=0,
|
||
stdout=(live if live is not None else head) + "\n",
|
||
stderr="")
|
||
if last.startswith("sh "):
|
||
return subprocess.CompletedProcess(
|
||
args=cmd, returncode=script_rc,
|
||
stdout="RELEASE_OK commit=%s" % head if script_rc == 0
|
||
else "boom",
|
||
stderr="remote-err" if script_rc != 0 else "")
|
||
return _cp(cmd)
|
||
|
||
return fake_run
|
||
|
||
def test_tar_deploy_success_and_failures(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
tarball, head = self._tarball(tmp)
|
||
for script_rc, live, ok in [(0, None, True),
|
||
(1, None, False),
|
||
(0, "0" * 40, False)]:
|
||
with self.subTest(script_rc=script_rc, live=live):
|
||
with mock.patch.object(
|
||
deploy_mod, "ssh_cmd",
|
||
side_effect=lambda h, u, c: ["ssh", "dummy", c]):
|
||
with mock.patch.object(deploy_mod, "transfer"):
|
||
with mock.patch.object(
|
||
deploy_mod.subprocess, "run",
|
||
side_effect=self._ssh_fake_main(
|
||
head, script_rc, live)):
|
||
if ok:
|
||
rc = deploy_mod.main(
|
||
["h.example", "--tarball", tarball])
|
||
self.assertEqual(rc, 0)
|
||
else:
|
||
with self.assertRaises(SystemExit):
|
||
deploy_mod.main(
|
||
["h.example", "--tarball",
|
||
tarball])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|