coulomb/repo-scoping
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge-duplicates slice and did a first polish

Browse Source
This commit is contained in:
Bernd Worsch
2026-04-25 23:50:58 +02:00
parent 1d6d103bc2
commit 19d34efa37
6 changed files with 788 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

148
tests/test_registry_service.py
View File

@@ -480,6 +480,154 @@ def test_relink_candidate_feature_and_evidence_to_another_capability(tmp_path):
}
def test_merge_candidate_ability_moves_capabilities_to_target(tmp_path):
source = tmp_path / "repo"
source.mkdir()
(source / "README.md").write_text("# Merge Ability\n", encoding="utf-8")
(source / "app.py").write_text(
"from fastapi import FastAPI\n"
"app = FastAPI()\n"
'@app.get("/health")\n'
"def health():\n"
" return {}\n",
encoding="utf-8",
)
service = make_service(tmp_path)
repository = service.register_repository(name="Merge Ability", url=str(source))
summary = service.analyze_repository(repository.id)
graph = service.candidate_graph(repository.id, summary.analysis_run.id)
source_ability = graph.abilities[0]
with service.store.connect() as connection:
cursor = connection.execute(
"""
INSERT INTO candidate_abilities
(repository_id, analysis_run_id, name, description, confidence)
VALUES (?, ?, ?, ?, ?)
""",
(
repository.id,
summary.analysis_run.id,
"Merged Operational Ability",
"Preferred duplicate ability.",
0.83,
),
)
target_ability_id = int(cursor.lastrowid)
graph = service.merge_candidate_ability(
repository.id,
summary.analysis_run.id,
source_ability.id,
target_ability_id=target_ability_id,
)
ability_map = service.approve_candidate_graph(repository.id, summary.analysis_run.id)
merged_source = [ability for ability in graph.abilities if ability.id == source_ability.id][0]
target = [ability for ability in graph.abilities if ability.id == target_ability_id][0]
assert merged_source.status == "merged"
assert target.capabilities
assert [ability.name for ability in ability_map.abilities] == [
"Merged Operational Ability"
]
assert ability_map.abilities[0].capabilities
def test_merge_candidate_capability_moves_children_to_target(tmp_path):
source = tmp_path / "repo"
source.mkdir()
(source / "README.md").write_text("# Merge Capability\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("/health")\n'
"def health():\n"
" return {}\n",
encoding="utf-8",
)
service = make_service(tmp_path)
repository = service.register_repository(name="Merge Capability", url=str(source))
summary = service.analyze_repository(repository.id)
graph = service.candidate_graph(repository.id, summary.analysis_run.id)
source_capability = graph.abilities[0].capabilities[0]
target_capability = graph.abilities[0].capabilities[1]
graph = service.merge_candidate_capability(
repository.id,
summary.analysis_run.id,
source_capability.id,
target_capability_id=target_capability.id,
)
ability_map = service.approve_candidate_graph(repository.id, summary.analysis_run.id)
merged_source = [
capability
for ability in graph.abilities
for capability in ability.capabilities
if capability.id == source_capability.id
][0]
target = [
capability
for ability in graph.abilities
for capability in ability.capabilities
if capability.id == target_capability.id
][0]
assert merged_source.status == "merged"
assert target.features
assert [capability.name for capability in ability_map.abilities[0].capabilities] == [
target_capability.name
]
assert ability_map.abilities[0].capabilities[0].features
def test_merge_candidate_feature_and_evidence_omits_duplicate_leaves(tmp_path):
source = tmp_path / "repo"
source.mkdir()
(source / "README.md").write_text("# Merge Leaves\n", encoding="utf-8")
(source / "tests").mkdir()
(source / "tests" / "test_health.py").write_text(
"def test_health(): pass\n",
encoding="utf-8",
)
(source / "app.py").write_text(
"from fastapi import FastAPI\n"
"app = FastAPI()\n"
'@app.get("/health")\n'
"def health():\n"
" return {}\n"
'@app.get("/ready")\n'
"def ready():\n"
" return {}\n",
encoding="utf-8",
)
service = make_service(tmp_path)
repository = service.register_repository(name="Merge Leaves", url=str(source))
summary = service.analyze_repository(repository.id)
graph = service.candidate_graph(repository.id, summary.analysis_run.id)
capability = graph.abilities[0].capabilities[0]
service.merge_candidate_feature(
repository.id,
summary.analysis_run.id,
capability.features[1].id,
target_feature_id=capability.features[0].id,
)
service.merge_candidate_evidence(
repository.id,
summary.analysis_run.id,
capability.evidence[1].id,
target_evidence_id=capability.evidence[0].id,
)
ability_map = service.approve_candidate_graph(repository.id, summary.analysis_run.id)
approved_capability = ability_map.abilities[0].capabilities[0]
assert len(approved_capability.features) == len(capability.features) - 1
assert len(approved_capability.evidence) == len(capability.evidence) - 1
def test_analyze_repository_failure_is_recorded(tmp_path):
service = make_service(tmp_path)
repository = service.register_repository(

92
tests/test_web_api.py
View File

@@ -271,6 +271,7 @@ def test_ui_register_analyze_and_approve_loop(tmp_path):
run_detail = client.get(run_path)
assert run_detail.status_code == 200
assert "Candidate Graph" in run_detail.text
assert "ID " in run_detail.text
approve_response = client.post(
f"{run_path}/candidate-graph/approve",
@@ -282,6 +283,10 @@ def test_ui_register_analyze_and_approve_loop(tmp_path):
assert approved_detail.status_code == 200
assert "Approved Ability Map" in approved_detail.text
assert "Review UI Repo Repository Usefulness" in approved_detail.text
search_response = client.get("/ui/search", params={"q": "repository"})
assert search_response.status_code == 200
assert "UI Repo" in search_response.text
finally:
app.dependency_overrides.clear()
@@ -443,3 +448,90 @@ def test_api_relinks_candidate_feature_and_evidence(tmp_path):
assert relinked_capabilities[1]["evidence"][0]["id"] == evidence_id
finally:
app.dependency_overrides.clear()
def test_api_merges_candidate_capability_feature_and_evidence(tmp_path):
source = tmp_path / "repo"
source.mkdir()
(source / "README.md").write_text("# API Merge\n", encoding="utf-8")
(source / "requirements.txt").write_text("fastapi\n", encoding="utf-8")
(source / "tests").mkdir()
(source / "tests" / "test_status.py").write_text(
"def test_status(): pass\n",
encoding="utf-8",
)
(source / "app.py").write_text(
"from fastapi import FastAPI\n"
"app = FastAPI()\n"
'@app.get("/status")\n'
"def status():\n"
" return {}\n"
'@app.get("/ready")\n'
"def ready():\n"
" return {}\n",
encoding="utf-8",
)
def override_settings():
return Settings(
database_path=str(tmp_path / "api-merge.sqlite3"),
checkout_root=str(tmp_path / "api-merge-checkouts"),
)
app.dependency_overrides[get_settings] = override_settings
client = TestClient(app)
try:
repository_response = client.post(
"/repos",
json={"name": "API Merge", "url": str(source)},
)
repository_id = repository_response.json()["id"]
run_response = client.post(f"/repos/{repository_id}/analysis-runs", json={})
run_id = run_response.json()["analysis_run"]["id"]
graph_response = client.get(
f"/repos/{repository_id}/analysis-runs/{run_id}/candidate-graph"
)
capabilities = graph_response.json()["abilities"][0]["capabilities"]
source_capability = capabilities[0]
target_capability = capabilities[1]
feature_response = client.post(
f"/repos/{repository_id}/analysis-runs/{run_id}"
f"/candidate-features/{source_capability['features'][1]['id']}/merge",
json={
"target_feature_id": source_capability["features"][0]["id"],
"notes": "Duplicate route",
},
)
assert feature_response.status_code == 200
assert (
feature_response.json()["abilities"][0]["capabilities"][0]["features"][1][
"status"
]
== "merged"
)
evidence_response = client.post(
f"/repos/{repository_id}/analysis-runs/{run_id}"
f"/candidate-evidence/{source_capability['evidence'][1]['id']}/merge",
json={
"target_evidence_id": source_capability["evidence"][0]["id"],
"notes": "Duplicate evidence",
},
)
assert evidence_response.status_code == 200
capability_response = client.post(
f"/repos/{repository_id}/analysis-runs/{run_id}"
f"/candidate-capabilities/{source_capability['id']}/merge",
json={
"target_capability_id": target_capability["id"],
"notes": "Duplicate capability",
},
)
assert capability_response.status_code == 200
capabilities = capability_response.json()["abilities"][0]["capabilities"]
assert capabilities[0]["status"] == "merged"
assert capabilities[1]["features"]
finally:
app.dependency_overrides.clear()