HistogramTestCase Class Reference

Inheritance diagram for HistogramTestCase:

Public Member Functions
def	setUp (self)
def	testEventPropagation (self)
def	check_histogram_repeated (self, file_name, expected_factor)

Public Attributes
	first_input_port
	first_input_port
	first_monitoring_port
	first_monitoring_port
	second_input_port
	second_input_port
	second_monitoring_port
	second_monitoring_port
	final_collector_input_port
	final_collector_input_port
	final_collector_monitoring_port
	final_collector_monitoring_port
	needed_programs
	needed_programs

Static Public Attributes
open	histogram_data = open(basf2.find_file("daq/hbasf2/tests/histos.raw"), "br").read()
	histogram_data
b	event_data
	event_data

Detailed Description

Test case

Definition at line 42 of file test_histogram.py.

Member Function Documentation

check_histogram_repeated()

def 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.

    def check_histogram_repeated(self, file_name, expected_factor):
        """Repeatedly call check_histogram_output 5 times until it is actually fulfilled"""
        tries = 0
        while tries < 5:
            self.assertHasOutputFile(file_name, unlink=False, timeout=2)
            if check_histogram_output(file_name, expected_factor):
                break
            tries += 1
        else:
            raise AssertionError("Even after retry, the output was not correct!")
 
 

setUp()

def setUp

(

self

)

Setup port numbers and necessary programs

Reimplemented from HLTZMQTestCase.

Definition at line 62 of file test_histogram.py.

    def setUp(self):
        """Setup port numbers and necessary programs"""
        
        self.first_input_port = HLTZMQTestCase.get_free_port()
        
        self.first_monitoring_port = HLTZMQTestCase.get_free_port()
 
        
        self.second_input_port = HLTZMQTestCase.get_free_port()
        
        self.second_monitoring_port = HLTZMQTestCase.get_free_port()
 
        
        self.final_collector_input_port = HLTZMQTestCase.get_free_port()
        
        self.final_collector_monitoring_port = HLTZMQTestCase.get_free_port()
        
        self.needed_programs = {
            "first": [
                "b2hlt_proxyhistoserver", "--input", f"tcp://*:{self.first_input_port}",
                "--output", f"tcp://localhost:{self.final_collector_input_port}",
                "--timeout", "1",
                "--monitor", f"tcp://*:{self.first_monitoring_port}"
            ],
            "second": [
                "b2hlt_proxyhistoserver",
                "--input", f"tcp://*:{self.second_input_port}",
                "--output", f"tcp://localhost:{self.final_collector_input_port}",
                "--timeout", "1",
                "--monitor", f"tcp://*:{self.second_monitoring_port}"
            ],
            "final_collector": [
                "b2hlt_finalhistoserver",
                "--input", f"tcp://*:{self.final_collector_input_port}",
                "--rootFileName", "outputFile.root",
                "--timeout", "1",
                "--monitor", f"tcp://*:{self.final_collector_monitoring_port}"
            ],
        }
        super().setUp()
 

testEventPropagation()

def testEventPropagation

(

self

)

test function

