Mirrors/ansible-playbook-grapher
2
0
Fork 0
mirror of https://github.com/haidaraM/ansible-playbook-grapher synced 2024-11-10 14:14:19 +00:00
Code Issues Projects Releases Packages Wiki Activity

fix: Remove counter from parser. It's the renderer's problem

Browse source
This commit is contained in:
Mohamed El Mouctar HAIDARA 2021-09-23 00:55:32 +02:00
parent 80630c7b1f
commit a86aa6f9c1
4 changed files with 50 additions and 45 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
ansibleplaybookgrapher/graph.py
View file

@ -15,7 +15,7 @@ class Node(ABC):
self.id = node_id
def __str__(self):
return f"{type(self).__name__}: {self.label} => {self.id}"
return f"{type(self).__name__}: label='{self.label}',id='{self.id}'"
def __eq__(self, other):
return self.id == other.id
@ -102,7 +102,7 @@ class PlaybookNode(CompositeNode):
:param edge_label:
:return:
"""
edge = EdgeNode(edge_label, self, play)
edge = EdgeNode(self, play, edge_label)
self.add_node("plays", edge)
return edge
@ -147,7 +147,7 @@ class EdgeNode(CompositeNode):
An edge between two nodes. It's a special case of composite node with only one composition with one element
"""
def __init__(self, node_label: str, source: Node, destination: Node, node_id: str = None):
def __init__(self, source: Node, destination: Node, node_label: str = "", node_id: str = None):
"""
:param node_label: The edge label

37
ansibleplaybookgrapher/parser.py
View file

@ -61,27 +61,26 @@ class BaseParser(ABC):
self.display.warning(ansible_error)
return data
def _add_task(self, task: Task, loop_counter: int, task_vars: Dict, node_type: str,
parent_node: CompositeNode) -> bool:
def _add_task(self, task: Task, task_vars: Dict, node_type: str, parent_node: CompositeNode) -> bool:
"""
Include the task in the graph.
:return: True if the task has been included, false otherwise
"""
self.display.vv(f"Adding {node_type} '{task.get_name()}' to the graph with counter {loop_counter}")
if not task.evaluate_tags(only_tags=self.tags, skip_tags=self.skip_tags, all_vars=task_vars):
self.display.vv(f"The task '{task.get_name()}' is skipped due to the tags.")
return False
task_edge_name = str(loop_counter)
self.display.vv(f"Adding {node_type} '{task.get_name()}' to the graph")
edge_label = ""
if len(task.when) > 0:
when = "".join(map(str, task.when))
task_edge_name += " [when: " + when + "]"
edge_label += " [when: " + when + "]"
task_name = clean_name(f"[{node_type}] " + self.template(task.get_name(), task_vars))
edge_node = EdgeNode(task_edge_name, parent_node, TaskNode(task_name, generate_id(f"{node_type}_")))
edge_node = EdgeNode(parent_node, TaskNode(task_name, generate_id(f"{node_type}_")), edge_label)
parent_node.add_node(target_composition=f"{node_type}s", node=edge_node)
return True
@ -127,7 +126,7 @@ class PlaybookParser(BaseParser):
"""
# loop through the plays
for play_counter, play in enumerate(self.playbook.get_plays(), 1):
for play in self.playbook.get_plays():
# the load basedir is relative to the playbook path
if play._included_path is not None:
@ -138,13 +137,13 @@ class PlaybookParser(BaseParser):
play_vars = self.variable_manager.get_vars(play)
play_hosts = [h.get_name() for h in self.inventory_manager.get_hosts(self.template(play.hosts, play_vars))]
play_name = "Play #{}: {} ({})".format(play_counter, clean_name(play.get_name()), len(play_hosts))
play_name = "Play: {} ({})".format(clean_name(play.get_name()), len(play_hosts))
play_name = self.template(play_name, play_vars)
self.display.banner("Parsing " + play_name)
play_node = PlayNode(play_name, hosts=play_hosts)
self.playbook_root_node.add_play(play_node, str(play_counter))
self.playbook_root_node.add_play(play_node, "")
# loop through the pre_tasks
self.display.v("Parsing pre_tasks...")
@ -155,7 +154,7 @@ class PlaybookParser(BaseParser):
# loop through the roles
self.display.v("Parsing roles...")
for role_counter, role in enumerate(play.get_roles(), 1):
for role in play.get_roles():
# Don't insert tasks from ``import/include_role``, preventing duplicate graphing
if role.from_include:
continue
@ -169,8 +168,7 @@ class PlaybookParser(BaseParser):
role_node = RoleNode(clean_name(role.get_name()))
# edge from play to role
play_node.add_node("roles",
EdgeNode(str(role_counter + len(play_node.pre_tasks)), play_node, role_node))
play_node.add_node("roles", EdgeNode(play_node, role_node))
if self.include_role_tasks:
# loop through the tasks of the roles
@ -234,9 +232,8 @@ class PlaybookParser(BaseParser):
f"An 'include_role' found. Including tasks from the role '{task_or_block.args['name']}'")
role_node = RoleNode(task_or_block.args['name'])
parent_nodes[-1].add_node("roles",
EdgeNode(str(parent_nodes[-1].total_length + 1), parent_nodes[-1],
role_node))
# TODO: add support for conditional (when) for include_role in the edge label
parent_nodes[-1].add_node("roles", EdgeNode(parent_nodes[-1], role_node))
if self.include_role_tasks:
# If we have an include_role and we want to include role tasks, the parent node now becomes
@ -257,8 +254,8 @@ class PlaybookParser(BaseParser):
self.display.warning(
f"Unable to translate the include task '{task_or_block.get_name()}' due to an undefined variable: {str(e)}. "
"Some variables are available only during the execution of the playbook.")
self._add_task(task=task_or_block, loop_counter=parent_nodes[-1].total_length + 1,
task_vars=task_vars, node_type=node_type, parent_node=parent_nodes[-1])
self._add_task(task=task_or_block, task_vars=task_vars, node_type=node_type,
parent_node=parent_nodes[-1])
continue
data = self.data_loader.load_from_file(include_file)
@ -291,5 +288,5 @@ class PlaybookParser(BaseParser):
# skipping
continue
self._add_task(task=task_or_block, loop_counter=parent_nodes[-1].total_length + 1, task_vars=play_vars,
node_type=node_type, parent_node=parent_nodes[-1])
self._add_task(task=task_or_block, task_vars=play_vars, node_type=node_type,
parent_node=parent_nodes[-1])

