agent-exec transport b9c7fb196afb6984f4367f30ec8ec18a/162bb30bb46e

2026-10-18 00:49:09 +00:00
parent b103f3c58a
commit 2757b6f68e
11 changed files with 325 additions and 24 deletions
@@ -1490,6 +1490,115 @@ class TaskLifecycleTests(unittest.TestCase):
        self.assertIn("已包含于 origin/main", archived)
        self.assertIn("PR #5", archived)

+    # ---- close --self：合入前在任务分支内闭合本卡 ----
+
+    def self_close_env(self):
+        return self.close_env() | {"GITEA_API_URL": self.gitea.url, "GITEA_REPO": "fixture/repo"}
+
+    def start_standalone(self, name="feature/solo"):
+        receipt = self.capture_precheck()
+        triage = self.triage_receipt()
+        self.task(self.repo, "start", name,
+                  env=self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage)})
+        return self.root / ("repo-wt-" + name.replace("/", "-"))
+
+    def open_pr(self, wt, branch, base="main"):
+        self.git(wt, "push", "-q", "-u", "origin", branch)
+        number = len(self.gitea.prs) + 1
+        sha = self.git(wt, "rev-parse", "HEAD").stdout.strip()
+        self.gitea.prs.append(FakeGitea.pr(number, branch, sha, base=base))
+        return number
+
+    def test_close_self_archives_own_card_inside_its_pr(self):
+        wt = self.start_standalone()
+        (wt / "work").write_text("task result\n")
+        self.git(wt, "add", "work")
+        self.git(wt, "commit", "-m", "Task work")
+        number = self.open_pr(wt, "feature/solo")
+        result = self.task(wt, "close", "--self", env=self.self_close_env())
+        self.assertIn(f"[closed] .task/feature-solo.md 已在 PR #{number} 合入前闭合并归档；合入后无需 task-close",
+                      result.stdout)
+        self.assertNotIn(self.gitea.token, result.stdout)
+        self.assertFalse((wt / ".task/feature-solo.md").exists())
+        text = (wt / ".task/archive/feature-solo.md").read_text()
+        self.assertIn("；状态：已完成（合入前闭合）；", text)
+        self.assertIn(f"任务主记录：PR #{number} {self.gitea.web}/{self.root.name}/remote/pulls/{number}；", text)
+        self.assertIn("## 结果与验证\n", text)
+        self.assertNotIn("## 结果与验证（完成后填写）", text)
+        self.assertIn("## 闭合记录（合入前折叠）", text)
+        self.assertIn(f"PR #{number}", text.split("## 闭合记录（合入前折叠）", 1)[1])
+        self.assertEqual(self.git(wt, "log", "-1", "--format=%s").stdout.strip(),
+                         "Fold-close the task card before merge")
+        body = self.git(wt, "log", "-1", "--format=%B").stdout
+        for field in ("Why:", "Evidence:", "Impact:", "Refs: task.sh close --self"):
+            self.assertIn(field, body)
+        self.assertEqual(self.git(wt, "status", "--porcelain").stdout, "")
+        head = self.git(wt, "rev-parse", "HEAD").stdout.strip()
+        self.assertEqual(self.git(self.remote, "rev-parse", "refs/heads/feature/solo").stdout.strip(), head)
+        self.assertEqual(self.git(self.remote, "show", "feature/solo:.task/archive/feature-solo.md").stdout, text)
+        again = self.task(wt, "close", "--self", check=False, env=self.self_close_env())
+        self.assertNotEqual(again.returncode, 0, again.stdout)
+        self.assertEqual(self.git(wt, "rev-parse", "HEAD").stdout.strip(), head)
+
+    def test_close_self_accepts_explicit_pr_number_without_lookup(self):
+        wt = self.start_standalone()
+        env = self.close_env() | {"TASK_PR": "42", "GITEA_API_URL": "http://127.0.0.1:9/api/v1"}
+        result = self.task(wt, "close", "--self", env=env)
+        self.assertIn("PR #42 合入前闭合", result.stdout)
+        self.assertIn("任务主记录：PR #42 ", (wt / ".task/archive/feature-solo.md").read_text())
+
+    def test_close_self_refuses_main_detached_dirty_and_missing_pr(self):
+        main_head = self.git(self.repo, "rev-parse", "HEAD").stdout.strip()
+        on_main = self.task(self.repo, "close", "--self", check=False, env=self.self_close_env())
+        self.assertNotEqual(on_main.returncode, 0, on_main.stdout)
+        self.assertIn("不能在 main 上运行", on_main.stdout)
+        self.assertEqual(self.git(self.repo, "rev-parse", "HEAD").stdout.strip(), main_head)
+        wt = self.start_standalone()
+        head = self.git(wt, "rev-parse", "HEAD").stdout.strip()
+        card = wt / ".task/feature-solo.md"
+        original = card.read_text()
+
+        def assert_refused(result, reason):
+            self.assertNotEqual(result.returncode, 0, result.stdout)
+            self.assertIn(reason, result.stdout)
+            self.assertNotIn(self.gitea.token, result.stdout)
+            self.assertEqual(self.git(wt, "rev-parse", "HEAD").stdout.strip(), head)
+            self.assertEqual(card.read_text(), original)
+            self.assertFalse((wt / ".task/archive").exists())
+
+        (wt / "stray").write_text("uncommitted\n")
+        assert_refused(self.task(wt, "close", "--self", check=False, env=self.self_close_env()), "不干净")
+        (wt / "stray").unlink()
+        assert_refused(self.task(wt, "close", "--self", check=False, env=self.self_close_env()),
+                       "需要恰好一个 head 为 feature/solo 的开放 PR")
+        assert_refused(self.task(wt, "close", "--self", check=False,
+                                 env=self.self_close_env() | {"TASK_PR": "7; rm"}), "TASK_PR")
+        self.git(wt, "checkout", "-q", "--detach")
+        assert_refused(self.task(wt, "close", "--self", check=False, env=self.self_close_env()), "detached HEAD")
+
+    def test_close_self_child_keeps_initiative_target_from_archive(self):
+        parent = self.start_initiative()
+        wt, _ = self.start_child()
+        number = self.open_pr(wt, "feature/alpha-login", base="feature/alpha")
+        self.task(wt, "close", "--self", env=self.self_close_env())
+        archived = wt / ".task/archive/feature-alpha-login.md"
+        self.assertIn("immediate_target：feature/alpha", archived.read_text())
+        self.assertIn(f"PR #{number}", archived.read_text())
+        (parent / "parent-later").write_text("later initiative work\n")
+        self.git(parent, "add", "parent-later")
+        self.git(parent, "commit", "-m", "Later initiative work")
+        self.publish_initiative("feature/alpha")
+        parent_head = self.git(parent, "rev-parse", "HEAD").stdout.strip()
+        synced = self.task(wt, "sync")
+        self.assertIn("sync_target=feature/alpha", synced.stdout)
+        self.assertIn(f"sync_base={parent_head}", synced.stdout)
+        self.assertIn(f"| 同步（记录前对象） | {parent_head} |", archived.read_text())
+        self.git(parent, "merge", "--no-ff", "feature/alpha-login", "-m", "Merge child into initiative")
+        self.publish_initiative("feature/alpha")
+        finished = self.task(wt, "finish")
+        self.assertIn("[verified] feature/alpha-login is included in parent initiative feature/alpha",
+                      finished.stdout)
+

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