Coverage for archdocs/features/messaging_queue/renderer.py: 100%

1import typing 

2 

3from archdocs import diagram_model 

4from archdocs.features.messaging_queue.const import BrokerFlow, MQFeatures 

5 

6 

7def _render_flow_edges( 

8 service_node: diagram_model.DiagramNode, 

9 one_flow: BrokerFlow, 

10 /, 

11) -> tuple[diagram_model.DiagramEdge, ...]: 

12 broker_node: typing.Final = diagram_model.build_diagram_node( 

13 one_flow.broker_name, 

14 one_flow.broker_name, 

15 diagram_model.NodeGroup.messaging_and_tasks, 

16 ) 

17 all_edges: typing.Final[list[diagram_model.DiagramEdge]] = [] 

18 if one_flow.consumes: 

19 all_edges.append( 

20 diagram_model.DiagramEdge( 

21 source_node=broker_node, 

22 target_node=service_node, 

23 edge_label=", ".join(one_flow.consumed_topics), 

24 ), 

25 ) 

26 if one_flow.produces: 

27 all_edges.append( 

28 diagram_model.DiagramEdge( 

29 source_node=service_node, 

30 target_node=broker_node, 

31 edge_label=", ".join(one_flow.produced_topics), 

32 ), 

33 ) 

34 return tuple(all_edges) 

35 

36 

37def render_mq_features( 

38 service_node: diagram_model.DiagramNode, 

39 features_to_draw: MQFeatures, 

40 /, 

41) -> tuple[diagram_model.DiagramEdge, ...]: 

42 return tuple( 

43 one_edge 

44 for one_flow in features_to_draw.broker_flows 

45 for one_edge in _render_flow_edges(service_node, one_flow) 

46 )