HLTZMQTestCase Class Reference

Inheritance diagram for HLTZMQTestCase:

Public Member Functions
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
zmq	ctx = zmq.Context()
	The ZMQ context.
dict	needed_programs = dict()
	The dict name -> cmd args of the programs to start, needs to be set in each test.

Protected Member Functions
def	_is_running (self, name)

Detailed Description

Base class for all HLT ZMQ tests helping to start the needed programs,
create ZMQ sockets and send/receive messages easily.
Has also some functionality for asserting test-correctness

Definition at line 29 of file test_support.py.

Member Function Documentation

_is_running()

def _is_running (   self,   name  )

protected

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

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.

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

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.

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

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.

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

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

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

def assertIsRunning	(		self,
			name
	)

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

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

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

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)

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

def assertNothingMore	(		self,
			socket
	)

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

def create_router_socket (   port )

static

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

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

static

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

def get_free_port ( )

static

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
 

recv()

def recv (   socket )

static

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

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

static

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

def setUp

(

self

)

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

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

Definition at line 50 of file test_support.py.

    def setUp(self):
        """
        Custom setUp function to go into a temporary folder
        and start the needed programs.
        """
        
        self.test_dir = tempfile.mkdtemp()
        
        self.previous_dir = os.getcwd()
        os.chdir(self.test_dir)
 
        
        self.started_programs = dict()
        for name, command in self.needed_programs.items():
            self.started_programs[name] = subprocess.Popen(command, start_new_session=True)
            self.assertIsRunning(name)
 
        atexit._clear()
        atexit.register(self.tearDown)
 

tearDown()

def tearDown

(

self

)

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

Member Data Documentation

ctx

zmq ctx = zmq.Context()

static

The ZMQ context.

Definition at line 36 of file test_support.py.

needed_programs

dict needed_programs = dict()

static

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.

previous_dir

previous_dir

remember current working directory

Definition at line 58 of file test_support.py.

started_programs

started_programs

dict for all started programs

Definition at line 62 of file test_support.py.

test_dir

test_dir

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/scripts/zmq_daq/test_support.py