"""deploy.py 的单测：只测纯逻辑与发布包结构，不碰真实主机。 主机侧 host_deploy.sh 的仿真见 test_host_deploy.py（假 systemctl/curl）。 """ import hashlib import os import subprocess import sys import tarfile import tempfile import unittest 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 class TestDeployPure(unittest.TestCase): def test_release_tarball_name_is_commit_scoped(self): name = deploy_mod.release_tarball_name("abc123") self.assertEqual(name, "codex-app-server-abc123.tar.gz") # 绝不再用通用名：不同 commit 绝不可能撞到同一个远端路径。 self.assertNotEqual( deploy_mod.release_tarball_name("abc123"), deploy_mod.release_tarball_name("def456")) def test_parse_release_ok_requires_matching_commit(self): out = "部署成功\nRELEASE_OK commit=abc sha=dead backup=/tmp/x.tar.gz\n" self.assertTrue(deploy_mod.parse_release_ok(out, "abc")) self.assertFalse(deploy_mod.parse_release_ok(out, "other")) self.assertFalse(deploy_mod.parse_release_ok("部署成功，health 通过\n", "abc")) def test_sha256_of(self): with tempfile.NamedTemporaryFile(delete=False) as f: f.write(b"hello") path = f.name try: self.assertEqual(deploy_mod.sha256_of(path), hashlib.sha256(b"hello").hexdigest()) finally: os.unlink(path) def test_tarball_commit_missing_manifest_refuses(self): with tempfile.NamedTemporaryFile(suffix=".tar.gz", delete=False) as f: path = f.name try: with tarfile.open(path, "w:gz") as tf: pass with self.assertRaises(SystemExit): deploy_mod.tarball_commit(path) finally: os.unlink(path) def test_remote_sha_ignores_ssh_warning_lines(self): # 2026-10-27 新链路首推实测：pty helper 把 ssh 的 Warning 行 # 合并进 stdout 放在 sha 行前面，旧 split()[0] 取到 Warning: # 导致传输明明成功却被判验真失败、两路都“失败”。 sha = "5d0645b85a65b2d58d0d5812f9e8020bdafe15d60dc440e763cfcc7cd178148b" fake = subprocess.CompletedProcess( args=[], returncode=0, stdout="Warning: Permanently added 'x' (ED25519) to the list of known hosts.\n" + sha + " /tmp/x.tar.gz\n", stderr="") with mock.patch.object(deploy_mod.subprocess, "run", return_value=fake): self.assertEqual(deploy_mod.remote_sha("h", "u", "/tmp/x"), sha) def test_remote_sha_none_without_hash(self): fake = subprocess.CompletedProcess(args=[], returncode=0, stdout="Warning: something\n", stderr="") with mock.patch.object(deploy_mod.subprocess, "run", return_value=fake): self.assertIsNone(deploy_mod.remote_sha("h", "u", "/tmp/x")) def test_live_commit_ignores_ssh_warning_lines(self): commit = "32125f7a980e24fd28664a60701d7c1123beb63e" out = "Warning: Permanently added 'x' (ED25519) to the list of known hosts.\n" \ + commit + "\n" self.assertEqual(deploy_mod.live_commit(out), commit) self.assertIsNone(deploy_mod.live_commit("Warning: something\n")) class TestGitModeHelpers(unittest.TestCase): """--source git 的纯逻辑门禁：完整 sha、SSH 形式、origin/main 归属。主机只能拉到已推的提交，未推的 HEAD 必须在 VM 侧拦住。""" def test_is_full_sha(self): self.assertTrue(deploy_mod.is_full_sha("a" * 40)) self.assertFalse(deploy_mod.is_full_sha("a" * 7)) self.assertFalse(deploy_mod.is_full_sha("g" * 40)) self.assertFalse(deploy_mod.is_full_sha("")) self.assertFalse(deploy_mod.is_full_sha(None)) def test_is_ssh_git_url(self): self.assertTrue(deploy_mod.is_ssh_git_url( "git@github.com:openqj/codex-app-server.git")) self.assertTrue(deploy_mod.is_ssh_git_url( "ssh://git@github.com/openqj/codex-app-server.git")) self.assertFalse(deploy_mod.is_ssh_git_url( "https://github.com/openqj/codex-app-server.git")) self.assertFalse(deploy_mod.is_ssh_git_url("")) def test_commit_on_origin_main_uses_merge_base(self): calls = [] def fake_run(cmd, cwd=None, check=True): calls.append(cmd) return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="") with mock.patch.object(deploy_mod, "run", side_effect=fake_run): self.assertTrue(deploy_mod.commit_on_origin_main("a" * 40)) self.assertIn("merge-base", calls[0]) self.assertIn("origin/main", calls[0]) def test_transfer_falls_back_to_chunk_key(self): """scp 失败后走 chunk_key 分片（密钥版），不依赖 SSH_PASSWORD； 验远端 sha 通过即成功，不应退出。""" calls = [] def fake_run(cmd, cwd=None, check=True, **kwargs): calls.append(cmd) if cmd[0] == "scp": return subprocess.CompletedProcess(args=cmd, returncode=1, stdout="", stderr="reset") return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="") def fake_helper(name): return "/helpers/chunk_send_key_explicit.py" \ if name == "chunk_send_key_explicit.py" else None with mock.patch.object(deploy_mod.subprocess, "run", side_effect=fake_run): with mock.patch.object(deploy_mod, "helper_path", side_effect=fake_helper): with mock.patch.object(deploy_mod, "ssh_cmd", return_value=["ssh", "dummy"]): with mock.patch.object( deploy_mod, "remote_sha", return_value="deadbeef"): deploy_mod.transfer("h", "u", "/tmp/a", "/tmp/b", "deadbeef") self.assertTrue(any("chunk_send_key_explicit.py" in x for c in calls for x in c)) class TestDefaultTestSuites(unittest.TestCase): """P2-ops 回归：部署前默认清单必须含 deploy/tests， 且 run_tests() 确实按 discover -s 逐套跑（不是包路径）。""" def test_default_suites_include_deploy_tests(self): self.assertIn("deploy/tests", deploy_mod.DEFAULT_TEST_SUITES) self.assertIn("tests", deploy_mod.DEFAULT_TEST_SUITES) self.assertIn("cloud/tests", deploy_mod.DEFAULT_TEST_SUITES) def test_run_tests_discovers_every_suite(self): calls = [] def fake_run(cmd, cwd=None, check=True): calls.append(cmd) return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="") with mock.patch.object(deploy_mod, "run", side_effect=fake_run): deploy_mod.run_tests() discovered = [c[c.index("-s") + 1] for c in calls if "discover" in c] self.assertEqual(discovered, list(deploy_mod.DEFAULT_TEST_SUITES)) # 没有任何一套走包路径（deploy.tests 会盖掉 deploy.py）。 for c in calls: self.assertNotIn("deploy.tests", c) class TestBuildRelease(unittest.TestCase): def test_build_from_temp_git_repo(self): with tempfile.TemporaryDirectory() as tmp: repo = os.path.join(tmp, "repo") os.makedirs(repo) subprocess.run(["git", "init", "-q"], cwd=repo, check=True) subprocess.run(["git", "config", "user.email", "t@t"], cwd=repo, check=True) subprocess.run(["git", "config", "user.name", "t"], cwd=repo, check=True) with open(os.path.join(repo, "hello.py"), "w") as f: f.write("x = 1\n") subprocess.run(["git", "add", "."], cwd=repo, check=True) subprocess.run(["git", "commit", "-qm", "init"], cwd=repo, check=True) commit = subprocess.run( ["git", "rev-parse", "HEAD"], cwd=repo, capture_output=True, text=True, check=True).stdout.strip() # 把 deploy 模块的 REPO_ROOT 临时指到这个小仓库。 old = deploy_mod.REPO_ROOT deploy_mod.REPO_ROOT = repo try: out_dir = os.path.join(tmp, "out") tarball, sha, got = deploy_mod.build_release("HEAD", out_dir) finally: deploy_mod.REPO_ROOT = old self.assertEqual(got, commit) self.assertEqual(deploy_mod.sha256_of(tarball), sha) self.assertEqual(deploy_mod.tarball_commit(tarball), commit) with tarfile.open(tarball, "r:gz") as tf: names = tf.getnames() self.assertIn("DEPLOYED_COMMIT", names) self.assertIn("RELEASE.json", names) self.assertIn("hello.py", names) class TestSshScpCmd(unittest.TestCase): """ssh_cmd/scp_cmd 的三路分发：有密码走 helper、无密码走 密钥直连、无 helper 无代理时直连。2026-10-27 密钥补丁曾只 放在 env-console 工作区未提交、切分支后丢失，本批把它收进 main；测试钉住分发逻辑，不依赖真实主机与真实密钥文件。""" def setUp(self): self._env = dict(os.environ) os.environ.pop("SSH_PASSWORD", None) os.environ.pop("SSH_PROXY_CMD", None) os.environ.pop("CODEX_DEPLOY_KEY", None) def tearDown(self): os.environ.clear() os.environ.update(self._env) def _cmds(self, helper=None, proxy_file=False, key_file=False): with mock.patch.object(deploy_mod, "helper_path", return_value=helper): with mock.patch.object(deploy_mod.os.path, "isfile", return_value=(proxy_file or key_file)): # isfile 对 proxy 与 key 都返回同一值：测试只断言 # 分支形状（是否出现 ProxyCommand/-i），不区分两者。 ssh = deploy_mod.ssh_cmd("h", "u", "true") scp = deploy_mod.scp_cmd("h", "u", "a", "b") return ssh, scp def test_password_uses_helper_even_without_proxy(self): os.environ["SSH_PASSWORD"] = "x" ssh, scp = self._cmds(helper="/helpers/ssh_run.py") self.assertEqual(ssh[:4], ["python3", "/helpers/ssh_run.py", "ssh", "h"]) self.assertEqual(scp[:4], ["python3", "/helpers/ssh_run.py", "scp", "h"]) def test_no_password_never_uses_helper(self): # 无密码时 helper（pty 喂密码）不可用：即使 helper 文件 # 存在也不走它，否则会在密码提示处卡住。 ssh, scp = self._cmds(helper="/helpers/ssh_run.py", proxy_file=True, key_file=True) self.assertNotIn("/helpers/ssh_run.py", ssh) self.assertNotIn("/helpers/ssh_run.py", scp) # ProxyCommand 与参数拼在一个元素里，查前缀而不是整元素。 self.assertTrue(any(x.startswith("ProxyCommand") for x in ssh)) self.assertIn("-i", ssh) self.assertIn("-i", scp) def test_no_password_no_helpers_goes_direct(self): ssh, scp = self._cmds(helper=None, proxy_file=False, key_file=False) self.assertEqual(ssh[0], "ssh") self.assertNotIn("ProxyCommand", ssh) self.assertNotIn("-i", ssh) self.assertEqual(scp[0], "scp") def test_explicit_proxy_cmd_wins(self): os.environ["SSH_PROXY_CMD"] = "proxy-cmd %h %p" ssh, _ = self._cmds(helper=None, proxy_file=False) self.assertIn("ProxyCommand proxy-cmd %h %p", ssh) def test_explicit_key_env_wins(self): os.environ["CODEX_DEPLOY_KEY"] = "/keys/d" ssh, _ = self._cmds(helper=None, proxy_file=False, key_file=False) self.assertIn("/keys/d", ssh) if __name__ == "__main__": unittest.main()