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

Public Member Functions

 setUp (self)
 
 testEventPropagation (self)
 
 check_histogram_repeated (self, file_name, expected_factor)
 
 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

 first_input_port = HLTZMQTestCase.get_free_port()
 first_input_port
 
 first_monitoring_port = HLTZMQTestCase.get_free_port()
 first_monitoring_port
 
 second_input_port = HLTZMQTestCase.get_free_port()
 second_input_port
 
 second_monitoring_port = HLTZMQTestCase.get_free_port()
 second_monitoring_port
 
 final_collector_input_port = HLTZMQTestCase.get_free_port()
 final_collector_input_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

 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 42 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

◆ check_histogram_repeated()

check_histogram_repeated ( self,
file_name,
expected_factor )
Repeatedly call check_histogram_output 5 times until it is actually fulfilled

Definition at line 203 of file test_histogram.py.

203 def check_histogram_repeated(self, file_name, expected_factor):
204 """Repeatedly call check_histogram_output 5 times until it is actually fulfilled"""
205 tries = 0
206 while tries < 5:
207 self.assertHasOutputFile(file_name, unlink=False, timeout=2)
208 if check_histogram_output(file_name, expected_factor):
209 break
210 tries += 1
211 else:
212 raise AssertionError("Even after retry, the output was not correct!")
213
214

◆ 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 62 of file test_histogram.py.

62 def setUp(self):
63 """Setup port numbers and necessary programs"""
64
65 self.first_input_port = HLTZMQTestCase.get_free_port()
66
67 self.first_monitoring_port = HLTZMQTestCase.get_free_port()
68
69
70 self.second_input_port = HLTZMQTestCase.get_free_port()
71
72 self.second_monitoring_port = HLTZMQTestCase.get_free_port()
73
74
75 self.final_collector_input_port = HLTZMQTestCase.get_free_port()
76
77 self.final_collector_monitoring_port = HLTZMQTestCase.get_free_port()
78
79 self.needed_programs = {
80 "first": [
81 "b2hlt_proxyhistoserver", "--input", f"tcp://*:{self.first_input_port}",
82 "--output", f"tcp://localhost:{self.final_collector_input_port}",
83 "--timeout", "1",
84 "--monitor", f"tcp://*:{self.first_monitoring_port}"
85 ],
86 "second": [
87 "b2hlt_proxyhistoserver",
88 "--input", f"tcp://*:{self.second_input_port}",
89 "--output", f"tcp://localhost:{self.final_collector_input_port}",
90 "--timeout", "1",
91 "--monitor", f"tcp://*:{self.second_monitoring_port}"
92 ],
93 "final_collector": [
94 "b2hlt_finalhistoserver",
95 "--input", f"tcp://*:{self.final_collector_input_port}",
96 "--rootFileName", "outputFile.root",
97 "--timeout", "1",
98 "--monitor", f"tcp://*:{self.final_collector_monitoring_port}"
99 ],
100 }
101 super().setUp()
102

◆ 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

◆ testEventPropagation()

testEventPropagation ( self)
test function

Definition at line 103 of file test_histogram.py.

