Belle II Software development
DyingHLTTestCase Class Reference
Inheritance diagram for DyingHLTTestCase:
HLTZMQTestCase

Public Member Functions

 setUp (self)
 
 testFullRun (self)
 
 tearDown (self)
 
 assertIsDown (self, name, timeout=5, minimum_delay=0.1)
 
 assertIsRunning (self, name)
 
 assertMonitoring (self, socket, search_key, search_value, timeout=10)
 
 assertIsAndGet (self, socket, message_type, final=True, router=False)
 
 assertIsMsgType (self, socket, message_type, final=True, router=False)
 
 assertNothingMore (self, socket)
 
 assertHasOutputFile (self, output_file, unlink=True, timeout=0.5, minimum_delay=0.1)
 
 assertNotHasOutputFile (self, output_file, timeout=0.5)
 

Static Public Member Functions

 get_free_port ()
 
 create_socket (port, socket_type=zmq.DEALER, identity="socket", bind=False)
 
 create_router_socket (port)
 
 send (socket, message_type, first_data=b"", second_data=b"", identity="")
 
 recv (socket)
 

Public Attributes

 distributor_input_port = HLTZMQTestCase.get_free_port()
 distributor_input_port
 
 distributor_output_port = HLTZMQTestCase.get_free_port()
 distributor_output_port
 
 distributor_monitoring_port = HLTZMQTestCase.get_free_port()
 distributor_monitoring_port
 
 final_collector_input_port = HLTZMQTestCase.get_free_port()
 final_collector_input_port
 
 final_collector_output_port = HLTZMQTestCase.get_free_port()
 final_collector_output_port
 
 final_collector_monitoring_port = HLTZMQTestCase.get_free_port()
 final_collector_monitoring_port
 
 test_dir = tempfile.mkdtemp()
 use a temporary folder for testing
 
 previous_dir = os.getcwd()
 remember current working directory
 
 started_programs = dict()
 dict for all started programs
 
 tearDown = subprocess.Popen(command, start_new_session=True)
 

Static Public Attributes

 event_data = open(basf2.find_file("daq/hbasf2/tests/out.raw"), "br").read()
 event_data
 
 ctx = zmq.Context()
 The ZMQ context.
 
 needed_programs = dict()
 The dict name -> cmd args of the programs to start, needs to be set in each test.
 

Protected Member Functions

 _is_running (self, name)
 

Detailed Description

Test case

Definition at line 262 of file test_hlt.py.

Member Function Documentation

◆ _is_running()

_is_running ( self,
name )
protectedinherited
Check if a given program is still running.

Definition at line 84 of file test_support.py.

84 def _is_running(self, name):
85 """
86 Check if a given program is still running.
87 """
88 process = self.started_programs[name]
89 pid, sts = process._try_wait(os.WNOHANG)
90 assert pid == process.pid or pid == 0
91 return pid == 0
92

◆ assertHasOutputFile()

assertHasOutputFile ( self,
output_file,
unlink = True,
timeout = 0.5,
minimum_delay = 0.1 )
inherited
Assert that - at least after the given timeout - the output file
is present. If unlink is set to True, remove the file after checking.

Definition at line 227 of file test_support.py.

227 def assertHasOutputFile(self, output_file, unlink=True, timeout=0.5, minimum_delay=0.1):
228 """
229 Assert that - at least after the given timeout - the output file
230 is present. If unlink is set to True, remove the file after checking.
231 """
232 endtime = time() + timeout
233
234 while True:
235 if os.path.exists(output_file):
236 if unlink:
237 os.unlink(output_file)
238 return
239
240 remaining = endtime - time()
241 self.assertFalse(remaining <= 0)
242
243 sleep(minimum_delay)
244

◆ assertIsAndGet()

assertIsAndGet ( self,
socket,
message_type,
final = True,
router = False )
inherited
Assert that the next message received on the socket has the given message type.
If final is set to True, also assert that there is no additional message on the socket.
Use router only for router sockets.

