generated from coulomb/repo-seed
2029 lines
77 KiB
Python
2029 lines
77 KiB
Python
from fastapi.testclient import TestClient
|
|
|
|
from repo_registry.web_api import app as app_module
|
|
from repo_registry.web_api.app import Settings, app, get_service, get_settings
|
|
|
|
|
|
def test_openapi_groups_agent_facing_endpoints():
|
|
client = TestClient(app)
|
|
|
|
response = client.get("/openapi.json")
|
|
|
|
assert response.status_code == 200
|
|
schema = response.json()
|
|
assert schema["info"]["title"] == "Repository Ability Registry"
|
|
assert schema["info"]["version"] == "0.1.0"
|
|
assert {tag["name"] for tag in schema["tags"]} >= {
|
|
"repositories",
|
|
"analysis",
|
|
"review",
|
|
"registry",
|
|
"search",
|
|
"discovery",
|
|
}
|
|
search_operation = schema["paths"]["/search"]["get"]
|
|
assert search_operation["tags"] == ["search"]
|
|
search_response = search_operation["responses"]["200"]["content"][
|
|
"application/json"
|
|
]["schema"]
|
|
assert search_response["items"]["$ref"].endswith("/SearchResultResponse")
|
|
assert {
|
|
parameter["name"]: parameter["description"]
|
|
for parameter in search_operation["parameters"]
|
|
}["q"].startswith("Natural-language")
|
|
ability_map_response = schema["paths"]["/repos/{repository_id}/ability-map"][
|
|
"get"
|
|
]["responses"]["200"]["content"]["application/json"]["schema"]
|
|
assert ability_map_response["$ref"].endswith("/RepositoryAbilityMapResponse")
|
|
components = schema["components"]["schemas"]
|
|
assert components["SearchResultResponse"]["examples"][0]["match_type"] == (
|
|
"capability"
|
|
)
|
|
assert components["RepositoryAbilityMapResponse"]["examples"][0]["abilities"][0][
|
|
"confidence_label"
|
|
] == "high"
|
|
assert components["CandidateGraphResponse"]["examples"][0]["abilities"][0][
|
|
"status"
|
|
] == "pending"
|
|
assert (
|
|
"/repos/{repository_id}/analysis-runs/{base_analysis_run_id}/diff/"
|
|
"{target_analysis_run_id}"
|
|
) in schema["paths"]
|
|
assert (
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/changes/approve"
|
|
) in schema["paths"]
|
|
assert components["ErrorResponse"]["examples"][0]["detail"]
|
|
assert (
|
|
schema["paths"]["/repos/{repository_id}"]["get"]["responses"]["404"][
|
|
"content"
|
|
]["application/json"]["schema"]["$ref"]
|
|
).endswith("/ErrorResponse")
|
|
assert (
|
|
schema["paths"][
|
|
"/repos/{repository_id}/analysis-runs/{base_analysis_run_id}/diff/"
|
|
"{target_analysis_run_id}"
|
|
]["get"]["responses"]["200"]["content"]["application/json"]["schema"]["$ref"]
|
|
).endswith("/AnalysisRunDiffResponse")
|
|
assert (
|
|
components["RepositoryComparisonResponse"]["examples"][0][
|
|
"unique_capabilities"
|
|
][0]["capability_name"]
|
|
== "Classify Incoming Email"
|
|
)
|
|
assert (
|
|
components["CapabilityGapResponse"]["examples"][0]["missing_capabilities"][0]
|
|
== "Route Email to Team"
|
|
)
|
|
|
|
|
|
def test_openapi_contract_snapshot_for_stable_agent_paths():
|
|
client = TestClient(app)
|
|
|
|
schema = client.get("/openapi.json").json()
|
|
|
|
def success_schema(operation):
|
|
responses = operation["responses"]
|
|
status_code = "201" if "201" in responses else "200"
|
|
content = responses.get(status_code, {}).get("content", {})
|
|
if "application/json" in content:
|
|
response_schema = content["application/json"]["schema"]
|
|
if "$ref" in response_schema:
|
|
return response_schema["$ref"].split("/")[-1]
|
|
if response_schema.get("type") == "array":
|
|
items = response_schema["items"]
|
|
if "$ref" in items:
|
|
return f"list[{items['$ref'].split('/')[-1]}]"
|
|
return "list"
|
|
return response_schema.get("type")
|
|
if "application/x-yaml" in content:
|
|
return "application/x-yaml"
|
|
return None
|
|
|
|
stable_contract = {
|
|
path: {
|
|
method: {
|
|
"tags": operation["tags"],
|
|
"success_schema": success_schema(operation),
|
|
}
|
|
for method, operation in sorted(methods.items())
|
|
}
|
|
for path, methods in sorted(schema["paths"].items())
|
|
if not path.startswith("/ui")
|
|
}
|
|
|
|
assert stable_contract == {
|
|
"/abilities": {
|
|
"get": {"tags": ["search"], "success_schema": "list[AbilitySummaryResponse]"}
|
|
},
|
|
"/capabilities": {
|
|
"get": {
|
|
"tags": ["search"],
|
|
"success_schema": "list[CapabilitySummaryResponse]",
|
|
}
|
|
},
|
|
"/capability-gaps": {
|
|
"post": {"tags": ["discovery"], "success_schema": "CapabilityGapResponse"}
|
|
},
|
|
"/health": {"get": {"tags": ["health"], "success_schema": "object"}},
|
|
"/repos": {
|
|
"get": {"tags": ["repositories"], "success_schema": "list[RepositoryResponse]"},
|
|
"post": {"tags": ["repositories"], "success_schema": "RepositoryResponse"},
|
|
},
|
|
"/repos/{repository_id}": {
|
|
"delete": {"tags": ["repositories"], "success_schema": None},
|
|
"get": {"tags": ["repositories"], "success_schema": "RepositoryResponse"},
|
|
"patch": {"tags": ["repositories"], "success_schema": "RepositoryResponse"},
|
|
},
|
|
"/repos/{repository_id}/abilities": {
|
|
"post": {"tags": ["registry"], "success_schema": "IdResponse"}
|
|
},
|
|
"/repos/{repository_id}/abilities/{ability_id}": {
|
|
"delete": {
|
|
"tags": ["registry"],
|
|
"success_schema": "RepositoryAbilityMapResponse",
|
|
},
|
|
"patch": {
|
|
"tags": ["registry"],
|
|
"success_schema": "RepositoryAbilityMapResponse",
|
|
},
|
|
},
|
|
"/repos/{repository_id}/ability-map": {
|
|
"get": {"tags": ["registry"], "success_schema": "RepositoryAbilityMapResponse"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs": {
|
|
"get": {"tags": ["analysis"], "success_schema": "list[AnalysisRunResponse]"},
|
|
"post": {"tags": ["analysis"], "success_schema": "ScanSummaryResponse"},
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}": {
|
|
"get": {"tags": ["analysis"], "success_schema": "AnalysisRunResponse"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-graph": {
|
|
"get": {"tags": ["review"], "success_schema": "CandidateGraphResponse"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-graph/approve": {
|
|
"post": {
|
|
"tags": ["review"],
|
|
"success_schema": "RepositoryAbilityMapResponse",
|
|
}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/changes/approve": {
|
|
"post": {
|
|
"tags": ["review"],
|
|
"success_schema": "RepositoryAbilityMapResponse",
|
|
}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/content-chunks": {
|
|
"get": {"tags": ["analysis"], "success_schema": "list[ContentChunkResponse]"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/review-decisions": {
|
|
"get": {"tags": ["review"], "success_schema": "list[ReviewDecisionResponse]"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{base_analysis_run_id}/diff/{target_analysis_run_id}": {
|
|
"get": {"tags": ["review"], "success_schema": "AnalysisRunDiffResponse"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-abilities/{candidate_ability_id}": {
|
|
"patch": {"tags": ["review"], "success_schema": "CandidateGraphResponse"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-abilities/{candidate_ability_id}/reject": {
|
|
"post": {"tags": ["review"], "success_schema": "CandidateGraphResponse"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-abilities/{source_ability_id}/merge": {
|
|
"post": {"tags": ["review"], "success_schema": "CandidateGraphResponse"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-capabilities/{candidate_capability_id}": {
|
|
"patch": {"tags": ["review"], "success_schema": "CandidateGraphResponse"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-capabilities/{candidate_capability_id}/reject": {
|
|
"post": {"tags": ["review"], "success_schema": "CandidateGraphResponse"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-capabilities/{candidate_capability_id}/relink": {
|
|
"post": {"tags": ["review"], "success_schema": "CandidateGraphResponse"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-capabilities/{source_capability_id}/merge": {
|
|
"post": {"tags": ["review"], "success_schema": "CandidateGraphResponse"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-evidence/{candidate_evidence_id}/reject": {
|
|
"post": {"tags": ["review"], "success_schema": "CandidateGraphResponse"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-evidence/{candidate_evidence_id}/relink": {
|
|
"post": {"tags": ["review"], "success_schema": "CandidateGraphResponse"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-evidence/{source_evidence_id}/merge": {
|
|
"post": {"tags": ["review"], "success_schema": "CandidateGraphResponse"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-features/{candidate_feature_id}/reject": {
|
|
"post": {"tags": ["review"], "success_schema": "CandidateGraphResponse"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-features/{candidate_feature_id}/relink": {
|
|
"post": {"tags": ["review"], "success_schema": "CandidateGraphResponse"}
|
|
},
|
|
"/repos/{repository_id}/analysis-runs/{analysis_run_id}/candidate-features/{source_feature_id}/merge": {
|
|
"post": {"tags": ["review"], "success_schema": "CandidateGraphResponse"}
|
|
},
|
|
"/repos/{repository_id}/capabilities": {
|
|
"post": {"tags": ["registry"], "success_schema": "IdResponse"}
|
|
},
|
|
"/repos/{repository_id}/capabilities/{capability_id}": {
|
|
"delete": {
|
|
"tags": ["registry"],
|
|
"success_schema": "RepositoryAbilityMapResponse",
|
|
},
|
|
"patch": {
|
|
"tags": ["registry"],
|
|
"success_schema": "RepositoryAbilityMapResponse",
|
|
},
|
|
},
|
|
"/repos/{repository_id}/content-chunks": {
|
|
"get": {"tags": ["analysis"], "success_schema": "list[ContentChunkResponse]"}
|
|
},
|
|
"/repos/{repository_id}/evidence": {
|
|
"post": {"tags": ["registry"], "success_schema": "IdResponse"}
|
|
},
|
|
"/repos/{repository_id}/evidence/{evidence_id}": {
|
|
"delete": {
|
|
"tags": ["registry"],
|
|
"success_schema": "RepositoryAbilityMapResponse",
|
|
},
|
|
"patch": {
|
|
"tags": ["registry"],
|
|
"success_schema": "RepositoryAbilityMapResponse",
|
|
},
|
|
},
|
|
"/repos/{repository_id}/export": {
|
|
"get": {"tags": ["discovery"], "success_schema": "application/x-yaml"}
|
|
},
|
|
"/repos/{repository_id}/expectation-gaps": {
|
|
"get": {"tags": ["review"], "success_schema": "list[ExpectationGapResponse]"},
|
|
"post": {"tags": ["review"], "success_schema": "ExpectationGapResponse"},
|
|
},
|
|
"/repos/{repository_id}/features": {
|
|
"post": {"tags": ["registry"], "success_schema": "IdResponse"}
|
|
},
|
|
"/repos/{repository_id}/features/{feature_id}": {
|
|
"delete": {
|
|
"tags": ["registry"],
|
|
"success_schema": "RepositoryAbilityMapResponse",
|
|
},
|
|
"patch": {
|
|
"tags": ["registry"],
|
|
"success_schema": "RepositoryAbilityMapResponse",
|
|
},
|
|
},
|
|
"/repos/{repository_id}/observed-facts": {
|
|
"get": {"tags": ["analysis"], "success_schema": "list[ObservedFactResponse]"}
|
|
},
|
|
"/repos/{repository_id}/review-decisions": {
|
|
"get": {"tags": ["review"], "success_schema": "list[ReviewDecisionResponse]"}
|
|
},
|
|
"/repository-comparisons": {
|
|
"get": {
|
|
"tags": ["discovery"],
|
|
"success_schema": "RepositoryComparisonResponse",
|
|
}
|
|
},
|
|
"/search": {
|
|
"get": {"tags": ["search"], "success_schema": "list[SearchResultResponse]"}
|
|
},
|
|
}
|
|
|
|
|
|
def test_docs_endpoint_is_available():
|
|
client = TestClient(app)
|
|
|
|
response = client.get("/docs")
|
|
|
|
assert response.status_code == 200
|
|
assert "Repository Ability Registry" in response.text
|
|
assert "openapi.json" in response.text
|
|
|
|
|
|
def test_health_reports_database_and_checkout_root(tmp_path):
|
|
def override_settings():
|
|
return Settings(
|
|
database_path=str(tmp_path / "health.sqlite3"),
|
|
checkout_root=str(tmp_path / "checkouts"),
|
|
)
|
|
|
|
app.dependency_overrides[get_settings] = override_settings
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.get("/health")
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["status"] == "ok"
|
|
assert body["database"]["reachable"] is True
|
|
assert body["database"]["error"] is None
|
|
assert body["database"]["path"].endswith("health.sqlite3")
|
|
assert body["checkout_root"]["path"].endswith("checkouts")
|
|
assert body["checkout_root"]["exists"] is False
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_api_manual_registry_loop(tmp_path):
|
|
def override_settings():
|
|
return Settings(
|
|
database_path=str(tmp_path / "api.sqlite3"),
|
|
checkout_root=str(tmp_path / "checkouts"),
|
|
)
|
|
|
|
app.dependency_overrides[get_settings] = override_settings
|
|
client = TestClient(app)
|
|
try:
|
|
repository_response = client.post(
|
|
"/repos",
|
|
json={
|
|
"name": "MailRouter",
|
|
"url": "https://example.com/mail-router.git",
|
|
"description": "Routes incoming customer email",
|
|
},
|
|
)
|
|
assert repository_response.status_code == 201
|
|
repository_id = repository_response.json()["id"]
|
|
|
|
update_response = client.patch(
|
|
f"/repos/{repository_id}",
|
|
json={
|
|
"name": "MailRouter Updated",
|
|
"description": "Updated mail routing summary.",
|
|
},
|
|
)
|
|
assert update_response.status_code == 200
|
|
assert update_response.json()["name"] == "MailRouter Updated"
|
|
|
|
ability_response = client.post(
|
|
f"/repos/{repository_id}/abilities",
|
|
json={
|
|
"name": "Business Email Routing",
|
|
"description": "Route inbound messages.",
|
|
},
|
|
)
|
|
assert ability_response.status_code == 201
|
|
ability_id = ability_response.json()["id"]
|
|
|
|
capability_response = client.post(
|
|
f"/repos/{repository_id}/capabilities",
|
|
json={
|
|
"ability_id": ability_id,
|
|
"name": "Classify Incoming Email",
|
|
"inputs": ["subject", "body"],
|
|
"outputs": ["intent"],
|
|
},
|
|
)
|
|
assert capability_response.status_code == 201
|
|
capability_id = capability_response.json()["id"]
|
|
|
|
feature_response = client.post(
|
|
f"/repos/{repository_id}/features",
|
|
json={
|
|
"capability_id": capability_id,
|
|
"name": "POST /api/classify-email",
|
|
"type": "REST endpoint",
|
|
"location": "src/routes/classify_email.py",
|
|
},
|
|
)
|
|
assert feature_response.status_code == 201
|
|
feature_id = feature_response.json()["id"]
|
|
|
|
evidence_response = client.post(
|
|
f"/repos/{repository_id}/evidence",
|
|
json={
|
|
"capability_id": capability_id,
|
|
"type": "unit_test",
|
|
"reference": "tests/test_email_classification.py",
|
|
},
|
|
)
|
|
assert evidence_response.status_code == 201
|
|
evidence_id = evidence_response.json()["id"]
|
|
|
|
ability_update_response = client.patch(
|
|
f"/repos/{repository_id}/abilities/{ability_id}",
|
|
json={"name": "Business Email Routing Updated"},
|
|
)
|
|
assert ability_update_response.status_code == 200
|
|
assert ability_update_response.json()["abilities"][0]["name"] == (
|
|
"Business Email Routing Updated"
|
|
)
|
|
|
|
capability_update_response = client.patch(
|
|
f"/repos/{repository_id}/capabilities/{capability_id}",
|
|
json={"name": "Classify Incoming Email Updated"},
|
|
)
|
|
assert capability_update_response.status_code == 200
|
|
assert capability_update_response.json()["abilities"][0]["capabilities"][0][
|
|
"name"
|
|
] == "Classify Incoming Email Updated"
|
|
|
|
feature_update_response = client.patch(
|
|
f"/repos/{repository_id}/features/{feature_id}",
|
|
json={"location": "src/routes/updated.py"},
|
|
)
|
|
assert feature_update_response.status_code == 200
|
|
evidence_update_response = client.patch(
|
|
f"/repos/{repository_id}/evidence/{evidence_id}",
|
|
json={"strength": "strong"},
|
|
)
|
|
assert evidence_update_response.status_code == 200
|
|
|
|
map_response = client.get(f"/repos/{repository_id}/ability-map")
|
|
assert map_response.status_code == 200
|
|
ability_map = map_response.json()
|
|
assert ability_map["repository"]["name"] == "MailRouter Updated"
|
|
assert ability_map["abilities"][0]["capabilities"][0]["name"] == (
|
|
"Classify Incoming Email Updated"
|
|
)
|
|
|
|
assert (
|
|
client.delete(f"/repos/{repository_id}/features/{feature_id}").status_code
|
|
== 200
|
|
)
|
|
assert (
|
|
client.delete(f"/repos/{repository_id}/evidence/{evidence_id}").status_code
|
|
== 200
|
|
)
|
|
|
|
search_response = client.get("/search", params={"q": "email"})
|
|
assert search_response.status_code == 200
|
|
assert search_response.json()
|
|
|
|
delete_response = client.delete(f"/repos/{repository_id}")
|
|
assert delete_response.status_code == 204
|
|
missing_response = client.get(f"/repos/{repository_id}")
|
|
assert missing_response.status_code == 404
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_api_compare_gap_and_export_use_cases(tmp_path):
|
|
def override_settings():
|
|
return Settings(
|
|
database_path=str(tmp_path / "discovery.sqlite3"),
|
|
checkout_root=str(tmp_path / "discovery-checkouts"),
|
|
)
|
|
|
|
app.dependency_overrides[get_settings] = override_settings
|
|
client = TestClient(app)
|
|
try:
|
|
first = client.post(
|
|
"/repos",
|
|
json={
|
|
"name": "MailRouter",
|
|
"url": "https://example.com/mail-router.git",
|
|
"description": "Routes customer email.",
|
|
},
|
|
).json()
|
|
second = client.post(
|
|
"/repos",
|
|
json={
|
|
"name": "SupportRouter",
|
|
"url": "https://example.com/support-router.git",
|
|
"description": "Routes support requests.",
|
|
},
|
|
).json()
|
|
|
|
first_ability = client.post(
|
|
f"/repos/{first['id']}/abilities",
|
|
json={
|
|
"name": "Business Email Routing",
|
|
"description": "Route inbound messages.",
|
|
"confidence": 0.92,
|
|
},
|
|
).json()["id"]
|
|
first_classify = client.post(
|
|
f"/repos/{first['id']}/capabilities",
|
|
json={
|
|
"ability_id": first_ability,
|
|
"name": "Classify Incoming Email",
|
|
"description": "Classify messages by intent.",
|
|
"confidence": 0.88,
|
|
},
|
|
).json()["id"]
|
|
client.post(
|
|
f"/repos/{first['id']}/evidence",
|
|
json={
|
|
"capability_id": first_classify,
|
|
"type": "unit_test",
|
|
"reference": "tests/test_classify.py",
|
|
"strength": "strong",
|
|
},
|
|
)
|
|
client.post(
|
|
f"/repos/{first['id']}/capabilities",
|
|
json={
|
|
"ability_id": first_ability,
|
|
"name": "Route Email to Team",
|
|
"description": "Route messages to owning teams.",
|
|
"confidence": 0.7,
|
|
},
|
|
)
|
|
|
|
second_ability = client.post(
|
|
f"/repos/{second['id']}/abilities",
|
|
json={
|
|
"name": "Business Email Routing",
|
|
"description": "Support routing workflows.",
|
|
"confidence": 0.8,
|
|
},
|
|
).json()["id"]
|
|
second_classify = client.post(
|
|
f"/repos/{second['id']}/capabilities",
|
|
json={
|
|
"ability_id": second_ability,
|
|
"name": "Classify Incoming Email",
|
|
"description": "Classify support requests.",
|
|
"confidence": 0.6,
|
|
},
|
|
).json()["id"]
|
|
client.post(
|
|
f"/repos/{second['id']}/evidence",
|
|
json={
|
|
"capability_id": second_classify,
|
|
"type": "documentation",
|
|
"reference": "README.md",
|
|
"strength": "medium",
|
|
},
|
|
)
|
|
client.post(
|
|
f"/repos/{second['id']}/capabilities",
|
|
json={
|
|
"ability_id": second_ability,
|
|
"name": "Archive Email",
|
|
"description": "Archive resolved messages.",
|
|
"confidence": 0.75,
|
|
},
|
|
)
|
|
|
|
comparison_response = client.get(
|
|
"/repository-comparisons",
|
|
params=[
|
|
("repository_ids", first["id"]),
|
|
("repository_ids", second["id"]),
|
|
],
|
|
)
|
|
assert comparison_response.status_code == 200
|
|
comparison = comparison_response.json()
|
|
assert {repo["name"] for repo in comparison["repositories"]} == {
|
|
"MailRouter",
|
|
"SupportRouter",
|
|
}
|
|
ability_entry = comparison["abilities"][0]
|
|
assert ability_entry["name"] == "Business Email Routing"
|
|
assert {repo["repository_name"] for repo in ability_entry["repositories"]} == {
|
|
"MailRouter",
|
|
"SupportRouter",
|
|
}
|
|
assert {
|
|
item["capability_name"] for item in comparison["unique_capabilities"]
|
|
} >= {"Route Email to Team", "Archive Email"}
|
|
|
|
gaps_response = client.post(
|
|
"/capability-gaps",
|
|
json={
|
|
"desired_ability": "Business Email Routing",
|
|
"desired_capabilities": [
|
|
"Classify Incoming Email",
|
|
"Route Email to Team",
|
|
"German Benchmark Evaluation",
|
|
],
|
|
"repository_ids": [first["id"], second["id"]],
|
|
},
|
|
)
|
|
assert gaps_response.status_code == 200
|
|
gaps = gaps_response.json()
|
|
assert gaps["missing_capabilities"] == ["German Benchmark Evaluation"]
|
|
assert gaps["matched_capabilities"][0]["capability"] == (
|
|
"Classify Incoming Email"
|
|
)
|
|
assert {
|
|
weak["capability"] for weak in gaps["weakly_evidenced_capabilities"]
|
|
} >= {"Classify Incoming Email", "Route Email to Team"}
|
|
assert gaps["duplicate_capabilities"][0]["capability"] == (
|
|
"classify incoming email"
|
|
)
|
|
|
|
export_response = client.get(f"/repos/{first['id']}/export")
|
|
assert export_response.status_code == 200
|
|
assert export_response.headers["content-type"].startswith("application/x-yaml")
|
|
assert 'name: "MailRouter"' in export_response.text
|
|
assert 'name: "Classify Incoming Email"' in export_response.text
|
|
assert 'reference: "tests/test_classify.py"' in export_response.text
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_api_registers_repository_from_url_metadata(tmp_path):
|
|
source = tmp_path / "metadata-api"
|
|
source.mkdir()
|
|
(source / "package.json").write_text(
|
|
'{"name":"metadata-api","description":"Imported through the API."}',
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def override_settings():
|
|
return Settings(
|
|
database_path=str(tmp_path / "metadata-api.sqlite3"),
|
|
checkout_root=str(tmp_path / "metadata-api-checkouts"),
|
|
)
|
|
|
|
app.dependency_overrides[get_settings] = override_settings
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.post("/repos", json={"url": str(source)})
|
|
|
|
assert response.status_code == 201
|
|
repository = response.json()
|
|
assert repository["name"] == "metadata-api"
|
|
assert repository["description"] == "Imported through the API."
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_api_service_settings_can_enable_llm_extractor(monkeypatch, tmp_path):
|
|
class Response:
|
|
content = '{"abilities": [{"name": "Configured LLM Ability"}]}'
|
|
|
|
class Adapter:
|
|
def execute_prompt(self, prompt, config):
|
|
return Response()
|
|
|
|
calls = []
|
|
|
|
def fake_create_adapter(provider, model=None):
|
|
calls.append((provider, model))
|
|
return Adapter()
|
|
|
|
monkeypatch.setattr(app_module, "create_llm_connect_adapter", fake_create_adapter)
|
|
service = get_service(
|
|
Settings(
|
|
database_path=str(tmp_path / "llm-settings.sqlite3"),
|
|
checkout_root=str(tmp_path / "checkouts"),
|
|
llm_provider="mock",
|
|
llm_model="demo-model",
|
|
)
|
|
)
|
|
|
|
assert calls == [("mock", "demo-model")]
|
|
assert service.llm_extractor is not None
|
|
|
|
|
|
def test_api_service_settings_can_enable_hashing_embedding_provider(tmp_path):
|
|
service = get_service(
|
|
Settings(
|
|
database_path=str(tmp_path / "embedding-settings.sqlite3"),
|
|
checkout_root=str(tmp_path / "checkouts"),
|
|
embedding_provider="hashing",
|
|
)
|
|
)
|
|
|
|
assert service.embedding_provider is not None
|
|
assert service.embedding_provider.name == "hashing-v1"
|
|
|
|
|
|
def test_settings_can_load_from_environment(monkeypatch):
|
|
monkeypatch.setenv("REPO_REGISTRY_DATABASE_PATH", "var/env.sqlite3")
|
|
monkeypatch.setenv("REPO_REGISTRY_CHECKOUT_ROOT", "var/env-checkouts")
|
|
monkeypatch.setenv("REPO_REGISTRY_LLM_PROVIDER", "mock")
|
|
monkeypatch.setenv("REPO_REGISTRY_LLM_MODEL", "demo-model")
|
|
monkeypatch.setenv("REPO_REGISTRY_EMBEDDING_PROVIDER", "hashing")
|
|
monkeypatch.setenv("REPO_REGISTRY_LOG_LEVEL", "DEBUG")
|
|
|
|
settings = Settings()
|
|
|
|
assert settings.database_path == "var/env.sqlite3"
|
|
assert settings.checkout_root == "var/env-checkouts"
|
|
assert settings.llm_provider == "mock"
|
|
assert settings.llm_model == "demo-model"
|
|
assert settings.embedding_provider == "hashing"
|
|
assert settings.log_level == "DEBUG"
|
|
|
|
|
|
def test_api_analysis_run_loop(tmp_path):
|
|
source = tmp_path / "repo"
|
|
source.mkdir()
|
|
(source / "README.md").write_text("# Searchable\n", encoding="utf-8")
|
|
(source / "package.json").write_text(
|
|
'{"dependencies":{"react":"latest","vite":"latest"}}',
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def override_settings():
|
|
return Settings(
|
|
database_path=str(tmp_path / "api-analysis.sqlite3"),
|
|
checkout_root=str(tmp_path / "api-checkouts"),
|
|
)
|
|
|
|
app.dependency_overrides[get_settings] = override_settings
|
|
client = TestClient(app)
|
|
try:
|
|
repository_response = client.post(
|
|
"/repos",
|
|
json={"name": "Frontend", "url": str(source)},
|
|
)
|
|
repository_id = repository_response.json()["id"]
|
|
|
|
run_response = client.post(f"/repos/{repository_id}/analysis-runs", json={})
|
|
assert run_response.status_code == 201
|
|
run = run_response.json()
|
|
assert run["analysis_run"]["status"] == "completed"
|
|
assert run["snapshot"]["file_count"] == 2
|
|
|
|
get_run_response = client.get(
|
|
f"/repos/{repository_id}/analysis-runs/{run['analysis_run']['id']}"
|
|
)
|
|
assert get_run_response.status_code == 200
|
|
assert get_run_response.json()["id"] == run["analysis_run"]["id"]
|
|
|
|
candidate_response = client.get(
|
|
f"/repos/{repository_id}/analysis-runs/"
|
|
f"{run['analysis_run']['id']}/candidate-graph"
|
|
)
|
|
assert candidate_response.status_code == 200
|
|
candidate_graph = candidate_response.json()
|
|
assert candidate_graph["abilities"][0]["name"] == "Support Frontend"
|
|
candidate_ability_id = candidate_graph["abilities"][0]["id"]
|
|
candidate_capability_id = candidate_graph["abilities"][0]["capabilities"][0]["id"]
|
|
|
|
reject_response = client.post(
|
|
f"/repos/{repository_id}/analysis-runs/"
|
|
f"{run['analysis_run']['id']}/candidate-abilities/"
|
|
f"{candidate_ability_id}/reject",
|
|
json={"notes": "Reject once to exercise review correction."},
|
|
)
|
|
assert reject_response.status_code == 200
|
|
assert reject_response.json()["abilities"][0]["status"] == "rejected"
|
|
decisions_response = client.get(f"/repos/{repository_id}/review-decisions")
|
|
assert decisions_response.status_code == 200
|
|
assert decisions_response.json()[0]["action"] == "reject_candidate_ability"
|
|
run_decisions_response = client.get(
|
|
f"/repos/{repository_id}/analysis-runs/"
|
|
f"{run['analysis_run']['id']}/review-decisions"
|
|
)
|
|
assert run_decisions_response.status_code == 200
|
|
assert run_decisions_response.json()[0]["notes"] == (
|
|
"Reject once to exercise review correction."
|
|
)
|
|
gap_response = client.post(
|
|
f"/repos/{repository_id}/expectation-gaps",
|
|
json={
|
|
"analysis_run_id": run["analysis_run"]["id"],
|
|
"expected_type": "capability",
|
|
"expected_name": "Use OpenRouter Models",
|
|
"source": "human",
|
|
"notes": "Expected provider capability was missing.",
|
|
},
|
|
)
|
|
assert gap_response.status_code == 201
|
|
assert gap_response.json()["expected_name"] == "Use OpenRouter Models"
|
|
gaps_response = client.get(
|
|
f"/repos/{repository_id}/expectation-gaps",
|
|
params={"analysis_run_id": run["analysis_run"]["id"]},
|
|
)
|
|
assert gaps_response.status_code == 200
|
|
assert gaps_response.json()[0]["source"] == "human"
|
|
|
|
run_response = client.post(f"/repos/{repository_id}/analysis-runs", json={})
|
|
assert run_response.status_code == 201
|
|
run = run_response.json()
|
|
candidate_response = client.get(
|
|
f"/repos/{repository_id}/analysis-runs/"
|
|
f"{run['analysis_run']['id']}/candidate-graph"
|
|
)
|
|
candidate_graph = candidate_response.json()
|
|
candidate_ability_id = candidate_graph["abilities"][0]["id"]
|
|
candidate_capability_id = candidate_graph["abilities"][0]["capabilities"][0]["id"]
|
|
|
|
ability_edit_response = client.patch(
|
|
f"/repos/{repository_id}/analysis-runs/"
|
|
f"{run['analysis_run']['id']}/candidate-abilities/"
|
|
f"{candidate_ability_id}",
|
|
json={
|
|
"name": "Frontend Delivery",
|
|
"description": "Serve a browser frontend.",
|
|
"confidence": 0.9,
|
|
"notes": "API edit test",
|
|
},
|
|
)
|
|
assert ability_edit_response.status_code == 200
|
|
assert ability_edit_response.json()["abilities"][0]["name"] == (
|
|
"Frontend Delivery"
|
|
)
|
|
|
|
capability_edit_response = client.patch(
|
|
f"/repos/{repository_id}/analysis-runs/"
|
|
f"{run['analysis_run']['id']}/candidate-capabilities/"
|
|
f"{candidate_capability_id}",
|
|
json={
|
|
"name": "Describe Frontend Stack",
|
|
"description": "Capture React and Vite usage.",
|
|
"confidence": 0.8,
|
|
},
|
|
)
|
|
assert capability_edit_response.status_code == 200
|
|
|
|
approve_response = client.post(
|
|
f"/repos/{repository_id}/analysis-runs/"
|
|
f"{run['analysis_run']['id']}/candidate-graph/approve",
|
|
json={"notes": "Approved in API test"},
|
|
)
|
|
assert approve_response.status_code == 200
|
|
ability_map = approve_response.json()
|
|
assert ability_map["repository"]["status"] == "indexed"
|
|
assert ability_map["abilities"][0]["name"] == "Frontend Delivery"
|
|
assert ability_map["abilities"][0]["confidence_label"] in {
|
|
"low",
|
|
"medium",
|
|
"high",
|
|
}
|
|
assert ability_map["abilities"][0]["capabilities"][0]["name"] == (
|
|
"Describe Frontend Stack"
|
|
)
|
|
|
|
search_response = client.get("/search", params={"q": "frontend"})
|
|
assert search_response.status_code == 200
|
|
assert search_response.json()
|
|
assert "matched_field" in search_response.json()[0]
|
|
assert "confidence_label" in search_response.json()[0]
|
|
|
|
filtered_search_response = client.get(
|
|
"/search",
|
|
params={
|
|
"q": "frontend",
|
|
"status": "indexed",
|
|
"ability": "Frontend",
|
|
"capability": "Frontend Stack",
|
|
},
|
|
)
|
|
assert filtered_search_response.status_code == 200
|
|
assert filtered_search_response.json()
|
|
|
|
abilities_response = client.get("/abilities")
|
|
assert abilities_response.status_code == 200
|
|
assert abilities_response.json()[0]["name"] == "Frontend Delivery"
|
|
assert abilities_response.json()[0]["repository_name"] == "Frontend"
|
|
assert abilities_response.json()[0]["confidence_label"] in {
|
|
"low",
|
|
"medium",
|
|
"high",
|
|
}
|
|
|
|
capabilities_response = client.get("/capabilities")
|
|
assert capabilities_response.status_code == 200
|
|
assert capabilities_response.json()[0]["name"] == "Describe Frontend Stack"
|
|
assert capabilities_response.json()[0]["ability_name"] == "Frontend Delivery"
|
|
|
|
facts_response = client.get(f"/repos/{repository_id}/observed-facts")
|
|
assert facts_response.status_code == 200
|
|
fact_names = {
|
|
(fact["kind"], fact["name"], fact["path"])
|
|
for fact in facts_response.json()
|
|
}
|
|
assert ("documentation", "README", "README.md") in fact_names
|
|
assert ("framework", "React", "package.json") in fact_names
|
|
assert ("framework", "Vite", "package.json") in fact_names
|
|
|
|
chunks_response = client.get(
|
|
f"/repos/{repository_id}/analysis-runs/"
|
|
f"{run['analysis_run']['id']}/content-chunks"
|
|
)
|
|
assert chunks_response.status_code == 200
|
|
assert {
|
|
(chunk["kind"], chunk["path"]) for chunk in chunks_response.json()
|
|
} >= {("documentation", "README.md"), ("manifest", "package.json")}
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_api_source_linked_candidate_and_repo_update_loop(tmp_path):
|
|
source = tmp_path / "source-linked"
|
|
source.mkdir()
|
|
(source / "README.md").write_text(
|
|
"# Source Linked\n\nProvides operational HTTP status checks.\n",
|
|
encoding="utf-8",
|
|
)
|
|
(source / "docs").mkdir()
|
|
(source / "docs" / "usage.md").write_text(
|
|
"# Usage\n\nCall the status endpoint before routing traffic.\n",
|
|
encoding="utf-8",
|
|
)
|
|
(source / "examples").mkdir()
|
|
(source / "examples" / "status_client.py").write_text(
|
|
"print('GET /status')\n",
|
|
encoding="utf-8",
|
|
)
|
|
(source / "tests").mkdir()
|
|
(source / "tests" / "test_status.py").write_text(
|
|
"def test_status(): pass\n",
|
|
encoding="utf-8",
|
|
)
|
|
(source / "requirements.txt").write_text("fastapi\npytest\n", encoding="utf-8")
|
|
app_file = source / "app.py"
|
|
app_file.write_text(
|
|
"from fastapi import FastAPI\n"
|
|
"app = FastAPI()\n"
|
|
'@app.get("/status")\n'
|
|
"def status():\n"
|
|
" return {'status': 'ok'}\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def override_settings():
|
|
return Settings(
|
|
database_path=str(tmp_path / "source-linked.sqlite3"),
|
|
checkout_root=str(tmp_path / "source-linked-checkouts"),
|
|
)
|
|
|
|
app.dependency_overrides[get_settings] = override_settings
|
|
client = TestClient(app)
|
|
try:
|
|
repository_response = client.post(
|
|
"/repos",
|
|
json={"name": "Source Linked", "url": str(source)},
|
|
)
|
|
assert repository_response.status_code == 201
|
|
repository_id = repository_response.json()["id"]
|
|
|
|
first_run_response = client.post(
|
|
f"/repos/{repository_id}/analysis-runs",
|
|
json={},
|
|
)
|
|
assert first_run_response.status_code == 201
|
|
first_run = first_run_response.json()
|
|
first_run_id = first_run["analysis_run"]["id"]
|
|
assert first_run["snapshot"]["file_count"] == 6
|
|
|
|
facts_response = client.get(f"/repos/{repository_id}/observed-facts")
|
|
assert facts_response.status_code == 200
|
|
facts = facts_response.json()
|
|
fact_keys = {(fact["kind"], fact["name"], fact["path"]) for fact in facts}
|
|
assert ("documentation", "README", "README.md") in fact_keys
|
|
assert ("documentation", "usage.md", "docs/usage.md") in fact_keys
|
|
assert ("example", "status_client.py", "examples/status_client.py") in fact_keys
|
|
assert ("test", "test_status.py", "tests/test_status.py") in fact_keys
|
|
assert ("framework", "FastAPI", "requirements.txt") in fact_keys
|
|
assert ("interface", "python route decorator", "app.py") in fact_keys
|
|
|
|
candidate_response = client.get(
|
|
f"/repos/{repository_id}/analysis-runs/{first_run_id}/candidate-graph"
|
|
)
|
|
assert candidate_response.status_code == 200
|
|
candidate_graph = candidate_response.json()
|
|
interface_capability = next(
|
|
capability
|
|
for capability in candidate_graph["abilities"][0]["capabilities"]
|
|
if capability["name"] == "Expose Repository Interface"
|
|
)
|
|
feature = next(
|
|
item
|
|
for item in interface_capability["features"]
|
|
if item["name"] == "GET /status"
|
|
)
|
|
assert feature["type"] == "API"
|
|
assert feature["location"] == "app.py"
|
|
assert feature["source_refs"][0]["path"] == "app.py"
|
|
assert feature["source_refs"][0]["line"] == 3
|
|
evidence_refs = {
|
|
(evidence["type"], evidence["reference"], evidence["strength"])
|
|
for evidence in interface_capability["evidence"]
|
|
}
|
|
assert ("test", "tests/test_status.py", "strong") in evidence_refs
|
|
assert ("example", "examples/status_client.py", "strong") in evidence_refs
|
|
assert ("documentation", "README.md", "medium") in evidence_refs
|
|
|
|
approve_response = client.post(
|
|
f"/repos/{repository_id}/analysis-runs/{first_run_id}"
|
|
"/candidate-graph/approve",
|
|
json={"notes": "Approved source-linked e2e fixture"},
|
|
)
|
|
assert approve_response.status_code == 200
|
|
approved_map = approve_response.json()
|
|
approved_capability = next(
|
|
capability
|
|
for capability in approved_map["abilities"][0]["capabilities"]
|
|
if capability["name"] == "Expose Repository Interface"
|
|
)
|
|
approved_feature = next(
|
|
item
|
|
for item in approved_capability["features"]
|
|
if item["name"] == "GET /status"
|
|
)
|
|
assert approved_feature["source_refs"][0]["line"] == 3
|
|
assert {
|
|
evidence["reference"] for evidence in approved_capability["evidence"]
|
|
} >= {"tests/test_status.py", "examples/status_client.py", "README.md"}
|
|
|
|
evidence_search = client.get(
|
|
"/search",
|
|
params={"q": "test_status.py", "status": "indexed"},
|
|
)
|
|
assert evidence_search.status_code == 200
|
|
evidence_result = evidence_search.json()[0]
|
|
assert evidence_result["match_type"] == "evidence"
|
|
assert evidence_result["evidence_level"] == "strong"
|
|
assert evidence_result["source_reference"] == "tests/test_status.py"
|
|
|
|
app_file.write_text(
|
|
"from fastapi import FastAPI\n"
|
|
"app = FastAPI()\n"
|
|
'@app.get("/status")\n'
|
|
"def status():\n"
|
|
" return {'status': 'ok'}\n\n"
|
|
'@app.get("/ready")\n'
|
|
"def ready():\n"
|
|
" return {'ready': True}\n",
|
|
encoding="utf-8",
|
|
)
|
|
second_run_response = client.post(
|
|
f"/repos/{repository_id}/analysis-runs",
|
|
json={},
|
|
)
|
|
assert second_run_response.status_code == 201
|
|
second_run_id = second_run_response.json()["analysis_run"]["id"]
|
|
assert second_run_id != first_run_id
|
|
|
|
second_candidate_response = client.get(
|
|
f"/repos/{repository_id}/analysis-runs/{second_run_id}/candidate-graph"
|
|
)
|
|
assert second_candidate_response.status_code == 200
|
|
second_features = {
|
|
feature["name"]
|
|
for ability in second_candidate_response.json()["abilities"]
|
|
for capability in ability["capabilities"]
|
|
for feature in capability["features"]
|
|
}
|
|
assert any(
|
|
"GET /status" in feature_name and "GET /ready" in feature_name
|
|
for feature_name in second_features
|
|
)
|
|
|
|
approved_after_reanalysis = client.get(f"/repos/{repository_id}/ability-map")
|
|
assert approved_after_reanalysis.status_code == 200
|
|
approved_features = {
|
|
feature["name"]
|
|
for ability in approved_after_reanalysis.json()["abilities"]
|
|
for capability in ability["capabilities"]
|
|
for feature in capability["features"]
|
|
}
|
|
assert "GET /status" in approved_features
|
|
assert "GET /ready" not in approved_features
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_ui_register_analyze_and_approve_loop(tmp_path):
|
|
source = tmp_path / "repo"
|
|
source.mkdir()
|
|
(source / "README.md").write_text(
|
|
"# UI Repo\nReports service status through API and CLI entry points.\n",
|
|
encoding="utf-8",
|
|
)
|
|
(source / "requirements.txt").write_text("fastapi\n", encoding="utf-8")
|
|
(source / "app.py").write_text(
|
|
"from fastapi import FastAPI\n"
|
|
"app = FastAPI()\n"
|
|
'@app.get("/status")\n'
|
|
"def status():\n"
|
|
" return {}\n",
|
|
encoding="utf-8",
|
|
)
|
|
(source / "cli.py").write_text(
|
|
"import click\n\n"
|
|
"@click.command()\n"
|
|
"def status():\n"
|
|
" click.echo('ok')\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def override_settings():
|
|
return Settings(
|
|
database_path=str(tmp_path / "ui.sqlite3"),
|
|
checkout_root=str(tmp_path / "ui-checkouts"),
|
|
)
|
|
|
|
app.dependency_overrides[get_settings] = override_settings
|
|
client = TestClient(app)
|
|
try:
|
|
index_response = client.get("/ui")
|
|
assert index_response.status_code == 200
|
|
assert "Register Repository" in index_response.text
|
|
assert "Registering repository..." in index_response.text
|
|
assert "Password or access token" in index_response.text
|
|
assert "Explore after registration" in index_response.text
|
|
assert "Use LLM assistance if configured" in index_response.text
|
|
assert "Trusted auto-populate after analysis" in index_response.text
|
|
|
|
create_response = client.post(
|
|
"/ui/repos",
|
|
data={
|
|
"url": str(source),
|
|
"branch": "main",
|
|
"access_username": "",
|
|
"access_password": "",
|
|
"explore_after_registration": "",
|
|
"use_llm_assistance": "1",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert create_response.status_code == 303
|
|
repository_path = create_response.headers["location"]
|
|
repository_id = int(repository_path.rsplit("/", 1)[-1])
|
|
|
|
detail_response = client.get(repository_path)
|
|
assert detail_response.status_code == 200
|
|
assert "Run Analysis" in detail_response.text
|
|
assert "Running analysis..." in detail_response.text
|
|
assert "Analyze cached checkout without fetching upstream" in detail_response.text
|
|
assert "Use LLM assistance if configured" in detail_response.text
|
|
assert "Trusted auto-populate after analysis" in detail_response.text
|
|
assert "Repository Metadata" in detail_response.text
|
|
|
|
edit_repository_response = client.post(
|
|
f"{repository_path}/edit",
|
|
data={
|
|
"name": "UI Repo Edited",
|
|
"description": "Edited in the UI.",
|
|
"branch": "develop",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert edit_repository_response.status_code == 303
|
|
detail_response = client.get(repository_path)
|
|
assert "UI Repo Edited" in detail_response.text
|
|
assert "develop" in detail_response.text
|
|
|
|
run_response = client.post(
|
|
f"{repository_path}/analysis-runs",
|
|
data={
|
|
"source_path": "",
|
|
"use_llm_assistance": "1",
|
|
"access_username": "",
|
|
"access_password": "",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert run_response.status_code == 303
|
|
run_path = run_response.headers["location"]
|
|
first_run_id = int(run_path.rsplit("/", 1)[-1])
|
|
|
|
run_detail = client.get(run_path)
|
|
assert run_detail.status_code == 200
|
|
assert "Candidate Graph" in run_detail.text
|
|
assert "1 abilities" in run_detail.text
|
|
assert "2 capabilities" in run_detail.text
|
|
assert "2 features" in run_detail.text
|
|
assert "7 facts" in run_detail.text
|
|
assert "Content Chunks" in run_detail.text
|
|
assert "README.md:1-2" in run_detail.text
|
|
assert "ID " in run_detail.text
|
|
assert "No review decisions yet." in run_detail.text
|
|
assert "Expectation Gaps" in run_detail.text
|
|
assert "Record Gap" in run_detail.text
|
|
|
|
gap_response = client.post(
|
|
f"{run_path}/expectation-gaps",
|
|
data={
|
|
"expected_type": "capability",
|
|
"expected_name": "Use OpenRouter Models",
|
|
"source": "human",
|
|
"notes": "Expected from provider docs.",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert gap_response.status_code == 303
|
|
run_detail = client.get(run_path)
|
|
assert "Use OpenRouter Models" in run_detail.text
|
|
assert "Expected from provider docs." in run_detail.text
|
|
|
|
approve_response = client.post(
|
|
f"{run_path}/candidate-graph/approve",
|
|
follow_redirects=False,
|
|
)
|
|
assert approve_response.status_code == 303
|
|
|
|
approved_detail = client.get(approve_response.headers["location"])
|
|
assert approved_detail.status_code == 200
|
|
assert "Approved Ability Map" in approved_detail.text
|
|
assert "1 abilities" in approved_detail.text
|
|
assert "2 capabilities" in approved_detail.text
|
|
assert "2 features" in approved_detail.text
|
|
assert "Latest Candidate Graph" in approved_detail.text
|
|
assert "1 candidate abilities" in approved_detail.text
|
|
assert "2 candidate capabilities" in approved_detail.text
|
|
assert "2 candidate features" in approved_detail.text
|
|
assert "7 candidate facts" in approved_detail.text
|
|
assert "Use Approved Registry" in approved_detail.text
|
|
assert "Search Profile" in approved_detail.text
|
|
assert "Discovery" in approved_detail.text
|
|
assert "Export" in approved_detail.text
|
|
assert "Approved Elements" in approved_detail.text
|
|
assert "q=Report+Service+Status" in approved_detail.text
|
|
assert (
|
|
f"/ui/repos/{repository_id}/elements?scope=approved&type=abilities"
|
|
in approved_detail.text
|
|
)
|
|
assert (
|
|
f"/ui/repos/{repository_id}/elements?scope=candidate&analysis_run_id={first_run_id}&type=features"
|
|
in approved_detail.text
|
|
)
|
|
assert (
|
|
f"/ui/repos/{repository_id}/elements?scope=facts&analysis_run_id={first_run_id}&type=facts"
|
|
in approved_detail.text
|
|
)
|
|
assert "Report Service Status Through API And CLI Entry" in approved_detail.text
|
|
assert "Language: Python" in approved_detail.text
|
|
assert "Framework: FastAPI" in approved_detail.text
|
|
assert "interface:app.py:3" in approved_detail.text
|
|
assert "approve_candidate_graph" in approved_detail.text
|
|
|
|
approved_listing = client.get(
|
|
f"/ui/repos/{repository_id}/elements",
|
|
params={"scope": "approved", "type": "capabilities"},
|
|
)
|
|
assert approved_listing.status_code == 200
|
|
assert "Approved Capabilities" in approved_listing.text
|
|
assert "Expose Repository Interface" in approved_listing.text
|
|
|
|
candidate_listing = client.get(
|
|
f"/ui/repos/{repository_id}/elements",
|
|
params={
|
|
"scope": "candidate",
|
|
"analysis_run_id": first_run_id,
|
|
"type": "features",
|
|
},
|
|
)
|
|
assert candidate_listing.status_code == 200
|
|
assert "Candidate Features" in candidate_listing.text
|
|
assert "Search" in candidate_listing.text
|
|
assert "Class" in candidate_listing.text
|
|
assert "GET /status" in candidate_listing.text
|
|
|
|
filtered_candidate_listing = client.get(
|
|
f"/ui/repos/{repository_id}/elements",
|
|
params={
|
|
"scope": "candidate",
|
|
"analysis_run_id": first_run_id,
|
|
"type": "features",
|
|
"q": "status",
|
|
"class_filter": "API",
|
|
},
|
|
)
|
|
assert filtered_candidate_listing.status_code == 200
|
|
assert "1 of 2 shown" in filtered_candidate_listing.text
|
|
assert "GET /status" in filtered_candidate_listing.text
|
|
assert "CLI command status" not in filtered_candidate_listing.text
|
|
|
|
fact_listing = client.get(
|
|
f"/ui/repos/{repository_id}/elements",
|
|
params={
|
|
"scope": "facts",
|
|
"analysis_run_id": first_run_id,
|
|
"type": "facts",
|
|
},
|
|
)
|
|
assert fact_listing.status_code == 200
|
|
assert "Observed Facts" in fact_listing.text
|
|
assert "python route decorator" in fact_listing.text
|
|
|
|
filtered_fact_listing = client.get(
|
|
f"/ui/repos/{repository_id}/elements",
|
|
params={
|
|
"scope": "facts",
|
|
"analysis_run_id": first_run_id,
|
|
"type": "facts",
|
|
"class_filter": "framework",
|
|
},
|
|
)
|
|
assert filtered_fact_listing.status_code == 200
|
|
assert "1 of 7 shown" in filtered_fact_listing.text
|
|
assert "FastAPI" in filtered_fact_listing.text
|
|
assert "python route decorator" not in filtered_fact_listing.text
|
|
|
|
(source / "app.py").write_text(
|
|
"from fastapi import FastAPI\n"
|
|
"app = FastAPI()\n"
|
|
'@app.get("/status")\n'
|
|
"def status():\n"
|
|
" return {}\n\n"
|
|
'@app.get("/ready")\n'
|
|
"def ready():\n"
|
|
" return {}\n",
|
|
encoding="utf-8",
|
|
)
|
|
second_run_response = client.post(
|
|
f"{repository_path}/analysis-runs",
|
|
data={"source_path": "", "use_llm_assistance": "1"},
|
|
follow_redirects=False,
|
|
)
|
|
assert second_run_response.status_code == 303
|
|
second_run_path = second_run_response.headers["location"]
|
|
|
|
second_run_detail = client.get(second_run_path)
|
|
assert second_run_detail.status_code == 200
|
|
assert "Compare to #" in second_run_detail.text
|
|
|
|
second_run_id = int(second_run_path.rsplit("/", 1)[-1])
|
|
diff_response = client.get(
|
|
f"{repository_path}/analysis-runs/{first_run_id}/diff/{second_run_id}"
|
|
)
|
|
assert diff_response.status_code == 200
|
|
assert "Change Review" in diff_response.text
|
|
assert "Approved Registry Impact" in diff_response.text
|
|
assert "Candidate Claims" in diff_response.text
|
|
assert "GET /ready" in diff_response.text
|
|
|
|
change_approval = client.post(
|
|
f"{second_run_path}/changes/approve",
|
|
data={"notes": "Accept UI change review."},
|
|
follow_redirects=False,
|
|
)
|
|
assert change_approval.status_code == 303
|
|
changed_detail = client.get(change_approval.headers["location"])
|
|
assert "GET /ready" in changed_detail.text
|
|
assert "approve_analysis_run_changes" in changed_detail.text
|
|
|
|
search_response = client.get("/ui/search", params={"q": "repository"})
|
|
assert search_response.status_code == 200
|
|
assert "UI Repo" in search_response.text
|
|
assert "Field" in search_response.text
|
|
assert "#capability-" in search_response.text or "#ability-" in search_response.text
|
|
|
|
filtered_search_response = client.get(
|
|
"/ui/search",
|
|
params={
|
|
"q": "repository",
|
|
"status": "indexed",
|
|
"language": "Python",
|
|
"ability": "Report Service Status",
|
|
"capability": "Repository",
|
|
},
|
|
)
|
|
assert filtered_search_response.status_code == 200
|
|
assert "UI Repo" in filtered_search_response.text
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_ui_register_and_explore_lands_on_analysis_result(tmp_path):
|
|
source = tmp_path / "explore-repo"
|
|
source.mkdir()
|
|
(source / "README.md").write_text("# Explore Repo\n", encoding="utf-8")
|
|
(source / "pyproject.toml").write_text(
|
|
"[project]\nname = \"explore-repo\"\ndescription = \"Explorable repo.\"\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def override_settings():
|
|
return Settings(
|
|
database_path=str(tmp_path / "ui-explore.sqlite3"),
|
|
checkout_root=str(tmp_path / "ui-explore-checkouts"),
|
|
)
|
|
|
|
app.dependency_overrides[get_settings] = override_settings
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.post(
|
|
"/ui/repos",
|
|
data={
|
|
"url": str(source),
|
|
"branch": "main",
|
|
"access_username": "",
|
|
"access_password": "",
|
|
"explore_after_registration": "1",
|
|
"use_llm_assistance": "",
|
|
"trusted_auto_approve": "1",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert response.headers["location"].endswith("/analysis-runs/1")
|
|
|
|
result = client.get(response.headers["location"])
|
|
assert result.status_code == 200
|
|
assert "Candidate Graph" in result.text
|
|
assert "approved" in result.text
|
|
assert "Observed Facts" in result.text
|
|
assert "trusted_auto_approve_candidate_graph" in result.text
|
|
|
|
repository_detail = client.get("/ui/repos/1")
|
|
assert repository_detail.status_code == 200
|
|
assert "Use Approved Registry" in repository_detail.text
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_ui_registration_failure_returns_feedback(tmp_path):
|
|
def override_settings():
|
|
return Settings(
|
|
database_path=str(tmp_path / "ui-error.sqlite3"),
|
|
checkout_root=str(tmp_path / "ui-error-checkouts"),
|
|
)
|
|
|
|
app.dependency_overrides[get_settings] = override_settings
|
|
client = TestClient(app)
|
|
try:
|
|
response = client.post(
|
|
"/ui/repos",
|
|
data={
|
|
"url": str(tmp_path / "missing-repo"),
|
|
"branch": "main",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "Registration failed." in response.text
|
|
assert "git clone" in response.text
|
|
assert "Register Repository" in response.text
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_ui_manual_registry_entry_loop(tmp_path):
|
|
source = tmp_path / "manual-repo"
|
|
source.mkdir()
|
|
(source / "README.md").write_text("# Manual Repo\n", encoding="utf-8")
|
|
|
|
def override_settings():
|
|
return Settings(
|
|
database_path=str(tmp_path / "ui-manual.sqlite3"),
|
|
checkout_root=str(tmp_path / "ui-manual-checkouts"),
|
|
)
|
|
|
|
app.dependency_overrides[get_settings] = override_settings
|
|
client = TestClient(app)
|
|
try:
|
|
create_response = client.post(
|
|
"/ui/repos",
|
|
data={"url": str(source), "branch": "main"},
|
|
follow_redirects=False,
|
|
)
|
|
assert create_response.status_code == 303
|
|
repository_path = create_response.headers["location"]
|
|
repository_id = int(repository_path.rsplit("/", 1)[-1])
|
|
|
|
detail_response = client.get(repository_path)
|
|
assert detail_response.status_code == 200
|
|
assert "Manual Registry Entry" in detail_response.text
|
|
|
|
ability_response = client.post(
|
|
f"{repository_path}/abilities",
|
|
data={
|
|
"name": "Manual Ability",
|
|
"description": "Curated by hand.",
|
|
"confidence": "0.95",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert ability_response.status_code == 303
|
|
ability_id = client.get(f"/repos/{repository_id}/ability-map").json()[
|
|
"abilities"
|
|
][0]["id"]
|
|
|
|
capability_response = client.post(
|
|
f"{repository_path}/capabilities",
|
|
data={
|
|
"ability_id": str(ability_id),
|
|
"name": "Manual Capability",
|
|
"description": "Curated capability.",
|
|
"inputs": "request, context",
|
|
"outputs": "response",
|
|
"confidence": "0.9",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert capability_response.status_code == 303
|
|
capability_id = client.get(f"/repos/{repository_id}/ability-map").json()[
|
|
"abilities"
|
|
][0]["capabilities"][0]["id"]
|
|
|
|
feature_response = client.post(
|
|
f"{repository_path}/features",
|
|
data={
|
|
"capability_id": str(capability_id),
|
|
"name": "Manual API",
|
|
"type": "REST endpoint",
|
|
"location": "src/manual.py",
|
|
"confidence": "0.88",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert feature_response.status_code == 303
|
|
|
|
evidence_response = client.post(
|
|
f"{repository_path}/evidence",
|
|
data={
|
|
"capability_id": str(capability_id),
|
|
"type": "documentation",
|
|
"reference": "README.md",
|
|
"strength": "medium",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert evidence_response.status_code == 303
|
|
|
|
detail_response = client.get(repository_path)
|
|
assert "Manual Ability" in detail_response.text
|
|
assert "Manual Capability" in detail_response.text
|
|
assert "Manual API" in detail_response.text
|
|
assert "README.md" in detail_response.text
|
|
assert "ID " in detail_response.text
|
|
assert "Save Ability" in detail_response.text
|
|
|
|
edit_ability_response = client.post(
|
|
f"{repository_path}/abilities/{ability_id}/edit",
|
|
data={
|
|
"name": "Edited Manual Ability",
|
|
"description": "Edited by hand.",
|
|
"confidence": "0.8",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert edit_ability_response.status_code == 303
|
|
|
|
edit_capability_response = client.post(
|
|
f"{repository_path}/capabilities/{capability_id}/edit",
|
|
data={
|
|
"name": "Edited Manual Capability",
|
|
"description": "Edited capability.",
|
|
"inputs": "ticket",
|
|
"outputs": "decision",
|
|
"confidence": "0.75",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert edit_capability_response.status_code == 303
|
|
|
|
ability_map = client.get(f"/repos/{repository_id}/ability-map").json()
|
|
feature_id = ability_map["abilities"][0]["capabilities"][0]["features"][0]["id"]
|
|
evidence_id = ability_map["abilities"][0]["capabilities"][0]["evidence"][0]["id"]
|
|
|
|
edit_feature_response = client.post(
|
|
f"{repository_path}/features/{feature_id}/edit",
|
|
data={
|
|
"name": "Edited Manual API",
|
|
"type": "HTTP endpoint",
|
|
"location": "src/edited.py",
|
|
"confidence": "0.7",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert edit_feature_response.status_code == 303
|
|
|
|
edit_evidence_response = client.post(
|
|
f"{repository_path}/evidence/{evidence_id}/edit",
|
|
data={
|
|
"type": "test",
|
|
"reference": "tests/test_manual.py",
|
|
"strength": "strong",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert edit_evidence_response.status_code == 303
|
|
|
|
detail_response = client.get(repository_path)
|
|
assert "Edited Manual Ability" in detail_response.text
|
|
assert "Edited Manual Capability" in detail_response.text
|
|
assert "Edited Manual API" in detail_response.text
|
|
assert "tests/test_manual.py" in detail_response.text
|
|
|
|
delete_feature_response = client.post(
|
|
f"{repository_path}/features/{feature_id}/delete",
|
|
follow_redirects=False,
|
|
)
|
|
assert delete_feature_response.status_code == 303
|
|
delete_evidence_response = client.post(
|
|
f"{repository_path}/evidence/{evidence_id}/delete",
|
|
follow_redirects=False,
|
|
)
|
|
assert delete_evidence_response.status_code == 303
|
|
|
|
detail_response = client.get(repository_path)
|
|
assert "Edited Manual API" not in detail_response.text
|
|
assert "tests/test_manual.py" not in detail_response.text
|
|
|
|
failed_delete_response = client.post(
|
|
f"{repository_path}/delete",
|
|
data={"confirm_name": "wrong name"},
|
|
follow_redirects=False,
|
|
)
|
|
assert failed_delete_response.status_code == 303
|
|
assert client.get(repository_path).status_code == 200
|
|
|
|
delete_repository_response = client.post(
|
|
f"{repository_path}/delete",
|
|
data={"confirm_name": "Manual Repo"},
|
|
follow_redirects=False,
|
|
)
|
|
assert delete_repository_response.status_code == 303
|
|
assert delete_repository_response.headers["location"] == "/ui"
|
|
assert client.get(repository_path).status_code == 404
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_ui_discovery_compare_gap_and_export(tmp_path):
|
|
def override_settings():
|
|
return Settings(
|
|
database_path=str(tmp_path / "ui-discovery.sqlite3"),
|
|
checkout_root=str(tmp_path / "ui-discovery-checkouts"),
|
|
)
|
|
|
|
app.dependency_overrides[get_settings] = override_settings
|
|
client = TestClient(app)
|
|
try:
|
|
first = client.post(
|
|
"/repos",
|
|
json={
|
|
"name": "MailRouter",
|
|
"url": "https://example.com/mail-router-ui.git",
|
|
"description": "Routes customer email.",
|
|
},
|
|
).json()
|
|
second = client.post(
|
|
"/repos",
|
|
json={
|
|
"name": "SupportRouter",
|
|
"url": "https://example.com/support-router-ui.git",
|
|
"description": "Routes support requests.",
|
|
},
|
|
).json()
|
|
empty = client.post(
|
|
"/repos",
|
|
json={
|
|
"name": "EmptyProfile",
|
|
"url": "https://example.com/empty-profile-ui.git",
|
|
"description": "No approved entries yet.",
|
|
},
|
|
).json()
|
|
|
|
first_ability = client.post(
|
|
f"/repos/{first['id']}/abilities",
|
|
json={
|
|
"name": "Business Email Routing",
|
|
"description": "Route inbound messages.",
|
|
"confidence": 0.9,
|
|
},
|
|
).json()["id"]
|
|
first_capability = client.post(
|
|
f"/repos/{first['id']}/capabilities",
|
|
json={
|
|
"ability_id": first_ability,
|
|
"name": "Classify Incoming Email",
|
|
"description": "Classify messages by intent.",
|
|
"confidence": 0.8,
|
|
},
|
|
).json()["id"]
|
|
client.post(
|
|
f"/repos/{first['id']}/evidence",
|
|
json={
|
|
"capability_id": first_capability,
|
|
"type": "unit_test",
|
|
"reference": "tests/test_classify.py",
|
|
"strength": "strong",
|
|
},
|
|
)
|
|
client.post(
|
|
f"/repos/{first['id']}/capabilities",
|
|
json={
|
|
"ability_id": first_ability,
|
|
"name": "Route Email to Team",
|
|
"description": "Route messages to owning teams.",
|
|
},
|
|
)
|
|
|
|
second_ability = client.post(
|
|
f"/repos/{second['id']}/abilities",
|
|
json={
|
|
"name": "Business Email Routing",
|
|
"description": "Support routing workflows.",
|
|
"confidence": 0.7,
|
|
},
|
|
).json()["id"]
|
|
second_capability = client.post(
|
|
f"/repos/{second['id']}/capabilities",
|
|
json={
|
|
"ability_id": second_ability,
|
|
"name": "Classify Incoming Email",
|
|
"description": "Classify support requests.",
|
|
"confidence": 0.6,
|
|
},
|
|
).json()["id"]
|
|
client.post(
|
|
f"/repos/{second['id']}/evidence",
|
|
json={
|
|
"capability_id": second_capability,
|
|
"type": "documentation",
|
|
"reference": "README.md",
|
|
"strength": "medium",
|
|
},
|
|
)
|
|
client.post(
|
|
f"/repos/{second['id']}/capabilities",
|
|
json={
|
|
"ability_id": second_ability,
|
|
"name": "Archive Email",
|
|
"description": "Archive resolved messages.",
|
|
},
|
|
)
|
|
|
|
discovery = client.get("/ui/discovery")
|
|
assert discovery.status_code == 200
|
|
assert "Compare Repositories" in discovery.text
|
|
assert "Capability Gap Report" in discovery.text
|
|
assert "EmptyProfile" in discovery.text
|
|
assert "No approved profile" in discovery.text
|
|
|
|
comparison = client.get(
|
|
"/ui/discovery/compare",
|
|
params=[
|
|
("repository_ids", first["id"]),
|
|
("repository_ids", second["id"]),
|
|
],
|
|
)
|
|
assert comparison.status_code == 200
|
|
assert "Repository Comparison" in comparison.text
|
|
assert "Business Email Routing" in comparison.text
|
|
assert "Route Email to Team" in comparison.text
|
|
assert "Archive Email" in comparison.text
|
|
|
|
gaps = client.post(
|
|
"/ui/discovery/gaps",
|
|
data={
|
|
"desired_ability": "Business Email Routing",
|
|
"desired_capabilities": (
|
|
"Classify Incoming Email\n"
|
|
"Route Email to Team\n"
|
|
"German Benchmark Evaluation"
|
|
),
|
|
"repository_ids": [str(first["id"]), str(second["id"])],
|
|
},
|
|
)
|
|
assert gaps.status_code == 200
|
|
assert "Capability Gap Report" in gaps.text
|
|
assert "German Benchmark Evaluation" in gaps.text
|
|
assert "Weak Evidence" in gaps.text
|
|
assert "Duplicates" in gaps.text
|
|
|
|
detail = client.get(f"/ui/repos/{first['id']}")
|
|
assert detail.status_code == 200
|
|
assert f'/ui/repos/{first["id"]}/export' in detail.text
|
|
|
|
export = client.get(f"/ui/repos/{first['id']}/export")
|
|
assert export.status_code == 200
|
|
assert export.headers["content-type"].startswith("application/x-yaml")
|
|
assert 'name: "MailRouter"' in export.text
|
|
assert 'name: "Classify Incoming Email"' in export.text
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_api_rejects_candidate_capability_feature_and_evidence(tmp_path):
|
|
source = tmp_path / "repo"
|
|
source.mkdir()
|
|
(source / "README.md").write_text("# API Reject Leaves\n", encoding="utf-8")
|
|
(source / "tests").mkdir()
|
|
(source / "tests" / "test_status.py").write_text(
|
|
"def test_status(): pass\n",
|
|
encoding="utf-8",
|
|
)
|
|
(source / "app.py").write_text(
|
|
"from fastapi import FastAPI\n"
|
|
"app = FastAPI()\n"
|
|
'@app.get("/status")\n'
|
|
"def status():\n"
|
|
" return {}\n",
|
|
encoding="utf-8",
|
|
)
|
|
def override_settings():
|
|
return Settings(
|
|
database_path=str(tmp_path / "api-reject.sqlite3"),
|
|
checkout_root=str(tmp_path / "api-reject-checkouts"),
|
|
)
|
|
|
|
app.dependency_overrides[get_settings] = override_settings
|
|
client = TestClient(app)
|
|
try:
|
|
repository_response = client.post(
|
|
"/repos",
|
|
json={"name": "API Reject Leaves", "url": str(source)},
|
|
)
|
|
repository_id = repository_response.json()["id"]
|
|
run_response = client.post(f"/repos/{repository_id}/analysis-runs", json={})
|
|
run_id = run_response.json()["analysis_run"]["id"]
|
|
graph_response = client.get(
|
|
f"/repos/{repository_id}/analysis-runs/{run_id}/candidate-graph"
|
|
)
|
|
capability = graph_response.json()["abilities"][0]["capabilities"][0]
|
|
feature_id = capability["features"][0]["id"]
|
|
evidence_id = capability["evidence"][0]["id"]
|
|
|
|
feature_response = client.post(
|
|
f"/repos/{repository_id}/analysis-runs/{run_id}"
|
|
f"/candidate-features/{feature_id}/reject",
|
|
json={"notes": "Noisy interface"},
|
|
)
|
|
assert feature_response.status_code == 200
|
|
assert (
|
|
feature_response.json()["abilities"][0]["capabilities"][0]["features"][0][
|
|
"status"
|
|
]
|
|
== "rejected"
|
|
)
|
|
|
|
evidence_response = client.post(
|
|
f"/repos/{repository_id}/analysis-runs/{run_id}"
|
|
f"/candidate-evidence/{evidence_id}/reject",
|
|
json={"notes": "Weak evidence"},
|
|
)
|
|
assert evidence_response.status_code == 200
|
|
assert (
|
|
evidence_response.json()["abilities"][0]["capabilities"][0]["evidence"][0][
|
|
"status"
|
|
]
|
|
== "rejected"
|
|
)
|
|
|
|
run_response = client.post(f"/repos/{repository_id}/analysis-runs", json={})
|
|
run_id = run_response.json()["analysis_run"]["id"]
|
|
graph_response = client.get(
|
|
f"/repos/{repository_id}/analysis-runs/{run_id}/candidate-graph"
|
|
)
|
|
capability_id = graph_response.json()["abilities"][0]["capabilities"][0]["id"]
|
|
capability_response = client.post(
|
|
f"/repos/{repository_id}/analysis-runs/{run_id}"
|
|
f"/candidate-capabilities/{capability_id}/reject",
|
|
json={"notes": "Reject whole capability"},
|
|
)
|
|
assert capability_response.status_code == 200
|
|
assert (
|
|
capability_response.json()["abilities"][0]["capabilities"][0]["status"]
|
|
== "rejected"
|
|
)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_api_relinks_candidate_feature_and_evidence(tmp_path):
|
|
source = tmp_path / "repo"
|
|
source.mkdir()
|
|
(source / "README.md").write_text("# API Relink Leaves\n", encoding="utf-8")
|
|
(source / "requirements.txt").write_text("fastapi\n", encoding="utf-8")
|
|
(source / "tests").mkdir()
|
|
(source / "tests" / "test_status.py").write_text(
|
|
"def test_status(): pass\n",
|
|
encoding="utf-8",
|
|
)
|
|
(source / "app.py").write_text(
|
|
"from fastapi import FastAPI\n"
|
|
"app = FastAPI()\n"
|
|
'@app.get("/status")\n'
|
|
"def status():\n"
|
|
" return {}\n",
|
|
encoding="utf-8",
|
|
)
|
|
def override_settings():
|
|
return Settings(
|
|
database_path=str(tmp_path / "api-relink.sqlite3"),
|
|
checkout_root=str(tmp_path / "api-relink-checkouts"),
|
|
)
|
|
|
|
app.dependency_overrides[get_settings] = override_settings
|
|
client = TestClient(app)
|
|
try:
|
|
repository_response = client.post(
|
|
"/repos",
|
|
json={"name": "API Relink Leaves", "url": str(source)},
|
|
)
|
|
repository_id = repository_response.json()["id"]
|
|
run_response = client.post(f"/repos/{repository_id}/analysis-runs", json={})
|
|
run_id = run_response.json()["analysis_run"]["id"]
|
|
graph_response = client.get(
|
|
f"/repos/{repository_id}/analysis-runs/{run_id}/candidate-graph"
|
|
)
|
|
capabilities = graph_response.json()["abilities"][0]["capabilities"]
|
|
source_capability = capabilities[0]
|
|
target_capability = capabilities[1]
|
|
feature_id = source_capability["features"][0]["id"]
|
|
evidence_id = source_capability["evidence"][0]["id"]
|
|
|
|
feature_response = client.post(
|
|
f"/repos/{repository_id}/analysis-runs/{run_id}"
|
|
f"/candidate-features/{feature_id}/relink",
|
|
json={
|
|
"target_capability_id": target_capability["id"],
|
|
"notes": "Move feature",
|
|
},
|
|
)
|
|
assert feature_response.status_code == 200
|
|
relinked_capabilities = feature_response.json()["abilities"][0]["capabilities"]
|
|
assert relinked_capabilities[0]["features"] == []
|
|
assert relinked_capabilities[1]["features"][0]["id"] == feature_id
|
|
|
|
evidence_response = client.post(
|
|
f"/repos/{repository_id}/analysis-runs/{run_id}"
|
|
f"/candidate-evidence/{evidence_id}/relink",
|
|
json={
|
|
"target_capability_id": target_capability["id"],
|
|
"notes": "Move evidence",
|
|
},
|
|
)
|
|
assert evidence_response.status_code == 200
|
|
relinked_capabilities = evidence_response.json()["abilities"][0]["capabilities"]
|
|
assert relinked_capabilities[1]["evidence"][0]["id"] == evidence_id
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_api_merges_candidate_capability_feature_and_evidence(tmp_path):
|
|
source = tmp_path / "repo"
|
|
source.mkdir()
|
|
(source / "README.md").write_text("# API Merge\n", encoding="utf-8")
|
|
(source / "requirements.txt").write_text("fastapi\n", encoding="utf-8")
|
|
(source / "tests").mkdir()
|
|
(source / "tests" / "test_status.py").write_text(
|
|
"def test_status(): pass\n",
|
|
encoding="utf-8",
|
|
)
|
|
(source / "app.py").write_text(
|
|
"from fastapi import FastAPI\n"
|
|
"app = FastAPI()\n"
|
|
'@app.get("/status")\n'
|
|
"def status():\n"
|
|
" return {}\n"
|
|
'@app.get("/ready")\n'
|
|
"def ready():\n"
|
|
" return {}\n",
|
|
encoding="utf-8",
|
|
)
|
|
(source / "cli.py").write_text(
|
|
"import click\n\n"
|
|
"@click.command()\n"
|
|
"def status_cli():\n"
|
|
" click.echo('ok')\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def override_settings():
|
|
return Settings(
|
|
database_path=str(tmp_path / "api-merge.sqlite3"),
|
|
checkout_root=str(tmp_path / "api-merge-checkouts"),
|
|
)
|
|
|
|
app.dependency_overrides[get_settings] = override_settings
|
|
client = TestClient(app)
|
|
try:
|
|
repository_response = client.post(
|
|
"/repos",
|
|
json={"name": "API Merge", "url": str(source)},
|
|
)
|
|
repository_id = repository_response.json()["id"]
|
|
run_response = client.post(f"/repos/{repository_id}/analysis-runs", json={})
|
|
run_id = run_response.json()["analysis_run"]["id"]
|
|
graph_response = client.get(
|
|
f"/repos/{repository_id}/analysis-runs/{run_id}/candidate-graph"
|
|
)
|
|
capabilities = graph_response.json()["abilities"][0]["capabilities"]
|
|
source_capability = capabilities[0]
|
|
target_capability = capabilities[1]
|
|
|
|
feature_response = client.post(
|
|
f"/repos/{repository_id}/analysis-runs/{run_id}"
|
|
f"/candidate-features/{source_capability['features'][1]['id']}/merge",
|
|
json={
|
|
"target_feature_id": source_capability["features"][0]["id"],
|
|
"notes": "Duplicate route",
|
|
},
|
|
)
|
|
assert feature_response.status_code == 200
|
|
assert (
|
|
feature_response.json()["abilities"][0]["capabilities"][0]["features"][1][
|
|
"status"
|
|
]
|
|
== "merged"
|
|
)
|
|
|
|
evidence_response = client.post(
|
|
f"/repos/{repository_id}/analysis-runs/{run_id}"
|
|
f"/candidate-evidence/{source_capability['evidence'][1]['id']}/merge",
|
|
json={
|
|
"target_evidence_id": source_capability["evidence"][0]["id"],
|
|
"notes": "Duplicate evidence",
|
|
},
|
|
)
|
|
assert evidence_response.status_code == 200
|
|
|
|
capability_response = client.post(
|
|
f"/repos/{repository_id}/analysis-runs/{run_id}"
|
|
f"/candidate-capabilities/{source_capability['id']}/merge",
|
|
json={
|
|
"target_capability_id": target_capability["id"],
|
|
"notes": "Duplicate capability",
|
|
},
|
|
)
|
|
assert capability_response.status_code == 200
|
|
capabilities = capability_response.json()["abilities"][0]["capabilities"]
|
|
assert capabilities[0]["status"] == "merged"
|
|
assert capabilities[1]["features"]
|
|
finally:
|
|
app.dependency_overrides.clear()
|