codex-app-server/deploy/tests/test_host_deploy.py

"""host_deploy.sh 的仿真测试：假 systemctl / 假 curl，不碰真实主机。

每个用例都在临时目录里建一个假 APP_DIR，host_deploy.sh 的
CODEX_DEPLOY_* 环境变量把它指过去。生产默认值不受影响。
"""
import hashlib
import json
import os
import subprocess
import tarfile
import tempfile
import unittest

REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(
    os.path.abspath(__file__))))
HOST_SCRIPT = os.path.join(REPO_ROOT, "deploy", "host_deploy.sh")


def make_tarball(source_dir, tarball):
    with tarfile.open(tarball, "w:gz", format=tarfile.PAX_FORMAT) as tf:
        for root, dirs, files in os.walk(source_dir):
            dirs.sort()
            for name in sorted(files):
                full = os.path.join(root, name)
                tf.add(full, arcname=os.path.relpath(full, source_dir))
    with open(tarball, "rb") as f:
        return hashlib.sha256(f.read()).hexdigest()


class HostDeploySim(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        base = self.tmp.name
        self.app = os.path.join(base, "codex-app-server")
        os.makedirs(os.path.join(self.app, "work"))
        with open(os.path.join(self.app, "DEPLOYED_COMMIT"), "w") as f:
            f.write("oldcommit\n")
        with open(os.path.join(self.app, "work", "keep.txt"), "w") as f:
            f.write("keep\n")
        with open(os.path.join(self.app, "old.py"), "w") as f:
            f.write("x = 0\n")
        # 假 systemctl：is-active 一律失败（走 cloud-worker 分支），其余成功。
        fake_bin = os.path.join(base, "fakebin")
        os.makedirs(fake_bin)
        with open(os.path.join(fake_bin, "systemctl"), "w") as f:
            f.write("#!/bin/sh\n"
                    "echo \"$@\" >> \"$FAKE_SYSTEMCTL_LOG\"\n"
                    "if [ \"$1\" = \"is-active\" ]; then exit 1; fi\n"
                    "exit 0\n")
        with open(os.path.join(fake_bin, "curl"), "w") as f:
            f.write("#!/bin/sh\ncat \"$FAKE_HEALTH_JSON\"\n")
        os.chmod(os.path.join(fake_bin, "systemctl"), 0o755)
        os.chmod(os.path.join(fake_bin, "curl"), 0o755)
        self.fake_bin = fake_bin
        self.health_path = os.path.join(base, "health.json")
        self.log_path = os.path.join(base, "systemctl.log")
        open(self.log_path, "w").close()
        self.env = dict(os.environ)
        self.env["PATH"] = fake_bin + os.pathsep + os.environ["PATH"]
        self.env["CODEX_DEPLOY_APP_DIR"] = self.app
        self.env["CODEX_DEPLOY_BACKUP_DIR"] = os.path.join(base, "backups")
        self.env["CODEX_DEPLOY_STAGING_DIR"] = os.path.join(base, "staging")
        self.env["CODEX_DEPLOY_OLD_DIR"] = os.path.join(base, "old")
        self.env["CODEX_DEPLOY_FAILED_DIR"] = os.path.join(base, "failed")
        self.env["CODEX_DEPLOY_PRESERVE_DIR"] = os.path.join(base, "preserve")
        self.env["CODEX_DEPLOY_HEALTH_INTERVAL"] = "0.1"
        self.env["CODEX_DEPLOY_HEALTH_ATTEMPTS"] = "3"
        self.env["FAKE_HEALTH_JSON"] = self.health_path
        self.env["FAKE_SYSTEMCTL_LOG"] = self.log_path

    def tearDown(self):
        self.tmp.cleanup()

    def set_health(self, payload):
        with open(self.health_path, "w") as f:
            json.dump(payload, f)

    def build_new_tarball(self, commit="newcommit"):
        src = os.path.join(self.tmp.name, "src")
        os.makedirs(src, exist_ok=True)
        with open(os.path.join(src, "DEPLOYED_COMMIT"), "w") as f:
            f.write(commit + "\n")
        with open(os.path.join(src, "new.py"), "w") as f:
            f.write("x = 1\n")
        tarball = os.path.join(self.tmp.name, "release.tar.gz")
        sha = make_tarball(src, tarball)
        return tarball, sha

    def run_host(self, tarball, sha, commit):
        return subprocess.run(
            ["sh", HOST_SCRIPT, tarball, sha, commit],
            capture_output=True, text=True, env=self.env)

    def live_commit(self):
        with open(os.path.join(self.app, "DEPLOYED_COMMIT")) as f:
            return f.read().strip()

    def test_success_swaps_and_preserves_work(self):
        self.set_health({"ok": True, "backend": "postgres", "db_ok": True,
                         "commit": "newcommit"})
        tarball, sha = self.build_new_tarball()
        r = self.run_host(tarball, sha, "newcommit")
        self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
        self.assertIn("RELEASE_OK", r.stdout)
        self.assertIn("commit=newcommit", r.stdout)
        self.assertEqual(self.live_commit(), "newcommit")
        with open(os.path.join(self.app, "work", "keep.txt")) as f:
            self.assertEqual(f.read().strip(), "keep")
        self.assertTrue(os.path.exists(os.path.join(self.app, "new.py")))
        self.assertFalse(os.path.exists(os.path.join(self.app, "old.py")))
        backups = os.listdir(self.env["CODEX_DEPLOY_BACKUP_DIR"])
        self.assertEqual(len(backups), 1)
        self.assertIn("oldcommit", backups[0])

    def test_bad_sha_never_touches_live(self):
        self.set_health({"ok": True, "backend": "postgres", "db_ok": True,
                         "commit": "newcommit"})
        tarball, _sha = self.build_new_tarball()
        r = self.run_host(tarball, "0" * 64, "newcommit")
        self.assertNotEqual(r.returncode, 0)
        self.assertEqual(self.live_commit(), "oldcommit")
        self.assertFalse(os.path.exists(self.env["CODEX_DEPLOY_BACKUP_DIR"])
                         and os.listdir(self.env["CODEX_DEPLOY_BACKUP_DIR"]))

    def test_manifest_mismatch_never_touches_live(self):
        self.set_health({"ok": True, "backend": "postgres", "db_ok": True,
                         "commit": "newcommit"})
        tarball, sha = self.build_new_tarball(commit="othercommit")
        r = self.run_host(tarball, sha, "newcommit")
        self.assertNotEqual(r.returncode, 0)
        self.assertEqual(self.live_commit(), "oldcommit")

    def test_health_failure_rolls_back(self):
        # health 通过但 commit 是旧的：这是旧链路会误报成功的情况，
        # 新链路必须回滚，live 回到 oldcommit。
        self.set_health({"ok": True, "backend": "postgres", "db_ok": True,
                         "commit": "oldcommit"})
        tarball, sha = self.build_new_tarball()
        r = self.run_host(tarball, sha, "newcommit")
        self.assertNotEqual(r.returncode, 0)
        self.assertEqual(self.live_commit(), "oldcommit")
        with open(os.path.join(self.app, "work", "keep.txt")) as f:
            self.assertEqual(f.read().strip(), "keep")


if __name__ == "__main__":
    unittest.main()