Definition at line 200 of file test_support.py.

200 def assertIsAndGet(self, socket, message_type, final=True, router=False):
201 """
202 Assert that the next message received on the socket has the given message type.
203 If final is set to True, also assert that there is no additional message on the socket.
204 Use router only for router sockets.
205 """
206 answer = HLTZMQTestCase.recv(socket)
207 type_index = 0
208 if router:
209 type_index = 1
210 self.assertEqual(answer[type_index], message_type.encode())
211 if final:
212 self.assertNothingMore(socket)
213 return answer
214

◆ assertIsDown()

assertIsDown ( self,
name,
timeout = 5,
minimum_delay = 0.1 )
inherited
Test helper to assert the given program has terminated - at least after timeout in seconds has passed.
Checks every "minimal_delay seconds.

Definition at line 93 of file test_support.py.

93 def assertIsDown(self, name, timeout=5, minimum_delay=0.1):
94 """
95 Test helper to assert the given program has terminated - at least after timeout in seconds has passed.
96 Checks every "minimal_delay seconds.
97 """
98 endtime = time() + timeout
99 while True:
100 if not self._is_running(name):
101 return
102
103 remaining = endtime - time()
104 self.assertFalse(remaining <= 0)
105
106 sleep(minimum_delay)
107

◆ assertIsMsgType()

assertIsMsgType ( self,
socket,
message_type,
final = True,
router = False )
inherited
Deprecated copy of "assertIsAndGet".

Definition at line 215 of file test_support.py.

215 def assertIsMsgType(self, socket, message_type, final=True, router=False):
216 """
217 Deprecated copy of "assertIsAndGet".
218 """
219 return self.assertIsAndGet(socket, message_type, final=final, router=router)
220

◆ assertIsRunning()

assertIsRunning ( self,
name )
inherited
Assert that a given program is still running.

Definition at line 108 of file test_support.py.

108 def assertIsRunning(self, name):
109 """
110 Assert that a given program is still running.
111 """
112 self.assertTrue(self._is_running(name))
113

◆ assertMonitoring()

assertMonitoring ( self,
socket,
search_key,
search_value,
timeout = 10 )
inherited
Ask the given socket for a monitoring JSON and make sure, the value related to "search_key"
is set to "search_value" - at least after the given timeout.
The search key can should be in the form "<category>.<key>".

Definition at line 169 of file test_support.py.

169 def assertMonitoring(self, socket, search_key, search_value, timeout=10):
170 """
171 Ask the given socket for a monitoring JSON and make sure, the value related to "search_key"
172 is set to "search_value" - at least after the given timeout.
173 The search key can should be in the form "<category>.<key>".
174 """
175 end_time = time() + timeout
176 monitoring = dict()
177 while time() < end_time:
178 HLTZMQTestCase.send(socket, "m")
179 answer = self.assertIsAndGet(socket, "c")
180
181 dict_monitoring = json.loads(answer[1])
182 for parent_key, parent_dict in dict_monitoring.items():
183 for key, value in parent_dict.items():
184 monitoring[parent_key + "." + key] = value
185
186 if search_key in monitoring and monitoring[search_key] == search_value:
187 break
188 else:
189 if monitoring:
190 if search_key not in monitoring:
191 raise AssertionError(f"Monitoring did not have a result with key {search_key}")
192 else:
193 raise AssertionError(
194 f"Monitoring did not show the result {search_value} for {search_key}, instead {monitoring[search_key]}")
195 else:
196 raise AssertionError("Monitoring did not answer in time.")
197
198 self.assertNothingMore(socket)
199

◆ assertNotHasOutputFile()

assertNotHasOutputFile ( self,
output_file,
timeout = 0.5 )
inherited
Assert that after the timeout the given file is not present
(a.k.a. no process has created it)

Definition at line 245 of file test_support.py.

245 def assertNotHasOutputFile(self, output_file, timeout=0.5):
246 """
247 Assert that after the timeout the given file is not present
248 (a.k.a. no process has created it)
249 """
250 sleep(timeout)
251 self.assertFalse(os.path.exists(output_file))
252
253

