generated from coulomb/repo-seed
1310 lines
49 KiB
Python
1310 lines
49 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"
|
|
|
|
|
|
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_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_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")
|
|
|
|
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"
|
|
|
|
|
|
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"] == (
|
|
"Review Frontend Repository Usefulness"
|
|
)
|
|
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."
|
|
)
|
|
|
|
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 {"GET /status", "GET /ready"} <= 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\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",
|
|
)
|
|
|
|
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
|
|
|
|
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"]
|
|
|
|
detail_response = client.get(repository_path)
|
|
assert detail_response.status_code == 200
|
|
assert "Run 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": ""},
|
|
follow_redirects=False,
|
|
)
|
|
assert run_response.status_code == 303
|
|
run_path = run_response.headers["location"]
|
|
|
|
run_detail = client.get(run_path)
|
|
assert run_detail.status_code == 200
|
|
assert "Candidate Graph" in run_detail.text
|
|
assert "Content Chunks" in run_detail.text
|
|
assert "README.md:1-1" in run_detail.text
|
|
assert "ID " in run_detail.text
|
|
assert "No review decisions yet." 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 "Review UI Repo Edited Repository Usefulness" 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
|
|
|
|
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": "Repository Usefulness",
|
|
"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_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_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",
|
|
)
|
|
|
|
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()
|