diff --git a/apps/python-app/services/graph-service/src/graph_service/app.py b/apps/python-app/services/graph-service/src/graph_service/app.py index 872f86b..e834bbe 100644 --- a/apps/python-app/services/graph-service/src/graph_service/app.py +++ b/apps/python-app/services/graph-service/src/graph_service/app.py @@ -131,6 +131,19 @@ def artifact_referrers( return Response(content=queries.to_mermaid(subgraph), media_type="text/plain") return subgraph + @app.get("/artifacts/provenance") + def artifact_provenance( + family: str = Query(..., description="Repository family (trailing image name, e.g. python)."), + format: str = Query("json", pattern="^(json|cytoscape|mermaid)$"), + ) -> Any: + with reading() as store: + subgraph = queries.provenance(store, family) + if format == "cytoscape": + return queries.to_cytoscape(subgraph) + if format == "mermaid": + return Response(content=queries.to_mermaid(subgraph), media_type="text/plain") + return subgraph + @app.get("/repositories/tags/history") def tag_history(ref: str = Query(...), tag: str = Query(...)) -> dict[str, Any]: with reading() as store: diff --git a/apps/python-app/services/graph-service/tests/test_app.py b/apps/python-app/services/graph-service/tests/test_app.py index ab761aa..fdd0f89 100644 --- a/apps/python-app/services/graph-service/tests/test_app.py +++ b/apps/python-app/services/graph-service/tests/test_app.py @@ -230,6 +230,25 @@ def test_referrers_endpoint_accepts_no_rollup(client: TestClient) -> None: ) +def test_provenance_endpoint_json(client: TestClient) -> None: + body = client.get("/artifacts/provenance", params={"family": "python"}).json() + assert body["nodes"] and body["edges"] + assert any(n.get("state") == "present-golden" for n in body["nodes"]) + assert "built" in {e["type"] for e in body["edges"]} + + +def test_provenance_endpoint_mermaid(client: TestClient) -> None: + resp = client.get("/artifacts/provenance", params={"family": "python", "format": "mermaid"}) + assert resp.status_code == 200 + assert resp.text.strip().startswith("flowchart") + + +def test_provenance_endpoint_unknown_family_is_empty(client: TestClient) -> None: + body = client.get("/artifacts/provenance", params={"family": "does-not-exist"}).json() + assert body == {"nodes": [], "edges": []} + + + def test_depth_is_capped(client: TestClient, tmp_path: Path) -> None: # A max_depth of 1 must stop bases traversal at the first hop. root = _write_data_root(tmp_path / "capped", RECORDS)