Belle II Software development
Public Member Functions | List of all members
NormalWorkerTestCase Class Reference
Inheritance diagram for NormalWorkerTestCase:
WorkerTestCase HLTZMQTestCase

Public Member Functions

def testInitialization (self)
 
def testRunSending (self)
 
def testEndRun (self)
 

Detailed Description

Tests for normal worker behavior

Definition at line 58 of file test_worker.py.

Member Function Documentation

◆ testEndRun()

def testEndRun (   self) 
test function

Definition at line 84 of file test_worker.py.

84 def testEndRun(self):
85 """test function"""
86 self.start()
87
88 # TODO: do I want to test which run was ended?
89 # end run trigger
90 self.send(self.input_socket, "l", identity=self.input_identity)
91 self.assertHasOutputFile("endrun_called", timeout=2)
92 self.assertIsMsgType(self.output_socket, "l", router=True)
93 self.send(self.output_socket, "c", identity=self.output_identity)
94
95 # Also the second one should give us an end run
96 self.send(self.input_socket, "l", identity=self.input_identity)
97 self.assertHasOutputFile("endrun_called", timeout=1)
98 self.assertIsMsgType(self.output_socket, "l", router=True)
99 self.send(self.output_socket, "c", identity=self.output_identity)
100
101 # Sneak in an event in between -> should give beginRun
102 self.send(self.input_socket, "u", first_data=self.first_run_event_data[0], identity=self.input_identity)
103 self.assertHasOutputFile("beginrun_called", timeout=1)
104 self.assertIsMsgType(self.output_socket, "w", router=True)
105 self.send(self.output_socket, "c", identity=self.output_identity)
106
107 # And end the run again
108 self.send(self.input_socket, "l", identity=self.input_identity)
109 self.assertHasOutputFile("endrun_called", timeout=1)
110 self.assertIsMsgType(self.output_socket, "l", router=True)
111 self.send(self.output_socket, "c", identity=self.output_identity)
112
113 # A second time...
114 self.send(self.input_socket, "l", identity=self.input_identity)
115 self.assertHasOutputFile("endrun_called", timeout=1)
116 self.assertIsMsgType(self.output_socket, "l", router=True)
117 self.send(self.output_socket, "c", identity=self.output_identity)
118
119 # Sneak in a second event in between -> should give beginRun (as it is a new run)
120 self.send(self.input_socket, "u", first_data=self.second_run_event_data[0], identity=self.input_identity)
121 self.assertHasOutputFile("beginrun_called", timeout=1)
122 self.assertIsMsgType(self.output_socket, "w", router=True)
123 self.send(self.output_socket, "c", identity=self.output_identity)
124
125 # And end the run again
126 self.send(self.input_socket, "l", identity=self.input_identity)
127 self.assertHasOutputFile("endrun_called", timeout=1)
128 self.assertIsMsgType(self.output_socket, "l", router=True)
129 self.send(self.output_socket, "c", identity=self.output_identity)
130
131 # Termination should also work
132 self.send(self.input_socket, "x", identity=self.input_identity)
133 # Attention: terminate is called in the different order
134 self.assertIsMsgType(self.output_socket, "x", router=True)
135 self.send(self.output_socket, "c", identity=self.output_identity)
136 self.assertHasOutputFile("terminate_called", timeout=1)
137
138 # And the termination should cause the process to go down
139 self.assertIsDown("worker", timeout=200)
140
141

◆ testInitialization()

def testInitialization (   self) 
test function

Definition at line 61 of file test_worker.py.

61 def testInitialization(self):
62 """test function"""
63 self.start()
64
65 # Initialisation should be called
66 self.assertHasOutputFile("initialize_called", timeout=1)
67

◆ testRunSending()

def testRunSending (   self) 
test function

Definition at line 68 of file test_worker.py.

68 def testRunSending(self):
69 """test function"""
70 self.start()
71
72 # Send first event (should trigger begin run again)
73 self.send(self.input_socket, "u", first_data=self.first_run_event_data[0], identity=self.input_identity)
74 self.assertHasOutputFile("beginrun_called", timeout=0.5)
75 self.assertIsMsgType(self.output_socket, "w", router=True)
76 self.send(self.output_socket, "c", identity=self.output_identity)
77
78 # Send second event (should not trigger begin run again)
79 self.send(self.input_socket, "u", first_data=self.first_run_event_data[1], identity=self.input_identity)
80 self.assertNotHasOutputFile("beginrun_called", timeout=0.5)
81 self.assertIsMsgType(self.output_socket, "w", router=True)
82 self.send(self.output_socket, "c", identity=self.output_identity)
83
The documentation for this class was generated from the following file: