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