◆ assertNothingMore()

assertNothingMore ( self,
socket )
inherited
Assert that there is no pending message to be received on the socket.

Definition at line 221 of file test_support.py.

221 def assertNothingMore(self, socket):
222 """
223 Assert that there is no pending message to be received on the socket.
224 """
225 self.assertFalse(socket.poll(0))
226

◆ create_router_socket()

create_router_socket ( port)
staticinherited
Shortcut to create a ROUTER type socket with the typical parameters
binding to the given port.

Definition at line 140 of file test_support.py.

140 def create_router_socket(port):
141 """
142 Shortcut to create a ROUTER type socket with the typical parameters
143 binding to the given port.
144 """
145 return HLTZMQTestCase.create_socket(port, socket_type=zmq.ROUTER, identity="", bind=True)
146

◆ create_socket()

create_socket ( port,
socket_type = zmq.DEALER,
identity = "socket",
bind = False )
staticinherited
Create and return a ZMQ socket with the given type and identity and
bind or connect it to localhost and the given port.

Definition at line 115 of file test_support.py.

115 def create_socket(port, socket_type=zmq.DEALER, identity="socket", bind=False):
116 """
117 Create and return a ZMQ socket with the given type and identity and
118 bind or connect it to localhost and the given port.
119 """
120 socket = HLTZMQTestCase.ctx.socket(socket_type)
121 socket.rcvtimeo = 10000
122 socket.linger = 0
123 if identity:
124 socket.setsockopt_string(zmq.IDENTITY, identity)
125 if bind:
126 if port is None:
127 port = socket.bind_to_random_port("tcp://*")
128 return socket, port
129 else:
130 socket.bind(f"tcp://*:{port}")
131 else:
132 if port is None:
133 raise RuntimeError("Cannot connect to unknown port")
134
135 socket.connect(f"tcp://localhost:{port}")
136
137 return socket
138

◆ get_free_port()

get_free_port ( )
staticinherited
Get a free port number by reusing ZMQ's function for this.

Definition at line 41 of file test_support.py.

41 def get_free_port():
42 """
43 Get a free port number by reusing ZMQ's function for this.
44 """
45 socket = HLTZMQTestCase.ctx.socket(zmq.ROUTER)
46 port = socket.bind_to_random_port("tcp://*")
47 socket.close()
48 return port
49

◆ recv()

recv ( socket)
staticinherited
Try to receive a message from the socket (or throw an assertion error if none comes after the set timeout
of the socket).

Definition at line 159 of file test_support.py.

159 def recv(socket):
160 """
161 Try to receive a message from the socket (or throw an assertion error if none comes after the set timeout
162 of the socket).
163 """
164 try:
165 return socket.recv_multipart()
166 except zmq.error.Again:
167 raise AssertionError("No answer from socket")
168

◆ send()

send ( socket,
message_type,
first_data = b"",
second_data = b"",
identity = "" )
staticinherited
Send a message consisting of the message type, the first and the second data
either to the identity if given or without identity if omitted.

Definition at line 148 of file test_support.py.

148 def send(socket, message_type, first_data=b"", second_data=b"", identity=""):
149 """
150 Send a message consisting of the message type, the first and the second data
151 either to the identity if given or without identity if omitted.
152 """
153 if identity:
154 socket.send_multipart([identity.encode(), message_type.encode(), first_data, second_data])
155 else:
156 socket.send_multipart([message_type.encode(), first_data, second_data])
157

◆ setUp()

setUp ( self)
Setup port numbers and necessary programs

Reimplemented from HLTZMQTestCase.

Definition at line 267 of file test_hlt.py.

