codex-app-server/cloud/tests/test_pr_commit.py

"""pr.py 提交步骤单测：push 之前先把 worktree 改动提交到分支。

只用标准库，不联网、不跑真实 git/gh：git/gh 一律用假 runner
注入；worktree 用真实临时目录，只为验证 __pycache__ / *.pyc
清理（可再生的缓存）真的删、普通文件不动。
"""

import os
import sys
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace

try:
    from cloud.pr import PrStepError, create_task_pr
except ImportError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    from cloud.pr import PrStepError, create_task_pr


def _result(returncode=0, stdout="", stderr=""):
    return SimpleNamespace(returncode=returncode, stdout=stdout,
                           stderr=stderr)


class _Rec:
    """记录命令的假 runner：git/gh 分开注入，commit 行为可配。"""

    def __init__(self, branch="host/t1", commit="ok"):
        self.calls = []
        self.branch = branch
        self.commit = commit  # ok | nothing | fail | no-identity | add-fail

    def git(self, cmd, cwd):
        self.calls.append(list(cmd))
        if "remote" in cmd:
            return _result(0, "https://example.com/repo.git\n")
        if "branch" in cmd:
            return _result(0, self.branch + "\n")
        if "add" in cmd:
            if self.commit == "add-fail":
                return _result(1, "", "add failed")
            return _result(0, "")
        if "commit" in cmd:
            if "-c" in cmd:
                return _result(0, "")
            if self.commit == "nothing":
                return _result(1, "",
                               "nothing to commit, working tree clean")
            if self.commit == "fail":
                return _result(1, "", "fatal: commit exploded")
            if self.commit == "no-identity":
                return _result(1, "",
                               "Please tell me who you are")
            return _result(0, "")
        if "push" in cmd:
            return _result(0, "")
        return _result(0, "")

    def gh(self, cmd, cwd):
        self.calls.append(list(cmd))
        if cmd[:2] == ["gh", "auth"]:
            return _result(0, "", "")
        if cmd[:2] == ["gh", "pr"]:
            return _result(0, "https://example.com/repo/pull/7\n")
        return _result(0, "")

    def kinds(self):
        """git 命令按出现顺序归类：branch/add/commit/push。"""
        out = []
        for cmd in self.calls:
            if cmd[:1] == ["git"]:
                for k in ("branch", "add", "commit", "push"):
                    if k in cmd:
                        out.append(k)
                        break
        return out


class PrCommitTestCase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.wt = os.path.join(self.tmp.name, "t1")
        os.makedirs(self.wt)

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

    def _run(self, rec):
        return create_task_pr(
            "t1", 1, worktrees_root=self.tmp.name,
            git_runner=rec.git, gh_runner=rec.gh, gh_which="gh")

    def test_commit_before_push_and_pycache_cleaned(self):
        with open(os.path.join(self.wt, "file.txt"), "w") as f:
            f.write("x")
        cache = os.path.join(self.wt, "__pycache__")
        os.makedirs(cache)
        with open(os.path.join(cache, "x.pyc"), "wb") as f:
            f.write(b"\0")
        with open(os.path.join(self.wt, "stale.pyc"), "wb") as f:
            f.write(b"\0")
        rec = _Rec()
        res = self._run(rec)
        self.assertEqual(res["pr_url"], "https://example.com/repo/pull/7")
        self.assertEqual(res["branch"], "host/t1")
        kinds = rec.kinds()
        self.assertLess(kinds.index("branch"), kinds.index("add"))
        self.assertLess(kinds.index("add"), kinds.index("commit"))
        self.assertLess(kinds.index("commit"), kinds.index("push"))
        commit_cmd = [c for c in rec.calls if "commit" in c][0]
        self.assertIn("cloud: t1", commit_cmd)
        self.assertFalse(os.path.exists(cache))
        self.assertFalse(os.path.exists(os.path.join(self.wt, "stale.pyc")))
        self.assertTrue(os.path.exists(os.path.join(self.wt, "file.txt")))

    def test_nothing_to_commit_proceeds_to_push(self):
        rec = _Rec(commit="nothing")
        res = self._run(rec)
        self.assertEqual(res["pr_number"], 7)
        self.assertIn("push", rec.kinds())

    def test_wrong_branch_fails_commit_without_push(self):
        rec = _Rec(branch="main")
        with self.assertRaises(PrStepError) as ctx:
            self._run(rec)
        self.assertEqual(ctx.exception.step, "commit")
        self.assertNotIn("push", rec.kinds())
        self.assertNotIn("commit", rec.kinds())

    def test_commit_failure_is_commit_step_without_push(self):
        rec = _Rec(commit="fail")
        with self.assertRaises(PrStepError) as ctx:
            self._run(rec)
        self.assertEqual(ctx.exception.step, "commit")
        self.assertNotIn("push", rec.kinds())

    def test_add_failure_is_commit_step_without_push(self):
        rec = _Rec(commit="add-fail")
        with self.assertRaises(PrStepError) as ctx:
            self._run(rec)
        self.assertEqual(ctx.exception.step, "commit")
        self.assertNotIn("push", rec.kinds())

    def test_identity_fallback_retries_with_cloud_identity(self):
        rec = _Rec(commit="no-identity")
        res = self._run(rec)
        self.assertEqual(res["pr_number"], 7)
        commits = [c for c in rec.calls if "commit" in c]
        self.assertEqual(len(commits), 2)
        self.assertIn("-c", commits[1])
        self.assertIn("user.name=cloud", commits[1])
        self.assertIn("push", rec.kinds())


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