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

1import typing 

2 

3from archdocs import diagram_model 

4from archdocs.features.kubernetes import const 

5 

6 

7_INGRESS_ENTRYPOINT_LABEL: typing.Final = "Ingress" 

8_SECURE_SCHEME: typing.Final = "HTTPS" 

9_PLAIN_SCHEME: typing.Final = "HTTP" 

10_PERSISTENT_VOLUME_NODE: typing.Final = diagram_model.build_diagram_node( 

11 "PersistentVolume", 

12 "Persistent volume", 

13 diagram_model.NodeGroup.data_stores, 

14) 

15 

16 

17def _render_amounts(resource_name: str, amounts_to_draw: const.ResourceAmounts, /) -> str: 

18 if not amounts_to_draw.requested_amount: 

19 return f"{resource_name} up to {amounts_to_draw.limited_amount}" if amounts_to_draw.limited_amount else "" 

20 if amounts_to_draw.limited_amount in {"", amounts_to_draw.requested_amount}: 

21 return f"{resource_name} {amounts_to_draw.requested_amount}" 

22 return f"{resource_name} {amounts_to_draw.requested_amount}-{amounts_to_draw.limited_amount}" 

23 

24 

25def render_kubernetes_node_annotations(features_to_draw: const.KubernetesFeatures) -> tuple[str, ...]: 

26 scaling_to_draw: typing.Final = features_to_draw.scaling_features 

27 resources_to_draw: typing.Final = features_to_draw.resource_features 

28 return tuple( 

29 filter( 

30 None, 

31 ( 

32 "" if scaling_to_draw.workload_kind == const.DEFAULT_WORKLOAD_KIND else scaling_to_draw.workload_kind, 

33 f"replicas {scaling_to_draw.replica_count}" if scaling_to_draw.replica_count else "", 

34 f"HPA {scaling_to_draw.min_replicas}-{scaling_to_draw.max_replicas}" 

35 if scaling_to_draw.max_replicas 

36 else "", 

37 f"target CPU {scaling_to_draw.target_cpu_utilization}%" 

38 if scaling_to_draw.target_cpu_utilization 

39 else "", 

40 _render_amounts("cpu", resources_to_draw.cpu_amounts), 

41 _render_amounts("RAM", resources_to_draw.memory_amounts), 

42 _render_amounts("GPU", resources_to_draw.gpu_amounts), 

43 ), 

44 ), 

45 ) 

46 

47 

48def _render_entrypoint_label(traffic_to_draw: const.TrafficFeatures, /) -> str: 

49 if traffic_to_draw.ingress_enabled: 

50 return _INGRESS_ENTRYPOINT_LABEL 

51 if traffic_to_draw.service_type not in const.EXPOSED_SERVICE_TYPES: 

52 return "" 

53 if not traffic_to_draw.service_port: 

54 return traffic_to_draw.service_type 

55 return f"{traffic_to_draw.service_type}, port {traffic_to_draw.service_port}" 

56 

57 

58def _render_traffic_edges( 

59 service_node: diagram_model.DiagramNode, 

60 traffic_to_draw: const.TrafficFeatures, 

61 /, 

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

63 if traffic_to_draw.ingress_enabled and traffic_to_draw.ingress_hosts: 

64 ingress_scheme: typing.Final = _SECURE_SCHEME if traffic_to_draw.ingress_tls_enabled else _PLAIN_SCHEME 

65 return tuple( 

66 diagram_model.DiagramEdge( 

67 source_node=diagram_model.EXTERNAL_CLIENT_NODE, 

68 target_node=service_node, 

69 edge_label=f"{ingress_scheme} {one_ingress_host}", 

70 ) 

71 for one_ingress_host in traffic_to_draw.ingress_hosts 

72 ) 

73 entrypoint_label: typing.Final = _render_entrypoint_label(traffic_to_draw) 

74 if not entrypoint_label: 

75 return () 

76 return ( 

77 diagram_model.DiagramEdge( 

78 source_node=diagram_model.EXTERNAL_CLIENT_NODE, 

79 target_node=service_node, 

80 edge_label=entrypoint_label, 

81 ), 

82 ) 

83 

84 

85def _render_configuration_edges( 

86 service_node: diagram_model.DiagramNode, 

87 all_configuration_sources: tuple[const.ConfigurationSource, ...], 

88 /, 

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

90 return tuple( 

91 diagram_model.DiagramEdge( 

92 source_node=diagram_model.build_diagram_node( 

93 f"{one_source.source_kind} {one_source.source_name}", 

94 f"{one_source.source_name} ({one_source.source_kind})", 

95 diagram_model.NodeGroup.configuration, 

96 ), 

97 target_node=service_node, 

98 edge_label=one_source.attachment_kind, 

99 ) 

100 for one_source in all_configuration_sources 

101 ) 

102 

103 

104def _render_storage_edges( 

105 service_node: diagram_model.DiagramNode, 

106 resources_to_draw: const.ResourceFeatures, 

107 /, 

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

109 if not resources_to_draw.persistence_enabled: 

110 return () 

111 storage_label: typing.Final = " ".join( 

112 filter(None, (const.VOLUME_ATTACHMENT, resources_to_draw.persistence_size)), 

113 ) 

114 return ( 

115 diagram_model.DiagramEdge( 

116 source_node=service_node, 

117 target_node=_PERSISTENT_VOLUME_NODE, 

118 edge_label=storage_label, 

119 ), 

120 ) 

121 

122 

123def render_kubernetes_features( 

124 service_node: diagram_model.DiagramNode, 

125 features_to_draw: const.KubernetesFeatures, 

126 /, 

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

128 return ( 

129 *_render_traffic_edges(service_node, features_to_draw.traffic_features), 

130 *_render_configuration_edges(service_node, features_to_draw.configuration_sources), 

131 *_render_storage_edges(service_node, features_to_draw.resource_features), 

132 )