host
420a7fecfd
背景：2026-10-27 另一条线给 deploy.py 打的密钥补丁只放在 env-console 工作区未提交，切分支后丢失；部署一度只能靠它 + --allow-dirty。本批把逻辑收进 main，并加一处加固。 分发：有 SSH_PASSWORD 时走 helper（pty 喂密码，实测最稳的一 路）；无密码时 helper 不可用（会卡在密码提示），改走密钥直连： ProxyCommand 显式环境变量优先、其次 helper 目录的 proxy_tunnel.py（存在才用），密钥按 CODEX_DEPLOY_KEY → /root/.ssh/id_codex_deploy → ~/.ssh/id_codex_deploy 顺序显式传 -i——不能只依赖 /root/.ssh/config 的 Host 条目，那个目录会在 VM 回收时被清空（2026-10-27 05:04 实测），而 ~/.ssh 的那份 持久。两处都没有时不硬编造，靠 ssh 配置或 agent。 测试：deploy/tests 新增 TestSshScpCmd 5 个（有密码走 helper、 无密码绝不用 helper、无 helper 无代理时直连、显式代理优先、 显式密钥优先）；全量 deploy 19、host 62、cloud 598 全过。
190 lines
8.3 KiB
Python
190 lines
8.3 KiB
Python
"""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 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()
|