3 from unittest
import TestCase, main
7 from tempfile
import NamedTemporaryFile
12 class DerivedCalculationProcess(CalculationProcess):
14 def start_process(self, **kwargs):
15 self.result_queue.put(
"TestItem",
"TestValue")
16 self.result_queue.put(
"AnotherTestItem",
"AnotherTestValue")
19 class OngoingCalculationProcess(CalculationProcess):
21 def start_process(self, **kwargs):
26 class LoggingCalculationProcess(CalculationProcess):
28 def start_process(self, **kwargs):
29 with open(self.log_file_name,
"w")
as f:
30 f.write(
"Log File Content")
33 class TestCalculationProcess(
TestCase):
35 def get_terminated_process(self):
36 queue = CalculationQueue()
37 process = DerivedCalculationProcess(result_queue=queue, log_file_name=
None, parameters=
None)
42 def test_invalid_result_queue(self):
43 self.assertRaises(ValueError, DerivedCalculationProcess, result_queue=
None,
44 log_file_name=
None, parameters=
None)
46 def test_default_parameters(self):
47 queue = CalculationQueue()
48 process = DerivedCalculationProcess(result_queue=queue, log_file_name=
None, parameters=
None)
50 self.assertFalse(process.is_alive())
51 self.assertTrue(process.is_valid)
53 def test_get_keys(self):
54 process = self.get_terminated_process()
56 keys = process.get_keys()
58 self.assertIn(
"TestItem", keys)
59 self.assertIn(
"AnotherTestItem", keys)
62 process = self.get_terminated_process()
64 self.assertEqual(process.get(
"TestItem"),
"TestValue")
65 self.assertEqual(process.get(
"AnotherTestItem"),
"AnotherTestValue")
67 self.assertRaises(KeyError, process.get,
"UnknownItem")
69 def test_isAlive(self):
70 process = self.get_terminated_process()
72 self.assertFalse(process.is_alive())
75 class TestLoggingCalculationProcess(
TestCase):
78 queue = CalculationQueue()
80 self.tmp_file_name = NamedTemporaryFile(delete=
False).name
82 self.process = LoggingCalculationProcess(result_queue=queue, log_file_name=self.tmp_file_name,
89 self.process.terminate()
91 if os.path.exists(self.tmp_file_name):
92 os.unlink(self.tmp_file_name)
94 def test_get_log(self):
95 log_file_content = self.process.get_log()
96 self.assertEqual(log_file_content,
"Log File Content")
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)
102 self.process.save_log()
104 self.assertEqual(self.process.log_file_content,
"Log File Content")
105 self.assertIsNone(self.process.log_file_name)
107 self.assertFalse(os.path.exists(self.tmp_file_name))
109 def test_log_already_there(self):
110 self.process.log_file_content =
"Already there"
111 self.process.save_log()
113 self.assertEqual(self.process.log_file_content,
"Already there")
116 class TestOngoingCalculationProcess(
TestCase):
119 queue = CalculationQueue()
120 self.process = OngoingCalculationProcess(result_queue=queue, log_file_name=
None, parameters=
None)
123 self.process.terminate()
128 self.assertEqual(self.process.get_keys(),
None)
129 self.assertEqual(self.process.get(
"TestItem"),
None)
132 class TestOngoingLoggingCalculationProcess(
TestCase):
135 self.tmp_file_name = NamedTemporaryFile(delete=
False).name
137 queue = CalculationQueue()
138 self.process = OngoingCalculationProcess(result_queue=queue, log_file_name=self.tmp_file_name,
142 self.process.terminate()
144 if os.path.exists(self.tmp_file_name):
145 os.unlink(self.tmp_file_name)
147 def test_get_log(self):
150 self.assertEqual(self.process.get_log(),
"")
152 if __name__ ==
"__main__":