fix: Handle group nodes in graph sorting (#3929)
* Fix: Handle group nodes in graph sorting - Added `get_root_of_group_node` function to identify the root of a group node. - Updated `sort_up_to_vertex` to use `get_root_of_group_node` for handling group nodes. - Modified `__filter_vertices` to pass `parent_node_map` to `sort_up_to_vertex`. * Refactor: Update NodeStatus component to handle group nodes and improve build status handling * [autofix.ci] apply automated fixes * Update type hint for parent_node_map in sort_up_to_vertex function --------- Co-authored-by: anovazzi1 <otavio2204@gmail.com> Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
parent
3ddd123b66
commit
b34a7c7f02
6 changed files with 75 additions and 14 deletions
|
|
@ -1885,7 +1885,10 @@ class Graph:
|
|||
|
||||
def __filter_vertices(self, vertex_id: str, is_start: bool = False):
|
||||
dictionaryized_graph = self.__to_dict()
|
||||
vertex_ids = sort_up_to_vertex(dictionaryized_graph, vertex_id, is_start)
|
||||
parent_node_map = {vertex.id: vertex.parent_node_id for vertex in self.vertices}
|
||||
vertex_ids = sort_up_to_vertex(
|
||||
graph=dictionaryized_graph, vertex_id=vertex_id, parent_node_map=parent_node_map, is_start=is_start
|
||||
)
|
||||
return [self.get_vertex(vertex_id) for vertex_id in vertex_ids]
|
||||
|
||||
def sort_vertices(
|
||||
|
|
|
|||
|
|
@ -246,17 +246,46 @@ def get_successors(graph: dict[str, dict[str, list[str]]], vertex_id: str) -> li
|
|||
if current_id in visited:
|
||||
continue
|
||||
visited.add(current_id)
|
||||
successors_result.append(current_id)
|
||||
if current_id != vertex_id:
|
||||
successors_result.append(current_id)
|
||||
stack.extend(graph[current_id]["successors"])
|
||||
return successors_result
|
||||
|
||||
|
||||
def sort_up_to_vertex(graph: dict[str, dict[str, list[str]]], vertex_id: str, is_start: bool = False) -> list[str]:
|
||||
def get_root_of_group_node(
|
||||
graph: dict[str, dict[str, list[str]]], vertex_id: str, parent_node_map: dict[str, str | None]
|
||||
) -> str:
|
||||
"""Returns the root of a group node."""
|
||||
if vertex_id in parent_node_map.values():
|
||||
# Get all vertices with vertex_id as their parent node
|
||||
child_vertices = [v_id for v_id, parent_id in parent_node_map.items() if parent_id == vertex_id]
|
||||
|
||||
# Now go through successors of the child vertices
|
||||
# and get the one that none of its successors is in child_vertices
|
||||
for child_id in child_vertices:
|
||||
successors = get_successors(graph, child_id)
|
||||
if not any(successor in child_vertices for successor in successors):
|
||||
return child_id
|
||||
|
||||
raise ValueError(f"Vertex {vertex_id} is not a top level vertex or no root vertex found")
|
||||
|
||||
|
||||
def sort_up_to_vertex(
|
||||
graph: dict[str, dict[str, list[str]]],
|
||||
vertex_id: str,
|
||||
parent_node_map: dict[str, str | None] | None = None,
|
||||
is_start: bool = False,
|
||||
) -> list[str]:
|
||||
"""Cuts the graph up to a given vertex and sorts the resulting subgraph."""
|
||||
try:
|
||||
stop_or_start_vertex = graph[vertex_id]
|
||||
except KeyError:
|
||||
raise ValueError(f"Vertex {vertex_id} not found into graph")
|
||||
if parent_node_map is None:
|
||||
raise ValueError("Parent node map is required to find the root of a group node")
|
||||
vertex_id = get_root_of_group_node(graph=graph, vertex_id=vertex_id, parent_node_map=parent_node_map)
|
||||
if vertex_id not in graph:
|
||||
raise ValueError(f"Vertex {vertex_id} not found into graph")
|
||||
stop_or_start_vertex = graph[vertex_id]
|
||||
|
||||
visited, excluded = set(), set()
|
||||
stack = [vertex_id]
|
||||
|
|
|
|||
|
|
@ -45,7 +45,7 @@ def test_get_successors_a(graph):
|
|||
|
||||
result = utils.get_successors(graph, vertex_id)
|
||||
|
||||
assert set(result) == {"A", "B", "D", "E", "F", "H", "G"}
|
||||
assert set(result) == {"B", "D", "E", "F", "H", "G"}
|
||||
|
||||
|
||||
def test_get_successors_z(graph):
|
||||
|
|
@ -53,7 +53,7 @@ def test_get_successors_z(graph):
|
|||
|
||||
result = utils.get_successors(graph, vertex_id)
|
||||
|
||||
assert set(result) == {"Z"}
|
||||
assert len(result) == 0
|
||||
|
||||
|
||||
def test_sort_up_to_vertex_n_is_start(graph):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue