428 lines
17 KiB
Python
428 lines
17 KiB
Python
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 (
|
|
ACTIVE_LABELS,
|
|
READY_OR_ACTIVE_LABELS,
|
|
STATUS_LABELS,
|
|
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_waiting_status_is_not_claimable_or_active(self) -> None:
|
|
self.assertIn("status/waiting", STATUS_LABELS)
|
|
self.assertNotIn("status/waiting", READY_OR_ACTIVE_LABELS)
|
|
self.assertNotIn("status/waiting", ACTIVE_LABELS)
|
|
|
|
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()
|