CollectorTestCase Class Reference

Inheritance diagram for CollectorTestCase:

Public Member Functions
	setUp (self)
	create_output_socket (self)
	get_signal (self, output_socket, signal_type)
	get_event (self, output_socket)
	testHelloAndMessageTransmission (self)
	testWrongRegistration (self)
	testEventPropagation (self)
	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
	input_port = HLTZMQTestCase.get_free_port()
	input_port
	output_port = HLTZMQTestCase.get_free_port()
	output_port
	monitoring_port = HLTZMQTestCase.get_free_port()
	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
bool	final_collector = False
	final_collector
	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 14 of file test_collector.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.

    def _is_running(self, name):
        """
        Check if a given program is still running.
        """
        process = self.started_programs[name]
        pid, sts = process._try_wait(os.WNOHANG)
        assert pid == process.pid or pid == 0
        return pid == 0
 

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.

    def assertHasOutputFile(self, output_file, unlink=True, timeout=0.5, minimum_delay=0.1):
        """
        Assert that - at least after the given timeout - the output file
        is present. If unlink is set to True, remove the file after checking.
        """
        endtime = time() + timeout
 
        while True:
            if os.path.exists(output_file):
                if unlink:
                    os.unlink(output_file)
                return
 
            remaining = endtime - time()
            self.assertFalse(remaining <= 0)
 
            sleep(minimum_delay)
 

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.

    def assertIsAndGet(self, socket, message_type, final=True, router=False):
        """
        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.
        """
        answer = HLTZMQTestCase.recv(socket)
        type_index = 0
        if router:
            type_index = 1
        self.assertEqual(answer[type_index], message_type.encode())
        if final:
            self.assertNothingMore(socket)
        return answer
 

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.

    def assertIsDown(self, name, timeout=5, minimum_delay=0.1):
        """
        Test helper to assert the given program has terminated - at least after timeout in seconds has passed.
        Checks every "minimal_delay seconds.
        """
        endtime = time() + timeout
        while True:
            if not self._is_running(name):
                return
 
            remaining = endtime - time()
            self.assertFalse(remaining <= 0)
 
            sleep(minimum_delay)
 

assertIsMsgType()

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

inherited

Deprecated copy of "assertIsAndGet".

Definition at line 215 of file test_support.py.

    def assertIsMsgType(self, socket, message_type, final=True, router=False):
        """
        Deprecated copy of "assertIsAndGet".
        """
        return self.assertIsAndGet(socket, message_type, final=final, router=router)
 

assertIsRunning()

assertIsRunning ( self, name )

inherited

Assert that a given program is still running.

Definition at line 108 of file test_support.py.

    def assertIsRunning(self, name):
        """
        Assert that a given program is still running.
        """
        self.assertTrue(self._is_running(name))
 

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.

    def assertMonitoring(self, socket, search_key, search_value, timeout=10):
        """
        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>".
        """
        end_time = time() + timeout
        monitoring = dict()
        while time() < end_time:
            HLTZMQTestCase.send(socket, "m")
            answer = self.assertIsAndGet(socket, "c")
 
            dict_monitoring = json.loads(answer[1])
            for parent_key, parent_dict in dict_monitoring.items():
                for key, value in parent_dict.items():
                    monitoring[parent_key + "." + key] = value
 
            if search_key in monitoring and monitoring[search_key] == search_value:
                break
        else:
            if monitoring:
                if search_key not in monitoring:
                    raise AssertionError(f"Monitoring did not have a result with key {search_key}")
                else:
                    raise AssertionError(
                        f"Monitoring did not show the result {search_value} for {search_key}, instead {monitoring[search_key]}")
            else:
                raise AssertionError("Monitoring did not answer in time.")
 
        self.assertNothingMore(socket)
 

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.

    def assertNotHasOutputFile(self, output_file, timeout=0.5):
        """
        Assert that after the timeout the given file is not present
        (a.k.a. no process has created it)
        """
        sleep(timeout)
        self.assertFalse(os.path.exists(output_file))
 
 

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.

    def assertNothingMore(self, socket):
        """
        Assert that there is no pending message to be received on the socket.
        """
        self.assertFalse(socket.poll(0))
 

create_output_socket()

create_output_socket ( self )

inherited

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
 

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.

    def create_router_socket(port):
        """
        Shortcut to create a ROUTER type socket with the typical parameters
        binding to the given port.
        """
        return HLTZMQTestCase.create_socket(port, socket_type=zmq.ROUTER, identity="", bind=True)
 

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.

    def create_socket(port, socket_type=zmq.DEALER, identity="socket", bind=False):
        """
        Create and return a ZMQ socket with the given type and identity and
        bind or connect it to localhost and the given port.
        """
        socket = HLTZMQTestCase.ctx.socket(socket_type)
        socket.rcvtimeo = 10000
        socket.linger = 0
        if identity:
            socket.setsockopt_string(zmq.IDENTITY, identity)
        if bind:
            if port is None:
                port = socket.bind_to_random_port("tcp://*")
                return socket, port
            else:
                socket.bind(f"tcp://*:{port}")
        else:
            if port is None:
                raise RuntimeError("Cannot connect to unknown port")
 
            socket.connect(f"tcp://localhost:{port}")
 
        return socket
 

get_event()

get_event ( self, output_socket )

inherited

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

    def get_free_port():
        """
        Get a free port number by reusing ZMQ's function for this.
        """
        socket = HLTZMQTestCase.ctx.socket(zmq.ROUTER)
        port = socket.bind_to_random_port("tcp://*")
        socket.close()
        return port
 

get_signal()

get_signal ( self, output_socket, signal_type )

inherited

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)
 

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.

    def recv(socket):
        """
        Try to receive a message from the socket (or throw an assertion error if none comes after the set timeout
        of the socket).
        """
        try:
            return socket.recv_multipart()
        except zmq.error.Again:
            raise AssertionError("No answer from socket")
 

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.

    def send(socket, message_type, first_data=b"", second_data=b"", identity=""):
        """
        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.
        """
        if identity:
            socket.send_multipart([identity.encode(), message_type.encode(), first_data, second_data])
        else:
            socket.send_multipart([message_type.encode(), first_data, second_data])
 

setUp()

setUp ( self )

inherited

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()
 

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.

    def tearDown(self):
        """
        Custom tearDown function to kill the started programs if still present
        and remove the temporary folder again.
        """
        for name, process in self.started_programs.items():
            if self._is_running(name):
                os.killpg(process.pid, signal.SIGKILL)
            process.wait()
        os.chdir(self.previous_dir)
        shutil.rmtree(self.test_dir)
 
        atexit._clear()
 

testEventPropagation()

testEventPropagation ( self )

inherited

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()

testHelloAndMessageTransmission ( self )

inherited

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()

testWrongRegistration ( self )

inherited

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

ctx

ctx = zmq.Context()

staticinherited

The ZMQ context.

Definition at line 36 of file test_support.py.

final_collector

bool final_collector = False

staticinherited

final_collector

Definition at line 259 of file test_support.py.

input_port

input_port = HLTZMQTestCase.get_free_port()

inherited

input_port

Definition at line 264 of file test_support.py.

monitoring_port

monitoring_port = HLTZMQTestCase.get_free_port()

inherited

monitoring_port

Definition at line 268 of file test_support.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.

output_port

output_port = HLTZMQTestCase.get_free_port()

inherited

output_port

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

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:

• daq/hbasf2/tests/test_collector.py