Belle II Software light-2406-ragdoll
test_calculationProcess.py
1
8import os
9from time import sleep
10from unittest import TestCase, main
11
12from hep_ipython_tools.calculation_process import CalculationProcess
13from hep_ipython_tools.calculation_queue import CalculationQueue
14from tempfile import NamedTemporaryFile
15
16# @cond internal_test
17
18
19class DerivedCalculationProcess(CalculationProcess):
20
21 def start_process(self, **kwargs):
22 self.result_queue.put("TestItem", "TestValue")
23 self.result_queue.put("AnotherTestItem", "AnotherTestValue")
24
25
26class OngoingCalculationProcess(CalculationProcess):
27
28 def start_process(self, **kwargs):
29 while True:
30 sleep(1)
31
32
33class LoggingCalculationProcess(CalculationProcess):
34
35 def start_process(self, **kwargs):
36 with open(self.log_file_name, "w") as f:
37 f.write("Log File Content")
38
39
40class TestCalculationProcess(TestCase):
41
42 def get_terminated_process(self):
43 queue = CalculationQueue()
44 process = DerivedCalculationProcess(result_queue=queue, log_file_name=None, parameters=None)
45 process.start()
46 process.join()
47 return process
48
49 def test_invalid_result_queue(self):
50 self.assertRaises(ValueError, DerivedCalculationProcess, result_queue=None,
51 log_file_name=None, parameters=None)
52
53 def test_default_parameters(self):
54 queue = CalculationQueue()
55 process = DerivedCalculationProcess(result_queue=queue, log_file_name=None, parameters=None)
56
57 self.assertFalse(process.is_alive())
58 self.assertTrue(process.is_valid)
59
60 def test_get_keys(self):
61 process = self.get_terminated_process()
62
63 keys = process.get_keys()
64
65 self.assertIn("TestItem", keys)
66 self.assertIn("AnotherTestItem", keys)
67
68 def test_get(self):
69 process = self.get_terminated_process()
70
71 self.assertEqual(process.get("TestItem"), "TestValue")
72 self.assertEqual(process.get("AnotherTestItem"), "AnotherTestValue")
73
74 self.assertRaises(KeyError, process.get, "UnknownItem")
75
76 def test_isAlive(self):
77 process = self.get_terminated_process()
78
79 self.assertFalse(process.is_alive())
80
81
82class TestLoggingCalculationProcess(TestCase):
83
84 def setUp(self):
85 queue = CalculationQueue()
86
87 self.tmp_file_name = NamedTemporaryFile(delete=False).name
88
89 self.process = LoggingCalculationProcess(result_queue=queue, log_file_name=self.tmp_file_name,
90 parameters=None)
91
92 self.process.start()
93 self.process.join()
94
95 def tearDown(self):
96 self.process.terminate()
97
98 if os.path.exists(self.tmp_file_name):
99 os.unlink(self.tmp_file_name)
100
101 def test_get_log(self):
102 log_file_content = self.process.get_log()
103 self.assertEqual(log_file_content, "Log File Content")
104
105 def test_save_log(self):
106 self.assertIsNone(self.process.log_file_content)
107 self.assertEqual(self.tmp_file_name, self.process.log_file_name)
108
109 self.process.save_log()
110
111 self.assertEqual(self.process.log_file_content, "Log File Content")
112 self.assertIsNone(self.process.log_file_name)
113
114 self.assertFalse(os.path.exists(self.tmp_file_name))
115
116 def test_log_already_there(self):
117 self.process.log_file_content = "Already there"
118 self.process.save_log()
119
120 self.assertEqual(self.process.log_file_content, "Already there")
121
122
123class TestOngoingCalculationProcess(TestCase):
124
125 def setUp(self):
126 queue = CalculationQueue()
127 self.process = OngoingCalculationProcess(result_queue=queue, log_file_name=None, parameters=None)
128
129 def tearDown(self):
130 self.process.terminate()
131
132 def test_get(self):
133 self.process.start()
134
135 self.assertEqual(self.process.get_keys(), None)
136 self.assertEqual(self.process.get("TestItem"), None)
137
138
139class TestOngoingLoggingCalculationProcess(TestCase):
140
141 def setUp(self):
142 self.tmp_file_name = NamedTemporaryFile(delete=False).name
143
144 queue = CalculationQueue()
145 self.process = OngoingCalculationProcess(result_queue=queue, log_file_name=self.tmp_file_name,
146 parameters=None)
147
148 def tearDown(self):
149 self.process.terminate()
150
151 if os.path.exists(self.tmp_file_name):
152 os.unlink(self.tmp_file_name)
153
154 def test_get_log(self):
155 self.process.start()
156
157 self.assertEqual(self.process.get_log(), "")
158
159
160if __name__ == "__main__":
161 main()
162# @endcond
Definition: main.py:1