267 def setUp(self):
268 """Setup port numbers and necessary programs"""
269
270 self.distributor_input_port = HLTZMQTestCase.get_free_port()
271
272 self.distributor_output_port = HLTZMQTestCase.get_free_port()
273
274 self.distributor_monitoring_port = HLTZMQTestCase.get_free_port()
275
276
277 self.final_collector_input_port = HLTZMQTestCase.get_free_port()
278
279 self.final_collector_output_port = HLTZMQTestCase.get_free_port()
280
281 self.final_collector_monitoring_port = HLTZMQTestCase.get_free_port()
282
283
284 self.needed_programs = {
285 "distributor": [
286 "b2hlt_distributor",
287 "--input", f"tcp://localhost:{self.distributor_input_port}",
288 "--output", f"tcp://*:{self.distributor_output_port}",
289 "--monitor", f"tcp://*:{self.distributor_monitoring_port}"
290 ],
291 "final_collector": [
292 "b2hlt_finalcollector", "--input", f"tcp://*:{self.final_collector_input_port}",
293 "--output", f"tcp://localhost:{self.final_collector_output_port}",
294 "--monitor", f"tcp://*:{self.final_collector_monitoring_port}"
295 ],
296 "worker": [
297 "python3", basf2.find_file("daq/hbasf2/tests/passthrough.no_run_py"),
298 "--input", f"tcp://localhost:{self.distributor_output_port}",
299 "--output", f"tcp://localhost:{self.final_collector_input_port}"
300 ],
301 "dying_worker": [
302 "python3", basf2.find_file("daq/hbasf2/tests/passthrough.no_run_py"),
303 "--prefix", "dying_", "--exit",
304 "--input", f"tcp://localhost:{self.distributor_output_port}",
305 "--output", f"tcp://localhost:{self.final_collector_input_port}"
306 ],
307 }
308 super().setUp()
309

◆ tearDown()

tearDown ( self)
inherited
Custom tearDown function to kill the started programs if still present
and remove the temporary folder again.

Definition at line 70 of file test_support.py.

70 def tearDown(self):
71 """
72 Custom tearDown function to kill the started programs if still present
73 and remove the temporary folder again.
74 """
75 for name, process in self.started_programs.items():
76 if self._is_running(name):
77 os.killpg(process.pid, signal.SIGKILL)
78 process.wait()
79 os.chdir(self.previous_dir)
80 shutil.rmtree(self.test_dir)
81
82 atexit._clear()
83

◆ testFullRun()

testFullRun ( self)
test function

Definition at line 310 of file test_hlt.py.

