938 lines
34 KiB
Python
938 lines
34 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import yaml
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
ACK_DIR = REPO_ROOT / "skills" / "ack"
|
|
SCRIPTS_DIR = ACK_DIR / "scripts"
|
|
SCHEMA_PATH = ACK_DIR / "templates" / "knowledge.schema.json"
|
|
EXAMPLE_PATH = ACK_DIR / "examples" / "knowledge.example.yaml"
|
|
TASKS_EXAMPLE_PATH = ACK_DIR / "examples" / "tasks.example.yaml"
|
|
|
|
sys.path.insert(0, str(SCRIPTS_DIR))
|
|
import select_knowledge # noqa: E402
|
|
import validate_knowledge # noqa: E402
|
|
|
|
|
|
def example_data() -> dict:
|
|
return yaml.safe_load(EXAMPLE_PATH.read_text(encoding="utf-8"))
|
|
|
|
|
|
class AckKnowledgeTests(unittest.TestCase):
|
|
def test_schema_template_and_example_are_aligned(self) -> None:
|
|
schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8"))
|
|
template = yaml.safe_load(
|
|
(ACK_DIR / "templates" / "knowledge.template.yaml").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
schema["definitions"]["entry"]["properties"]["kind"]["enum"],
|
|
["guardrail", "pitfall", "verification"],
|
|
)
|
|
self.assertEqual(
|
|
schema["definitions"]["entry"]["properties"]["status"]["enum"],
|
|
["active", "stale", "superseded", "archived"],
|
|
)
|
|
self.assertEqual(template["entries"], [])
|
|
self.assertEqual(template["verificationRegistry"], {})
|
|
|
|
errors, _ = validate_knowledge.validate_all(
|
|
example_data(), SCHEMA_PATH, use_schema=False
|
|
)
|
|
self.assertEqual(errors, [])
|
|
self.assertEqual(
|
|
validate_knowledge.stable_ref(example_data()["entries"][0]), "K-001@1"
|
|
)
|
|
|
|
def test_semantics_run_even_when_schema_reports_no_errors(self) -> None:
|
|
data = example_data()
|
|
data["entries"].append(copy.deepcopy(data["entries"][0]))
|
|
with mock.patch.object(
|
|
validate_knowledge, "validate_with_schema", return_value=[]
|
|
):
|
|
errors, mode = validate_knowledge.validate_all(
|
|
data, SCHEMA_PATH, use_schema=True
|
|
)
|
|
self.assertIn("内置语义", mode)
|
|
self.assertTrue(any("稳定引用 K-001@1 重复" in error for error in errors))
|
|
|
|
def test_blank_required_text_fails_in_builtin_and_schema_modes(self) -> None:
|
|
data = example_data()
|
|
data["entries"][0]["title"] = " \t "
|
|
|
|
builtin_errors, _ = validate_knowledge.validate_all(
|
|
data, SCHEMA_PATH, use_schema=False
|
|
)
|
|
with mock.patch.object(
|
|
validate_knowledge, "validate_with_schema", return_value=[]
|
|
):
|
|
schema_errors, _ = validate_knowledge.validate_all(
|
|
data, SCHEMA_PATH, use_schema=True
|
|
)
|
|
|
|
self.assertTrue(any("title: 必须是非空字符串" in e for e in builtin_errors))
|
|
self.assertTrue(any("title: 必须是非空字符串" in e for e in schema_errors))
|
|
schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8"))
|
|
self.assertEqual(
|
|
schema["definitions"]["entry"]["properties"]["title"]["pattern"],
|
|
"\\S",
|
|
)
|
|
|
|
def test_temporary_entry_requires_removal_condition_and_review_date(self) -> None:
|
|
data = example_data()
|
|
entry = data["entries"][0]
|
|
entry["temporary"] = True
|
|
entry["removalCondition"] = None
|
|
entry["reviewAfter"] = None
|
|
errors = validate_knowledge.validate_semantics(data)
|
|
self.assertTrue(any("removalCondition" in error for error in errors))
|
|
self.assertTrue(any("reviewAfter" in error for error in errors))
|
|
|
|
def test_overdue_active_entry_requires_revalidation(self) -> None:
|
|
data = example_data()
|
|
data["entries"][0]["reviewAfter"] = "2020-01-01T00:00:00+00:00"
|
|
|
|
errors = validate_knowledge.validate_semantics(data)
|
|
|
|
self.assertTrue(any("必须重新验证或标记 stale" in error for error in errors))
|
|
|
|
def test_global_active_entry_requires_decision_owner_approval(self) -> None:
|
|
data = example_data()
|
|
data["entries"][0]["scope"] = {
|
|
"all": True,
|
|
"components": [],
|
|
"paths": [],
|
|
"dependencies": [],
|
|
"versions": [],
|
|
"tags": [],
|
|
"symbols": [],
|
|
"errorSignatures": [],
|
|
}
|
|
|
|
errors = validate_knowledge.validate_semantics(data)
|
|
|
|
self.assertTrue(any("Decision Owner approval" in error for error in errors))
|
|
|
|
def test_active_conflicts_and_duplicate_active_revision_are_rejected(self) -> None:
|
|
data = example_data()
|
|
conflicting = copy.deepcopy(data["entries"][0])
|
|
conflicting.update(
|
|
{
|
|
"id": "K-002",
|
|
"subject": "service-process-alignment",
|
|
"conflictsWith": ["K-001@1"],
|
|
}
|
|
)
|
|
data["entries"].append(conflicting)
|
|
errors = validate_knowledge.validate_semantics(data)
|
|
self.assertTrue(any("active 条目不能与 active K-001@1 冲突" in e for e in errors))
|
|
|
|
conflicting["id"] = "K-001"
|
|
conflicting["revision"] = 2
|
|
conflicting["conflictsWith"] = []
|
|
errors = validate_knowledge.validate_semantics(data)
|
|
self.assertTrue(any("多个 active revision" in error for error in errors))
|
|
|
|
def test_verification_ref_must_resolve_to_contained_registry_target(self) -> None:
|
|
data = example_data()
|
|
data["entries"][0]["verification"]["ref"] = "missing-check"
|
|
errors = validate_knowledge.validate_semantics(data)
|
|
self.assertTrue(any("未在 verificationRegistry 注册" in e for e in errors))
|
|
|
|
data = example_data()
|
|
data["verificationRegistry"]["service-worktree-alignment"]["path"] = "../run.sh"
|
|
errors = validate_knowledge.validate_semantics(data)
|
|
self.assertTrue(any("必须是仓库内相对路径" in error for error in errors))
|
|
|
|
def test_verification_registry_rejects_any_symlink_component(self) -> None:
|
|
data = example_data()
|
|
data["verificationRegistry"]["service-worktree-alignment"]["path"] = (
|
|
"checks/verify.py"
|
|
)
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
base = Path(temp_dir)
|
|
project = base / "project"
|
|
outside = base / "outside"
|
|
project.mkdir()
|
|
outside.mkdir()
|
|
(project / "checks").symlink_to(outside, target_is_directory=True)
|
|
|
|
errors = validate_knowledge.validate_semantics(
|
|
data, project_root=project
|
|
)
|
|
|
|
self.assertTrue(any("symlink" in error for error in errors))
|
|
|
|
data = example_data()
|
|
data["verificationRegistry"]["service-worktree-alignment"]["path"] = (
|
|
"checks/verify.py"
|
|
)
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
project = Path(temp_dir) / "project"
|
|
real_checks = project / "real-checks"
|
|
real_checks.mkdir(parents=True)
|
|
target = real_checks / "verify.py"
|
|
target.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
target.chmod(target.stat().st_mode | 0o111)
|
|
(project / "checks").symlink_to(
|
|
real_checks,
|
|
target_is_directory=True,
|
|
)
|
|
|
|
errors = validate_knowledge.validate_semantics(
|
|
data,
|
|
project_root=project,
|
|
)
|
|
|
|
self.assertTrue(any("symlink" in error for error in errors))
|
|
|
|
data = example_data()
|
|
data["verificationRegistry"]["service-worktree-alignment"]["path"] = (
|
|
"tests/check.sh;touch"
|
|
)
|
|
errors = validate_knowledge.validate_semantics(data)
|
|
self.assertTrue(any("必须是仓库内相对路径" in error for error in errors))
|
|
|
|
def test_verification_registry_requires_regular_executable_target(self) -> None:
|
|
data = example_data()
|
|
data["verificationRegistry"]["service-worktree-alignment"]["path"] = (
|
|
"checks/verify"
|
|
)
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
project = Path(temp_dir) / "project"
|
|
project.mkdir()
|
|
|
|
missing = validate_knowledge.validate_semantics(
|
|
data,
|
|
project_root=project,
|
|
)
|
|
self.assertTrue(any("不存在" in error for error in missing))
|
|
|
|
target = project / "checks" / "verify"
|
|
target.parent.mkdir()
|
|
target.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
not_executable = validate_knowledge.validate_semantics(
|
|
data,
|
|
project_root=project,
|
|
)
|
|
self.assertTrue(
|
|
any("不可执行" in error for error in not_executable)
|
|
)
|
|
|
|
target.chmod(target.stat().st_mode | 0o111)
|
|
valid = validate_knowledge.validate_semantics(
|
|
data,
|
|
project_root=project,
|
|
)
|
|
self.assertFalse(
|
|
any("verificationRegistry" in error for error in valid),
|
|
valid,
|
|
)
|
|
|
|
def test_free_command_fields_are_rejected_and_never_executed(self) -> None:
|
|
data = example_data()
|
|
data["entries"][0]["command"] = "echo unsafe"
|
|
errors = validate_knowledge.validate_semantics(data)
|
|
self.assertTrue(any("不保存或执行自由命令" in error for error in errors))
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
marker = Path(temp_dir) / "should-not-exist"
|
|
safe = example_data()
|
|
safe["entries"][0]["directive"] = f"touch {marker}"
|
|
selected = select_knowledge.select_entries(
|
|
safe,
|
|
{
|
|
"components": ["web"],
|
|
"paths": ["web/app.py"],
|
|
"dependencies": [],
|
|
"versions": [],
|
|
"tags": ["long-running-service"],
|
|
"symbols": [],
|
|
"errorSignatures": [],
|
|
},
|
|
)
|
|
self.assertEqual(len(selected), 1)
|
|
self.assertFalse(marker.exists())
|
|
|
|
def test_common_secret_material_is_rejected(self) -> None:
|
|
data = example_data()
|
|
data["entries"][0]["rationale"] = (
|
|
"debug token=sk-abcdefghijklmnopqrstuvwxyz123456"
|
|
)
|
|
|
|
errors = validate_knowledge.validate_semantics(data)
|
|
|
|
self.assertTrue(any("OpenAI-style token" in error for error in errors))
|
|
self.assertTrue(any("脱敏摘要" in error for error in errors))
|
|
|
|
def test_selection_is_active_deterministic_and_limited(self) -> None:
|
|
data = example_data()
|
|
stale = copy.deepcopy(data["entries"][0])
|
|
stale.update(
|
|
{
|
|
"id": "K-002",
|
|
"status": "stale",
|
|
"statusReason": "相关服务已移除",
|
|
}
|
|
)
|
|
global_entry = copy.deepcopy(data["entries"][0])
|
|
global_entry.update(
|
|
{
|
|
"id": "K-003",
|
|
"subject": "global-release-check",
|
|
"scope": {
|
|
"all": True,
|
|
"components": [],
|
|
"paths": [],
|
|
"dependencies": [],
|
|
"versions": [],
|
|
"tags": [],
|
|
"symbols": [],
|
|
"errorSignatures": [],
|
|
},
|
|
}
|
|
)
|
|
data["entries"].extend([stale, global_entry])
|
|
context = {
|
|
"components": ["web"],
|
|
"paths": ["web/app.py"],
|
|
"dependencies": [],
|
|
"versions": [],
|
|
"tags": ["long-running-service"],
|
|
"symbols": [],
|
|
"errorSignatures": [],
|
|
}
|
|
selected = select_knowledge.select_entries(data, context, limit=1)
|
|
self.assertEqual(
|
|
[validate_knowledge.stable_ref(entry) for entry in selected], ["K-003@1"]
|
|
)
|
|
self.assertFalse(
|
|
select_knowledge.scope_matches(
|
|
data["entries"][0]["scope"],
|
|
{**context, "tags": []},
|
|
)
|
|
)
|
|
|
|
def test_global_entries_are_never_silently_dropped_by_limit(self) -> None:
|
|
data = example_data()
|
|
for number in range(2, 13):
|
|
scoped = copy.deepcopy(data["entries"][0])
|
|
scoped.update(
|
|
{
|
|
"id": f"K-{number:03d}",
|
|
"subject": f"scoped-check-{number}",
|
|
}
|
|
)
|
|
data["entries"].append(scoped)
|
|
global_entry = copy.deepcopy(data["entries"][0])
|
|
global_entry.update(
|
|
{
|
|
"id": "K-999",
|
|
"subject": "global-release-check",
|
|
"scope": {
|
|
"all": True,
|
|
"components": [],
|
|
"paths": [],
|
|
"dependencies": [],
|
|
"versions": [],
|
|
"tags": [],
|
|
"symbols": [],
|
|
"errorSignatures": [],
|
|
},
|
|
}
|
|
)
|
|
data["entries"].append(global_entry)
|
|
context = {
|
|
"components": ["web"],
|
|
"paths": ["web/app.py"],
|
|
"dependencies": [],
|
|
"versions": [],
|
|
"tags": ["long-running-service"],
|
|
"symbols": [],
|
|
"errorSignatures": [],
|
|
}
|
|
|
|
selected = select_knowledge.select_entries(data, context, limit=10)
|
|
|
|
refs = [validate_knowledge.stable_ref(entry) for entry in selected]
|
|
self.assertEqual(len(refs), 10)
|
|
self.assertIn("K-999@1", refs)
|
|
|
|
second_global = copy.deepcopy(global_entry)
|
|
second_global.update({"id": "K-998", "subject": "global-security-check"})
|
|
data["entries"].append(second_global)
|
|
with self.assertRaisesRegex(ValueError, "不能静默丢弃"):
|
|
select_knowledge.select_entries(data, context, limit=1)
|
|
|
|
def test_path_glob_does_not_let_single_star_cross_directory(self) -> None:
|
|
self.assertTrue(select_knowledge._path_glob_matches("web/app.py", "web/*"))
|
|
self.assertFalse(
|
|
select_knowledge._path_glob_matches("web/pages/app.py", "web/*")
|
|
)
|
|
self.assertTrue(
|
|
select_knowledge._path_glob_matches("web/pages/app.py", "web/**")
|
|
)
|
|
|
|
def test_limit_prefers_narrow_scope_without_rewarding_or_patterns(self) -> None:
|
|
data = example_data()
|
|
narrow = copy.deepcopy(data["entries"][0])
|
|
narrow.update({"id": "K-002", "subject": "narrow-service-check"})
|
|
narrow["scope"]["paths"] = ["web/special/**"]
|
|
broad_or = copy.deepcopy(data["entries"][0])
|
|
broad_or.update({"id": "K-003", "subject": "broad-or-service-check"})
|
|
broad_or["scope"]["paths"] = ["web/special/**", "api/**"]
|
|
data["entries"].extend([narrow, broad_or])
|
|
context = {
|
|
"components": ["web"],
|
|
"paths": ["web/special/pages/app.py"],
|
|
"dependencies": [],
|
|
"versions": [],
|
|
"tags": ["long-running-service"],
|
|
"symbols": [],
|
|
"errorSignatures": [],
|
|
}
|
|
|
|
selected = select_knowledge.select_entries(data, context, limit=1)
|
|
|
|
self.assertEqual(validate_knowledge.stable_ref(selected[0]), "K-002@1")
|
|
|
|
def test_task_cross_validation_requires_active_exact_ref_and_passed_check(self) -> None:
|
|
data = example_data()
|
|
task = {
|
|
"id": "BUG-002",
|
|
"status": "verified",
|
|
"knowledgeRefs": ["K-001@1"],
|
|
"knowledgeApplied": [
|
|
{
|
|
"ref": "K-001@1",
|
|
"result": "applied",
|
|
"evidence": "developer report",
|
|
}
|
|
],
|
|
"knowledgeChecks": [
|
|
{
|
|
"ref": "K-001@1",
|
|
"result": "passed",
|
|
"evidence": "test report",
|
|
}
|
|
],
|
|
"dispatch": {
|
|
"rounds": [
|
|
{
|
|
"round": 1,
|
|
"attemptId": "BUG-002-A1",
|
|
"result": "failed",
|
|
},
|
|
{
|
|
"round": 2,
|
|
"attemptId": "BUG-002-A2",
|
|
"result": "passed",
|
|
}
|
|
]
|
|
},
|
|
}
|
|
tasks_data = {
|
|
"project": {
|
|
"name": "notes-web",
|
|
"knowledgeFile": "docs/ack/knowledge.yaml",
|
|
},
|
|
"tasks": [task],
|
|
}
|
|
self.assertEqual(
|
|
validate_knowledge.validate_task_references(data, tasks_data), []
|
|
)
|
|
|
|
task["knowledgeChecks"][0]["result"] = "not_applicable"
|
|
errors = validate_knowledge.validate_task_references(data, tasks_data)
|
|
self.assertTrue(any("缺少 passed knowledgeCheck" in error for error in errors))
|
|
|
|
data["entries"][0]["kind"] = "pitfall"
|
|
errors = validate_knowledge.validate_task_references(data, tasks_data)
|
|
self.assertTrue(any("缺少 passed knowledgeCheck" in error for error in errors))
|
|
|
|
task["knowledgeRefs"] = ["K-001@2"]
|
|
errors = validate_knowledge.validate_task_references(data, tasks_data)
|
|
self.assertTrue(any("找不到精确版本 K-001@2" in error for error in errors))
|
|
|
|
def test_task_cross_validation_rejects_another_project(self) -> None:
|
|
data = example_data()
|
|
tasks = {
|
|
"project": {
|
|
"name": "another-project",
|
|
"knowledgeFile": "docs/ack/knowledge.yaml",
|
|
},
|
|
"tasks": [],
|
|
}
|
|
|
|
errors = validate_knowledge.validate_task_references(data, tasks)
|
|
|
|
self.assertTrue(any("与知识库项目" in error for error in errors))
|
|
|
|
def test_cross_validation_requires_traceable_explicit_attempt_id(self) -> None:
|
|
data = example_data()
|
|
task = {
|
|
"id": "BUG-002",
|
|
"status": "open",
|
|
"dispatch": {
|
|
"rounds": [
|
|
{
|
|
"round": 1,
|
|
"attemptId": "BUG-002-A1",
|
|
"result": "failed",
|
|
},
|
|
{
|
|
"round": 2,
|
|
"attemptId": "orca-dispatch-91",
|
|
"result": "passed",
|
|
}
|
|
]
|
|
},
|
|
}
|
|
tasks_data = {
|
|
"project": {
|
|
"name": "notes-web",
|
|
"knowledgeFile": "docs/ack/knowledge.yaml",
|
|
},
|
|
"tasks": [task],
|
|
}
|
|
|
|
errors = validate_knowledge.validate_task_references(data, tasks_data)
|
|
|
|
self.assertTrue(any("必须精确等于 'BUG-002-A2'" in e for e in errors))
|
|
self.assertTrue(any("未命中任务 'BUG-002'" in e for e in errors))
|
|
|
|
task["dispatch"]["rounds"][1]["attemptId"] = "BUG-002-A2"
|
|
errors = validate_knowledge.validate_task_references(data, tasks_data)
|
|
self.assertFalse(any("provenance" in error for error in errors))
|
|
|
|
def test_cross_validation_rejects_non_contiguous_rounds(self) -> None:
|
|
data = example_data()
|
|
data["entries"][0]["provenance"]["attemptId"] = "BUG-002-A3"
|
|
tasks_data = {
|
|
"project": {
|
|
"name": "notes-web",
|
|
"knowledgeFile": "docs/ack/knowledge.yaml",
|
|
},
|
|
"tasks": [
|
|
{
|
|
"id": "BUG-002",
|
|
"status": "open",
|
|
"dispatch": {
|
|
"rounds": [
|
|
{
|
|
"round": 3,
|
|
"attemptId": "BUG-002-A3",
|
|
"result": "passed",
|
|
}
|
|
]
|
|
},
|
|
}
|
|
],
|
|
}
|
|
|
|
errors = validate_knowledge.validate_task_references(data, tasks_data)
|
|
|
|
self.assertTrue(any("round 必须按 1..N 连续" in error for error in errors))
|
|
|
|
def test_cross_validation_binds_configured_knowledge_file(self) -> None:
|
|
data = example_data()
|
|
task = {
|
|
"id": "BUG-002",
|
|
"status": "open",
|
|
"dispatch": {
|
|
"rounds": [
|
|
{
|
|
"round": 1,
|
|
"attemptId": "BUG-002-A1",
|
|
"result": "failed",
|
|
},
|
|
{
|
|
"round": 2,
|
|
"attemptId": "BUG-002-A2",
|
|
"result": "passed",
|
|
}
|
|
]
|
|
},
|
|
}
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
project = Path(temp_dir) / "project"
|
|
ack_dir = project / "docs" / "ack"
|
|
ack_dir.mkdir(parents=True)
|
|
knowledge_path = ack_dir / "knowledge.yaml"
|
|
tasks_path = ack_dir / "tasks.yaml"
|
|
knowledge_path.write_text(
|
|
yaml.safe_dump(data, allow_unicode=True), encoding="utf-8"
|
|
)
|
|
tasks_data = {
|
|
"project": {
|
|
"name": "notes-web",
|
|
"repoPath": str(project),
|
|
"knowledgeFile": "docs/ack/other.yaml",
|
|
},
|
|
"tasks": [task],
|
|
}
|
|
|
|
errors = validate_knowledge.validate_task_references(
|
|
data,
|
|
tasks_data,
|
|
knowledge_path=knowledge_path,
|
|
tasks_path=tasks_path,
|
|
)
|
|
self.assertTrue(any("与当前知识文件" in error for error in errors))
|
|
|
|
del tasks_data["project"]["knowledgeFile"]
|
|
errors = validate_knowledge.validate_task_references(data, tasks_data)
|
|
self.assertTrue(any("project.knowledgeFile 必填" in e for e in errors))
|
|
|
|
tasks_data["project"]["knowledgeFile"] = "docs/ack/knowledge.yaml"
|
|
staging_root = Path(temp_dir) / "staging"
|
|
staged = staging_root / "docs" / "ack" / "knowledge.yaml"
|
|
staged.parent.mkdir(parents=True)
|
|
staged.write_text(
|
|
yaml.safe_dump(data, allow_unicode=True), encoding="utf-8"
|
|
)
|
|
errors = validate_knowledge.validate_task_references(
|
|
data,
|
|
tasks_data,
|
|
knowledge_path=staged,
|
|
tasks_path=tasks_path,
|
|
project_root=staging_root,
|
|
project_root_is_explicit=True,
|
|
)
|
|
self.assertFalse(any("knowledgeFile" in error for error in errors))
|
|
|
|
def test_cli_fails_closed_for_missing_repo_root_and_wrong_binding(self) -> None:
|
|
data = example_data()
|
|
task = {
|
|
"id": "BUG-002",
|
|
"title": "source task",
|
|
"status": "open",
|
|
"dispatch": {
|
|
"rounds": [
|
|
{
|
|
"round": 1,
|
|
"attemptId": "BUG-002-A1",
|
|
"result": "failed",
|
|
},
|
|
{
|
|
"round": 2,
|
|
"attemptId": "BUG-002-A2",
|
|
"result": "passed",
|
|
}
|
|
]
|
|
},
|
|
}
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
base = Path(temp_dir)
|
|
project = base / "project"
|
|
ack_dir = project / "docs" / "ack"
|
|
outside = base / "outside"
|
|
ack_dir.mkdir(parents=True)
|
|
outside.mkdir()
|
|
(project / "checks").symlink_to(outside, target_is_directory=True)
|
|
data["verificationRegistry"]["service-worktree-alignment"]["path"] = (
|
|
"checks/verify.py"
|
|
)
|
|
knowledge_path = ack_dir / "knowledge.yaml"
|
|
knowledge_path.write_text(
|
|
yaml.safe_dump(data, allow_unicode=True), encoding="utf-8"
|
|
)
|
|
tasks_data = {
|
|
"version": 1,
|
|
"project": {
|
|
"name": "notes-web",
|
|
"repoPath": str(project / "missing"),
|
|
"knowledgeFile": "docs/ack/not-the-current-file.yaml",
|
|
},
|
|
"tasks": [task],
|
|
}
|
|
tasks_path = ack_dir / "tasks.yaml"
|
|
tasks_path.write_text(
|
|
yaml.safe_dump(tasks_data, allow_unicode=True), encoding="utf-8"
|
|
)
|
|
|
|
command = [
|
|
sys.executable,
|
|
str(SCRIPTS_DIR / "validate_knowledge.py"),
|
|
str(knowledge_path),
|
|
"--tasks",
|
|
str(tasks_path),
|
|
]
|
|
inferred = subprocess.run(
|
|
command,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
explicit = subprocess.run(
|
|
[*command, "--project-root", str(project)],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
self.assertEqual(inferred.returncode, 1)
|
|
self.assertIn("project.repoPath", inferred.stderr)
|
|
self.assertIn("与当前知识文件", inferred.stderr)
|
|
self.assertIn("symlink", inferred.stderr)
|
|
self.assertEqual(explicit.returncode, 1)
|
|
self.assertIn("与当前知识文件", explicit.stderr)
|
|
self.assertIn("symlink", explicit.stderr)
|
|
|
|
def test_cli_requires_root_for_nonempty_registry_outside_project_layout(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
knowledge_path = Path(temp_dir) / "knowledge.yaml"
|
|
knowledge_path.write_text(
|
|
EXAMPLE_PATH.read_text(encoding="utf-8"), encoding="utf-8"
|
|
)
|
|
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(SCRIPTS_DIR / "validate_knowledge.py"),
|
|
str(knowledge_path),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 1)
|
|
self.assertIn("无法确定项目根目录", result.stderr)
|
|
self.assertIn("--project-root", result.stderr)
|
|
|
|
def test_terminal_task_keeps_historical_ref_after_entry_becomes_stale(self) -> None:
|
|
data = example_data()
|
|
data["entries"][0]["status"] = "stale"
|
|
data["entries"][0]["statusReason"] = "service architecture changed"
|
|
task = {
|
|
"id": "BUG-002",
|
|
"status": "verified",
|
|
"knowledgeRefs": ["K-001@1"],
|
|
"knowledgeApplied": [],
|
|
"knowledgeChecks": [
|
|
{
|
|
"ref": "K-001@1",
|
|
"result": "passed",
|
|
"evidence": "historical test evidence",
|
|
}
|
|
],
|
|
"dispatch": {
|
|
"rounds": [
|
|
{
|
|
"round": 1,
|
|
"attemptId": "BUG-002-A1",
|
|
"result": "failed",
|
|
},
|
|
{
|
|
"round": 2,
|
|
"attemptId": "BUG-002-A2",
|
|
"result": "passed",
|
|
}
|
|
]
|
|
},
|
|
}
|
|
tasks_data = {
|
|
"project": {
|
|
"name": "notes-web",
|
|
"knowledgeFile": "docs/ack/knowledge.yaml",
|
|
},
|
|
"tasks": [task],
|
|
}
|
|
|
|
terminal_errors = validate_knowledge.validate_task_references(
|
|
data, tasks_data
|
|
)
|
|
self.assertFalse(any("必须 active" in error for error in terminal_errors))
|
|
|
|
task["status"] = "open"
|
|
active_errors = validate_knowledge.validate_task_references(
|
|
data, tasks_data
|
|
)
|
|
self.assertTrue(any("必须 active" in error for error in active_errors))
|
|
|
|
def test_cli_validates_tasks_and_selector_outputs_stable_refs(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
project = Path(temp_dir) / "project"
|
|
ack_dir = project / "docs" / "ack"
|
|
ack_dir.mkdir(parents=True)
|
|
knowledge_path = ack_dir / "knowledge.yaml"
|
|
knowledge_path.write_text(
|
|
EXAMPLE_PATH.read_text(encoding="utf-8"), encoding="utf-8"
|
|
)
|
|
target = project / "tests" / "ack" / "check_service_worktree.py"
|
|
target.parent.mkdir(parents=True)
|
|
target.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
target.chmod(target.stat().st_mode | 0o111)
|
|
tasks = {
|
|
"project": {
|
|
"name": "notes-web",
|
|
"repoPath": str(project),
|
|
"knowledgeFile": "docs/ack/knowledge.yaml",
|
|
},
|
|
"tasks": [
|
|
{
|
|
"id": "BUG-002",
|
|
"status": "verified",
|
|
"knowledgeRefs": ["K-001@1"],
|
|
"knowledgeApplied": [],
|
|
"knowledgeChecks": [
|
|
{
|
|
"ref": "K-001@1",
|
|
"result": "passed",
|
|
"evidence": "independent retest",
|
|
}
|
|
],
|
|
"dispatch": {
|
|
"rounds": [
|
|
{
|
|
"round": 1,
|
|
"attemptId": "BUG-002-A1",
|
|
"result": "failed",
|
|
},
|
|
{
|
|
"round": 2,
|
|
"attemptId": "BUG-002-A2",
|
|
"result": "passed",
|
|
}
|
|
]
|
|
},
|
|
}
|
|
],
|
|
}
|
|
tasks_path = ack_dir / "tasks.yaml"
|
|
tasks_path.write_text(
|
|
yaml.safe_dump(tasks, allow_unicode=True), encoding="utf-8"
|
|
)
|
|
validated = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(SCRIPTS_DIR / "validate_knowledge.py"),
|
|
str(knowledge_path),
|
|
"--tasks",
|
|
str(tasks_path),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
self.assertEqual(validated.returncode, 0, validated.stderr)
|
|
|
|
selected = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(SCRIPTS_DIR / "select_knowledge.py"),
|
|
str(knowledge_path),
|
|
"--component",
|
|
"web",
|
|
"--path",
|
|
"web/app.py",
|
|
"--tag",
|
|
"long-running-service",
|
|
"--limit",
|
|
"1",
|
|
"--format",
|
|
"refs",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
self.assertEqual(selected.returncode, 0, selected.stderr)
|
|
self.assertEqual(selected.stdout.strip(), "K-001@1")
|
|
|
|
selected_json = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(SCRIPTS_DIR / "select_knowledge.py"),
|
|
str(knowledge_path),
|
|
"--component",
|
|
"web",
|
|
"--path",
|
|
"web/app.py",
|
|
"--tag",
|
|
"long-running-service",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
self.assertEqual(selected_json.returncode, 0, selected_json.stderr)
|
|
payload = json.loads(selected_json.stdout)
|
|
self.assertEqual(
|
|
payload["entries"][0]["verificationTarget"]["path"],
|
|
"tests/ack/check_service_worktree.py",
|
|
)
|
|
|
|
def test_paired_examples_validate_when_installed_in_project_layout(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
project = Path(temp_dir) / "notes-web"
|
|
ack_dir = project / "docs" / "ack"
|
|
ack_dir.mkdir(parents=True)
|
|
knowledge_path = ack_dir / "knowledge.yaml"
|
|
tasks_path = ack_dir / "tasks.yaml"
|
|
knowledge_path.write_text(
|
|
EXAMPLE_PATH.read_text(encoding="utf-8"),
|
|
encoding="utf-8",
|
|
)
|
|
tasks = yaml.safe_load(
|
|
TASKS_EXAMPLE_PATH.read_text(encoding="utf-8")
|
|
)
|
|
tasks["project"]["repoPath"] = str(project)
|
|
tasks["project"]["devWorktree"] = str(project)
|
|
tasks_path.write_text(
|
|
yaml.safe_dump(tasks, allow_unicode=True, sort_keys=False),
|
|
encoding="utf-8",
|
|
)
|
|
target = project / "tests" / "ack" / "check_service_worktree.py"
|
|
target.parent.mkdir(parents=True)
|
|
target.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
target.chmod(target.stat().st_mode | 0o111)
|
|
|
|
tasks_result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(SCRIPTS_DIR / "validate_tasks.py"),
|
|
str(tasks_path),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
knowledge_result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(SCRIPTS_DIR / "validate_knowledge.py"),
|
|
str(knowledge_path),
|
|
"--tasks",
|
|
str(tasks_path),
|
|
"--project-root",
|
|
str(project),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
self.assertEqual(tasks_result.returncode, 0, tasks_result.stderr)
|
|
self.assertEqual(
|
|
knowledge_result.returncode,
|
|
0,
|
|
knowledge_result.stderr,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|