generated from coulomb/repo-seed
137 lines
4.6 KiB
Python
137 lines
4.6 KiB
Python
from repo_registry.acceptance import AgenticReviewDecision
|
|
from repo_registry.core.service import RegistryService
|
|
from repo_registry.repo_ingestion.git import GitIngestionService
|
|
from repo_registry.storage.sqlite import RegistryStore
|
|
|
|
|
|
class BoundaryApprovingReviewer:
|
|
reviewer_id = "boundary-agent"
|
|
policy_version = "agentic-review-policy/test"
|
|
|
|
def review(self, request):
|
|
return [
|
|
AgenticReviewDecision(
|
|
action="approve",
|
|
target_type="candidate_graph",
|
|
target_id=request.candidate_graph.analysis_run.id,
|
|
rationale="README and source refs support the generated API capability.",
|
|
criterion_ids=["RREG-QC-004"],
|
|
evidence_refs=["README.md", "app.py"],
|
|
)
|
|
]
|
|
|
|
|
|
def make_service(tmp_path, *, reviewer=None):
|
|
store = RegistryStore(tmp_path / "registry.sqlite3")
|
|
store.initialize()
|
|
return RegistryService(
|
|
store,
|
|
ingestion=GitIngestionService(tmp_path / "checkouts"),
|
|
agentic_reviewer=reviewer,
|
|
)
|
|
|
|
|
|
def write_api_repo(tmp_path):
|
|
source = tmp_path / "api-repo"
|
|
source.mkdir()
|
|
(source / "README.md").write_text("# API Repo\nReports health.\n", encoding="utf-8")
|
|
(source / "app.py").write_text('@app.get("/health")\ndef health():\n return {}\n', encoding="utf-8")
|
|
return source
|
|
|
|
|
|
def write_provider_repo(tmp_path):
|
|
source = tmp_path / "provider-repo"
|
|
source.mkdir()
|
|
(source / "README.md").write_text("# Provider Repo\n", encoding="utf-8")
|
|
(source / "providers.py").write_text(
|
|
"provider_registry = {'openrouter': OpenRouterAdapter}\n",
|
|
encoding="utf-8",
|
|
)
|
|
return source
|
|
|
|
|
|
def test_deterministic_analysis_leaves_candidates_pending(tmp_path):
|
|
service = make_service(tmp_path)
|
|
repository = service.register_repository(
|
|
name="API Repo",
|
|
url=str(write_api_repo(tmp_path)),
|
|
)
|
|
|
|
summary = service.analyze_repository(repository.id, use_llm_assistance=False)
|
|
|
|
graph = service.candidate_graph(repository.id, summary.analysis_run.id)
|
|
assert service.ability_map(repository.id).abilities == []
|
|
assert {
|
|
capability.status
|
|
for ability in graph.abilities
|
|
for capability in ability.capabilities
|
|
} == {"candidate"}
|
|
|
|
|
|
def test_deterministic_gates_flag_provider_regression_without_approval(tmp_path):
|
|
service = make_service(tmp_path)
|
|
repository = service.register_repository(
|
|
name="Provider Repo",
|
|
url=str(write_provider_repo(tmp_path)),
|
|
)
|
|
|
|
summary = service.analyze_repository(repository.id, use_llm_assistance=False)
|
|
|
|
graph = service.candidate_graph(repository.id, summary.analysis_run.id)
|
|
decisions = service.list_review_decisions(repository.id, summary.analysis_run.id)
|
|
assert service.ability_map(repository.id).abilities == []
|
|
assert graph.abilities[0].capabilities[0].status == "candidate"
|
|
assert any(
|
|
decision.action == "quality_gate_evaluation"
|
|
and "RREG-QC-002" in decision.criterion_ids
|
|
for decision in decisions
|
|
)
|
|
|
|
|
|
def test_agentic_review_is_only_automated_approval_path(tmp_path):
|
|
service = make_service(tmp_path, reviewer=BoundaryApprovingReviewer())
|
|
repository = service.register_repository(
|
|
name="Agent Approved Repo",
|
|
url=str(write_api_repo(tmp_path)),
|
|
)
|
|
|
|
summary = service.analyze_repository(
|
|
repository.id,
|
|
use_llm_assistance=False,
|
|
agentic_review=True,
|
|
)
|
|
|
|
decisions = service.list_review_decisions(repository.id, summary.analysis_run.id)
|
|
assert service.ability_map(repository.id).abilities
|
|
assert any(
|
|
decision.action == "agentic_approve_candidate_graph"
|
|
and decision.reviewer_type == "agent"
|
|
and decision.rationale
|
|
and decision.criteria_version == "repo-scoping-quality-criteria/v1"
|
|
and decision.evidence_refs == ["README.md", "app.py"]
|
|
for decision in decisions
|
|
)
|
|
|
|
|
|
def test_manual_approval_path_still_works(tmp_path):
|
|
service = make_service(tmp_path)
|
|
repository = service.register_repository(
|
|
name="Manual Review Repo",
|
|
url=str(write_api_repo(tmp_path)),
|
|
)
|
|
summary = service.analyze_repository(repository.id, use_llm_assistance=False)
|
|
|
|
service.approve_candidate_graph(
|
|
repository.id,
|
|
summary.analysis_run.id,
|
|
notes="Manual curator approval.",
|
|
)
|
|
|
|
decisions = service.list_review_decisions(repository.id, summary.analysis_run.id)
|
|
assert service.ability_map(repository.id).abilities
|
|
assert any(
|
|
decision.action == "approve_candidate_graph"
|
|
and decision.reviewer_type == "human"
|
|
for decision in decisions
|
|
)
|