Belle II Software prerelease-10-00-00a
HistogramStopTestCase Class Reference
Inheritance diagram for HistogramStopTestCase:
Collaboration diagram for HistogramStopTestCase:

Public Member Functions

 testStopPropagation (self)
 
 tearDown (self)
 
 setUp (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

 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

 input_port = HLTZMQTestCase.get_free_port()
 input_port
 
 monitoring_port = HLTZMQTestCase.get_free_port()
 monitoring_port
 
 histogram_data = open(basf2.find_file("daq/hbasf2/tests/histos.raw"), "br").read()
 histogram_data
 
str event_data
 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 215 of file test_histogram.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)
inherited
Custom setUp function to go into a temporary folder
and start the needed programs.

Reimplemented in DistributorTestCase, HistogramTestCase, DyingHLTTestCase, HLTTestCase, WorkerTestCase, and BaseCollectorTestCase.

Definition at line 50 of file test_support.py.

50 def setUp(self):
51 """
52 Custom setUp function to go into a temporary folder
53 and start the needed programs.
54 """
55
56 self.test_dir = tempfile.mkdtemp()
57
58 self.previous_dir = os.getcwd()
59 os.chdir(self.test_dir)
60
61
62 self.started_programs = dict()
63 for name, command in self.needed_programs.items():
64 self.started_programs[name] = subprocess.Popen(command, start_new_session=True)
65 self.assertIsRunning(name)
66
67 atexit._clear()
68 atexit.register(self.tearDown)
69

◆ 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

◆ testStopPropagation()

testStopPropagation ( self)
test function

Definition at line 247 of file test_histogram.py.

247 def testStopPropagation(self):
248 """test function"""
249 monitoring_socket = self.create_socket(self.monitoring_port)
250
251 input_socket = self.create_socket(self.input_port)
252 self.send(input_socket, "h")
253 self.assertIsMsgType(input_socket, "c")
254
255 second_input_socket = self.create_socket(self.input_port, identity="other_socket")
256 self.send(second_input_socket, "h")
257 self.assertIsMsgType(second_input_socket, "c")
258
259 # At the beginning, everything should be at normal state
260 self.assertMonitoring(monitoring_socket, "input.registered_workers", 2)
261 self.assertNotHasOutputFile("outputFile.root", timeout=1)
262
263 # So far no stop messages should be there
264 self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 0)
265 self.assertMonitoring(monitoring_socket, "input.all_stop_messages", False)
266
267 # the first stop message should not trigger a transmission
268 self.send(input_socket, "l")
269 self.assertIsMsgType(input_socket, "c")
270 self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 1)
271 self.assertMonitoring(monitoring_socket, "input.all_stop_messages", False)
272 self.assertNotHasOutputFile("outputFile.root", timeout=1)
273
274 # The second stop message should also not, as there are no histograms so far
275 self.send(second_input_socket, "l")
276 self.assertIsMsgType(second_input_socket, "c")
277 self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 2)
278 self.assertMonitoring(monitoring_socket, "input.all_stop_messages", True)
279 self.assertNotHasOutputFile("outputFile.root", timeout=1)
280
281 # Reset everything
282 self.send(monitoring_socket, "n")
283 self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 0)
284 self.assertMonitoring(monitoring_socket, "input.all_stop_messages", False)
285
286 # Now lets send some events
287 self.send(input_socket, "v", self.histogram_data, self.event_data)
288 self.assertIsMsgType(input_socket, "c")
289
290 self.send(input_socket, "v", self.histogram_data, self.event_data)
291 self.assertIsMsgType(input_socket, "c")
292
293 self.send(second_input_socket, "v", self.histogram_data, self.event_data)
294 self.assertIsMsgType(second_input_socket, "c")
295
296 self.send(input_socket, "v", self.histogram_data, self.event_data)
297 self.assertIsMsgType(input_socket, "c")
298
299 self.send(second_input_socket, "v", self.histogram_data, self.event_data)
300 self.assertIsMsgType(second_input_socket, "c")
301
302 self.send(second_input_socket, "v", self.histogram_data, self.event_data)
303 self.assertIsMsgType(second_input_socket, "c")
304
305 # This should not be enough to trigger a merge
306 self.assertMonitoring(monitoring_socket, "input.received_events", 6)
307 self.assertNotHasOutputFile("outputFile.root", timeout=1)
308
309 # But if we again send the stop messages
310 self.send(input_socket, "l")
311 self.assertIsMsgType(input_socket, "c")
312 self.send(second_input_socket, "l")
313 self.assertIsMsgType(second_input_socket, "c")
314
315 # .. it should have merged it. We expect 2 entries, as we have 2 clients (no matter how often they sent)
316 self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 2)
317 self.assertMonitoring(monitoring_socket, "input.all_stop_messages", True)
318 self.assertHasOutputFile("outputFile.root", unlink=False)
319 self.assertTrue(check_histogram_output("outputFile.root", 2))
320
321 # Now send a terminate message
322 self.send(input_socket, "x")
323 self.assertIsMsgType(input_socket, "c")
324 self.send(second_input_socket, "x")
325 self.assertIsMsgType(second_input_socket, "c")
326
327 # There should be no merge happening, as the files are already written
328 self.assertNotHasOutputFile("outputFile.root")
329
330 self.assertIsDown("histoserver")
331
332

Member Data Documentation

◆ ctx

ctx = zmq.Context()
staticinherited

The ZMQ context.

Definition at line 36 of file test_support.py.

◆ event_data

event_data
static
Initial value:
= b"""{
"_typename" : "Belle2::EventMetaData",
"fUniqueID" : 0,
"fBits" : 33554432,
"m_event" : 1,
"m_run" : 1,
"m_subrun" : 0,
"m_experiment" : 1,
"m_production" : 0,
"m_time" : 0,
"m_parentLfn" : "",
"m_generatedWeight" : 1,
"m_errorFlag" : 0
}"""

event_data

Definition at line 232 of file test_histogram.py.

◆ histogram_data

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

histogram_data

Definition at line 230 of file test_histogram.py.

◆ input_port

input_port = HLTZMQTestCase.get_free_port()
static

input_port

Definition at line 218 of file test_histogram.py.

◆ monitoring_port

monitoring_port = HLTZMQTestCase.get_free_port()
static

monitoring_port

Definition at line 220 of file test_histogram.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: