test_distributor.py

 
 
from unittest import main
import basf2
 
import zmq
 
from zmq_daq.test_support import HLTZMQTestCase, ZMQ_TEST_FOR_LOOPS, ZMQ_TEST_MAX_FAILURES
 
 
class DistributorTestCase(HLTZMQTestCase):
    """Test case"""
    
    event_data = open(basf2.find_file("daq/hbasf2/tests/out.raw"), "br").read()
 
    def setUp(self):
        """Setup port numbers and necessary programs"""
        
        self.input_portinput_port = HLTZMQTestCase.get_free_port()
        
        self.output_portoutput_port = HLTZMQTestCase.get_free_port()
        
        self.monitoring_portmonitoring_port = HLTZMQTestCase.get_free_port()
        
        self.needed_programsneeded_programsneeded_programs = {
            "distributor": [
                "b2hlt_distributor",
                "--input", f"tcp://localhost:{self.input_port}",
                "--output", f"tcp://*:{self.output_port}",
                "--monitor", f"tcp://*:{self.monitoring_port}",
                "--stopWaitingTime", "1"]}
        super().setUp()
 
    def testConnectAndDisconnect(self):
        """test function"""
        monitoring_socket = self.create_socketcreate_socket(self.monitoring_portmonitoring_port)
 
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_state", "disconnected")
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_connects", 0)
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_disconnects", 0)
 
        # To make it actually look for input messages, it needs at least a single worker
        output_socket = self.create_socketcreate_socket(self.output_portoutput_port)
        for _ in range(5):
            self.sendsend(output_socket, "r")
 
        # Now we can go on
        input_socket = self.create_socketcreate_socket(self.input_portinput_port, socket_type=zmq.STREAM, bind=True)
 
        # Still no connection
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_state", "disconnected")
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_connects", 0)
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_disconnects", 0)
 
        # This should initiate a connection
        identity, _ = self.recvrecv(input_socket)
 
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_state", "connected")
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_connects", 1)
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_disconnects", 0)
 
        # And we can also close it again
        input_socket.close()
 
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_state", "disconnected")
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_connects", 1)
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_disconnects", 1)
 
        # or open
        input_socket = self.create_socketcreate_socket(self.input_portinput_port, socket_type=zmq.STREAM, bind=True)
        identity, _ = self.recvrecv(input_socket)
 
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_state", "connected")
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_connects", 2)
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_disconnects", 1)
 
        # sending an event should not change anything
        input_socket.send_multipart([identity, self.event_dataevent_data])
        input_socket.send_multipart([identity, self.event_dataevent_data])
        input_socket.send_multipart([identity, self.event_dataevent_data])
        input_socket.send_multipart([identity, self.event_dataevent_data])
 
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.received_events", 4)
 
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_state", "connected")
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_connects", 2)
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_disconnects", 1)
 
        input_socket.close()
 
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_state", "disconnected")
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_connects", 2)
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_disconnects", 2)
 
    def testEvents(self):
        """test function"""
        monitoring_socket = self.create_socketcreate_socket(self.monitoring_portmonitoring_port)
 
        # connect the input
        input_socket = self.create_socketcreate_socket(self.input_portinput_port, socket_type=zmq.STREAM, bind=True)
        identity, _ = self.recvrecv(input_socket)
 
        # Start two workers
        output_socket = self.create_socketcreate_socket(self.output_portoutput_port)
        second_output_socket = self.create_socketcreate_socket(self.output_portoutput_port, identity="other_socket")
 
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.ready_queue_size", 0)
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.registered_workers", 0)
 
        # Every ready should give us an event at exactly this worker, when there is an event
 
        # 1. only send ready message
        self.sendsend(output_socket, "r")
 
        # No event
        self.assertNothingMoreassertNothingMore(output_socket)
        self.assertNothingMoreassertNothingMore(second_output_socket)
 
        # But ready worker
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.ready_queue_size", 1)
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.registered_workers", 1)
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.ready_messages[socket]", 1)
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_state", "connected")
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_connects", 1)
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_disconnects", 0)
 
        # Now the event
        input_socket.send_multipart([identity, self.event_dataevent_data])
 
        self.assertIsMsgTypeassertIsMsgType(output_socket, "u")
        self.assertNothingMoreassertNothingMore(second_output_socket)
 
        # And no ready worker
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.ready_queue_size", 0)
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.registered_workers", 1)
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.ready_messages[socket]", 0)
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_state", "connected")
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_connects", 1)
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_disconnects", 0)
 
        # 2. Try out sending the event first
        input_socket.send_multipart([identity, self.event_dataevent_data])
 
        # Still no event
        self.assertNothingMoreassertNothingMore(output_socket)
        self.assertNothingMoreassertNothingMore(second_output_socket)
 
        # And no updated monitoring
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.ready_queue_size", 0)
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.registered_workers", 1)
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.ready_messages[socket]", 0)
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_state", "connected")
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_connects", 1)
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_disconnects", 0)
 
        # but with a ready message
        self.sendsend(second_output_socket, "r")
 
        self.assertNothingMoreassertNothingMore(output_socket)
        self.assertIsMsgTypeassertIsMsgType(second_output_socket, "u")
 
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.ready_queue_size", 0)
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.registered_workers", 2)
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_state", "connected")
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_connects", 1)
        self.assertMonitoringassertMonitoring(monitoring_socket, "input.socket_disconnects", 0)
 
        # and again
        input_socket.send_multipart([identity, self.event_dataevent_data])
 
        self.sendsend(second_output_socket, "r")
        self.assertNothingMoreassertNothingMore(output_socket)
        self.assertIsMsgTypeassertIsMsgType(second_output_socket, "u")
 
        # As we have answered all ready messages, nothing should be in the queue
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.ready_queue_size", 0)
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.registered_workers", 2)
 
        # until we start sending more ready messages
        self.sendsend(output_socket, "r")
        self.sendsend(output_socket, "r")
        self.sendsend(second_output_socket, "r")
        self.sendsend(second_output_socket, "r")
 
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.ready_queue_size", 4)
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.registered_workers", 2)
 
    def testEndRun(self):
        """test function"""
        monitoring_socket = self.create_socketcreate_socket(self.monitoring_portmonitoring_port)
 
        # Start two workers
        output_socket = self.create_socketcreate_socket(self.output_portoutput_port)
        second_output_socket = self.create_socketcreate_socket(self.output_portoutput_port, identity="other_socket")
 
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.ready_queue_size", 0)
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.registered_workers", 0)
 
        # and register them by sending a ready
        self.sendsend(output_socket, "r")
        self.sendsend(second_output_socket, "r")
 
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.ready_queue_size", 2)
        self.assertMonitoringassertMonitoring(monitoring_socket, "output.registered_workers", 2)
 
        # a stop run should be sent to all workers
        self.sendsend(monitoring_socket, "l")
 
        self.assertIsMsgTypeassertIsMsgType(output_socket, "l")
        self.assertIsMsgTypeassertIsMsgType(second_output_socket, "l")
 
        # also multiple times
        self.sendsend(monitoring_socket, "l")
 
        self.assertIsMsgTypeassertIsMsgType(output_socket, "l")
        self.assertIsMsgTypeassertIsMsgType(second_output_socket, "l")
 
        # and if there are events in between they should not be transported
        input_socket = self.create_socketcreate_socket(self.input_portinput_port, socket_type=zmq.STREAM, bind=True)
        identity, _ = self.recvrecv(input_socket)
        input_socket.send_multipart([identity, self.event_dataevent_data])
        input_socket.send_multipart([identity, self.event_dataevent_data])
 
        self.assertNothingMoreassertNothingMore(output_socket)
        self.assertNothingMoreassertNothingMore(second_output_socket)
 
        # it shouldn't matter
        self.sendsend(monitoring_socket, "l")
 
        self.assertIsMsgTypeassertIsMsgType(output_socket, "l")
        self.assertIsMsgTypeassertIsMsgType(second_output_socket, "l")
 
        # A terminate message should also be sent
        self.sendsend(monitoring_socket, "x")
 
        self.assertIsMsgTypeassertIsMsgType(output_socket, "x")
        self.assertIsMsgTypeassertIsMsgType(second_output_socket, "x")
 
        # and the distributor should go down
        self.assertIsDownassertIsDown("distributor")
 
 
if __name__ == '__main__':
    
    number_of_failures = 0
 
    for i in range(ZMQ_TEST_FOR_LOOPS):
        try:
            main(exit=False)
        except AssertionError:
            number_of_failures += 1
 
    
    message = f'Number of failed for loops: {number_of_failures}/{ZMQ_TEST_FOR_LOOPS}'
    if number_of_failures <= ZMQ_TEST_MAX_FAILURES:
        basf2.B2INFO(message)
    else:
        basf2.B2FATAL(message)