feat(graph): pass design_dir from router to build_panorama
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
50e38c64a6
commit
e386a59336
|
|
@ -40,17 +40,17 @@ def _get_or_trigger_scan(project_id: str):
|
||||||
@router.get("")
|
@router.get("")
|
||||||
def get_graph(project_id: str):
|
def get_graph(project_id: str):
|
||||||
"""Build and return the full panorama graph for a project."""
|
"""Build and return the full panorama graph for a project."""
|
||||||
_project_service.get_project(project_id) # Ensure project exists (raises 404)
|
project = _project_service.get_project(project_id) # Ensure project exists (raises 404)
|
||||||
scan_result = _get_or_trigger_scan(project_id)
|
scan_result = _get_or_trigger_scan(project_id)
|
||||||
view = _graph_service.build_panorama(scan_result)
|
view = _graph_service.build_panorama(scan_result, design_dir=project.design_dir)
|
||||||
return asdict(view)
|
return asdict(view)
|
||||||
|
|
||||||
|
|
||||||
@router.get("/nodes/{node_id}/neighbors")
|
@router.get("/nodes/{node_id}/neighbors")
|
||||||
def get_neighbors(project_id: str, node_id: str):
|
def get_neighbors(project_id: str, node_id: str):
|
||||||
"""Return the subgraph of neighbors for a given node."""
|
"""Return the subgraph of neighbors for a given node."""
|
||||||
_project_service.get_project(project_id) # Ensure project exists (raises 404)
|
project = _project_service.get_project(project_id) # Ensure project exists (raises 404)
|
||||||
scan_result = _get_or_trigger_scan(project_id)
|
scan_result = _get_or_trigger_scan(project_id)
|
||||||
view = _graph_service.build_panorama(scan_result)
|
view = _graph_service.build_panorama(scan_result, design_dir=project.design_dir)
|
||||||
neighbors = _graph_service.get_neighbors(view, node_id)
|
neighbors = _graph_service.get_neighbors(view, node_id)
|
||||||
return asdict(neighbors)
|
return asdict(neighbors)
|
||||||
|
|
|
||||||
|
|
@ -27,8 +27,8 @@ def graph_service():
|
||||||
return GraphService()
|
return GraphService()
|
||||||
|
|
||||||
|
|
||||||
def test_panorama_has_groups(graph_service, scan_result):
|
def test_panorama_has_groups(graph_service, scan_result, design_dir):
|
||||||
view = graph_service.build_panorama(scan_result)
|
view = graph_service.build_panorama(scan_result, design_dir=design_dir)
|
||||||
group_ids = {g.id for g in view.groups}
|
group_ids = {g.id for g in view.groups}
|
||||||
assert "business" in group_ids
|
assert "business" in group_ids
|
||||||
assert "application" in group_ids
|
assert "application" in group_ids
|
||||||
|
|
@ -37,42 +37,42 @@ def test_panorama_has_groups(graph_service, scan_result):
|
||||||
assert "cross-layer" in group_ids
|
assert "cross-layer" in group_ids
|
||||||
|
|
||||||
|
|
||||||
def test_panorama_has_capability_nodes(graph_service, scan_result):
|
def test_panorama_has_capability_nodes(graph_service, scan_result, design_dir):
|
||||||
view = graph_service.build_panorama(scan_result)
|
view = graph_service.build_panorama(scan_result, design_dir=design_dir)
|
||||||
cap_nodes = [n for n in view.nodes if n.type == "capability"]
|
cap_nodes = [n for n in view.nodes if n.type == "capability"]
|
||||||
assert len(cap_nodes) > 0
|
assert len(cap_nodes) > 0
|
||||||
assert all(n.group_id == "business" for n in cap_nodes)
|
assert all(n.group_id == "business" for n in cap_nodes)
|
||||||
|
|
||||||
|
|
||||||
def test_panorama_has_module_nodes(graph_service, scan_result):
|
def test_panorama_has_module_nodes(graph_service, scan_result, design_dir):
|
||||||
view = graph_service.build_panorama(scan_result)
|
view = graph_service.build_panorama(scan_result, design_dir=design_dir)
|
||||||
mod_nodes = [n for n in view.nodes if n.type == "module"]
|
mod_nodes = [n for n in view.nodes if n.type == "module"]
|
||||||
assert len(mod_nodes) > 0
|
assert len(mod_nodes) > 0
|
||||||
assert all(n.group_id == "application" for n in mod_nodes)
|
assert all(n.group_id == "application" for n in mod_nodes)
|
||||||
|
|
||||||
|
|
||||||
def test_panorama_has_entity_nodes(graph_service, scan_result):
|
def test_panorama_has_entity_nodes(graph_service, scan_result, design_dir):
|
||||||
view = graph_service.build_panorama(scan_result)
|
view = graph_service.build_panorama(scan_result, design_dir=design_dir)
|
||||||
ent_nodes = [n for n in view.nodes if n.type == "entity"]
|
ent_nodes = [n for n in view.nodes if n.type == "entity"]
|
||||||
assert len(ent_nodes) > 0
|
assert len(ent_nodes) > 0
|
||||||
assert all(n.group_id == "data" for n in ent_nodes)
|
assert all(n.group_id == "data" for n in ent_nodes)
|
||||||
|
|
||||||
|
|
||||||
def test_panorama_has_edges(graph_service, scan_result):
|
def test_panorama_has_edges(graph_service, scan_result, design_dir):
|
||||||
view = graph_service.build_panorama(scan_result)
|
view = graph_service.build_panorama(scan_result, design_dir=design_dir)
|
||||||
assert len(view.edges) > 0
|
assert len(view.edges) > 0
|
||||||
relations = {e.relation for e in view.edges}
|
relations = {e.relation for e in view.edges}
|
||||||
assert "traces_to" in relations
|
assert "traces_to" in relations
|
||||||
|
|
||||||
|
|
||||||
def test_panorama_depends_on_edges(graph_service, scan_result):
|
def test_panorama_depends_on_edges(graph_service, scan_result, design_dir):
|
||||||
view = graph_service.build_panorama(scan_result)
|
view = graph_service.build_panorama(scan_result, design_dir=design_dir)
|
||||||
dep_edges = [e for e in view.edges if e.relation == "depends_on"]
|
dep_edges = [e for e in view.edges if e.relation == "depends_on"]
|
||||||
assert len(dep_edges) > 0
|
assert len(dep_edges) > 0
|
||||||
|
|
||||||
|
|
||||||
def test_neighbors_returns_subgraph(graph_service, scan_result):
|
def test_neighbors_returns_subgraph(graph_service, scan_result, design_dir):
|
||||||
view = graph_service.build_panorama(scan_result)
|
view = graph_service.build_panorama(scan_result, design_dir=design_dir)
|
||||||
# Use a known capability node
|
# Use a known capability node
|
||||||
neighbors = graph_service.get_neighbors(view, "CAP-PROJ-REG")
|
neighbors = graph_service.get_neighbors(view, "CAP-PROJ-REG")
|
||||||
assert len(neighbors.nodes) > 0
|
assert len(neighbors.nodes) > 0
|
||||||
|
|
@ -80,15 +80,15 @@ def test_neighbors_returns_subgraph(graph_service, scan_result):
|
||||||
assert len(neighbors.edges) > 0
|
assert len(neighbors.edges) > 0
|
||||||
|
|
||||||
|
|
||||||
def test_neighbors_unknown_node(graph_service, scan_result):
|
def test_neighbors_unknown_node(graph_service, scan_result, design_dir):
|
||||||
view = graph_service.build_panorama(scan_result)
|
view = graph_service.build_panorama(scan_result, design_dir=design_dir)
|
||||||
neighbors = graph_service.get_neighbors(view, "NONEXISTENT")
|
neighbors = graph_service.get_neighbors(view, "NONEXISTENT")
|
||||||
assert len(neighbors.nodes) == 0
|
assert len(neighbors.nodes) == 0
|
||||||
assert len(neighbors.edges) == 0
|
assert len(neighbors.edges) == 0
|
||||||
|
|
||||||
|
|
||||||
def test_graph_node_has_parent_field(graph_service, scan_result):
|
def test_graph_node_has_parent_field(graph_service, scan_result, design_dir):
|
||||||
view = graph_service.build_panorama(scan_result)
|
view = graph_service.build_panorama(scan_result, design_dir=design_dir)
|
||||||
for node in view.nodes:
|
for node in view.nodes:
|
||||||
assert hasattr(node, 'parent')
|
assert hasattr(node, 'parent')
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue
Block a user