BaseCollectorTestCase Class Reference

Inheritance diagram for BaseCollectorTestCase:

Public Member Functions
def	setUp (self)
def	create_output_socket (self)
def	get_signal (self, output_socket, signal_type)
def	get_event (self, output_socket)
def	testHelloAndMessageTransmission (self)
def	testWrongRegistration (self)
def	testEventPropagation (self)

Public Attributes
	input_port
	input_port
	output_port
	output_port
	monitoring_port
	monitoring_port
	needed_programs

Static Public Attributes
bool	final_collector = False
	final_collector

Detailed Description

As the collectors are mostly equal, use a common base test case class

Definition at line 254 of file test_support.py.

Member Function Documentation

create_output_socket()

def create_output_socket

(

self

)

create the output socket depending if final collector or not

Definition at line 283 of file test_support.py.

    def create_output_socket(self):
        """create the output socket depending if final collector or not"""
        if self.final_collector:
            output_socket = self.create_socket(self.output_port, socket_type=zmq.STREAM, bind=True)
            output_socket.send(b"")
            self.recv(output_socket)
        else:
            output_socket = self.create_socket(self.output_port)
            self.send(output_socket, "r")
        return output_socket
 

get_event()

def get_event	(		self,
			output_socket
	)

get an event from the socket depending if final collector or not

Definition at line 301 of file test_support.py.

    def get_event(self, output_socket):
        """get an event from the socket depending if final collector or not"""
        if self.final_collector:
            self.recv(output_socket)
            self.assertNothingMore(output_socket)
        else:
            self.assertIsMsgType(output_socket, "u")
 

get_signal()

def get_signal	(		self,
			output_socket,
			signal_type
	)

get a signal from the socket depending if final collector or not

Definition at line 294 of file test_support.py.

    def get_signal(self, output_socket, signal_type):
        """get a signal from the socket depending if final collector or not"""
        if self.final_collector:
            self.assertNothingMore(output_socket)
        else:
            self.assertIsMsgType(output_socket, signal_type, final=True)
 

setUp()

def setUp

(

self

)

Setup port numbers and necessary programs

Reimplemented from HLTZMQTestCase.

Definition at line 261 of file test_support.py.

    def setUp(self):
        """Setup port numbers and necessary programs"""
        
        self.input_port = HLTZMQTestCase.get_free_port()
        
        self.output_port = HLTZMQTestCase.get_free_port()
        
        self.monitoring_port = HLTZMQTestCase.get_free_port()
 
        command = "b2hlt_finalcollector" if self.final_collector else "b2hlt_collector"
        output = "localhost" if self.final_collector else "*"
        self.needed_programs = {
            "collector": [
                command,
                "--input", f"tcp://*:{self.input_port}",
                "--output", f"tcp://{output}:{self.output_port}",
                "--monitor", f"tcp://*:{self.monitoring_port}"
            ]
        }
        # programs are setup, call parent setup function now
        super().setUp()
 

testEventPropagation()

def testEventPropagation

(

self

)

test function