44
ansibleplaybookgrapher/renderer.py
View file

@ -44,7 +44,7 @@ class GraphvizRenderer:
graph_attr=graph_attr or GraphvizRenderer.DEFAULT_GRAPH_ATTR,
edge_attr=edge_attr or GraphvizRenderer.DEFAULT_EDGE_ATTR)
def _add_task(self, graph: GraphvizCustomDigraph, parent_node: Node, edge: EdgeNode, color: str,
def _add_task(self, graph: GraphvizCustomDigraph, parent_node: Node, edge: EdgeNode, color: str, task_counter: int,
shape: str = "octagon"):
"""
Add a task in the given graph
@ -58,8 +58,9 @@ class GraphvizRenderer:
destination_node = edge.destination
graph.node(destination_node.id, label=destination_node.label, shape=shape, id=destination_node.id,
tooltip=destination_node.label)
graph.edge(parent_node.id, destination_node.id, label=edge.label, color=color, fontcolor=color, style="bold",
id=edge.id, tooltip=edge.label, labeltooltip=edge.label)
edge_label = f"{task_counter} {edge.label}"
graph.edge(parent_node.id, destination_node.id, label=edge_label, color=color, fontcolor=color, style="bold",
id=edge.id, tooltip=edge_label, labeltooltip=edge_label)
def _convert_to_graphviz(self):
"""
@ -69,7 +70,7 @@ class GraphvizRenderer:
# root node
self.graphviz.node(self.playbook_node.label, style="dotted", id="root_node")
for play_edge in self.playbook_node.plays:
for play_counter, play_edge in enumerate(self.playbook_node.plays, 1):
# noinspection PyTypeChecker
play = play_edge.destination # type: PlayNode
with self.graphviz.subgraph(name=play.label) as play_subgraph:
@ -79,37 +80,44 @@ class GraphvizRenderer:
self.graphviz.node(play.id, id=play.id, label=play.label, style="filled", shape="box", color=color,
fontcolor=play_font_color, tooltip=play_tooltip)
# edge from root node to play
playbook_to_play_label = f"{play_counter} {play_edge.label}"
self.graphviz.edge(self.playbook_node.label, play.id, id=play_edge.id, style="bold",
label=play_edge.label, color=color, fontcolor=color, tooltip=play_edge.label,
labeltooltip=play_edge.label)
label=playbook_to_play_label, color=color, fontcolor=color,
tooltip=playbook_to_play_label, labeltooltip=playbook_to_play_label)
# pre_tasks
for pre_task_edge in play.pre_tasks:
self._add_task(play_subgraph, play, pre_task_edge, color)
for pre_task_counter, pre_task_edge in enumerate(play.pre_tasks, 1):
self._add_task(graph=play_subgraph, parent_node=play, edge=pre_task_edge, color=color,
task_counter=pre_task_counter)
# roles
for role_edge in play.roles:
for role_counter, role_edge in enumerate(play.roles, 1):
# noinspection PyTypeChecker
role = role_edge.destination # type: RoleNode
role_edge_label = f"{role_counter + len(play.pre_tasks)} {role_edge.label}"
with self.graphviz.subgraph(name=role.label, node_attr={}) as role_subgraph:
# from play to role
role_subgraph.node(role.id, id=role.id, label=f"[role] {role.label}", tooltip=role.label)
play_subgraph.edge(play.id, role.id, label=role_edge.label, color=color, fontcolor=color,
style="bold", id=role_edge.id, tooltip=role_edge.label,
labeltooltip=role_edge.label)
play_subgraph.edge(play.id, role.id, label=role_edge_label, color=color, fontcolor=color,
style="bold", id=role_edge.id, tooltip=role_edge_label,
labeltooltip=role_edge_label)
# role tasks
for role_task_edge in role.tasks:
self._add_task(role_subgraph, role, role_task_edge, color)
for role_task_counter, role_task_edge in enumerate(role.tasks, 1):
self._add_task(role_subgraph, role, role_task_edge, color, task_counter=role_task_counter)
# tasks
for task_edge in play.tasks:
self._add_task(play_subgraph, play, task_edge, color)
for task_counter, task_edge in enumerate(play.tasks, 1):
self._add_task(play_subgraph, play, task_edge, color,
task_counter=len(play.pre_tasks) + len(play.roles) + task_counter)
# post_tasks
for post_task_edge in play.post_tasks:
self._add_task(play_subgraph, play, post_task_edge, color)
for post_task_counter, post_task_edge in enumerate(play.post_tasks, 1):
self._add_task(play_subgraph, play, post_task_edge, color,
task_counter=len(play.pre_tasks) + len(play.roles) + len(
play.tasks) + post_task_counter)
def render(self, output_filename: str, save_dot_file=False) -> str:
"""

8
tests/test_graph.py
View file

@ -9,21 +9,21 @@ def test_links_structure():
play = PlayNode("composite_node")
role = RoleNode("my_role_1")
edge_role = EdgeNode("from play to role", play, role)
edge_role = EdgeNode(play, role, "from play to role")
play.add_node("roles", edge_role)
# play -> role -> edge 1 -> task 1
task_1 = TaskNode("task 1")
edge_1 = EdgeNode("from role1 to task 1", role, task_1)
edge_1 = EdgeNode(role, task_1, "from role1 to task 1")
role.add_node("tasks", edge_1)
# play -> role -> edge 2 -> task 2
task_2 = TaskNode("task 2")
edge_2 = EdgeNode("from role1 to task 2", role, task_2)
edge_2 = EdgeNode(role, task_2, "from role1 to task 2")
role.add_node("tasks", edge_2)
# play -> edge 3 -> task 3
task_3 = TaskNode("task 3")
edge_3 = EdgeNode("from play to task 3", play, task_3)
edge_3 = EdgeNode(play, task_3, "from play to task 3")
play.add_node("tasks", edge_3)
all_links = play.links_structure()