generated from coulomb/repo-seed
250 lines
8.8 KiB
Python
250 lines
8.8 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"]
|
|
|
|
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"
|
|
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()
|
|
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
|
|
|
|
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"]
|
|
|
|
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"
|
|
|
|
run_response = client.post(f"/repos/{repository_id}/analysis-runs", json={})
|
|
assert run_response.status_code == 201
|
|
run = run_response.json()
|
|
|
|
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"] == (
|
|
"Review Frontend Repository Usefulness"
|
|
)
|
|
|
|
search_response = client.get("/search", params={"q": "structure"})
|
|
assert search_response.status_code == 200
|
|
assert search_response.json()
|
|
|
|
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 / "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
|
|
|
|
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
|
|
finally:
|
|
app.dependency_overrides.clear()
|