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

1import dataclasses 

2import typing 

3 

4from archdocs.features.kubernetes import const, values, values_lookup 

5 

6 

7type _KindNames = tuple[str, ...] 

8type _ConfigurationPaths = tuple[tuple[values.ValuePath, str, str], ...] 

9 

10_NAME_KEY: typing.Final = "name" 

11_TRUE_VALUES: typing.Final = frozenset(("true", "yes", "on")) 

12_CONFIGURATION_SOURCE_PATHS: typing.Final[_ConfigurationPaths] = ( 

13 (("configMapRef", _NAME_KEY), const.CONFIG_MAP_KIND, const.ENVIRONMENT_ATTACHMENT), 

14 (("configMapKeyRef", _NAME_KEY), const.CONFIG_MAP_KIND, const.ENVIRONMENT_ATTACHMENT), 

15 (("existingConfigMap",), const.CONFIG_MAP_KIND, const.ENVIRONMENT_ATTACHMENT), 

16 (("secretRef", _NAME_KEY), const.SECRET_KIND, const.ENVIRONMENT_ATTACHMENT), 

17 (("secretKeyRef", _NAME_KEY), const.SECRET_KIND, const.ENVIRONMENT_ATTACHMENT), 

18 (("existingSecret",), const.SECRET_KIND, const.ENVIRONMENT_ATTACHMENT), 

19 (("configMap", _NAME_KEY), const.CONFIG_MAP_KIND, const.VOLUME_ATTACHMENT), 

20 (("secret", "secretName"), const.SECRET_KIND, const.VOLUME_ATTACHMENT), 

21) 

22 

23 

24def _read_toggle( 

25 all_values: values.ManifestValues, 

26 toggle_path: values.ValuePath, 

27 kind_name: str, 

28 all_kinds: _KindNames, 

29 /, 

30) -> bool: 

31 toggle_value: typing.Final = values_lookup.read_first_value(all_values, toggle_path) 

32 if toggle_value: 

33 return toggle_value.lower() in _TRUE_VALUES 

34 return kind_name in all_kinds 

35 

36 

37def _read_amounts(all_values: values.ManifestValues, /, *resource_keys: str) -> const.ResourceAmounts: 

38 return const.ResourceAmounts( 

39 requested_amount=values_lookup.read_first_value( 

40 all_values, 

41 *[("resources", "requests", one_resource_key) for one_resource_key in resource_keys], 

42 ), 

43 limited_amount=values_lookup.read_first_value( 

44 all_values, 

45 *[("resources", "limits", one_resource_key) for one_resource_key in resource_keys], 

46 ), 

47 ) 

48 

49 

50def _read_traffic(all_values: values.ManifestValues, all_kinds: _KindNames, /) -> const.TrafficFeatures: 

51 ingress_enabled: typing.Final = _read_toggle(all_values, ("ingress", "enabled"), const.INGRESS_KIND, all_kinds) 

52 return const.TrafficFeatures( 

53 ingress_enabled=ingress_enabled, 

54 ingress_hosts=values_lookup.read_values(all_values, ("hosts", "host"), ("rules", "host")) 

55 if ingress_enabled 

56 else (), 

57 ingress_tls_enabled=ingress_enabled 

58 and values_lookup.has_any_block(all_values, ("ingress", "tls"), ("spec", "tls")), 

59 service_type=values_lookup.read_first_value(all_values, ("service", "type"), ("spec", "type")), 

60 service_port=values_lookup.read_int_value(all_values, ("service", "port"), ("ports", "port")), 

61 ) 

62 

63 

64def _read_scaling(all_values: values.ManifestValues, all_kinds: _KindNames, /) -> const.ScalingFeatures: 

65 plain_scaling: typing.Final = const.ScalingFeatures( 

66 workload_kind=next((one_kind for one_kind in all_kinds if one_kind in const.WORKLOAD_KINDS), ""), 

67 replica_count=values_lookup.read_int_value(all_values, ("replicaCount",), ("spec", "replicas")), 

68 ) 

69 if not _read_toggle(all_values, ("autoscaling", "enabled"), const.AUTOSCALER_KIND, all_kinds): 

70 return plain_scaling 

71 return dataclasses.replace( 

72 plain_scaling, 

73 min_replicas=values_lookup.read_int_value(all_values, ("minReplicas",)), 

74 max_replicas=values_lookup.read_int_value(all_values, ("maxReplicas",)), 

75 target_cpu_utilization=values_lookup.read_int_value( 

76 all_values, 

77 ("targetCPUUtilizationPercentage",), 

78 ("target", "averageUtilization"), 

79 ), 

80 ) 

81 

82 

83def _read_resources(all_values: values.ManifestValues, all_kinds: _KindNames, /) -> const.ResourceFeatures: 

84 return const.ResourceFeatures( 

85 cpu_amounts=_read_amounts(all_values, "cpu"), 

86 memory_amounts=_read_amounts(all_values, "memory"), 

87 gpu_amounts=_read_amounts(all_values, *const.GPU_RESOURCE_KEYS), 

88 persistence_enabled=_read_toggle(all_values, ("persistence", "enabled"), const.VOLUME_CLAIM_KIND, all_kinds) 

89 or values_lookup.has_any_block(all_values, ("volumeClaimTemplates",)), 

90 persistence_size=values_lookup.read_first_value(all_values, ("persistence", "size"), ("requests", "storage")), 

91 ) 

92 

93 

94def _read_configuration(all_values: values.ManifestValues, /) -> tuple[const.ConfigurationSource, ...]: 

95 return tuple( 

96 dict.fromkeys( 

97 const.ConfigurationSource( 

98 source_kind=one_source_kind, 

99 source_name=one_source_name, 

100 attachment_kind=one_attachment_kind, 

101 ) 

102 for one_value_path, one_source_kind, one_attachment_kind in _CONFIGURATION_SOURCE_PATHS 

103 for one_source_name in values_lookup.read_values(all_values, one_value_path) 

104 ), 

105 ) 

106 

107 

108def find_kubernetes_features(raw_source: str) -> const.KubernetesFeatures: 

109 all_values: typing.Final = values.read_manifest_values(raw_source) 

110 all_kinds: typing.Final = values_lookup.read_values(all_values, ("kind",)) 

111 return const.KubernetesFeatures( 

112 traffic_features=_read_traffic(all_values, all_kinds), 

113 scaling_features=_read_scaling(all_values, all_kinds), 

114 resource_features=_read_resources(all_values, all_kinds), 

115 configuration_sources=_read_configuration(all_values), 

116 )