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

"""host_git_fetch.sh 的仿真测试：本地 git 仓库代替 GitHub，不联网、
不碰真实主机、不改生产。

每个用例都在临时目录里建一个小 git 仓库（一个提交），用假
host_deploy.sh 代替真实换目录脚本：假脚本只验 tar sha 与
DEPLOYED_COMMIT，打印 RELEASE_OK，不碰 systemctl、不动 live。
host_git_fetch.sh 的 CODEX_GIT_* 环境变量把工作区指到临时目录。
"""
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__))))
GIT_FETCH_SCRIPT = os.path.join(REPO_ROOT, "deploy", "host_git_fetch.sh")


def _run_git(cwd, *args):
    subprocess.run(["git"] + list(args), cwd=cwd, check=True,
                   capture_output=True, text=True)


def _make_repo(base):
    repo = os.path.join(base, "repo")
    os.makedirs(repo)
    _run_git(repo, "init", "-q")
    _run_git(repo, "config", "user.email", "t@t")
    _run_git(repo, "config", "user.name", "t")
    with open(os.path.join(repo, "hello.py"), "w") as f:
        f.write("x = 1\n")
    _run_git(repo, "add", ".")
    _run_git(repo, "commit", "-qm", "init")
    commit = subprocess.run(["git", "rev-parse", "HEAD"], cwd=repo,
                            capture_output=True, text=True,
                            check=True).stdout.strip()
    return repo, commit


def _make_fake_host_deploy(base):
    """假 host_deploy.sh：验 tar sha + 包内清单，打印 RELEASE_OK。

    签名与真实脚本一致：sh host_deploy.sh <tarball> <sha> <commit>。
    """
    path = os.path.join(base, "fake_host_deploy.sh")
    with open(path, "w") as f:
        f.write("#!/bin/sh\n"
                "set -eu\n"
                "TARBALL=$1; EXPECTED_SHA=$2; EXPECTED_COMMIT=$3\n"
                "ACTUAL_SHA=$(sha256sum \"$TARBALL\" | awk '{print $1}')\n"
                "if [ \"$ACTUAL_SHA\" != \"$EXPECTED_SHA\" ]; then\n"
                "  echo \"sha mismatch\" >&2; exit 2\n"
                "fi\n"
                "TARBALL_COMMIT=$(tar -xzf \"$TARBALL\" -O DEPLOYED_COMMIT "
                "| tr -d ' \\n\\r')\n"
                "if [ \"$TARBALL_COMMIT\" != \"$EXPECTED_COMMIT\" ]; then\n"
                "  echo \"commit mismatch\" >&2; exit 2\n"
                "fi\n"
                "echo \"RELEASE_OK commit=$EXPECTED_COMMIT "
                "sha=$EXPECTED_SHA backup=/tmp/fake-backup.tar.gz\"\n")
    os.chmod(path, 0o644)
    return path


class HostGitFetchSim(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.base = self.tmp.name
        self.repo, self.commit = _make_repo(self.base)
        self.fake_deploy = _make_fake_host_deploy(self.base)
        self.key = os.path.join(self.base, "id_test")
        with open(self.key, "w") as f:
            f.write("fake-key-material-not-a-real-key\n")
        os.chmod(self.key, 0o600)
        self.workdir = os.path.join(self.base, "workdir")
        self.env = dict(os.environ)
        self.env["CODEX_GIT_ALLOW_LOCAL"] = "1"
        self.env["CODEX_GIT_WORKDIR"] = self.workdir
        self.env["CODEX_GIT_TIMEOUT"] = "30"
        self.env["CODEX_GIT_ATTEMPTS"] = "1"
        self.env["CODEX_GIT_SLEEP"] = "0"

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

    def _run_fetch(self, repo, commit, key=None, deploy=None):
        return subprocess.run(
            ["sh", GIT_FETCH_SCRIPT, repo, commit,
             key or self.key, deploy or self.fake_deploy],
            capture_output=True, text=True, env=self.env)

    def test_success_builds_tar_with_clean_member_names(self):
        r = self._run_fetch(self.repo, self.commit)
        self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
        self.assertIn("RELEASE_OK commit=%s" % self.commit, r.stdout)
        tarball = os.path.join(
            self.workdir, "codex-app-server-%s.tar.gz" % self.commit)
        self.assertTrue(os.path.isfile(tarball))
        with tarfile.open(tarball, "r:gz") as tf:
            names = tf.getnames()
        # 关键：成员名不带 ./ 前缀，否则真实 host_deploy.sh 的
        # tar -O DEPLOYED_COMMIT 找不到成员、会误判清单不符。
        self.assertIn("DEPLOYED_COMMIT", names)
        self.assertIn("RELEASE.json", names)
        self.assertIn("hello.py", names)
        self.assertNotIn(".git", names)
        self.assertFalse(any(n.startswith("./") and n != "./"
                           for n in names))

    def test_unknown_commit_fails_before_host_deploy(self):
        # 40 位完整 sha 但不在仓库里：checkout 失败，host_deploy
        # 绝不能被调到（假脚本被调到时会在 workdir 留标记文件，
        # 这里直接断言退出码非零即可，live 未动是脚本结构保证）。
        r = self._run_fetch(self.repo, "0" * 40)
        self.assertNotEqual(r.returncode, 0)

    def test_key_with_wrong_permissions_refuses(self):
        os.chmod(self.key, 0o644)
        r = self._run_fetch(self.repo, self.commit)
        self.assertEqual(r.returncode, 2)
        self.assertIn("0600", r.stderr + r.stdout)

    def test_missing_key_refuses(self):
        r = self._run_fetch(self.repo, self.commit,
                            key=os.path.join(self.base, "no-such-key"))
        self.assertEqual(r.returncode, 2)

    def test_https_repo_refuses(self):
        r = self._run_fetch(
            "https://github.com/openqj/codex-app-server.git", self.commit)
        self.assertEqual(r.returncode, 2)

    def test_short_commit_refuses(self):
        r = self._run_fetch(self.repo, self.commit[:7])
        self.assertEqual(r.returncode, 2)


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