"""需求分流收据契约（scripts/task_triage.py）：远端发现、显式决策、核验拒绝。 全部对真实临时 Git 仓库（bare origin + clone）与进程内 Gitea 桩运行；不联网，不读取真实 ~/.secrets。 """ import hashlib import json import os import socket import subprocess import sys import tempfile import unittest from pathlib import Path from unittest import mock sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from test_task import FakeGitea # noqa: E402 SOURCE = Path(__file__).resolve().parents[1] SCRIPT = SOURCE / "scripts" / "task_triage.py" sys.path.insert(0, str(SOURCE / "scripts")) import task_triage # noqa: E402 def card_text(branch, kind="initiative", accepting="true", root="main", status="进行中"): slug = branch.replace("/", "-") return ( f"# {branch} 任务卡\n" f"任务主记录：fixture；状态：{status}；owner：Triage Test\n" f"- 权威仓：fixture；目标分支：main；R0（initial_base_sha）= 0\n" f"- 工作区：fixture（worktree）；分支：{branch}\n" f"- 分支类型：{kind}；initiative_id：{slug}；root_target：{root}；" f"immediate_target：{root}；accepting_requirements：{accepting}\n\n" "## 目标与非目标\n\n- 交付登录能力\n- 不做第三方登录\n\n## 其它\n\n- 无\n" ) class TriageTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory(prefix="triage-") self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name) self.remote = self.root / "remote.git" self.repo = self.root / "repo" self.receipt = self.root / "receipt.json" self.env = os.environ.copy() self.env.update({"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null", "GIT_ALLOW_PROTOCOL": "file", "GIT_TERMINAL_PROMPT": "0", "GIT_AUTHOR_NAME": "Triage Test", "GIT_AUTHOR_EMAIL": "triage@example.invalid", "GIT_COMMITTER_NAME": "Triage Test", "GIT_COMMITTER_EMAIL": "triage@example.invalid", "GIT_CONFIG_COUNT": "2", "GIT_CONFIG_KEY_0": "gc.auto", "GIT_CONFIG_VALUE_0": "0", "GIT_CONFIG_KEY_1": "maintenance.auto", "GIT_CONFIG_VALUE_1": "false", "NO_PROXY": "127.0.0.1,localhost", "no_proxy": "127.0.0.1,localhost", "GITEA_REPO": "fixture/repo", "GITEA_TOKEN": "dummy"}) for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GITEA_API_URL", "TASK_TARGET_BRANCH", "TRIAGE_RECEIPT"): self.env.pop(key, None) self.git(self.root, "init", "--bare", "--initial-branch=main", str(self.remote)) self.git(self.root, "init", "--initial-branch=main", str(self.repo)) (self.repo / "README").write_text("fixture\n") self.git(self.repo, "add", "README") self.git(self.repo, "commit", "-m", "Create fixture") self.git(self.repo, "remote", "add", "origin", str(self.remote)) self.git(self.repo, "push", "-u", "origin", "main") self.gitea = FakeGitea(token="dummy") self.addCleanup(self.gitea.close) self.env["GITEA_API_URL"] = self.gitea.url # ---- 夹具 ---- def git(self, cwd, *args): return subprocess.run(["git", *args], cwd=cwd, env=self.env, capture_output=True, check=True) def make_initiative(self, branch, **card): self.git(self.repo, "switch", "-c", branch, "main") path = self.repo / ".task" / (branch.replace("/", "-") + ".md") path.parent.mkdir(exist_ok=True) path.write_text(card_text(branch, **card), encoding="utf-8") self.git(self.repo, "add", ".task") self.git(self.repo, "commit", "-m", f"Record {branch}") self.git(self.repo, "push", "origin", branch) self.git(self.repo, "switch", "main") return self.head(branch) def advance_initiative(self, branch): self.git(self.repo, "switch", branch) (self.repo / "more-work.txt").write_text("later initiative work\n") self.git(self.repo, "add", "more-work.txt") self.git(self.repo, "commit", "-m", "More initiative work") self.git(self.repo, "push", "origin", branch) self.git(self.repo, "switch", "main") def head(self, ref): return subprocess.run(["git", "rev-parse", ref], cwd=self.repo, env=self.env, capture_output=True, text=True, check=True).stdout.strip() def open_pr(self, number, branch, base="main", state="open"): self.gitea.prs.append(FakeGitea.pr(number, branch, self.head(branch), base=base, state=state)) def run_script(self, *args, env=None): result = subprocess.run([sys.executable, str(SCRIPT), *args], cwd=self.repo, env=self.env | (env or {}), capture_output=True, text=True, timeout=60) result.out = result.stdout + result.stderr return result def discover(self, env=None, requirement="add login", receipt=None): return self.run_script("discover", "--receipt", str(receipt or self.receipt), "--requirement", requirement, env=env) def decide(self, verdict, reason="", initiative=None): args = ["decide", "--receipt", str(self.receipt), "--verdict", verdict, "--reason", reason] if initiative: args += ["--initiative", initiative] return self.run_script(*args) def verify(self, kind, initiative=None, max_age=None, env=None): args = ["verify", "--receipt", str(self.receipt), "--kind", kind] if initiative: args += ["--initiative", initiative] if max_age is not None: args += ["--max-age-seconds", str(max_age)] return self.run_script(*args, env=env) def load(self): return json.loads(self.receipt.read_text(encoding="utf-8")) def save(self, data): self.receipt.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") def discover_single(self, branch="feature/alpha"): self.make_initiative(branch) self.open_pr(len(self.gitea.prs) + 1, branch) result = self.discover() self.assertEqual(result.returncode, 5, result.out) return result # ---- 发现 ---- def test_remote_initiative_absent_from_main_tree_is_discovered(self): head = self.make_initiative("feature/alpha") self.open_pr(1, "feature/alpha") self.assertNotIn(".task/feature-alpha.md", subprocess.run(["git", "ls-tree", "-r", "--name-only", "main"], cwd=self.repo, env=self.env, capture_output=True, text=True).stdout) result = self.discover() self.assertEqual(result.returncode, 5, result.out) self.assertIn("[candidate] branch=feature/alpha", result.out) self.assertIn("[verdict] DECISION_REQUIRED", result.out) self.assertIn("make task-triage-decide", result.out) data = self.load() cand = data["candidates"][0] self.assertEqual(cand["head_sha"], head) self.assertEqual(cand["card_path"], ".task/feature-alpha.md") self.assertEqual(cand["pr_base"], "main") self.assertEqual(cand["pr_number"], 1) self.assertIn("交付登录能力", cand["goal_excerpt"]) blob = subprocess.run(["git", "cat-file", "blob", f"{head}:.task/feature-alpha.md"], cwd=self.repo, env=self.env, capture_output=True, check=True).stdout self.assertEqual(cand["card_sha256"], hashlib.sha256(blob).hexdigest()) self.assertEqual((data["final_verdict"], data["discovery_verdict"]), ("DECISION_REQUIRED", "DECISION_REQUIRED")) self.assertIsNone(data["decision"]) def test_zero_candidates_yields_new_without_selecting_an_initiative(self): result = self.discover() self.assertEqual(result.returncode, 0, result.out) self.assertIn("[verdict] NEW", result.out) data = self.load() self.assertEqual((data["discovery_verdict"], data["final_verdict"]), ("NEW", "NEW")) self.assertEqual(data["candidates"], []) self.assertEqual(data["decision"]["verdict"], "NEW") self.assertIsNone(data["decision"]["selected_initiative"]) self.assertEqual(data["decision"]["rationale"], "no accepting initiative") self.assertEqual(data["requirement"]["sha256"], hashlib.sha256(b"add login").hexdigest()) self.assertEqual(data["repo"]["repo_slug"], "fixture/repo") self.assertEqual(data["root_target_sha"], self.head("main")) def test_single_candidate_requires_decision_never_auto_join(self): self.discover_single() data = self.load() self.assertEqual(data["final_verdict"], "DECISION_REQUIRED") self.assertIsNone(data["decision"]) def test_multiple_candidates_stop_at_decision_required(self): self.make_initiative("feature/alpha") self.make_initiative("feature/beta") self.open_pr(1, "feature/alpha") self.open_pr(2, "feature/beta") result = self.discover() self.assertEqual(result.returncode, 5, result.out) self.assertIn("[candidate] branch=feature/alpha", result.out) self.assertIn("[candidate] branch=feature/beta", result.out) data = self.load() self.assertEqual([c["branch"] for c in data["candidates"]], ["feature/alpha", "feature/beta"]) self.assertEqual(data["final_verdict"], "DECISION_REQUIRED") def test_pr_base_not_main_and_closed_pr_are_ignored(self): self.make_initiative("feature/alpha") self.make_initiative("feature/beta") self.open_pr(1, "feature/alpha", base="release") self.open_pr(2, "feature/beta", state="closed") result = self.discover() self.assertEqual(result.returncode, 0, result.out) self.assertEqual(self.load()["final_verdict"], "NEW") def test_card_filters_ignore_closed_intake_done_plain_and_other_root(self): self.make_initiative("feature/active") self.make_initiative("feature/closed-intake", accepting="false") self.make_initiative("feature/done", status="已完成（回填）") self.make_initiative("feature/plain", kind="standalone") self.make_initiative("feature/other-root", root="release") for number, branch in enumerate(("feature/active", "feature/closed-intake", "feature/done", "feature/plain", "feature/other-root"), start=1): self.open_pr(number, branch) result = self.discover() self.assertEqual(result.returncode, 5, result.out) self.assertEqual([c["branch"] for c in self.load()["candidates"]], ["feature/active"]) def test_remote_discovery_paginates_until_empty(self): for number in range(1, 51): self.gitea.prs.append(FakeGitea.pr(number, f"feature/noise-{number}", "0" * 40, base="release")) self.make_initiative("feature/late") self.open_pr(51, "feature/late") result = self.discover() self.assertEqual(result.returncode, 5, result.out) self.assertEqual([c["branch"] for c in self.load()["candidates"]], ["feature/late"]) def test_api_unreachable_yields_remote_unverified_without_receipt(self): sock = socket.socket() sock.bind(("127.0.0.1", 0)) port = sock.getsockname()[1] sock.close() result = self.discover(env={"GITEA_API_URL": f"http://127.0.0.1:{port}/api/v1"}) self.assertEqual(result.returncode, 6, result.out) self.assertIn("[verdict] REMOTE_UNVERIFIED", result.out) self.assertFalse(self.receipt.exists()) def test_ls_remote_failure_yields_remote_unverified_without_receipt(self): self.git(self.repo, "remote", "set-url", "origin", str(self.root / "missing.git")) result = self.discover() self.assertEqual(result.returncode, 6, result.out) self.assertIn("[verdict] REMOTE_UNVERIFIED", result.out) self.assertFalse(self.receipt.exists()) def test_token_from_secrets_file_is_used_and_never_printed(self): home = self.root / "home" (home / ".secrets").mkdir(parents=True) (home / ".secrets" / "gitea").write_text( "URL_PREFIX=https://example.invalid\n- ACCESS_TOKEN=file-token-7f3a\nACCESS_TOKEN=second-token\n") self.gitea.token = "file-token-7f3a" self.make_initiative("feature/alpha") self.open_pr(1, "feature/alpha") result = self.discover(env={"HOME": str(home), "GITEA_TOKEN": ""}) self.assertEqual(result.returncode, 5, result.out) self.assertNotIn("file-token-7f3a", result.out) self.assertNotIn("file-token-7f3a", self.receipt.read_text(encoding="utf-8")) self.gitea.token = "other-token" rejected = self.discover(env={"HOME": str(home), "GITEA_TOKEN": ""}, receipt=self.root / "other.json") self.assertEqual(rejected.returncode, 6, rejected.out) self.assertIn("[verdict] REMOTE_UNVERIFIED", rejected.out) self.assertNotIn("file-token-7f3a", rejected.out) def test_repo_slug_is_derived_from_origin_when_gitea_repo_unset(self): owner_dir = self.root / "fixture-owner" owner_dir.mkdir() named = owner_dir / "fixture-repo.git" self.git(self.root, "init", "--bare", "--initial-branch=main", str(named)) self.git(self.repo, "remote", "set-url", "origin", str(named)) self.git(self.repo, "push", "-u", "origin", "main") result = self.discover(env={"GITEA_REPO": ""}) self.assertEqual(result.returncode, 0, result.out) self.assertEqual(self.load()["repo"]["repo_slug"], "fixture-owner/fixture-repo") def test_discover_requires_requirement_text(self): result = self.discover(requirement=" ") self.assertEqual(result.returncode, 2, result.out) # ---- 决策 ---- def test_join_decision_succeeds_and_verifies_child(self): head = self.make_initiative("feature/alpha") self.open_pr(1, "feature/alpha") self.discover() decided = self.decide("JOIN", "加入登录主线", initiative="feature/alpha") self.assertEqual(decided.returncode, 0, decided.out) data = self.load() self.assertEqual(data["final_verdict"], "JOIN") self.assertEqual(data["decision"]["selected_initiative"], "feature/alpha") verified = self.verify("child", initiative="feature/alpha") self.assertEqual(verified.returncode, 0, verified.out) self.assertIn("[triage-verified] verdict=JOIN selected_initiative=feature/alpha " "requirement_sha256=", verified.out) self.assertIn(f"[triage-head] selected_head_sha={head}", verified.out) def test_new_decision_for_single_non_fitting_candidate_requires_reason(self): self.discover_single() before = self.receipt.read_bytes() no_reason = self.decide("NEW", "") self.assertEqual(no_reason.returncode, 7, no_reason.out) self.assertEqual(self.receipt.read_bytes(), before) decided = self.decide("NEW", "不适配：独立的后台巡检任务") self.assertEqual(decided.returncode, 0, decided.out) self.assertEqual(self.load()["final_verdict"], "NEW") self.assertEqual(self.verify("standalone").returncode, 0) def test_decide_rejects_unknown_initiative_or_missing_reason_without_changing_receipt(self): self.make_initiative("feature/alpha") self.open_pr(1, "feature/alpha") self.discover() before = self.receipt.read_bytes() unknown = self.decide("JOIN", "理由", initiative="feature/beta") self.assertEqual(unknown.returncode, 7, unknown.out) no_reason = self.decide("JOIN", "", initiative="feature/alpha") self.assertEqual(no_reason.returncode, 7, no_reason.out) self.assertEqual(self.receipt.read_bytes(), before) def test_decide_rejects_candidate_head_change_and_keeps_receipt(self): self.discover_single() before = self.receipt.read_bytes() self.advance_initiative("feature/alpha") result = self.decide("JOIN", "加入", initiative="feature/alpha") self.assertEqual(result.returncode, 7, result.out) self.assertIn("candidate set changed", result.out) self.assertEqual(self.receipt.read_bytes(), before) # ---- 核验 ---- def test_verify_rejects_stale_receipt(self): self.assertEqual(self.discover().returncode, 0) self.assertEqual(self.verify("standalone").returncode, 0) data = self.load() data["discovered_at"] = "2001-01-01T00:00:00Z" self.save(data) result = self.verify("standalone") self.assertEqual(result.returncode, 8, result.out) self.assertIn("stale", result.out) def test_verify_rejects_changed_candidate_head(self): self.discover_single() self.decide("JOIN", "加入", initiative="feature/alpha") self.advance_initiative("feature/alpha") result = self.verify("child", initiative="feature/alpha") self.assertEqual(result.returncode, 8, result.out) self.assertIn("candidates changed", result.out) def test_verify_rejects_changed_card_digest(self): self.discover_single() self.decide("JOIN", "加入", initiative="feature/alpha") data = self.load() data["candidates"][0]["card_sha256"] = "0" * 64 self.save(data) result = self.verify("child", initiative="feature/alpha") self.assertEqual(result.returncode, 8, result.out) def test_verify_rejects_verdict_and_kind_mismatch(self): self.discover_single() self.decide("JOIN", "加入", initiative="feature/alpha") standalone = self.verify("standalone") self.assertEqual(standalone.returncode, 8, standalone.out) self.decide("NEW", "不适配") child = self.verify("child", initiative="feature/alpha") self.assertEqual(child.returncode, 8, child.out) def test_verify_rejects_child_with_wrong_initiative(self): self.make_initiative("feature/alpha") self.make_initiative("feature/beta") self.open_pr(1, "feature/alpha") self.open_pr(2, "feature/beta") self.discover() self.decide("JOIN", "加入", initiative="feature/alpha") result = self.verify("child", initiative="feature/beta") self.assertEqual(result.returncode, 8, result.out) def test_verify_rejects_root_target_mismatch(self): self.discover() result = self.verify("standalone", env={"TASK_TARGET_BRANCH": "release"}) self.assertEqual(result.returncode, 8, result.out) self.assertIn("root target", result.out) def test_verify_rejects_wrong_schema_and_missing_receipt(self): self.discover() data = self.load() data["schema"] = "agent-system/other/v9" self.save(data) self.assertEqual(self.verify("standalone").returncode, 8) self.receipt.unlink() missing = self.verify("standalone") self.assertEqual(missing.returncode, 8, missing.out) class RedactionTests(unittest.TestCase): def test_userinfo_is_stripped_from_urls_and_scp_forms(self): self.assertEqual(task_triage.redact_url("https://user:secret@git.example:30319/o/r.git"), "https://git.example:30319/o/r.git") self.assertEqual(task_triage.redact_url("git@git-ssh.example:weichen/Repo.git"), "git-ssh.example:weichen/Repo.git") self.assertEqual(task_triage.redact_url("/tmp/plain/remote.git"), "/tmp/plain/remote.git") class TokenSafetyTests(unittest.TestCase): FAKE_A = "a" * 40 FAKE_B = "b" * 40 def setUp(self): tmp = tempfile.TemporaryDirectory(prefix="triage-token-") self.addCleanup(tmp.cleanup) self.home = Path(tmp.name) def env(self, token=""): return mock.patch.dict(os.environ, {"HOME": str(self.home), "GITEA_TOKEN": token}) def write_secrets(self, text): (self.home / ".secrets").mkdir(exist_ok=True) (self.home / ".secrets" / "gitea").write_text(text, encoding="utf-8") def test_multiline_env_token_is_rejected_without_echoing_values(self): with self.env(f"{self.FAKE_A}\n{self.FAKE_B}"): with self.assertRaises(task_triage.Stop) as caught: task_triage.read_token() self.assertEqual(caught.exception.code, task_triage.EXIT_REMOTE) self.assertIn("malformed", caught.exception.message) for value in (self.FAKE_A, self.FAKE_B): self.assertNotIn(value, caught.exception.message) self.assertNotIn(value, str(caught.exception)) def test_secrets_file_with_sections_yields_home_value(self): self.write_secrets("IMC:\n- URL_PREFIX=https://imc.invalid\n- ACCESS_TOKEN=" + self.FAKE_B + "\n" "HOME:\n- URL_PREFIX=https://home.invalid\n- ACCESS_TOKEN=" + self.FAKE_A + "\n") with self.env(): self.assertEqual(task_triage.read_token(), self.FAKE_A) def test_secrets_file_without_home_access_token_returns_none(self): self.write_secrets("HOME:\n- URL_PREFIX=https://home.invalid\nIMC:\n- ACCESS_TOKEN=" + self.FAKE_B + "\n") with self.env(): self.assertIsNone(task_triage.read_token()) def test_missing_configuration_returns_none(self): with self.env(): self.assertIsNone(task_triage.read_token()) def test_request_value_error_does_not_leak_token(self): token = self.FAKE_A leak = ValueError(f"Invalid header value b'token {token}\\n{self.FAKE_B}'") with mock.patch.object(task_triage.urllib.request, "urlopen", side_effect=leak): with self.assertRaises(task_triage.Stop) as caught: task_triage.http_get_json("https://gitea.invalid/api/v1/x", token) self.assertEqual(caught.exception.code, task_triage.EXIT_REMOTE) self.assertEqual(caught.exception.message, "Gitea API request failed (ValueError)") self.assertIsNone(caught.exception.__cause__) self.assertTrue(caught.exception.__suppress_context__) for value in (token, self.FAKE_B): self.assertNotIn(value, caught.exception.message) if __name__ == "__main__": unittest.main()