Coverage for /Users/buh/.pyenv/versions/3.12.2/envs/es-testbed/lib/python3.12/site-packages/es_testbed/defaults.py: 100%

1"""Default values and constants""" 

2 

3import typing as t 

4 

5# pylint: disable=missing-function-docstring 

6 

7EPILOG: str = 'Learn more at https://github.com/untergeek/es-testbed' 

8 

9HELP_OPTIONS: dict = {'help_option_names': ['-h', '--help']} 

10 

11ARGSCLASSES: list = ['IlmBuilder', 'IlmExplain', 'TestPlan'] 

12 

13COLD_PREFIX: str = 'restored-' 

14FROZEN_PREFIX: str = 'partial-' 

15 

16SS_PREFIX: t.Dict[str, str] = {'cold': COLD_PREFIX, 'frozen': FROZEN_PREFIX} 

17 

18MAPPING: dict = { 

19 'properties': { 

20 '@timestamp': {'type': 'date'}, 

21 'message': {'type': 'keyword'}, 

22 'number': {'type': 'long'}, 

23 'nested': {'properties': {'key': {'type': 'keyword'}}}, 

24 'deep': { 

25 'properties': { 

26 'l1': { 

27 'properties': {'l2': {'properties': {'l3': {'type': 'keyword'}}}} 

28 } 

29 } 

30 }, 

31 } 

32} 

33 

34NAMEMAPPER: t.Dict[str, str] = { 

35 'index': 'idx', 

36 'data_stream': 'ds', 

37 'component': 'cmp', 

38 'ilm': 'ilm', 

39 'template': 'tmpl', 

40 'snapshot': 'snp', 

41} 

42 

43PAUSE_DEFAULT: str = '0.25' 

44PAUSE_ENVVAR: str = 'ES_TESTBED_PAUSE' 

45 

46PLURALMAP: t.Dict[str, str] = { 

47 'ilm': 'ILM Policie', 

48 'index': 'indice', 

49} 

50 

51TESTPLAN: dict = { 

52 'type': 'indices', 

53 'prefix': 'es-testbed', 

54 'repository': None, 

55 'rollover_alias': None, 

56 'ilm': { 

57 'enabled': False, 

58 'tiers': ['hot', 'delete'], 

59 'forcemerge': False, 

60 'max_num_segments': 1, 

61 }, 

62 'defaults': { 

63 'entity_count': 3, 

64 'docs': 10, 

65 'match': True, 

66 'searchable': None, 

67 }, 

68 'entities': [], 

69} 

70 

71TIER: dict = { 

72 'hot': {'pref': 'data_hot,data_content'}, 

73 'warm': {'pref': 'data_warm,data_hot,data_content'}, 

74 'cold': { 

75 'pref': 'data_cold,data_warm,data_hot,data_content', 

76 'prefix': 'restored', 

77 'storage': 'full_copy', 

78 }, 

79 'frozen': { 

80 'pref': 'data_frozen', 

81 'prefix': 'partial', 

82 'storage': 'shared_cache', 

83 }, 

84} 

85 

86TIMEOUT_DEFAULT: str = '30' 

87TIMEOUT_ENVVAR: str = 'ES_TESTBED_TIMEOUT' 

88 

89IlmPhase: t.TypeAlias = t.Dict[ 

90 str, t.Union[str, t.Dict[str, str], t.Dict[str, t.Dict[str, t.Dict[str, str]]]] 

91] 

92 

93 

94def ilmhot() -> IlmPhase: 

95 return {'actions': {'rollover': {'max_primary_shard_size': '1gb', 'max_age': '1d'}}} 

96 

97 

98def ilmwarm() -> IlmPhase: 

99 return {'min_age': '2d', 'actions': {}} 

100 

101 

102def ilmcold() -> IlmPhase: 

103 return {'min_age': '3d', 'actions': {}} 

104 

105 

106def ilmfrozen() -> IlmPhase: 

107 return {'min_age': '4d', 'actions': {}} 

108 

109 

110def ilmdelete() -> IlmPhase: 

111 return {'min_age': '5d', 'actions': {'delete': {}}} 

112 

113 

114def ilm_phase(tier): 

115 """Return the default phase step based on 'tier'""" 

116 phase_map = { 

117 'hot': ilmhot(), 

118 'warm': ilmwarm(), 

119 'cold': ilmcold(), 

120 'frozen': ilmfrozen(), 

121 'delete': ilmdelete(), 

122 } 

123 return {tier: phase_map[tier]} 

124 

125 

126def ilm_force_merge(max_num_segments=1): 

127 """Return an ILM policy force merge action block using max_num_segments""" 

128 return {'forcemerge': {'max_num_segments': max_num_segments}}