Definition at line 473 of file test_support.py.

    def testEventPropagation(self):
        """test function"""
        monitoring_socket = self.create_socket(self.monitoring_port)
 
        # Send two ready messages from the first socket
        output_socket = self.create_output_socket()
        if not self.final_collector:
            self.send(output_socket, "r")
 
            self.assertMonitoring(monitoring_socket, "output.ready_queue_size", 2)
            self.assertMonitoring(monitoring_socket, "output.registered_workers", 1)
 
        if not self.final_collector:
            # Send two ready message from the second socket (the last one is needed to keep the collector listening)
            second_output_socket = self.create_socket(self.output_port, identity="other_socket")
            self.send(second_output_socket, "r")
            self.send(second_output_socket, "r")
 
            self.assertMonitoring(monitoring_socket, "output.ready_queue_size", 4)
            self.assertMonitoring(monitoring_socket, "output.registered_workers", 2)
 
        # Register two workers
        input_socket = self.create_socket(self.input_port)
        self.send(input_socket, "h")
        self.assertIsMsgType(input_socket, "c")
        self.assertMonitoring(monitoring_socket, "input.registered_workers", 1)
 
        second_input_socket = self.create_socket(self.input_port, identity="other_socket")
        self.send(second_input_socket, "h")
        self.assertIsMsgType(second_input_socket, "c")
        self.assertMonitoring(monitoring_socket, "input.registered_workers", 2)
 
        # The first event should go to the first worker
        self.send(input_socket, "u", b"event data")
        self.assertIsMsgType(input_socket, "c")
 
        self.get_event(output_socket)
        if not self.final_collector:
            self.assertNothingMore(second_output_socket)
 
        # The second also
        self.send(second_input_socket, "u", b"event data")
        self.assertIsMsgType(second_input_socket, "c")
 
        self.get_event(output_socket)
        if not self.final_collector:
            self.assertNothingMore(second_output_socket)
 
        # But the third to the second worker
        self.send(input_socket, "u", b"event data")
        self.assertIsMsgType(input_socket, "c")
 
        if not self.final_collector:
            self.assertNothingMore(output_socket)
            self.get_event(second_output_socket)
        else:
            self.get_event(output_socket)
 
        # A stop message should be sent to all workers
        self.send(input_socket, "l")
        self.assertIsMsgType(input_socket, "c")
 
        # But only if it is complete...
        self.assertNothingMore(output_socket)
        if not self.final_collector:
            self.assertNothingMore(second_output_socket)
 
        self.send(second_input_socket, "l")
        self.assertIsMsgType(second_input_socket, "c")
 
        self.get_signal(output_socket, "l")
        if not self.final_collector:
            self.get_signal(second_output_socket, "l")
 
        # TODO: Test and implement: should not transmit events after stop
 
        # As well as a terminate message
        self.send(input_socket, "x")
        self.assertIsMsgType(input_socket, "c")
 
        self.assertNothingMore(output_socket)
        if not self.final_collector:
            self.assertNothingMore(second_output_socket)
 
        self.send(second_input_socket, "x")
        self.assertIsMsgType(second_input_socket, "c")
 
        self.get_signal(output_socket, "x")
        if not self.final_collector:
            self.get_signal(second_output_socket, "x")
 
        self.assertIsDown("collector")

testHelloAndMessageTransmission()

def testHelloAndMessageTransmission

(

self

)

test function

