"""覆盖率 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()