Inheritance diagram for HistogramStopTestCase:

Collaboration diagram for HistogramStopTestCase:

[legend]

Public Member Functions
def	testStopPropagation (self)

def	setUp (self)

def	tearDown (self)

def	assertIsDown (self, name, timeout=5, minimum_delay=0.1)

def	assertIsRunning (self, name)

def	assertMonitoring (self, socket, search_key, search_value, timeout=10)

def	assertIsAndGet (self, socket, message_type, final=True, router=False)

def	assertIsMsgType (self, socket, message_type, final=True, router=False)

def	assertNothingMore (self, socket)

def	assertHasOutputFile (self, output_file, unlink=True, timeout=0.5, minimum_delay=0.1)

def	assertNotHasOutputFile (self, output_file, timeout=0.5)

Static Public Member Functions
def	get_free_port ()

def	create_socket (port, socket_type=zmq.DEALER, identity="socket", bind=False)

def	create_router_socket (port)

def	send (socket, message_type, first_data=b"", second_data=b"", identity="")

def	recv (socket)

Public Attributes
	test_dir
	use a temporary folder for testing

	previous_dir
	remember current working directory

	started_programs
	dict for all started programs

Static Public Attributes
	input_port = HLTZMQTestCase.get_free_port()
	input_port

	monitoring_port = HLTZMQTestCase.get_free_port()
	monitoring_port

dictionary	needed_programs
	needed_programs More...

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

string	event_data
	event_data More...

	ctx = zmq.Context()
	The ZMQ context.

Private Member Functions
def	_is_running (self, name)

Detailed Description

Test case

Definition at line 210 of file test_histogram.py.

Member Function Documentation

◆ _is_running()

def _is_running	(	self,
		name
	)

privateinherited

Check if a given program is still running.

Definition at line 72 of file test_support.py.

◆ assertHasOutputFile()

def 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 215 of file test_support.py.

◆ assertIsAndGet()

def 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 188 of file test_support.py.

◆ assertIsDown()

def 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 81 of file test_support.py.

◆ assertIsMsgType()

def assertIsMsgType	(	self,
		socket,
		message_type,
		final = `True`,
		router = `False`
	)

inherited

Deprecated copy of "assertIsAndGet".

Definition at line 203 of file test_support.py.

◆ assertIsRunning()

def assertIsRunning	(	self,
		name
	)

inherited

Assert that a given program is still running.

Definition at line 96 of file test_support.py.

◆ assertMonitoring()

def 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 157 of file test_support.py.

◆ assertNotHasOutputFile()

def 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 233 of file test_support.py.

◆ assertNothingMore()

def assertNothingMore	(	self,
		socket
	)

inherited

Assert that there is no pending message to be received on the socket.

Definition at line 209 of file test_support.py.

◆ create_router_socket()

def create_router_socket ( port )

staticinherited

Shortcut to create a ROUTER type socket with the typical parameters
binding to the given port.

Definition at line 128 of file test_support.py.

◆ create_socket()

def 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 103 of file test_support.py.

◆ get_free_port()

def get_free_port ( )

staticinherited

Get a free port number by reusing ZMQ's function for this.

Definition at line 29 of file test_support.py.

◆ recv()

def 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 147 of file test_support.py.

◆ send()

def 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 136 of file test_support.py.

◆ setUp()

def setUp ( self )

inherited

Custom setUp function to go into a temporary folder
and start the needed programs.

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

Definition at line 38 of file test_support.py.

◆ tearDown()

def tearDown ( self )

inherited

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

Definition at line 58 of file test_support.py.

◆ testStopPropagation()

def testStopPropagation ( self )

test function