Definition at line 309 of file test_support.py.

    def testHelloAndMessageTransmission(self):
        """test function"""
        monitoring_socket = self.create_socket(self.monitoring_port)
 
        # Register first worker
        input_socket = self.create_socket(self.input_port)
        self.send(input_socket, "h")
 
        # However the collector is only polling for ready now...
        self.assertNothingMore(input_socket)
        self.assertMonitoring(monitoring_socket, "input.registered_workers", 0)
 
        # To make it actually look for input messages, it needs at least a single worker
        output_socket = self.create_output_socket()
 
        # Now we can go on
        self.assertIsMsgType(input_socket, "c")
        self.assertMonitoring(monitoring_socket, "input.registered_workers", 1)
 
        # Register second worker
        second_input_socket = self.create_socket(self.input_port, identity="other_socket")
        self.send(second_input_socket, "h")
        self.assertIsMsgType(second_input_socket, "c")
        self.assertMonitoring(monitoring_socket, "input.registered_workers", 2)
 
        # 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.assertNothingMore(output_socket)
 
        # The second stop message should
        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.get_signal(output_socket, "l")
 
        # Another stop message should not change anything
        self.send(input_socket, "l")
        self.assertIsMsgType(input_socket, "c")
        self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 2)
        self.assertMonitoring(monitoring_socket, "input.all_stop_messages", True)
        self.assertNothingMore(output_socket)
 
        # But if we reset...
        self.send(monitoring_socket, "n")
        self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 0)
        self.assertMonitoring(monitoring_socket, "input.all_stop_messages", False)
 
        # .. it should
        self.send(input_socket, "l")
        self.assertIsMsgType(input_socket, "c")
        self.send(second_input_socket, "l")
        self.assertIsMsgType(second_input_socket, "c")
 
        self.get_signal(output_socket, "l")
 
        # Now we reset again
        self.send(monitoring_socket, "n")
        self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 0)
        self.assertMonitoring(monitoring_socket, "input.all_stop_messages", False)
 
        # send just one stop message
        self.send(input_socket, "l")
        self.assertIsMsgType(input_socket, "c")
 
        # and unregister the second
        self.send(second_input_socket, "d", b"other_socket")
        self.assertIsMsgType(second_input_socket, "c")
        self.assertMonitoring(monitoring_socket, "input.registered_workers", 1)
 
        # which should also give us a stop message
        self.get_signal(output_socket, "l")
        self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 1)
        self.assertMonitoring(monitoring_socket, "input.all_stop_messages", True)
 
        # lets register and reset it again
        self.send(second_input_socket, "h")
        self.assertIsMsgType(second_input_socket, "c")
        self.send(monitoring_socket, "n")
        self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 0)
        self.assertMonitoring(monitoring_socket, "input.all_stop_messages", False)
        self.assertMonitoring(monitoring_socket, "input.registered_workers", 2)
 
        # and try the other way round: first unregister, then send stop
        self.send(second_input_socket, "d", b"other_socket")
        self.assertIsMsgType(second_input_socket, "c")
        self.send(input_socket, "l")
        self.assertIsMsgType(input_socket, "c")
        self.assertMonitoring(monitoring_socket, "input.registered_workers", 1)
        # which should also give us a stop message
        self.get_signal(output_socket, "l")
        self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 1)
        self.assertMonitoring(monitoring_socket, "input.all_stop_messages", True)
 
        # reset the state
        self.send(second_input_socket, "h")
        self.assertIsMsgType(second_input_socket, "c")
        self.send(monitoring_socket, "n")
        self.assertMonitoring(monitoring_socket, "input.received_stop_messages", 0)
        self.assertMonitoring(monitoring_socket, "input.all_stop_messages", False)
        self.assertMonitoring(monitoring_socket, "input.registered_workers", 2)
 
        # The same applies to terminate messages:
        # Nothing at the beginning
        self.assertMonitoring(monitoring_socket, "input.received_terminate_messages", 0)
        self.assertMonitoring(monitoring_socket, "input.all_terminate_messages", False)
 
        # the first terminate message should not trigger a transmission
        self.send(input_socket, "x")
        self.assertIsMsgType(input_socket, "c")
        self.assertMonitoring(monitoring_socket, "input.received_terminate_messages", 1)
        self.assertMonitoring(monitoring_socket, "input.all_terminate_messages", False)
        self.assertNothingMore(output_socket)
 
        # Another terminate message should not change anything
        self.send(input_socket, "x")
        self.assertIsMsgType(input_socket, "c")
        self.assertMonitoring(monitoring_socket, "input.received_terminate_messages", 1)
        self.assertMonitoring(monitoring_socket, "input.all_terminate_messages", False)
        self.assertNothingMore(output_socket)
 
        # But if we reset...
        self.send(monitoring_socket, "n")
        self.assertMonitoring(monitoring_socket, "input.received_terminate_messages", 0)
        self.assertMonitoring(monitoring_socket, "input.all_terminate_messages", False)
 
        # ... and send again ...
        self.send(input_socket, "x")
        self.assertIsMsgType(input_socket, "c")
        self.send(second_input_socket, "x")
        self.assertIsMsgType(second_input_socket, "c")
 
        self.get_signal(output_socket, "x")
 
        # ... and the collector should have terminated
        self.assertIsDown("collector")
 
        # TODO: test and implement: timeout in wait for stop or terminate messages
 

testWrongRegistration()

def testWrongRegistration

(

self

)

test function

Definition at line 456 of file test_support.py.

    def testWrongRegistration(self):
        """test function"""
        # To make it actually look for input messages, it needs at least a single worker
        self.create_output_socket()
 
        # Register first worker
        input_socket = self.create_socket(self.input_port)
        self.send(input_socket, "h")
        self.assertIsMsgType(input_socket, "c")
 
        # Send with a second, unregistered worker
        second_input_socket = self.create_socket(self.input_port, identity="other_socket")
        self.send(second_input_socket, "l")
 
        # The collector should die
        self.assertIsDown("collector")
 

Member Data Documentation

final_collector

bool final_collector = False

static

final_collector

Definition at line 259 of file test_support.py.

input_port

input_port

input_port

Definition at line 264 of file test_support.py.

monitoring_port

monitoring_port

monitoring_port

Definition at line 268 of file test_support.py.

needed_programs

needed_programs

Definition at line 272 of file test_support.py.

output_port

output_port

output_port

Definition at line 266 of file test_support.py.

---

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

• daq/hbasf2/scripts/zmq_daq/test_support.py