Files
repo-scoping/tests/test_web_api.py
2026-04-26 00:29:07 +02:00

614 lines
23 KiB
Python

from fastapi.testclient import TestClient
from repo_registry.web_api.app import Settings, app, get_settings
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
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"
)
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_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_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]["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]
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"
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
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
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 "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 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_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()