from __future__ import annotations import base64 import sys import tempfile import unittest from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).resolve().parents[1] SCRIPTS = ROOT / "scripts" if str(SCRIPTS) not in sys.path: sys.path.insert(0, str(SCRIPTS)) from audit_gitea_coordination import ( audit_repository, latest_claim, paged, parse_datetime, parse_write_paths, select_latest_claim, ) from setup_gitea_labels import LABELS, NoRedirect, build_plan from test_gitea_claim_race import PROBE_PREFIX, new_probe_branch from validate_agent_context import validate_manifest from validate_harness_governance import ( validate_markdown_links, validate_navigation, validate_repository, validate_secrets, validate_tasks, is_safe_repo_path, scopes_overlap, ) class RepositoryIntegrationTests(unittest.TestCase): def test_repository_governance_passes(self) -> None: self.assertEqual([], validate_repository(ROOT)) def test_context_manifest_passes(self) -> None: self.assertEqual([], validate_manifest(ROOT)) class OfflineRuleTests(unittest.TestCase): def test_broken_markdown_link_is_reported(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) page = root / "page.md" page.write_text("[missing](missing.md)\n", encoding="utf-8") findings = validate_markdown_links(root, [page]) self.assertEqual("markdown-link", findings[0].rule) def test_navigation_omission_is_reported(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) (root / "docs").mkdir() (root / "README.md").write_text("# root\n", encoding="utf-8") (root / "docs" / "README.md").write_text("# docs\n", encoding="utf-8") (root / "docs" / "new.md").write_text("# new\n", encoding="utf-8") findings = validate_navigation(root) self.assertTrue(any(item.path == "README.md" for item in findings)) self.assertTrue(any(item.path == "docs/README.md" for item in findings)) def test_task_status_and_scope_errors_are_reported(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) task_dir = root / "docs" / "tasks" task_dir.mkdir(parents=True) template = (ROOT / "docs" / "tasks" / "_template.md").read_text( encoding="utf-8" ) (task_dir / "_template.md").write_text(template, encoding="utf-8") task = template.replace("T-XXX", "T-001").replace( "status: TODO", "status: INVALID" ).replace("title: 一句话任务名", "title: []").replace( "phase: 1", "phase: banana" ).replace("created: 【日期】", "created: nonsense").replace( "issue: null", "issue: 0" ) (task_dir / "T-001.md").write_text(task, encoding="utf-8") rules = {finding.rule for finding in validate_tasks(root)} self.assertIn("task-status", rules) self.assertIn("task-scope", rules) self.assertIn("task-metadata", rules) self.assertIn("task-issue", rules) def test_done_task_requires_done_dependencies(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) task_dir = root / "docs" / "tasks" task_dir.mkdir(parents=True) template = (ROOT / "docs" / "tasks" / "_template.md").read_text( encoding="utf-8" ) (task_dir / "_template.md").write_text(template, encoding="utf-8") base = ( template.replace("created: 【日期】", "created: 2026-07-14") .replace(" - 【允许修改的仓库相对路径】\n", "") .replace( "（做完在此记录：改了哪些文件、跑的验证命令与结果、阻塞、关键决策。\n执行记录只写进本任务文件，不逐任务追加共享的 `progress.md`，避免多 agent 抢改共享文件。）", "验证：python -m unittest，结果通过。", ) ) first = base.replace("T-XXX", "T-001") second = ( base.replace("T-XXX", "T-002") .replace("deps: []", "deps: [T-001]") .replace("status: TODO", "status: DONE") ) (task_dir / "T-001.md").write_text(first, encoding="utf-8") (task_dir / "T-002.md").write_text(second, encoding="utf-8") findings = validate_tasks(root) self.assertTrue( any(item.rule == "task-deps" and item.path.endswith("T-002.md") for item in findings) ) def test_scope_prefix_overlap(self) -> None: self.assertTrue(scopes_overlap("src/api/", "src/api/users.py")) self.assertTrue(scopes_overlap("README.md", "README.md")) self.assertTrue(scopes_overlap(".", "src/api/users.py")) self.assertTrue(scopes_overlap("Src/API", "src/api/users.py")) self.assertFalse(scopes_overlap("src/api/", "src/ui/")) self.assertFalse(is_safe_repo_path("src/**")) self.assertFalse(is_safe_repo_path("src/\tapi")) def test_secret_finding_does_not_echo_value(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) path = root / "tracked.txt" secret = "private-" + "credential-value" path.write_text("GITEA_" + "TOKEN=" + secret + "\n", encoding="utf-8") findings = validate_secrets(root, [path]) rendered = "\n".join(finding.render() for finding in findings) self.assertTrue(findings) self.assertNotIn(secret, rendered) def test_secret_formats_are_detected(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) key = "GITEA_" + "TOKEN" secret = "another-" + "private-value" authorization = "Author" + "ization" paths = [] for name, content in ( (".env", f"{key}={secret}\n"), ("config.ps1", f"$env:{key} = '{secret}'\n"), ("config.json", f'{{"{key}": "{secret}"}}\n'), ("config.yml", f"{key}: {secret}\n"), ("defaults.env", f"{key}=${{TOKEN:-{secret}}}\n"), ("configure.cmd", f'set "{key}={secret}"\n'), ("headers.txt", f"{authorization}: Basic dXNl" + "cjpwYXNz\n"), ): path = root / name path.write_text(content, encoding="utf-8") paths.append(path) findings = validate_secrets(root, paths) self.assertGreaterEqual(len(findings), 7) self.assertNotIn(secret, "\n".join(item.render() for item in findings)) def test_windows_token_setters_and_utf16_are_detected(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) key = "GITEA_" + "TOKEN" secret = "windows-" + "private-value" setter = "[Environment]::SetEnvironmentVariable" path = root / "configure.ps1" path.write_text( f'{setter}("{key}", "{secret}", "User")\n' f'setx {key} {secret}\n', encoding="utf-16", ) findings = validate_secrets(root, [path]) self.assertGreaterEqual(len(findings), 2) self.assertNotIn(secret, "\n".join(item.render() for item in findings)) class GiteaHelperTests(unittest.TestCase): def test_label_plan_detects_exclusive_change(self) -> None: desired = next(label for label in LABELS if label["name"] == "status/todo") existing = { desired["name"]: { "id": 1, "name": desired["name"], "color": desired["color"], "description": desired["description"], "exclusive": False, } } actions = {item[1]["name"]: item[0] for item in build_plan(existing)} self.assertEqual("update", actions["status/todo"]) def test_redirect_handler_refuses_redirect(self) -> None: handler = NoRedirect() self.assertIsNone( handler.redirect_request(None, None, 302, "Found", {}, "http://example.invalid") ) def test_claim_parsers(self) -> None: comment = """CLAIM task: T-123 claimed_by: worker-1 allocated_by: dispatcher-1 context_ref: aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa claim_branch: claims/T-123 work_branch: agent/worker-1/T-123 write_paths: - docs/tasks/T-123.md - src/api/ claimed_at: 2029-05-31T12:00:00Z lease_until: 2029-06-01T12:00:00Z """ comments = [ {"body": "note"}, {"id": 1, "body": comment, "user": {"login": "dispatcher-1"}}, ] self.assertEqual(comment, latest_claim(comments, "T-123", "dispatcher-1")) self.assertEqual(["docs/tasks/T-123.md", "src/api/"], parse_write_paths(comment)) self.assertGreater( parse_datetime("2030-01-01T00:00:00Z"), datetime(2029, 1, 1, tzinfo=timezone.utc), ) self.assertIsNone(parse_datetime("2030-01-01")) quoted = {"id": 99, "body": "Discussion quoted CLAIM and task: T-123"} self.assertEqual( comment, latest_claim( [ { "id": 1, "body": comment, "user": {"login": "dispatcher-1"}, }, quoted, ], "T-123", "dispatcher-1", ), ) renewal = comment.replace("CLAIM\n", "CLAIM RENEWAL\n", 1).replace( "claimed_at: 2029-05-31T12:00:00Z", "claimed_at: 2029-06-01T00:00:00Z", ) selected, errors = select_latest_claim( comments + [ { "id": 2, "body": renewal, "user": {"login": "dispatcher-1"}, } ], "T-123", "dispatcher-1", ) self.assertEqual(renewal, selected) self.assertEqual([], errors) changed_identity = renewal.replace("claimed_by: worker-1", "claimed_by: worker-2") selected, errors = select_latest_claim( comments + [ {"id": 2, "body": renewal, "user": {"login": "worker-1"}}, { "id": 3, "body": changed_identity, "user": {"login": "dispatcher-1"}, }, ], "T-123", "dispatcher-1", ) self.assertEqual(comment, selected) self.assertGreaterEqual(len(errors), 2) def test_pagination_reads_until_empty_page(self) -> None: class FakePagedClient: def __init__(self) -> None: self.pages: list[int] = [] def request(self, method: str, path: str) -> object: page = int(path.rsplit("page=", 1)[1]) self.pages.append(page) if page == 1: return [{"id": number} for number in range(1, 21)] if page == 2: return [{"id": 21}] return [] client = FakePagedClient() values = paged(client, "/items") # type: ignore[arg-type] self.assertEqual(21, len(values)) self.assertEqual([1, 2, 3], client.pages) def test_probe_branch_is_unique_and_scoped(self) -> None: first = new_probe_branch() second = new_probe_branch() self.assertTrue(first.startswith(PROBE_PREFIX)) self.assertNotEqual(first, second) def test_valid_active_remote_task_audits_cleanly(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) task_dir = root / "docs" / "tasks" task_dir.mkdir(parents=True) template = (ROOT / "docs" / "tasks" / "_template.md").read_text( encoding="utf-8" ) default_task = template.replace("T-XXX", "T-123").replace( "issue: null", "issue: 1" ) (task_dir / "T-123.md").write_text(default_task, encoding="utf-8") sha = "a" * 40 work_task = ( default_task.replace("status: TODO", "status: DOING") .replace("context_ref: null", f"context_ref: {sha}") .replace("claim_branch: null", "claim_branch: claims/T-123") .replace("work_branch: null", "work_branch: agent/worker-1/T-123") .replace(" - 【允许修改的仓库相对路径】\n", "") ) claim = f"""CLAIM task: T-123 claimed_by: worker-1 allocated_by: dispatcher-1 context_ref: {sha} claim_branch: claims/T-123 work_branch: agent/worker-1/T-123 write_paths: - docs/tasks/T-123.md claimed_at: 2029-05-31T12:00:00Z lease_until: 2029-06-01T12:00:00Z """ labels = [ {"name": "kind/task"}, {"name": "type/code"}, {"name": "priority/p1"}, {"name": "status/doing"}, ] class FakeClient: def list_labels(self) -> dict[str, dict[str, object]]: return { label["name"]: { "name": label["name"], "exclusive": label["exclusive"], } for label in LABELS } def request(self, method: str, path: str, payload: object = None) -> object: self.assert_get(method) page = int(path.rsplit("page=", 1)[1]) if "page=" in path else 1 if page > 1: return [] if path.startswith("/issues?state=all"): return [ { "number": 1, "title": "[T-123] valid", "body": "- task_id: `T-123`\n- task_file: `docs/tasks/T-123.md`\n- write_paths:\n - `docs/tasks/T-123.md`\n", "state": "open", "labels": labels, } ] if path.startswith("/branches?"): return [ {"name": "claims/T-123", "commit": {"id": sha}}, {"name": "agent/worker-1/T-123", "commit": {"id": "b" * 40}}, ] if path.startswith("/pulls?"): return [] if path.startswith("/issues/1/comments?"): return [ { "body": claim, "user": {"login": "dispatcher-1"}, } ] if path.startswith("/contents/docs/tasks/T-123.md?"): return { "content": base64.b64encode(work_task.encode("utf-8")).decode( "ascii" ) } self.fail(f"unexpected path: {path}") def assert_get(self, method: str) -> None: if method != "GET": self.fail("audit attempted a write") def fail(self, message: str) -> None: raise AssertionError(message) findings, count = audit_repository( root, FakeClient(), # type: ignore[arg-type] datetime(2029, 6, 1, tzinfo=timezone.utc), "dispatcher-1", ) self.assertEqual(1, count) self.assertEqual([], findings) (task_dir / "T-123.md").write_text( default_task.replace("deps: []", "deps: [T-122]"), encoding="utf-8", ) findings, _ = audit_repository( root, FakeClient(), # type: ignore[arg-type] datetime(2029, 6, 1, tzinfo=timezone.utc), "dispatcher-1", ) self.assertTrue(any(item.rule == "dependency" for item in findings)) if __name__ == "__main__": unittest.main()