Coverage for tests/unit/test_utils.py: 100%

1"""Test functions in es_testbed.helpers.utils""" 

2 

3# pylint: disable=missing-function-docstring,redefined-outer-name 

4import typing as t 

5import pytest 

6from dotmap import DotMap 

7from es_testbed.defaults import ( 

8 ilmhot, 

9 ilmwarm, 

10 ilmcold, 

11 ilmfrozen, 

12 ilmdelete, 

13) 

14from es_testbed.exceptions import TestbedMisconfig 

15from es_testbed.helpers.utils import build_ilm_phase, build_ilm_policy, doc_gen 

16 

17FMAP: t.Dict[str, t.Dict] = { 

18 'hot': ilmhot(), 

19 'warm': ilmwarm(), 

20 'cold': ilmcold(), 

21 'frozen': ilmfrozen(), 

22 'delete': ilmdelete(), 

23} 

24REPO: str = 'repo' 

25TIERS: t.Sequence[str] = ['hot', 'warm', 'cold', 'frozen', 'delete'] 

26TREPO: t.Dict[str, t.Union[str, None]] = { 

27 'hot': None, 

28 'warm': None, 

29 'cold': REPO, 

30 'frozen': REPO, 

31 'delete': None, 

32} 

33 

34 

35def searchable(repo: str = None) -> t.Union[t.Dict[str, t.Dict[str, str]], None]: 

36 if repo: 

37 return {'searchable_snapshot': {'snapshot_repository': repo}} 

38 return {} 

39 

40 

41def forcemerge( 

42 fm: bool = False, mns: int = 1 

43) -> t.Union[t.Dict[str, t.Dict[str, str]], None]: 

44 if fm: 

45 return {'forcemerge': {'max_num_segments': mns}} 

46 return {} 

47 

48 

49@pytest.fixture 

50def tiertestval(): 

51 def _tiertestval(tier: str, repo: str = None, fm: bool = False, mns: int = 1): 

52 retval = {tier: FMAP[tier]} 

53 retval[tier]['actions'].update(searchable(repo)) 

54 retval[tier]['actions'].update(forcemerge(fm=fm, mns=mns)) 

55 return retval 

56 

57 return _tiertestval 

58 

59 

60@pytest.fixture 

61def builtphase(): 

62 def _builtphase(tier: str, repo: str = None, fm: bool = False, mns: int = 1): 

63 return build_ilm_phase( 

64 tier, actions=forcemerge(fm=fm, mns=mns), repository=repo 

65 ) 

66 

67 return _builtphase 

68 

69 

70def test_build_ilm_phase_defaults(builtphase, tiertestval): 

71 for tier in TIERS: 

72 assert builtphase(tier, repo=TREPO[tier]) == tiertestval(tier, repo=TREPO[tier]) 

73 

74 

75def test_build_ilm_phase_add_action(): 

76 expected = {'foo': 'bar'} 

77 tier = 'warm' 

78 assert build_ilm_phase(tier, actions=expected)[tier]['actions'] == expected 

79 

80 

81def test_build_ilm_phase_fail_repo(builtphase): 

82 with pytest.raises(TestbedMisconfig): 

83 builtphase('cold', repo=None) 

84 

85 

86# This allows me to run multiple testing scenarios in the same test space 

87def test_build_ilm_policy(tiertestval): 

88 # 3 tests building different ILM policies with different tiers 

89 tgroups = [ 

90 ['hot', 'delete'], 

91 ['hot', 'frozen', 'delete'], 

92 ['hot', 'cold', 'delete'], 

93 ] 

94 # Each tier group corresponds to a forcemerge plan by list index, with each index a 

95 # tuple for forcemerge True/False and max_num_segment count 

96 fmerge = [(False, 0), (False, 0), (True, 3)] 

97 for idx, tgrp in enumerate(tgroups): # Iterate over testing scenarios 

98 phases = {} # Build out the phase dict for each scenario 

99 fm, mns = fmerge[idx] # Extract whether to use forcemerge by index/tuple 

100 for tier in tgrp: # Iterate over each tier in the testing scenario 

101 phases.update( 

102 tiertestval(tier, repo=TREPO[tier]) 

103 ) # Update with values per tier 

104 if fm: # If we're doing forcemerge 

105 phases['hot']['actions'].update( 

106 forcemerge(fm=fm, mns=mns) 

107 ) # Update the hot tier 

108 # To keep the line more readable, build the kwargs as a dict first 

109 kwargs = {'repository': REPO, 'forcemerge': fm, 'max_num_segments': mns} 

110 # Then pass it as **kwargs 

111 assert build_ilm_policy(tgrp, **kwargs) == {'phases': phases} 

112 # Our policy is easier to build at the last minute rather than constantly 

113 # passing dict['phases']['tier'] 

114 

115 

116def test_build_ilm_policy_fail_repo(): 

117 with pytest.raises(TestbedMisconfig): 

118 build_ilm_policy(['hot', 'frozen'], repository=None) 

119 

120 

121@pytest.fixture 

122def fieldmatch(): 

123 def _fieldmatch(val: str, num: int): 

124 return f'{val}{num}' 

125 

126 return _fieldmatch 

127 

128 

129def test_doc_gen_matching(fieldmatch): 

130 i = 0 

131 for res in doc_gen(count=3, start_at=0, match=True): 

132 doc = DotMap(res) 

133 tests = [ 

134 (doc.message, 'message'), 

135 (doc.nested.key, 'nested'), 

136 (doc.deep.l1.l2.l3, 'deep'), 

137 ] 

138 for test in tests: 

139 dm, val = test 

140 assert dm == fieldmatch(val, i) 

141 i += 1