import json import sqlite3 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 add_candidate_capability(database_path, repository_id, analysis_run_id, name): with sqlite3.connect(database_path) as connection: ability_id = connection.execute( """ SELECT id FROM candidate_abilities WHERE repository_id = ? AND analysis_run_id = ? ORDER BY id LIMIT 1 """, (repository_id, analysis_run_id), ).fetchone()[0] cursor = connection.execute( """ INSERT INTO candidate_capabilities (repository_id, analysis_run_id, ability_id, name, description, inputs, outputs, primary_class, attributes, confidence, source_refs) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( repository_id, analysis_run_id, ability_id, name, "Review target capability inserted for API review workflow tests.", "[]", "[]", "test-capability", json.dumps(["test-review-target"]), 0.5, "[]", ), ) return int(cursor.lastrowid) 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 Scoping" assert schema["info"]["version"] == "0.1.0" assert {tag["name"] for tag in schema["tags"]} >= { "repositories", "analysis", "review", "registry", "scope", "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}/characteristics/rebuild": { "post": { "tags": ["analysis"], "success_schema": "CharacteristicRebuildResponse", } }, "/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/{repo_slug}/scope": { "get": {"tags": ["scope"], "success_schema": None} }, "/repos/{repo_slug}/scope/diff": { "get": {"tags": ["scope"], "success_schema": "object"} }, "/repos/{repo_slug}/scope/write": { "post": {"tags": ["scope"], "success_schema": "object"} }, "/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 Scoping" in response.text assert "openapi.json" in response.text def test_ui_uses_repository_scoping_brand(): client = TestClient(app) response = client.get("/ui") assert response.status_code == 200 assert "Repository Scoping" in response.text assert "Repository Ability Registry" not in response.text def test_ui_homepage_registry_panel_uses_directory_identifier_and_left_column(tmp_path): source = tmp_path / "short-dir" source.mkdir() (source / "README.md").write_text("# Short Dir\n", encoding="utf-8") def override_settings(): return Settings( database_path=str(tmp_path / "ui-home.sqlite3"), checkout_root=str(tmp_path / "ui-home-checkouts"), ) app.dependency_overrides[get_settings] = override_settings client = TestClient(app) try: response = client.post( "/repos", json={ "name": "Very Long Marketing Project Name", "url": str(source), }, ) assert response.status_code == 201 homepage = client.get("/ui") assert homepage.status_code == 200 assert homepage.text.index(">Registry<") < homepage.text.index(">Register Repository<") registry_panel = homepage.text[ homepage.text.index(">Registry<") : homepage.text.index(">Register Repository<") ] assert "short-dir" in registry_panel assert "Very Long Marketing Project Name" not in registry_panel assert "Discovery" not in registry_panel finally: app.dependency_overrides.clear() def test_ui_scope_page_presents_scope_md(): client = TestClient(app) response = client.get("/ui/scope") assert response.status_code == 200 assert "SCOPE.md" in response.text assert "Canonical scope summary for the repo-scoping repository." in response.text assert "scope.generate" in response.text assert "repo-scoping" 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_generates_diffs_and_writes_scope_md(tmp_path): source = tmp_path / "scope-repo" source.mkdir() def override_settings(): return Settings( database_path=str(tmp_path / "scope-api.sqlite3"), checkout_root=str(tmp_path / "checkouts"), ) app.dependency_overrides[get_settings] = override_settings client = TestClient(app) try: repository = client.post( "/repos", json={ "name": "Scope Repo", "url": str(source), "description": "Generates SCOPE.md through the API.", }, ).json() ability_id = client.post( f"/repos/{repository['id']}/abilities", json={ "name": "Maintain Repository Scope", "description": "Keeps repository utility understandable.", }, ).json()["id"] client.post( f"/repos/{repository['id']}/capabilities", json={ "ability_id": ability_id, "name": "Generate SCOPE.md", "description": "Renders SCOPE.md from approved characteristics.", "primary_class": "api", "attributes": ["scope", "generation"], }, ) preview = client.get("/repos/scope-repo/scope") assert preview.status_code == 200 assert preview.headers["content-type"].startswith("text/markdown") assert "# SCOPE" in preview.text assert "title: Generate SCOPE.md" in preview.text diff = client.get("/repos/scope-repo/scope/diff") assert diff.status_code == 200 assert diff.json()["needs_update"] is True assert {section["status"] for section in diff.json()["sections"]} == {"missing"} write = client.post("/repos/scope-repo/scope/write") assert write.status_code == 200 assert write.json() == {"written": True, "path": str(source / "SCOPE.md")} assert (source / "SCOPE.md").read_text(encoding="utf-8").startswith("# SCOPE") current = client.get("/repos/scope-repo/scope/diff") assert current.status_code == 200 assert current.json()["needs_update"] is False assert {section["status"] for section in current.json()["sections"]} == {"ok"} empty = client.post( "/repos", json={ "name": "Empty Scope", "url": "https://example.test/empty-scope.git", "description": "No approved characteristics yet.", }, ).json() assert client.get("/repos/empty-scope/scope").status_code == 404 remote = client.post( "/repos", json={ "name": "Remote Scope", "url": "https://example.test/remote-scope.git", "description": "Has no known local checkout path.", }, ).json() remote_ability = client.post( f"/repos/{remote['id']}/abilities", json={"name": "Remote Scope Generation"}, ).json()["id"] client.post( f"/repos/{remote['id']}/capabilities", json={ "ability_id": remote_ability, "name": "Generate Remote SCOPE.md", }, ) assert client.post("/repos/remote-scope/scope/write").status_code == 409 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", ) database_path = str(tmp_path / "api-analysis.sqlite3") def override_settings(): return Settings( database_path=database_path, 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"] add_candidate_capability( database_path, repository_id, run["analysis_run"]["id"], "Describe Frontend Stack", ) 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() add_candidate_capability( database_path, repository_id, run["analysis_run"]["id"], "Describe Frontend Stack", ) 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 / "SCOPE.md").write_text( "# SCOPE\n\n## One-liner\n\nUI Repo owns the status reporting scope.\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 assert 'SCOPE' not 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 assert ( f'SCOPE' in detail_response.text ) repo_scope_response = client.get(f"/ui/repos/{repository_id}/scope") assert repo_scope_response.status_code == 200 assert ( f'repo' in repo_scope_response.text ) assert "Canonical scope summary for the repo repository." in repo_scope_response.text assert "UI Repo owns the status reporting scope." in repo_scope_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 "Run Diagnostics" in run_detail.text assert "Analysis completed with reviewable results." in run_detail.text assert "Candidate Graph" in run_detail.text assert "1 abilities" in run_detail.text assert "1 capabilities" in run_detail.text assert "2 features" in run_detail.text assert "8 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 pending_candidate_listing = client.get( f"/ui/repos/{repository_id}/elements", params={ "scope": "candidate", "analysis_run_id": first_run_id, "type": "features", }, ) assert pending_candidate_listing.status_code == 200 assert "Accept" in pending_candidate_listing.text pending_support_listing = client.get( f"/ui/repos/{repository_id}/elements", params={ "scope": "candidate", "analysis_run_id": first_run_id, "type": "supports", }, ) assert pending_support_listing.status_code == 200 assert "Candidate Supports" in pending_support_listing.text assert "Accept" in pending_support_listing.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 ( f'repo' in approved_detail.text ) assert "Approved Characteristics" in approved_detail.text assert "Approved Characteristic Tree" in approved_detail.text assert 'UI Repo' in approved_detail.text assert "scope" in approved_detail.text assert "Evidence supporting this capability" in approved_detail.text assert "1 scope" in approved_detail.text assert "supports" in approved_detail.text assert "1 abilities" in approved_detail.text assert "1 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 "1 candidate capabilities" in approved_detail.text assert "2 candidate features" in approved_detail.text assert "8 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 "Elements" in approved_detail.text assert "q=Report+Service+Status" in approved_detail.text scope_listing = client.get( f"/ui/repos/{repository_id}/elements", params={"scope": "all", "type": "scopes", "entry_filter": "approved"}, ) assert scope_listing.status_code == 200 assert ( f'repo' in scope_listing.text ) assert f'UI Repo' in scope_listing.text assert ( f"/ui/repos/{repository_id}/elements?scope=all&entry_filter=approved&type=abilities" in approved_detail.text ) assert ( f"/ui/repos/{repository_id}/elements?scope=all&entry_filter=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": "all", "entry_filter": "approved", "type": "capabilities"}, ) assert approved_listing.status_code == 200 assert "Registry Capabilities" in approved_listing.text assert "Entry" in approved_listing.text assert "Approved only" in approved_listing.text assert "Expose Repository Interface" in approved_listing.text assert "Save" in approved_listing.text assert "Delete" in approved_listing.text scope_listing = client.get( f"/ui/repos/{repository_id}/elements", params={"scope": "all", "entry_filter": "approved", "type": "scopes"}, ) assert scope_listing.status_code == 200 assert "Registry Scopes" in scope_listing.text assert "Save" in scope_listing.text support_listing = client.get( f"/ui/repos/{repository_id}/elements", params={"scope": "all", "entry_filter": "approved", "type": "supports"}, ) assert support_listing.status_code == 200 assert "Registry Supports" in support_listing.text assert "supports capability" in support_listing.text assert "references source" in support_listing.text combined_listing = client.get( f"/ui/repos/{repository_id}/elements", params={ "scope": "all", "analysis_run_id": first_run_id, "type": "features", }, ) assert combined_listing.status_code == 200 assert "Registry Features" in combined_listing.text assert "Approved and candidate" in combined_listing.text assert ">approved<" in combined_listing.text assert ">candidate: approved<" in combined_listing.text approved_map = client.get(f"/repos/{repository_id}/ability-map").json() approved_capability = approved_map["abilities"][0]["capabilities"][0] tune_response = client.post( f"/ui/repos/{repository_id}/capabilities/{approved_capability['id']}/edit", data={ "name": "Expose Tuned Repository Interface", "description": approved_capability["description"], "inputs": ", ".join(approved_capability["inputs"]), "outputs": ", ".join(approved_capability["outputs"]), "confidence": str(approved_capability["confidence"]), }, follow_redirects=False, ) assert tune_response.status_code == 303 tuned_listing = client.get( f"/ui/repos/{repository_id}/elements", params={"scope": "all", "entry_filter": "approved", "type": "capabilities"}, ) assert "Expose Tuned Repository Interface" in tuned_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 8 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 final_map = client.get(f"/repos/{repository_id}/ability-map").json() feature_id = final_map["abilities"][0]["capabilities"][0]["features"][0]["id"] delete_feature_response = client.post( f"/ui/repos/{repository_id}/features/{feature_id}/delete", follow_redirects=False, ) assert delete_feature_response.status_code == 303 deleted_feature_listing = client.get( f"/ui/repos/{repository_id}/elements", params={"scope": "approved", "type": "features"}, ) assert f"/ui/repos/{repository_id}/features/{feature_id}/delete" not in ( deleted_feature_listing.text ) finally: app.dependency_overrides.clear() def test_ui_element_listing_hides_rejected_candidates_by_default(tmp_path): source = tmp_path / "repo" source.mkdir() (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-rejected.sqlite3"), checkout_root=str(tmp_path / "ui-rejected-checkouts"), ) app.dependency_overrides[get_settings] = override_settings client = TestClient(app) try: repository_response = client.post( "/repos", json={"name": "Rejected UI", "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"] candidate_graph = client.get( f"/repos/{repository_id}/analysis-runs/{run_id}/candidate-graph" ).json() candidate_feature = candidate_graph["abilities"][0]["capabilities"][0][ "features" ][0] candidate_feature_id = candidate_feature["id"] candidate_feature_name = candidate_feature["name"] reject_response = client.post( f"/ui/repos/{repository_id}/analysis-runs/{run_id}" f"/candidate-features/{candidate_feature_id}/reject", follow_redirects=False, ) assert reject_response.status_code == 303 repository_detail = client.get(f"/ui/repos/{repository_id}") assert repository_detail.status_code == 200 assert "0 candidate features" in repository_detail.text default_listing = client.get( f"/ui/repos/{repository_id}/elements", params={ "scope": "all", "analysis_run_id": run_id, "type": "features", }, ) assert default_listing.status_code == 200 assert "Hide rejected" in default_listing.text assert candidate_feature_name not in default_listing.text rejected_listing = client.get( f"/ui/repos/{repository_id}/elements", params={ "scope": "all", "analysis_run_id": run_id, "type": "features", "candidate_status_filter": "all", }, ) assert rejected_listing.status_code == 200 assert candidate_feature_name in rejected_listing.text assert "candidate: rejected" in rejected_listing.text assert "Accept" in rejected_listing.text finally: app.dependency_overrides.clear() def test_ui_analysis_run_diagnostics_explain_failures_and_empty_results(tmp_path): empty_source = tmp_path / "empty-repo" empty_source.mkdir() def override_settings(): return Settings( database_path=str(tmp_path / "ui-diagnostics.sqlite3"), checkout_root=str(tmp_path / "ui-diagnostics-checkouts"), ) app.dependency_overrides[get_settings] = override_settings client = TestClient(app) try: create_response = client.post( "/repos", json={ "url": str(empty_source), "name": "Diagnostics Repo", "description": "Used for UI diagnostics.", "branch": "main", }, ) assert create_response.status_code == 201 repository_id = create_response.json()["id"] failed_run = client.post( f"/ui/repos/{repository_id}/analysis-runs", data={ "source_path": str(tmp_path / "missing-repo"), "use_llm_assistance": "", }, follow_redirects=False, ) assert failed_run.status_code == 303 failed_detail = client.get(failed_run.headers["location"]) assert failed_detail.status_code == 200 assert "Run Diagnostics" in failed_detail.text assert "Analysis failed." in failed_detail.text assert "Verify the local path or Git URL" in failed_detail.text empty_run = client.post( f"/ui/repos/{repository_id}/analysis-runs", data={ "source_path": "", "use_llm_assistance": "", }, follow_redirects=False, ) assert empty_run.status_code == 303 empty_detail = client.get(empty_run.headers["location"]) assert empty_detail.status_code == 200 assert "No observed facts were found." in empty_detail.text assert "0 facts" in empty_detail.text assert "0 abilities" in empty_detail.text finally: app.dependency_overrides.clear() def test_ui_analysis_run_diagnostics_warn_when_only_baseline_context_exists(tmp_path): source = tmp_path / "dependency-only-ui" source.mkdir() (source / "README.md").write_text("# Dependency Only\nUses libraries.\n", encoding="utf-8") (source / "requirements.txt").write_text("fastapi\npytest\n", encoding="utf-8") def override_settings(): return Settings( database_path=str(tmp_path / "ui-baseline-diagnostics.sqlite3"), checkout_root=str(tmp_path / "ui-baseline-diagnostics-checkouts"), ) app.dependency_overrides[get_settings] = override_settings client = TestClient(app) try: repository = client.post( "/repos", json={ "url": str(source), "name": "Dependency Only UI", "description": "Used for baseline diagnostics.", }, ).json() run = client.post( f"/ui/repos/{repository['id']}/analysis-runs", data={"source_path": "", "use_llm_assistance": ""}, follow_redirects=False, ) detail = client.get(run.headers["location"]) assert detail.status_code == 200 assert "No domain capabilities were produced." in detail.text assert "only baseline context or weak documentation was available" in detail.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 "Latest Candidate Graph" in repository_detail.text assert "Rebuild Characteristics" in repository_detail.text assert "Use Approved Registry" not in repository_detail.text finally: app.dependency_overrides.clear() def test_rebuild_characteristics_endpoint_dry_run_and_confirm(tmp_path): source = tmp_path / "rebuild-api" source.mkdir() (source / "README.md").write_text("# Rebuild API\nReports status.\n", encoding="utf-8") (source / "app.py").write_text('@app.get("/status")\ndef status():\n return {}\n', encoding="utf-8") def override_settings(): return Settings( database_path=str(tmp_path / "rebuild-api.sqlite3"), checkout_root=str(tmp_path / "rebuild-api-checkouts"), ) app.dependency_overrides[get_settings] = override_settings client = TestClient(app) try: repository_id = client.post( "/repos", json={"name": "Rebuild API", "url": str(source)}, ).json()["id"] run_id = client.post( f"/repos/{repository_id}/analysis-runs", json={"source_path": str(source), "use_llm_assistance": False}, ).json()["analysis_run"]["id"] approve_response = client.post( f"/repos/{repository_id}/analysis-runs/{run_id}/candidate-graph/approve", json={"notes": "approve before rebuild"}, ) assert approve_response.status_code == 200 dry_run_response = client.post( f"/repos/{repository_id}/characteristics/rebuild", json={ "dry_run": True, "source_path": str(source), "use_llm_assistance": False, }, ) assert dry_run_response.status_code == 200 dry_run = dry_run_response.json() assert dry_run["cleared_approved"] is False assert dry_run["previous_counts"]["abilities"] == 1 assert dry_run["previous_ids"]["abilities"] rejected_response = client.post( f"/repos/{repository_id}/characteristics/rebuild", json={ "dry_run": False, "confirm": False, "source_path": str(source), "use_llm_assistance": False, }, ) assert rejected_response.status_code == 400 confirmed_response = client.post( f"/repos/{repository_id}/characteristics/rebuild", json={ "dry_run": False, "confirm": True, "source_path": str(source), "use_llm_assistance": False, }, ) assert confirmed_response.status_code == 200 confirmed = confirmed_response.json() assert confirmed["cleared_approved"] is True assert client.get(f"/repos/{repository_id}/ability-map").json()["abilities"] == [] finally: app.dependency_overrides.clear() def test_ui_rebuild_characteristics_form_runs_dry_run(tmp_path): source = tmp_path / "ui-rebuild" source.mkdir() (source / "README.md").write_text("# UI Rebuild\nReports status.\n", encoding="utf-8") def override_settings(): return Settings( database_path=str(tmp_path / "ui-rebuild.sqlite3"), checkout_root=str(tmp_path / "ui-rebuild-checkouts"), ) app.dependency_overrides[get_settings] = override_settings client = TestClient(app) try: repository_id = client.post( "/repos", json={"name": "UI Rebuild", "url": str(source)}, ).json()["id"] response = client.post( f"/ui/repos/{repository_id}/characteristics/rebuild", data={ "source_path": str(source), "dry_run": "1", "use_llm_assistance": "", "use_cached_checkout": "", }, follow_redirects=False, ) assert response.status_code == 303 assert f"/ui/repos/{repository_id}/analysis-runs/" in response.headers["location"] detail = client.get(response.headers["location"]) assert "dry_run_rebuild_characteristics_from_scratch" in 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 Characteristic Tuning" in detail_response.text assert "Add Capability Support" in detail_response.text assert "Supported characteristic kind" in detail_response.text assert "Reference kind" in detail_response.text assert "Save Scope" in detail_response.text scope_response = client.post( f"{repository_path}/scope/edit", data={ "name": "Manual Repository Scope", "description": "Root product scope.", "confidence": "0.91", }, follow_redirects=False, ) assert scope_response.status_code == 303 assert client.get(f"/repos/{repository_id}/ability-map").json()["scope"][ "name" ] == "Manual Repository Scope" ability_response = client.post( f"{repository_path}/abilities", data={ "name": "Manual Ability", "description": "Curated by hand.", "primary_class": "repository-intelligence", "attributes": "review, curation", "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", "primary_class": "review", "attributes": "ui, workflow", "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", "primary_class": "api", "attributes": "integration, review", "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", "reference_kind": "source", "strength": "medium", }, follow_redirects=False, ) assert evidence_response.status_code == 303 upward_support_response = client.post( f"{repository_path}/evidence", data={ "capability_id": str(capability_id), "target_kind": "feature", "target_id": "999", "type": "review-smell", "reference": "Manual Capability", "reference_kind": "capability", "strength": "weak", }, follow_redirects=False, ) assert upward_support_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 "supports capability" in detail_response.text assert "references source" in detail_response.text assert "downward support" in detail_response.text assert "upward support review" 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.", "primary_class": "workflow-automation", "attributes": "manual, curation", "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", "primary_class": "decisioning", "attributes": "workflow, review", "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", "primary_class": "ui", "attributes": "api, review", "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", "target_kind": "capability", "target_id": str(capability_id), "reference_kind": "feature", "reference_id": str(feature_id), "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 "Classification Quality Feedback" in detail_response.text assert "classification-primary" in detail_response.text assert "1 capabilities" in detail_response.text assert "1 features" in detail_response.text assert "2 supports" in detail_response.text assert "0 facts" in detail_response.text assert "tests/test_manual.py" in detail_response.text assert f"references feature #{feature_id}" in detail_response.text assert "downward support" in detail_response.text ability_map = client.get(f"/repos/{repository_id}/ability-map").json() edited_ability = ability_map["abilities"][0] edited_capability = edited_ability["capabilities"][0] edited_feature = edited_capability["features"][0] assert edited_ability["primary_class"] == "workflow-automation" assert edited_ability["attributes"] == ["manual", "curation"] assert edited_capability["primary_class"] == "decisioning" assert edited_capability["attributes"] == ["workflow", "review"] assert edited_feature["primary_class"] == "ui" assert edited_feature["attributes"] == ["api", "review"] filtered_feature_listing = client.get( f"/ui/repos/{repository_id}/elements", params={ "scope": "all", "entry_filter": "approved", "type": "features", "class_filter": "ui", "attribute_filter": "api", }, ) assert filtered_feature_listing.status_code == 200 assert "Attribute" in filtered_feature_listing.text assert "Repository" in filtered_feature_listing.text assert "Facts" in filtered_feature_listing.text assert "Edited Manual API" in filtered_feature_listing.text assert "1 of 1 shown" in filtered_feature_listing.text classification_gap_response = client.post( f"{repository_path}/expectation-gaps", data={ "expected_type": "classification-support", "expected_name": "Feature references another feature too broadly", "source": "human", "notes": "Same-level support should be reviewed.", }, follow_redirects=False, ) assert classification_gap_response.status_code == 303 detail_response = client.get(repository_path) assert "Feature references another feature too broadly" in detail_response.text assert "classification-support" in detail_response.text upward_support_listing = client.get( f"/ui/repos/{repository_id}/elements", params={ "scope": "all", "entry_filter": "approved", "type": "supports", "support_orientation_filter": "upward support review", }, ) assert upward_support_listing.status_code == 200 assert "Support orientation" in upward_support_listing.text assert "1 of 2 shown" in upward_support_listing.text assert "review-smell" in upward_support_listing.text assert "tests/test_manual.py" not in upward_support_listing.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 "empty-profile-ui" 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", ) database_path = str(tmp_path / "api-relink.sqlite3") def override_settings(): return Settings( database_path=database_path, 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"] add_candidate_capability( database_path, repository_id, run_id, "Review Target Capability", ) 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", ) database_path = str(tmp_path / "api-merge.sqlite3") def override_settings(): return Settings( database_path=database_path, 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"] add_candidate_capability( database_path, repository_id, run_id, "Review Target Capability", ) 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()