Belle II Software  release-05-01-25
test_calculation.py
1 from unittest import TestCase, main
2 from hep_ipython_tools.calculation import Calculation
3 from hep_ipython_tools.calculation_process import CalculationProcess
4 from hep_ipython_tools.calculation_queue import CalculationQueue
5 
6 # @cond internal_test
7 
8 
9 class TestCalculation(TestCase):
10 
11  def test_default(self):
12  calculation = Calculation()
13 
14  self.assertEqual(calculation._calculation_process_type, CalculationProcess)
15  self.assertEqual(calculation.process_list, [])
16 
17  calculations = Calculation([1, 2, 3])
18  self.assertEqual(calculations.process_list, [1, 2, 3])
19 
20  def test_iter(self):
21  calculations = Calculation([0, 1, 2])
22 
23  for i, calc in enumerate(calculations):
24  self.assertEqual(calc.process_list, [i])
25 
26  self.assertEqual(len(calculations), 3)
27 
28  def test_is_running(self):
29  calculations, process = self.create_calculations()
30 
31  process.is_alive = lambda: True
32  self.assertEqual(calculations.is_running(), [False, True, False])
33 
34  def test_is_finished(self):
35  calculations, process = self.create_calculations()
36 
37  process.already_run = lambda: True
38  self.assertEqual(calculations.is_finished(), [False, True, False])
39 
40  def test_map_on_process(self):
41  calculations, process = self.create_calculations()
42 
43  process.is_alive = lambda: True
44 
45  self.assertEqual(calculations.map_on_processes(lambda p: p.is_alive(), index=None), [False, True, False])
46  self.assertEqual(calculations.map_on_processes(lambda p: p.is_alive(), index=1), True)
47  self.assertEqual(calculations.map_on_processes(lambda p: p.is_alive(), index=0), False)
48  self.assertEqual(calculations.map_on_processes(lambda p: p.is_alive(), index=process), True)
49 
50  def create_calculations(self):
51  process1 = CalculationProcess(CalculationQueue(), None, None)
52  process2 = CalculationProcess(CalculationQueue(), None, None)
53  process3 = CalculationProcess(CalculationQueue(), None, None)
54 
55  calculations = Calculation([process1, process2, process3])
56 
57  return calculations, process2
58 
59 if __name__ == "__main__":
60  main()
61 # @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
hep_ipython_tools.calculation
Definition: calculation.py:1