host: GitHub PR/CI line (github-workflow, pr, pr-checks, pr-merge with checks gate)

tests/test_host.py

@@ -373,6 +373,154 @@ class HealthRestoreRecoverTest(unittest.TestCase):
             self.assertTrue(report["checks"]["server_ok"])
 
 
+class GithubPRTest(unittest.TestCase):
+    class FakeGh:
+        def __init__(self, checks_output="", pr_url=""):
+            self.checks_output = checks_output
+            self.pr_url = pr_url
+            self.calls = []
+
+        def __call__(self, args):
+            self.calls.append(list(args))
+            import types
+            if args[:2] == ["pr", "checks"]:
+                return types.SimpleNamespace(returncode=0,
+                                             stdout=self.checks_output,
+                                             stderr="")
+            if args[:2] == ["pr", "create"]:
+                return types.SimpleNamespace(returncode=0,
+                                             stdout=self.pr_url + "\n",
+                                             stderr="")
+            if args[:2] == ["pr", "merge"]:
+                return types.SimpleNamespace(returncode=0,
+                                             stdout="Merged\n", stderr="")
+            return types.SimpleNamespace(returncode=1, stdout="",
+                                         stderr="unknown fake gh call")
+
+    def test_workflow_content_and_idempotent(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            changed, path = host.generate_github_workflow(tmp)
+            self.assertTrue(changed)
+            content = Path(path).read_text()
+            self.assertIn("actions/checkout", content)
+            self.assertIn("setup-python", content)
+            self.assertIn("python3 -m unittest discover -s tests -v",
+                          content)
+            changed2, _ = host.generate_github_workflow(tmp)
+            self.assertFalse(changed2)
+            # 内容被改坏后再次生成应恢复
+            Path(path).write_text("garbage\n")
+            changed3, _ = host.generate_github_workflow(tmp)
+            self.assertTrue(changed3)
+            self.assertEqual(Path(path).read_text(), content)
+
+    def test_pr_checks_parse_pass_fail_mixed(self):
+        checks, all_pass = host.parse_pr_checks(
+            "build\tpass\t1m\ntest\tpass\t2m\n")
+        self.assertTrue(all_pass)
+        self.assertEqual([c["state"] for c in checks], ["pass", "pass"])
+        checks, all_pass = host.parse_pr_checks(
+            "✓ build pass\n✗ test fail\n")
+        self.assertFalse(all_pass)
+        self.assertEqual([c["name"] for c in checks], ["build", "test"])
+        checks, all_pass = host.parse_pr_checks("")
+        self.assertFalse(all_pass)
+        self.assertEqual(checks, [])
+
+    def test_pr_merge_refuses_when_not_all_pass(self):
+        fake = self.FakeGh(checks_output="build\tpass\ntest\tfail\n")
+        task = {"name": "t", "pr_url": "https://example.com/pull/7",
+                "pr_number": 7, "status": "completed"}
+        ok, msg = host.merge_pr(task, gh_runner=fake)
+        self.assertFalse(ok)
+        self.assertIn("未全过", msg)
+        self.assertNotIn(["pr", "merge", "7", "--merge"], fake.calls)
+        self.assertEqual(task["status"], "completed")
+
+    def test_pr_merge_only_when_all_pass(self):
+        fake = self.FakeGh(checks_output="build\tpass\ntest\tpass\n")
+        task = {"name": "t", "pr_url": "https://example.com/pull/8",
+                "pr_number": 8, "status": "completed"}
+        ok, msg = host.merge_pr(task, gh_runner=fake)
+        self.assertTrue(ok, msg)
+        self.assertIn(["pr", "merge", "8", "--merge"], fake.calls)
+        self.assertEqual(task["status"], "merged")
+
+    def test_pr_without_branch_or_wrong_status_rejected(self):
+        with self.assertRaises(ValueError):
+            host.create_pr("/tmp", {"name": "t", "status": "completed"})
+        with self.assertRaises(ValueError):
+            host.create_pr("/tmp", {"name": "t", "branch": "host/t",
+                                    "status": "running"})
+
+    def test_pr_without_remote_gives_clear_error(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            _init_git_repo(tmp)
+            Path(tmp, "f.txt").write_text("x\n")
+            subprocess.run(["git", "add", "-A"], cwd=tmp, check=True)
+            subprocess.run(["git", "commit", "-qm", "initial"],
+                           cwd=tmp, check=True)
+            task = {"name": "t", "branch": "host/t", "status": "completed",
+                    "prompt": "do it"}
+            with self.assertRaises(RuntimeError) as ctx:
+                host.create_pr(tmp, task, gh_runner=self.FakeGh())
+            self.assertIn("remote", str(ctx.exception))
+
+    def test_pr_push_to_local_bare_remote_then_fake_gh(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            repo = os.path.join(tmp, "repo")
+            os.makedirs(repo)
+            _init_git_repo(repo)
+            Path(repo, "f.txt").write_text("x\n")
+            subprocess.run(["git", "add", "-A"], cwd=repo, check=True)
+            subprocess.run(["git", "commit", "-qm", "initial"],
+                           cwd=repo, check=True)
+            subprocess.run(["git", "checkout", "-qb", "host/t1"],
+                           cwd=repo, check=True)
+            Path(repo, "f.txt").write_text("y\n")
+            subprocess.run(["git", "commit", "-qam", "change"],
+                           cwd=repo, check=True)
+            remote = os.path.join(tmp, "remote.git")
+            subprocess.run(["git", "init", "--bare", "-q", remote],
+                           check=True)
+            subprocess.run(["git", "remote", "add", "origin", remote],
+                           cwd=repo, check=True)
+            fake = self.FakeGh(
+                pr_url="https://github.com/example/repo/pull/42")
+            task = {"name": "t1", "branch": "host/t1",
+                    "status": "completed", "prompt": "change f"}
+            info = host.create_pr(repo, task, gh_runner=fake)
+            self.assertEqual(info["pr_number"], 42)
+            self.assertEqual(task["pr_url"],
+                             "https://github.com/example/repo/pull/42")
+            self.assertEqual(task["pr_number"], 42)
+            branches = subprocess.run(
+                ["git", "--git-dir", remote, "branch"], capture_output=True,
+                text=True).stdout
+            self.assertIn("host/t1", branches)
+
+    def test_pr_cli_without_remote_fails_with_clear_error(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            repo = os.path.join(tmp, "repo")
+            os.makedirs(repo)
+            _init_git_repo(repo)
+            state_path = os.path.join(tmp, "host-state.json")
+            with open(state_path, "w") as f:
+                json.dump({"tasks": {"t1": {
+                    "name": "t1", "branch": "host/t1",
+                    "status": "completed", "prompt": "do it",
+                    "repo": repo}}, "automations": []}, f)
+            cfg_path = os.path.join(tmp, "cfg.json")
+            with open(cfg_path, "w") as f:
+                json.dump({"repo": repo, "state": state_path}, f)
+            r = subprocess.run(
+                [sys.executable, str(REPO_ROOT / "host.py"),
+                 "--config", cfg_path, "pr", "--task", "t1"],
+                capture_output=True, text=True)
+            self.assertNotEqual(r.returncode, 0)
+            self.assertIn("remote", r.stdout + r.stderr)
+
+
 if __name__ == "__main__":
     unittest.main()