310 def testFullRun(self):
311 """test function"""
312 distributor_monitoring_socket = self.create_socket(self.distributor_monitoring_port)
313 final_collector_monitoring_socket = self.create_socket(self.final_collector_monitoring_port)
314
315 input_socket = self.create_socket(self.distributor_input_port, socket_type=zmq.STREAM, bind=True)
316 input_identity, _ = self.recv(input_socket)
317
318 output_socket = self.create_socket(self.final_collector_output_port, socket_type=zmq.STREAM, bind=True)
319 output_identity, _ = self.recv(output_socket)
320
321 # At the beginning, everything should be at normal state
322 self.assertMonitoring(distributor_monitoring_socket, "input.socket_state", "connected")
323 self.assertMonitoring(distributor_monitoring_socket, "input.socket_connects", 1)
324 self.assertMonitoring(distributor_monitoring_socket, "input.socket_disconnects", 0)
325 self.assertMonitoring(distributor_monitoring_socket, "output.ready_queue_size", 40)
326 self.assertMonitoring(distributor_monitoring_socket, "output.registered_workers", 2)
327
328 self.assertMonitoring(final_collector_monitoring_socket, "input.registered_workers", 2)
329
330 self.assertHasOutputFile("initialize_called", timeout=1)
331 self.assertHasOutputFile("dying_initialize_called", timeout=1)
332
333 # Now lets try sending some events
334 for _ in range(100):
335 input_socket.send_multipart([input_identity, self.event_data])
336
337 self.assertMonitoring(distributor_monitoring_socket, "input.socket_state", "connected")
338 self.assertMonitoring(distributor_monitoring_socket, "input.socket_connects", 1)
339 self.assertMonitoring(distributor_monitoring_socket, "input.socket_disconnects", 0)
340 self.assertMonitoring(distributor_monitoring_socket, "output.ready_queue_size", 40)
341 self.assertMonitoring(distributor_monitoring_socket, "output.registered_workers", 2)
342 self.assertMonitoring(distributor_monitoring_socket, "output.sent_events", 100)
343
344 self.assertMonitoring(final_collector_monitoring_socket, "input.registered_workers", 2)
345 self.assertMonitoring(final_collector_monitoring_socket, "input.received_events", 100)
346 self.assertMonitoring(final_collector_monitoring_socket, "output.sent_events", 100)
347
348 self.assertHasOutputFile("beginrun_called", timeout=1)
349 self.assertHasOutputFile("dying_beginrun_called", timeout=1)
350
351 buffer = b""
352 while output_socket.poll(0):
353 _, msg = self.recv(output_socket)
354 buffer += msg
355 self.assertNothingMore(output_socket)
356
357 # Data Size != raw data, as data is in different format, size taken from test itself
358 self.assertEqual(len(buffer), 122638 * 100)
359
360 # Now we kill one of the workers on purpose
361 Path("dying_exit_request").touch()
362
363 # And send some more events
364 for _ in range(100):
365 input_socket.send_multipart([input_identity, self.event_data])
366
367 self.assertHasOutputFile("dying_exit_called", timeout=1)
368 self.assertIsDown("dying_worker", timeout=10)
369
370 buffer = b""
371 while output_socket.poll(0):
372 _, msg = self.recv(output_socket)
373 buffer += msg
374 self.assertNothingMore(output_socket)
375
376 # We expect to have at least some events (but some will be missing)!
377 # 100 - 20 (queue size of dying worker) - 1 (the event that "caused" the problem) = 79
378 self.assertEqual(len(buffer), 122638 * 79)
379
380 # The collector should have it removed
381 self.assertMonitoring(final_collector_monitoring_socket, "input.registered_workers", 1)
382
383 # Also a stop signal should be transported correctly
384 self.send(distributor_monitoring_socket, "l")
385
386 self.assertMonitoring(distributor_monitoring_socket, "input.socket_state", "connected")
387 self.assertMonitoring(distributor_monitoring_socket, "input.socket_connects", 1)
388 self.assertMonitoring(distributor_monitoring_socket, "input.socket_disconnects", 0)
389 self.assertMonitoring(distributor_monitoring_socket, "output.ready_queue_size", 20)
390 # the distributor does not know about the dying worker, but this is also no problem
391 self.assertMonitoring(distributor_monitoring_socket, "output.registered_workers", 2)
392 self.assertMonitoring(distributor_monitoring_socket, "output.sent_events", 200)
393
394 self.assertMonitoring(final_collector_monitoring_socket, "input.registered_workers", 1)
395 self.assertMonitoring(final_collector_monitoring_socket, "input.received_events", 179)
396 self.assertMonitoring(final_collector_monitoring_socket, "input.received_stop_messages", 1)
397 self.assertMonitoring(final_collector_monitoring_socket, "input.all_stop_messages", 1)
398
399 # should go to the workers
400 self.assertHasOutputFile("endrun_called", timeout=1)
401
402 # Now lets restart the run
403 self.send(distributor_monitoring_socket, "n")
404 self.send(final_collector_monitoring_socket, "n")
405
406 self.assertMonitoring(distributor_monitoring_socket, "input.socket_state", "connected")
407 self.assertMonitoring(distributor_monitoring_socket, "input.socket_connects", 1)
408 self.assertMonitoring(distributor_monitoring_socket, "input.socket_disconnects", 0)
409 self.assertMonitoring(distributor_monitoring_socket, "output.ready_queue_size", 20)
410 self.assertMonitoring(distributor_monitoring_socket, "output.registered_workers", 2)
411
412 self.assertMonitoring(final_collector_monitoring_socket, "input.registered_workers", 1)
413 self.assertMonitoring(final_collector_monitoring_socket, "input.received_events", 179)
414 self.assertMonitoring(final_collector_monitoring_socket, "input.received_stop_messages", 0)
415 self.assertMonitoring(final_collector_monitoring_socket, "input.all_stop_messages", 0)
416
417 for _ in range(100):
418 input_socket.send_multipart([input_identity, self.event_data])
419
420 self.assertMonitoring(distributor_monitoring_socket, "input.socket_state", "connected")
421 self.assertMonitoring(distributor_monitoring_socket, "input.socket_connects", 1)
422 self.assertMonitoring(distributor_monitoring_socket, "input.socket_disconnects", 0)
423 self.assertMonitoring(distributor_monitoring_socket, "output.ready_queue_size", 20)
424 self.assertMonitoring(distributor_monitoring_socket, "output.registered_workers", 2)
425 self.assertMonitoring(distributor_monitoring_socket, "output.sent_events", 300)
426
427 self.assertMonitoring(final_collector_monitoring_socket, "input.registered_workers", 1)
428 self.assertMonitoring(final_collector_monitoring_socket, "input.received_events", 279)
429 self.assertMonitoring(final_collector_monitoring_socket, "output.sent_events", 279)
430
431 self.assertNotHasOutputFile("endrun_called", timeout=1)
432 self.assertHasOutputFile("beginrun_called", timeout=1)
433
434 buffer = b""
435 while output_socket.poll(0):
436 _, msg = self.recv(output_socket)
437 buffer += msg
438 self.assertNothingMore(output_socket)
439
440 # Now all events should be transported again
441 self.assertEqual(len(buffer), 122638 * 100)
442
443 # And finally: terminate everything
444 self.send(distributor_monitoring_socket, "x")
445
446 self.assertIsDown("final_collector")
447 self.assertIsDown("distributor")
448 self.assertIsDown("worker")
449
450

