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

1import typing 

2 

3from archdocs import diagram_model 

4from archdocs.features.task_queues.const import TaskQueueFeatures 

5 

6 

7_TASK_WORKER_NODE: typing.Final = diagram_model.build_diagram_node( 

8 "TaskQueue_Worker", 

9 "Task workers", 

10 diagram_model.NodeGroup.messaging_and_tasks, 

11) 

12 

13 

14def render_task_queue_features( 

15 service_node: diagram_model.DiagramNode, 

16 features_to_draw: TaskQueueFeatures, 

17 /, 

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

19 if not features_to_draw.queues_used: 

20 return () 

21 

22 properties_on_arrow: typing.Final = ", ".join( 

23 filter( 

24 None, 

25 [ 

26 ", ".join(sorted(features_to_draw.queues_used)), 

27 "workers" if features_to_draw.has_workers else "", 

28 ", ".join(sorted(features_to_draw.brokers_detected)) if features_to_draw.brokers_detected else "", 

29 ], 

30 ), 

31 ) 

32 

33 return ( 

34 diagram_model.DiagramEdge( 

35 source_node=service_node, 

36 target_node=_TASK_WORKER_NODE, 

37 edge_label=f"Tasks ({properties_on_arrow})", 

38 ), 

39 )