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

1"""ILM Policy Entity Manager Class""" 

2 

3import typing as t 

4from .entitymgr import EntityMgr 

5from ..exceptions import ResultNotExpected 

6from ..helpers.es_api import exists, put_ilm 

7from ..helpers.utils import build_ilm_policy, getlogger 

8 

9if t.TYPE_CHECKING: 9 ↛ 10line 9 didn't jump to line 10, because the condition on line 9 was never true

10 from elasticsearch8 import Elasticsearch 

11 from dotmap import DotMap 

12 

13# pylint: disable=missing-docstring 

14 

15 

16class IlmMgr(EntityMgr): 

17 kind = 'ilm' 

18 listname = 'ilm_policies' 

19 

20 def __init__( 

21 self, 

22 client: t.Union['Elasticsearch', None] = None, 

23 plan: t.Union['DotMap', None] = None, 

24 autobuild: t.Optional[bool] = True, 

25 ): 

26 self.logger = getlogger('es_testbed.IlmMgr') 

27 super().__init__(client=client, plan=plan, autobuild=autobuild) 

28 

29 def get_policy(self) -> t.Dict: 

30 """Return the configured ILM policy""" 

31 d = self.plan.ilm 

32 kwargs = { 

33 'tiers': d.tiers, 

34 'forcemerge': d.forcemerge, 

35 'max_num_segments': d.max_num_segments, 

36 'repository': self.plan.repository, 

37 } 

38 return build_ilm_policy(**kwargs) 

39 

40 def setup(self) -> None: 

41 """Setup the entity manager""" 

42 if self.plan.ilm.enabled: 

43 self.plan.ilm.policy = self.get_policy() 

44 put_ilm(self.client, self.name, policy=self.plan.ilm.policy) 

45 # Verify existence 

46 if not exists(self.client, 'ilm', self.name): 46 ↛ 47line 46 didn't jump to line 47, because the condition on line 46 was never true

47 raise ResultNotExpected( 

48 f'Unable to verify creation of ilm policy {self.name}' 

49 ) 

50 # This goes first because the length of entity_list determines the suffix 

51 self.appender(self.name) 

52 self.logger.info('Successfully created ILM policy: %s', self.last) 

53 else: 

54 self.appender(None) # This covers self.plan.ilm_policies[-1] 

55 self.logger.info('No ILM policy created.') 

56 self.success = True