host
f39ed3654e
基底：#251 头 174b631（堆叠于 #242 feat-ci-expand 之上）， 基线 origin/main aebc6238；本轨不改 #251 已定语义、不动 workflow 门槛、不删检查、不加 skip。 lint（ruff E,F@120）：109 -> 10 - F401 43 -> 0：删未用导入；try/except 双绑定的名字出现 “打地鼠”（删 except 分支后 try 分支又被标出），两分支 均已核对正文无引用后清掉；MCP 三模块的 Error 名不删， 改走 importlib 属性绑定（见下），两路径绑定都保留。 - E702 40 -> 0：全在 host.py main() 的 argparse 段，纯分行， tokenize 按分号 token 拆（含 2377 行多行语句只断首个分号）； 一次 5 空格缩进失误已回滚重做并以 ast + 全量 host 测试验证。 - E501 13 -> 0：host.py 5 项随 E702 分行连带清掉；mcp_server 3 项把描述拆成相邻字面量（AST 比对 24 条工具描述与原值 完全一致，注意 ruff 按东亚宽度计长，CJK 算 2）；test_host 5 项断开 git commit 参数表。 - E402 8 -> 10：生产侧 2 项清掉（db.py _re、github_app _base64 移到顶部，之前无引用、无重名）；剩余 10 项全是 tests/ 的 sys.path 前置 import（6 项旧 + 4 个新测试文件 沿用同一既定模式），移上去会破坏单文件直跑，有意保留。 - F841 4 -> 0：只去绑定、不删调用（run_task、DB 入队、 tarfile 上下文、git 提交），各有注释与原测试/新测试覆盖； host dispatch_from_branch 新增真实 git 仓库回归测试。 - F811 1 -> 0：server 本地 _row_get 与 peer_messages.row_get 逐行相同，删本地定义，全部调用点解析到同一 import； 新增 test_server_row_get（同一对象 + 三路行为）。 mypy（生产范围）：24 -> 0，零新增 type: ignore - no-redef 9 -> 0：MCP 三模块 try/except 两次 from-import 改 importlib.import_module + 属性绑定（cloud 优先、flat copy 回落、内层 ImportError 同样回落，语义与原来一致； flat-copy 子进程测试证实回落后 __module__ 为顶层名）； server _row_get 随 F811 消除。 - arg-type 10 -> 0：harness2_tasks shared 改 TypedDict 逐键精确类型；CodexProbeAdapter 补 no-op validate_tree （_adapter_validate 本就跳过缺失方法，行为一致）。 - var-annotated 3 -> 0：sub2api _SHARED_ADMIN/_login_slots、 verify-multiuser RESULTS 加精确注解。 - assignment/misc 2 -> 0：marker_tools ZoneInfo 先声明 Any 再 try import（草稿先验 mypy 通过）。 bandit：总数 286 不变，High 仍 4 项，全在 host.py，本轨不改 （属第 3 项白名单轨范围），只做 triage： - 1319 run_verify：无插值，整串验测命令，管道/&& 须 shell 语义（新增 test_host_run_verify 断言管道与 && 真实生效）。 - 1497 notify：模板来自算子配置，三处插值全部 shlex.quote， 证据为既有 test_notify_failure_quotes_shell_metacharacters （; touch 与 $(touch ) 不执行）。 - 2099/2103 goal verify：无插值，任务配置验测串，同 run_verify。 - 无新增 High；pip-audit 0 漏洞。 覆盖率（同口径三套件 coverage 计量）：75.96% -> 77.87% （22392 行/缺 5385 -> 22437 行/缺 4963，显示 76% -> 78%） - client.py 0% -> 90%（pipe 假子进程 + main 分支） - run_parallel.py 0% -> 97%、migrate 包装 0% -> 95% - playwright_mcp 37% -> 62%、imagegen_mcp 47% -> 79%、 websearch_mcp 51% -> 86%、deploy.py 56% -> 77% - verify-multiuser 0% -> 27%（仅 check/req helper；main() 是主机侧真实多用户实弹验证，按门禁不建生产任务、 不以 mock 冒充，仍为缺口） - host.py 55% -> 57%（文件 2020 -> 2060 行系 E702 分行 的计量口径变化，非代码膨胀） - 未到 85%：门槛保持 75% 不动，不删低覆盖模块出统计口径； 缺口主力仍是 server.py / worker.py / host.py 大文件， 下一阶段须针对性深测，非本轨快赢可及。 验证：三套件 1576 -> 1629 项全过（tests 127、cloud/tests 1449 skipped=4、deploy/tests 53）；ruff E9 通过；mypy 0； bandit High 4（同上）；pip-audit 0 漏洞；试跑产物已删， 提交后工作树干净。
283 lines
13 KiB
Python
283 lines
13 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:
|
||
# 空 tarball 文件本身是测试前置（tarball_commit 应拒绝）：上下文照跑，只去 as 绑定。
|
||
with tarfile.open(path, "w:gz"):
|
||
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()
|