openqjandhost
fc8e337d0f
* 部署自动化：唯一发布链路 deploy.py + host_deploy.sh，sha/commit 双重验真，失败回滚 旧链路三次栽过的坑：通用 tar 路径导致旧包误部署、传输不验 sha、 主机脚本多份漂移、health 通过不等于新代码上线。本单把四点变硬门禁： - 产物唯一命名 codex-app-server-<commit>.tar.gz + .sha256，远端唯一路径 - git archive 产物 + DEPLOYED_COMMIT 清单，默认拒绝脏工作区 - 传输链 scp -> scp_run -> chunk 分片，每路传完远端验 sha，不符删掉重来 - 主机侧唯一脚本：先验 sha/清单，staging 解包 + 编译检查，换目录上线， work/ 原样保留，health + 上线 commit 不符即回滚并退出非零 - deploy.py 成功后独立 ssh 再读一次 live 清单二次核对 - CI：host-verify 加 deploy/tests；release.yml 只在手动/tag + production environment 审批时部署，主干推送绝不自动部署生产 - deploy.sh 标记为旧链路弃用 - health 的 commit 字段已在父提交 HA 阶段 0 里随 server.py 落地， 本分支整体包含它；commit 为空时只靠文件核对，门禁同样成立 测试：host 32、cloud 487（4 跳过）、deploy 9，全部通过； host_deploy.sh 仿真覆盖成功/坏 sha/清单不符/health commit 不符回滚。 生产实机部署未做，等确认后再用新链路推。 * HA stage 1: worktree_host + queued cross-host PR jobs --------- Co-authored-by: host <host@test.local>
94 lines
3.7 KiB
Python
94 lines
3.7 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
|
|
|
|
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)
|
|
|
|
|
|
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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|