Definition at line 242 of file test_histogram.py.

     def testStopPropagation(self):
         """test function"""
         monitoring_socket = self.create_socket(self.monitoring_port)
  
         input_socket = self.create_socket(self.input_port)
         self.send(input_socket, "h")
         self.assertIsMsgType(input_socket, "c")
  
         second_input_socket = self.create_socket(self.input_port, identity="other_socket")
         self.send(second_input_socket, "h")
         self.assertIsMsgType(second_input_socket, "c")
  
         # At the beginning, everything should be at normal state
         self.assertMonitoring(monitoring_socket, "input.registered_workers", 2)
         self.assertNotHasOutputFile("outputFile.root", timeout=1)
  
         # So far no stop messages should be there
         self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 0)
         self.assertMonitoring(monitoring_socket, "input.all_stop_messages", False)
  
         # the first stop message should not trigger a transmission
         self.send(input_socket, "l")
         self.assertIsMsgType(input_socket, "c")
         self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 1)
         self.assertMonitoring(monitoring_socket, "input.all_stop_messages", False)
         self.assertNotHasOutputFile("outputFile.root", timeout=1)
  
         # The second stop message should also not, as there are no histograms so far
         self.send(second_input_socket, "l")
         self.assertIsMsgType(second_input_socket, "c")
         self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 2)
         self.assertMonitoring(monitoring_socket, "input.all_stop_messages", True)
         self.assertNotHasOutputFile("outputFile.root", timeout=1)
  
         # Reset everything
         self.send(monitoring_socket, "n")
         self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 0)
         self.assertMonitoring(monitoring_socket, "input.all_stop_messages", False)
  
         # Now lets send some events
         self.send(input_socket, "v", self.histogram_data, self.event_data)
         self.assertIsMsgType(input_socket, "c")
  
         self.send(input_socket, "v", self.histogram_data, self.event_data)
         self.assertIsMsgType(input_socket, "c")
  
         self.send(second_input_socket, "v", self.histogram_data, self.event_data)
         self.assertIsMsgType(second_input_socket, "c")
  
         self.send(input_socket, "v", self.histogram_data, self.event_data)
         self.assertIsMsgType(input_socket, "c")
  
         self.send(second_input_socket, "v", self.histogram_data, self.event_data)
         self.assertIsMsgType(second_input_socket, "c")
  
         self.send(second_input_socket, "v", self.histogram_data, self.event_data)
         self.assertIsMsgType(second_input_socket, "c")
  
         # This should not be enough to trigger a merge
         self.assertMonitoring(monitoring_socket, "input.received_events", 6)
         self.assertNotHasOutputFile("outputFile.root", timeout=1)
  
         # But if we again send the stop messages
         self.send(input_socket, "l")
         self.assertIsMsgType(input_socket, "c")
         self.send(second_input_socket, "l")
         self.assertIsMsgType(second_input_socket, "c")
  
         # .. it should have merged it. We expect 2 entries, as we have 2 clients (no matter how often they sent)
         self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 2)
         self.assertMonitoring(monitoring_socket, "input.all_stop_messages", True)
         self.assertHasOutputFile("outputFile.root", unlink=False)
         self.assertTrue(check_histogram_output("outputFile.root", 2))
  
         # Now send a terminate message
         self.send(input_socket, "x")
         self.assertIsMsgType(input_socket, "c")
         self.send(second_input_socket, "x")
         self.assertIsMsgType(second_input_socket, "c")
  
         # There should be no merge happening, as the files are already written
         self.assertNotHasOutputFile("outputFile.root")
  
         self.assertIsDown("histoserver")
  
  

Member Data Documentation

◆ event_data

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

◆ needed_programs

dictionary needed_programs

static

Initial value:

=  {"histoserver": ["b2hlt_finalhistoserver", "--input", f"tcp://*:{input_port}",
                                       "--rootFileName", "outputFile.root",
                                       "--timeout", "0",  # we remove the timeout on purpose
                                       "--monitor", f"tcp://*:{monitoring_port}"],
                       }

needed_programs

Definition at line 218 of file test_histogram.py.

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

daq/hbasf2/tests/test_histogram.py

Public Member Functions

Static Public Member Functions

Public Attributes

Static Public Attributes

Private Member Functions

Detailed Description

Member Function Documentation

◆ _is_running()

◆ assertHasOutputFile()

◆ assertIsAndGet()

◆ assertIsDown()

◆ assertIsMsgType()

◆ assertIsRunning()

◆ assertMonitoring()

◆ assertNotHasOutputFile()

◆ assertNothingMore()

◆ create_router_socket()

◆ create_socket()

◆ get_free_port()

◆ recv()

◆ send()

◆ setUp()

◆ tearDown()

◆ testStopPropagation()

Member Data Documentation

◆ event_data

◆ needed_programs