Inheritance diagram for WorkerTestCase:

[legend]

Collaboration diagram for WorkerTestCase:

[legend]

Public Member Functions
def	setUp (self)

def	start (self)

Public Attributes
	needed_programs
	input_socket

	output_identity
	output_identity

	input_identity
	input_identity

	first_run_event_data
	some data

	second_run_event_data
	some data

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

list	extra_arguments = []
	extra arguments to pass to the worker script

Detailed Description

Test case baseclass to spawn a worker

Definition at line 16 of file test_worker.py.

Member Function Documentation

◆ setUp()

def setUp ( self )

Setup necessary sockets and programs

Reimplemented from HLTZMQTestCase.

Definition at line 24 of file test_worker.py.

    def setUp(self):
        """Setup necessary sockets and programs"""
        
        self.input_socket, input_port = self.create_router_socket(None)
        
        self.output_socket, output_port = self.create_router_socket(None)
        # and set the list of necessary programs to use these sockets
        self.needed_programs = {
            "worker": [
                "python3", basf2.find_file("daq/hbasf2/tests/passthrough.no_run_py"),
                "--input", f"tcp://localhost:{input_port}",
                "--output", f"tcp://localhost:{output_port}",
            ] + self.extra_arguments
        }
        super().setUp()
 

◆ start()

def start ( self )

start the needed sockets and send some hello messages

Definition at line 40 of file test_worker.py.

    def start(self):
        """start the needed sockets and send some hello messages"""
        # There should be a hello message
        
        self.output_identity = self.assertIsMsgType(self.output_socket, "h", router=True)[0].decode()
        self.send(self.output_socket, "c", identity=self.output_identity)
 
        # There are probably many more ready messages, but we are only interested in at least one here
        
        self.input_identity = self.assertIsMsgType(self.input_socket, "r", router=True, final=False)[0].decode()
 
        # Store some example events
        
        self.first_run_event_data = [self.event_data, self.event_data]
        
        self.second_run_event_data = [self.event_data, self.event_data]

Member Data Documentation

◆ event_data

open event_data = open(basf2.find_file("daq/hbasf2/tests/out.raw"), "br").read()

static

event_data

Definition at line 19 of file test_worker.py.

◆ extra_arguments

list extra_arguments = []

static

extra arguments to pass to the worker script

Definition at line 22 of file test_worker.py.

◆ first_run_event_data

first_run_event_data

some data

Definition at line 53 of file test_worker.py.

◆ input_identity

input_identity

Definition at line 49 of file test_worker.py.

◆ needed_programs

needed_programs

input_socket

output_socket

Definition at line 31 of file test_worker.py.

◆ output_identity

output_identity

Definition at line 44 of file test_worker.py.

◆ second_run_event_data

second_run_event_data