458 lines
22 KiB
Python
458 lines
22 KiB
Python
"""需求分流收据契约（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()
|