Coverage for venv1 / lib / python3.10 / site-packages / combinatrix / test / unit / test_load_parameter_sets.py: 0%

1import unittest, os 

2from parameterized import parameterized 

3from combinatrix.testintegration import load_parameter_sets, rel2abs 

4from combinatrix.core import CombinatrixException 

5from combinatrix.test.fixtures.settings_bundle import SettingsBundleFactory 

6 

7EXCEPTIONS = { 

8 "CombinatrixException": CombinatrixException 

9} 

10 

11BUNDLE_PATHS = { 

12 "none": None, 

13 "not_exists": rel2abs(__file__, "path", "does", "not", "exist"), 

14 "exists": rel2abs(__file__, "..", "resources", "fixtures", "test_bundle") 

15} 

16 

17BUNDLE_NAMES = { 

18 "none": None, 

19 "not_exists": "not_exists", 

20 "exists": "test_bundle" 

21} 

22 

23NAME_FIELDS = { 

24 "none": None, 

25 "not_exists": "not_exists", 

26 "exists": "test_id" 

27} 

28 

29FILTERS = { 

30 "none": None, 

31 "single_filter": {"test_id": ["1"]}, 

32 "double_filter": {"test_id": ["1", "2", "3"], "field3": ["1"]} 

33} 

34 

35 

36def load_cases(): 

37 return load_parameter_sets(rel2abs(__file__, "..", "resources", "bundles", "load_parameter_sets"), 

38 "load_parameter_sets", "test_id", {"test_id": []}) 

39 

40 

41class TestLoadParameterSets(unittest.TestCase): 

42 

43 def setUp(self): 

44 super(TestLoadParameterSets, self).setUp() 

45 

46 def tearDown(self): 

47 SettingsBundleFactory.tear_down_bundle() 

48 super(TestLoadParameterSets, self).tearDown() 

49 

50 @parameterized.expand(load_cases) 

51 def test_load_parameter_sets(self, name, kwargs): 

52 bundle_path_arg = kwargs.get("bundle_path") 

53 bundle_name_arg = kwargs.get("bundle_name") 

54 name_field_arg = kwargs.get("name_field") 

55 filters_arg = kwargs.get("filters") 

56 settings_csv_arg = kwargs.get("settings_csv") 

57 settings_json_arg = kwargs.get("settings_json") 

58 matrix_arg = kwargs.get("matrix") 

59 settings_csv_age_arg = kwargs.get("settings_csv_age") 

60 settings_json_age_arg = kwargs.get("settings_json_age") 

61 matrix_age_arg = kwargs.get("matrix_age") 

62 raises_arg = kwargs.get("raises") 

63 

64 raises = EXCEPTIONS.get(raises_arg) 

65 bundle_path = BUNDLE_PATHS[bundle_path_arg] 

66 bundle_name = BUNDLE_NAMES[bundle_name_arg] 

67 name_field = NAME_FIELDS[name_field_arg] 

68 filters = FILTERS[filters_arg] 

69 

70 modified_order = {} 

71 if settings_csv_age_arg not in ["-"]: 

72 modified_order[int(settings_csv_age_arg)] = "settings_csv" 

73 if settings_json_age_arg not in ["-"]: 

74 modified_order[int(settings_json_age_arg)] = "settings_json" 

75 if matrix_age_arg not in ["-"]: 

76 modified_order[int(matrix_age_arg)] = "matrix" 

77 order_spec = list(modified_order.keys()) 

78 order_spec.sort() 

79 order = [] 

80 for o in order_spec: 

81 order.append(modified_order[o]) 

82 

83 bundle_info = SettingsBundleFactory.make_bundle(settings_csv_arg, settings_json_arg, matrix_arg, order) 

84 

85 if raises is not None: 

86 with self.assertRaises(raises): 

87 load_parameter_sets(bundle_path, bundle_name, name_field, filters) 

88 else: 

89 combinations = load_parameter_sets(bundle_path, bundle_name, name_field, filters) 

90 

91 # categorise the files into the three types: 

92 # 1. forward files - they should exist and have new modified dates 

93 # 2. current - it should exist and have an unmodified date 

94 # 3. past - may exist or not, and if exists should have an unmodified date 

95 forward = [] 

96 current = None 

97 past = [] 

98 

99 # seek through the ordered items, looking for the item that we would expect to be 

100 # the one the run picks up on as the primary artefact. This will be the earliest present 

101 # file 

102 seek = ["settings_csv", "settings_json", "matrix"] 

103 for o in seek: 

104 if bundle_info[o]["status"] == "present": 

105 current = o 

106 break 

107 

108 # sort the list of possible files into past and forward 

109 mode = "past" 

110 for stage in seek: 

111 if stage == current: 

112 mode = "forward" 

113 continue 

114 if mode == "past": 

115 past.append(stage) 

116 elif mode == "forward": 

117 forward.append(stage) 

118 

119 # now run the checks on each of these bins 

120 for p in past: 

121 assert not os.path.exists(bundle_info[p]["path"]) 

122 

123 assert os.path.exists(bundle_info[current]["path"]) 

124 mtime = os.path.getmtime(bundle_info[current]["path"]) 

125 assert mtime == bundle_info[current]["mtime"] 

126 

127 for f in forward: 

128 assert os.path.exists(bundle_info[f]["path"]) 

129 mtime = os.path.getmtime(bundle_info[f]["path"]) 

130 assert mtime > bundle_info[f]["mtime"] 

131 

132 if filters_arg == "none": 

133 assert len(combinations) == 8 

134 elif filters_arg == "single_filter": 

135 assert len(combinations) == 1 

136 elif filters_arg == "double_filter": 

137 assert len(combinations) == 2