Definition at line 103 of file test_histogram.py.

    def testEventPropagation(self):
        """test function"""
        self.assertNotHasOutputFile("outputFile.root")
 
        first_monitoring_socket = self.create_socket(self.first_monitoring_port)
        second_monitoring_socket = self.create_socket(self.second_monitoring_port)
        final_monitoring_socket = self.create_socket(self.final_collector_monitoring_port)
 
        input_sockets = [
            self.create_socket(self.first_input_port, identity="1"),
            self.create_socket(self.first_input_port, identity="2"),
            self.create_socket(self.second_input_port, identity="3"),
            self.create_socket(self.second_input_port, identity="4"),
            self.create_socket(self.second_input_port, identity="5"),
        ]
 
        for input_socket in input_sockets:
            self.send(input_socket, "h")
            self.assertIsMsgType(input_socket, "c")
 
        # At the beginning, everything should be at normal state
        self.assertMonitoring(first_monitoring_socket, "input.registered_workers", 2)
        self.assertMonitoring(second_monitoring_socket, "input.registered_workers", 3)
        self.assertMonitoring(final_monitoring_socket, "input.registered_workers", 2)
 
        # So far no stop messages should be there
        self.assertMonitoring(first_monitoring_socket, "input.received_stop_messages", 0)
        self.assertMonitoring(second_monitoring_socket, "input.received_stop_messages", 0)
        self.assertMonitoring(final_monitoring_socket, "input.received_stop_messages", 0)
 
        self.assertMonitoring(first_monitoring_socket, "input.all_stop_messages", False)
        self.assertMonitoring(second_monitoring_socket, "input.all_stop_messages", False)
        self.assertMonitoring(final_monitoring_socket, "input.all_stop_messages", False)
 
        # send some events, which should eventually trigger a merge
        for _ in range(10):
            for input_socket in input_sockets:
                self.send(input_socket, "v", self.histogram_data, self.event_data)
                self.assertIsMsgType(input_socket, "c")
 
        self.assertMonitoring(first_monitoring_socket, "input.received_events", 20)
        self.assertMonitoring(second_monitoring_socket, "input.received_events", 30)
        self.assertMonitoring(final_monitoring_socket, "input.received_events", 2)
 
        # Make sure to get the newest data
        self.check_histogram_repeated("outputFile.root", 5)
 
        # send out stop messages, which should also trigger a merge
        for input_socket in input_sockets:
            self.send(input_socket, "l")
            self.assertIsMsgType(input_socket, "c")
 
        self.assertMonitoring(first_monitoring_socket, "input.received_stop_messages", 2)
        self.assertMonitoring(first_monitoring_socket, "input.all_stop_messages", True)
        self.assertMonitoring(second_monitoring_socket, "input.received_stop_messages", 3)
        self.assertMonitoring(second_monitoring_socket, "input.all_stop_messages", True)
 
        self.assertMonitoring(final_monitoring_socket, "input.received_stop_messages", 2)
        self.assertMonitoring(final_monitoring_socket, "input.all_stop_messages", True)
        self.assertHasOutputFile("outputFile.root", unlink=False)
        self.check_histogram_repeated("outputFile.root", 5)
 
        # Now we clean up
        self.send(first_monitoring_socket, "n")
        self.send(second_monitoring_socket, "n")
        self.send(final_monitoring_socket, "n")
 
        # which should give us a clean root file again
        self.assertNotHasOutputFile("outputFile.root", timeout=1)
 
        self.assertMonitoring(first_monitoring_socket, "input.received_stop_messages", 0)
        self.assertMonitoring(first_monitoring_socket, "input.all_stop_messages", False)
        self.assertMonitoring(second_monitoring_socket, "input.received_stop_messages", 0)
        self.assertMonitoring(second_monitoring_socket, "input.all_stop_messages", False)
 
        self.assertMonitoring(final_monitoring_socket, "input.received_stop_messages", 0)
        self.assertMonitoring(final_monitoring_socket, "input.all_stop_messages", False)
 
        # and send some more events, but only to the first two clients
        for _ in range(5):
            for input_socket in input_sockets[:2]:
                self.send(input_socket, "v", self.histogram_data, self.event_data)
                self.assertIsMsgType(input_socket, "c")
 
        # this time we expect less entries
        self.check_histogram_repeated("outputFile.root", 2)
 
        # Now send a terminate message
        for input_socket in input_sockets:
            self.send(input_socket, "x")
            self.assertIsMsgType(input_socket, "c")
 
        # We expect another merge here
        self.assertHasOutputFile("outputFile.root", unlink=False)
        self.check_histogram_repeated("outputFile.root", 2)
 
        self.assertIsDown("first")
        self.assertIsDown("second")
        self.assertIsDown("final_collector")
 

Member Data Documentation

event_data

b 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

final_collector_input_port

Definition at line 75 of file test_histogram.py.

final_collector_monitoring_port

final_collector_monitoring_port

final_collector_monitoring_port

Definition at line 77 of file test_histogram.py.

first_input_port

first_input_port

first_input_port

Definition at line 65 of file test_histogram.py.

first_monitoring_port

first_monitoring_port

first_monitoring_port

Definition at line 67 of file test_histogram.py.

histogram_data

open 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

needed_programs

Definition at line 79 of file test_histogram.py.

second_input_port

second_input_port

second_input_port

Definition at line 70 of file test_histogram.py.

second_monitoring_port

second_monitoring_port

second_monitoring_port

Definition at line 72 of file test_histogram.py.

---

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

• daq/hbasf2/tests/test_histogram.py