103 def testEventPropagation(self):
104 """test function"""
105 self.assertNotHasOutputFile("outputFile.root")
106
107 first_monitoring_socket = self.create_socket(self.first_monitoring_port)
108 second_monitoring_socket = self.create_socket(self.second_monitoring_port)
109 final_monitoring_socket = self.create_socket(self.final_collector_monitoring_port)
110
111 input_sockets = [
112 self.create_socket(self.first_input_port, identity="1"),
113 self.create_socket(self.first_input_port, identity="2"),
114 self.create_socket(self.second_input_port, identity="3"),
115 self.create_socket(self.second_input_port, identity="4"),
116 self.create_socket(self.second_input_port, identity="5"),
117 ]
118
119 for input_socket in input_sockets:
120 self.send(input_socket, "h")
121 self.assertIsMsgType(input_socket, "c")
122
123 # At the beginning, everything should be at normal state
124 self.assertMonitoring(first_monitoring_socket, "input.registered_workers", 2)
125 self.assertMonitoring(second_monitoring_socket, "input.registered_workers", 3)
126 self.assertMonitoring(final_monitoring_socket, "input.registered_workers", 2)
127
128 # So far no stop messages should be there
129 self.assertMonitoring(first_monitoring_socket, "input.received_stop_messages", 0)
130 self.assertMonitoring(second_monitoring_socket, "input.received_stop_messages", 0)
131 self.assertMonitoring(final_monitoring_socket, "input.received_stop_messages", 0)
132
133 self.assertMonitoring(first_monitoring_socket, "input.all_stop_messages", False)
134 self.assertMonitoring(second_monitoring_socket, "input.all_stop_messages", False)
135 self.assertMonitoring(final_monitoring_socket, "input.all_stop_messages", False)
136
137 # send some events, which should eventually trigger a merge
138 for _ in range(10):
139 for input_socket in input_sockets:
140 self.send(input_socket, "v", self.histogram_data, self.event_data)
141 self.assertIsMsgType(input_socket, "c")
142
143 self.assertMonitoring(first_monitoring_socket, "input.received_events", 20)
144 self.assertMonitoring(second_monitoring_socket, "input.received_events", 30)
145 self.assertMonitoring(final_monitoring_socket, "input.received_events", 2)
146
147 # Make sure to get the newest data
148 self.check_histogram_repeated("outputFile.root", 5)
149
150 # send out stop messages, which should also trigger a merge
151 for input_socket in input_sockets:
152 self.send(input_socket, "l")
153 self.assertIsMsgType(input_socket, "c")
154
155 self.assertMonitoring(first_monitoring_socket, "input.received_stop_messages", 2)
156 self.assertMonitoring(first_monitoring_socket, "input.all_stop_messages", True)
157 self.assertMonitoring(second_monitoring_socket, "input.received_stop_messages", 3)
158 self.assertMonitoring(second_monitoring_socket, "input.all_stop_messages", True)
159
160 self.assertMonitoring(final_monitoring_socket, "input.received_stop_messages", 2)
161 self.assertMonitoring(final_monitoring_socket, "input.all_stop_messages", True)
162 self.assertHasOutputFile("outputFile.root", unlink=False)
163 self.check_histogram_repeated("outputFile.root", 5)
164
165 # Now we clean up
166 self.send(first_monitoring_socket, "n")
167 self.send(second_monitoring_socket, "n")
168 self.send(final_monitoring_socket, "n")
169
170 # which should give us a clean root file again
171 self.assertNotHasOutputFile("outputFile.root", timeout=1)
172
173 self.assertMonitoring(first_monitoring_socket, "input.received_stop_messages", 0)
174 self.assertMonitoring(first_monitoring_socket, "input.all_stop_messages", False)
175 self.assertMonitoring(second_monitoring_socket, "input.received_stop_messages", 0)
176 self.assertMonitoring(second_monitoring_socket, "input.all_stop_messages", False)
177
178 self.assertMonitoring(final_monitoring_socket, "input.received_stop_messages", 0)
179 self.assertMonitoring(final_monitoring_socket, "input.all_stop_messages", False)
180
181 # and send some more events, but only to the first two clients
182 for _ in range(5):
183 for input_socket in input_sockets[:2]:
184 self.send(input_socket, "v", self.histogram_data, self.event_data)
185 self.assertIsMsgType(input_socket, "c")
186
187 # this time we expect less entries
188 self.check_histogram_repeated("outputFile.root", 2)
189
190 # Now send a terminate message
191 for input_socket in input_sockets:
192 self.send(input_socket, "x")
193 self.assertIsMsgType(input_socket, "c")
194
195 # We expect another merge here
196 self.assertHasOutputFile("outputFile.root", unlink=False)
197 self.check_histogram_repeated("outputFile.root", 2)
198
199 self.assertIsDown("first")
200 self.assertIsDown("second")
201 self.assertIsDown("final_collector")
202

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 47 of file test_histogram.py.

◆ final_collector_input_port

final_collector_input_port = HLTZMQTestCase.get_free_port()

final_collector_input_port

Definition at line 75 of file test_histogram.py.

◆ final_collector_monitoring_port

final_collector_monitoring_port = HLTZMQTestCase.get_free_port()

final_collector_monitoring_port

Definition at line 77 of file test_histogram.py.

◆ first_input_port

first_input_port = HLTZMQTestCase.get_free_port()

first_input_port

Definition at line 65 of file test_histogram.py.

◆ first_monitoring_port

first_monitoring_port = HLTZMQTestCase.get_free_port()

first_monitoring_port

Definition at line 67 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 45 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.

◆ second_input_port

second_input_port = HLTZMQTestCase.get_free_port()

second_input_port

Definition at line 70 of file test_histogram.py.

◆ second_monitoring_port

second_monitoring_port = HLTZMQTestCase.get_free_port()

second_monitoring_port

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