Member Data Documentation

◆ ctx

ctx = zmq.Context()
staticinherited

The ZMQ context.

Definition at line 36 of file test_support.py.

◆ distributor_input_port

distributor_input_port = HLTZMQTestCase.get_free_port()

distributor_input_port

Definition at line 270 of file test_hlt.py.

◆ distributor_monitoring_port

distributor_monitoring_port = HLTZMQTestCase.get_free_port()

distributor_monitoring_port

Definition at line 274 of file test_hlt.py.

◆ distributor_output_port

distributor_output_port = HLTZMQTestCase.get_free_port()

distributor_output_port

Definition at line 272 of file test_hlt.py.

◆ event_data

event_data = open(basf2.find_file("daq/hbasf2/tests/out.raw"), "br").read()
static

event_data

Definition at line 265 of file test_hlt.py.

◆ final_collector_input_port

final_collector_input_port = HLTZMQTestCase.get_free_port()

final_collector_input_port

Definition at line 277 of file test_hlt.py.

◆ final_collector_monitoring_port

final_collector_monitoring_port = HLTZMQTestCase.get_free_port()

final_collector_monitoring_port

Definition at line 281 of file test_hlt.py.

◆ final_collector_output_port

final_collector_output_port = HLTZMQTestCase.get_free_port()

final_collector_output_port

Definition at line 279 of file test_hlt.py.

◆ needed_programs

needed_programs = dict()
staticinherited

The dict name -> cmd args of the programs to start, needs to be set in each test.

Definition at line 38 of file test_support.py.

◆ previous_dir

previous_dir = os.getcwd()
inherited

remember current working directory

Definition at line 58 of file test_support.py.

◆ started_programs

started_programs = dict()
inherited

dict for all started programs

Definition at line 62 of file test_support.py.

◆ tearDown

tearDown = subprocess.Popen(command, start_new_session=True)
inherited

Definition at line 68 of file test_support.py.

◆ test_dir

test_dir = tempfile.mkdtemp()
inherited

use a temporary folder for testing

Definition at line 56 of file test_support.py.


The documentation for this class was generated from the following file: