Belle II Software light-2406-ragdoll
test_IPythonHandler.py
1
8import os
9from unittest import TestCase, main
10
11from hep_ipython_tools.calculation_process import CalculationProcess
12from hep_ipython_tools.calculation import Calculation
13from hep_ipython_tools.calculation_queue import CalculationQueue
14from hep_ipython_tools.information import EnvironmentInformation
15from hep_ipython_tools.ipython_handler import IPythonHandler
16from hep_ipython_tools.tests.fixtures import MockQueue
17
18# @cond internal_test
19
20
21class DeriviedCalculationProcess(CalculationProcess):
22
23 def __init__(self, result_queue, log_file_name, parameters, some_variable, other_variable):
24 CalculationProcess.__init__(self, result_queue, log_file_name, parameters)
25
26 self.some_variable = some_variable
27 self.other_variable = other_variable
28
29 def start_process(self):
30 pass
31
32
33class DerivedCalculation(Calculation):
34
35 def __init__(self):
36 Calculation.__init__(self)
37 self._calculation_process_type = DeriviedCalculationProcess
38
39
40class DerivedIPythonHandler(IPythonHandler):
41
42 def __init__(self):
43 IPythonHandler.__init__(self)
44
45 self._calculation_type = DerivedCalculation
46
47
48class TestIPythonHandler(TestCase):
49
50 def test_default(self):
51 handler = IPythonHandler()
52
53 self.assertEqual(handler.log_files, [])
54 self.assertEqual(handler._calculation_type, Calculation)
55 self.assertIsInstance(handler.information, EnvironmentInformation)
56
57 def test_create_queue(self):
58 queue = DerivedIPythonHandler.create_queue()
59
60 self.assertIsInstance(queue, CalculationQueue)
61
62 self.assertEqual(queue.get_keys(), [])
63 self.assertEqual(queue.results, {})
64
65 def test_next_log_file(self):
66
67 handler = DerivedIPythonHandler()
68
69 self.assertEqual(handler.log_files, [])
70
71 next_log_file = handler.next_log_file_name()
72
73 try:
74 self.assertEqual(len(handler.log_files), 1)
75 self.assertEqual(handler.log_files[0][1], next_log_file)
76
77 finally:
78 os.unlink(next_log_file)
79
80 def test_many_next_log_files(self):
81 handler = DerivedIPythonHandler()
82
83 self.assertEqual(handler.log_files, [])
84
85 next_log_files = [handler.next_log_file_name() for _ in range(200)]
86
87 try:
88 self.assertEqual(len(handler.log_files), 20)
89
90 names = [log_file[1] for log_file in handler.log_files]
91 self.assertEqual(names, next_log_files[-20:])
92
93 for closed_log_file in next_log_files[:-20]:
94 self.assertFalse(os.path.exists(closed_log_file))
95
96 finally:
97 for log_file_name in next_log_files:
98 if os.path.exists(log_file_name):
99 os.unlink(log_file_name)
100
101 def test_many_log_files_closed(self):
102 handler = DerivedIPythonHandler()
103
104 self.assertEqual(handler.log_files, [])
105
106 next_log_files = [handler.next_log_file_name() for _ in range(100)]
107
108 for log_file_name in next_log_files:
109 if os.path.exists(log_file_name):
110 os.unlink(log_file_name)
111
112 next_log_files = [handler.next_log_file_name() for _ in range(100)]
113
114 try:
115 self.assertEqual(len(handler.log_files), 20)
116
117 names = [log_file[1] for log_file in handler.log_files]
118 self.assertEqual(names, next_log_files[-20:])
119
120 for closed_log_file in next_log_files[:-20]:
121 self.assertFalse(os.path.exists(closed_log_file))
122
123 finally:
124 for log_file_name in next_log_files:
125 if os.path.exists(log_file_name):
126 os.unlink(log_file_name)
127
128 def test_process_result_queue(self):
129 handler = DerivedIPythonHandler()
130
131 queue = CalculationQueue()
132 queue.queue = MockQueue()
133
134 queue.put("TestItem", "TestValue")
135
136 calculation = handler.process(queue, some_variable=42, other_variable=21)
137
138 self.assertEqual(len(calculation.process_list), 1)
139
140 added_calculation = calculation.process_list[0]
141
142 self.assertEqual(added_calculation.log_file_name, handler.log_files[0][1])
143 self.assertEqual(added_calculation.result_queue, queue)
144 self.assertEqual(added_calculation.result_queue.get_keys(), ["TestItem"])
145 self.assertEqual(added_calculation.result_queue.get("TestItem"), "TestValue")
146 self.assertEqual(added_calculation.parameters, None)
147 self.assertEqual(added_calculation.some_variable, 42)
148 self.assertEqual(added_calculation.other_variable, 21)
149
150 self.assertEqual(calculation.get_keys(), ["TestItem"])
151 self.assertEqual(calculation.get("TestItem"), "TestValue")
152 self.assertEqual(calculation.get_parameters(), None)
153
154 def test_process_no_result_queue(self):
155 handler = DerivedIPythonHandler()
156
157 calculation = handler.process(some_variable=42, other_variable=21)
158
159 self.assertEqual(len(calculation.process_list), 1)
160
161 added_calculation = calculation.process_list[0]
162
163 result_queue = added_calculation.result_queue
164
165 self.assertIsInstance(result_queue, CalculationQueue)
166
167 def test_parameter_space_no_queue(self):
168 handler = DerivedIPythonHandler()
169
170 def creator_function(some_variable, other_variable):
171 return dict(some_variable=some_variable + 1, other_variable=other_variable + "c")
172
173 calculations = handler.process_parameter_space(creator_function,
174 some_variable=[1, 2], other_variable=["a", "b"])
175
176 self.assertEqual(len(calculations.process_list), 4)
177
178 parameters = calculations.get_parameters()
179
180 self.assertIn({"some_variable": 1, "other_variable": "a"}, parameters)
181 self.assertIn({"some_variable": 1, "other_variable": "b"}, parameters)
182 self.assertIn({"some_variable": 2, "other_variable": "a"}, parameters)
183 self.assertIn({"some_variable": 2, "other_variable": "b"}, parameters)
184
185 def assert_is_in_processes(some_variable, other_variable):
186 for process in calculations.process_list:
187 if process.some_variable == some_variable and process.other_variable == other_variable:
188 return True
189
190 self.fail()
191
192 assert_is_in_processes(2, "ac")
193 assert_is_in_processes(3, "ac")
194 assert_is_in_processes(2, "bc")
195 assert_is_in_processes(3, "bc")
196
197
198if __name__ == "__main__":
199 main()
200# @endcond
Definition: main.py:1