Coverage for tests/integration/__init__.py: 100%

1"""Integration Test Setup""" 

2 

3import logging 

4import pytest 

5from es_testbed import PlanBuilder, TestBed 

6 

7 

8class TestAny: 

9 """ 

10 Test Any index or data_stream 

11 

12 Set this up by setting class variables, as below. 

13 """ 

14 

15 sstier = 'hot' 

16 kind = 'data_stream' 

17 roll = False 

18 repo_test = False 

19 ilm = { 

20 'enabled': False, 

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

22 'forcemerge': False, 

23 'max_num_segments': 1, 

24 } 

25 logger = logging.getLogger(__name__) 

26 

27 @pytest.fixture(scope="class") 

28 def tb(self, client, settings, skip_no_repo): 

29 """TestBed setup/teardown""" 

30 skip_no_repo(self.repo_test) 

31 cfg = settings( 

32 plan_type=self.kind, 

33 rollover_alias=self.roll, 

34 ilm=self.ilm, 

35 sstier=self.sstier, 

36 ) 

37 theplan = PlanBuilder(settings=cfg).plan 

38 teebee = TestBed(client, plan=theplan) 

39 teebee.setup() 

40 yield teebee 

41 teebee.teardown() 

42 

43 def test_entity_count(self, entity_count, entitymgr, tb): 

44 """Count the number of entities (index or data_stream)""" 

45 assert len(entitymgr(tb).entity_list) == entity_count(self.kind) 

46 

47 def test_name(self, actual_rollover, rollovername, tb): 

48 """ 

49 Verify the name of a rollover alias or data_stream 

50 

51 Will still return True if not a rollover or data_stream enabled test 

52 """ 

53 assert actual_rollover(tb) == rollovername(tb.plan) 

54 

55 def test_first_index(self, actual_index, first, index_name, tb): 

56 """Assert that the first index matches the expected name""" 

57 expected = index_name(which=first, plan=tb.plan, tier=self.sstier) 

58 actual = actual_index(tb, first) 

59 assert actual == expected 

60 

61 def test_last_index(self, actual_index, index_name, last, tb): 

62 """Assert that the last index matches the expected name""" 

63 expected = index_name(which=last, plan=tb.plan, tier=self.sstier) 

64 actual = actual_index(tb, last) 

65 assert actual == expected 

66 

67 def test_write_index(self, tb, actual_write_index, write_index_name): 

68 """ 

69 Test that the write index or current data_stream target is correct 

70 

71 Will still return True if not a rollover or data_stream enabled test 

72 """ 

73 expected = write_index_name(plan=tb.plan, tier=self.sstier) 

74 self.logger.debug('PLAN: %s', tb.plan) 

75 self.logger.debug('expected: %s', expected) 

76 actual = actual_write_index(tb) 

77 self.logger.debug('actual: %s', actual) 

78 assert actual == expected 

79 

80 def test_index_template(self, tb, components, get_template, template): 

81 """Assert that the index template and component templates are correct""" 

82 assert tb.componentmgr.entity_list == components 

83 assert tb.templatemgr.last == template 

84 result = get_template(tb.client) 

85 assert len(result) == 1 

86 assert result[0]['index_template']['composed_of'] == components