test_worker.py

 
 
from pathlib import Path
from unittest import main
import basf2
 
from zmq_daq.test_support import HLTZMQTestCase, ZMQ_TEST_FOR_LOOPS, ZMQ_TEST_MAX_FAILURES
 
 
class WorkerTestCase(HLTZMQTestCase):
    """Test case baseclass to spawn a worker"""
    
    event_data = open(basf2.find_file("daq/hbasf2/tests/out.raw"), "br").read()
 
    
    extra_arguments = []
 
    def setUp(self):
        """Setup necessary sockets and programs"""
        
        self.input_socket, input_port = self.create_router_socketcreate_router_socket(None)
        
        self.output_socket, output_port = self.create_router_socketcreate_router_socket(None)
        # and set the list of necessary programs to use these sockets
        self.needed_programsneeded_programsneeded_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_argumentsextra_arguments
        }
        super().setUp()
 
    def start(self):
        """start the needed sockets and send some hello messages"""
        # There should be a hello message
        
        self.output_identityoutput_identity = self.assertIsMsgTypeassertIsMsgType(self.output_socket, "h", router=True)[0].decode()
        self.sendsend(self.output_socket, "c", identity=self.output_identityoutput_identity)
 
        # There are probably many more ready messages, but we are only interested in at least one here
        
        self.input_identityinput_identity = self.assertIsMsgTypeassertIsMsgType(self.input_socket, "r", router=True, final=False)[0].decode()
 
        # Store some example events
        
        self.first_run_event_datafirst_run_event_data = [self.event_dataevent_data, self.event_dataevent_data]
        
        self.second_run_event_datasecond_run_event_data = [self.event_dataevent_data, self.event_dataevent_data]
 
 
class NormalWorkerTestCase(WorkerTestCase):
    """Tests for normal worker behavior"""
 
    def testInitialization(self):
        """test function"""
        self.startstart()
 
        # Initialisation should be called
        self.assertHasOutputFileassertHasOutputFile("initialize_called", timeout=1)
 
    def testRunSending(self):
        """test function"""
        self.startstart()
 
        # Send first event (should trigger begin run again)
        self.sendsend(self.input_socket, "u", first_data=self.first_run_event_datafirst_run_event_data[0], identity=self.input_identityinput_identity)
        self.assertHasOutputFileassertHasOutputFile("beginrun_called", timeout=0.5)
        self.assertIsMsgTypeassertIsMsgType(self.output_socket, "w", router=True)
        self.sendsend(self.output_socket, "c", identity=self.output_identityoutput_identity)
 
        # Send second event (should not trigger begin run again)
        self.sendsend(self.input_socket, "u", first_data=self.first_run_event_datafirst_run_event_data[1], identity=self.input_identityinput_identity)
        self.assertNotHasOutputFileassertNotHasOutputFile("beginrun_called", timeout=0.5)
        self.assertIsMsgTypeassertIsMsgType(self.output_socket, "w", router=True)
        self.sendsend(self.output_socket, "c", identity=self.output_identityoutput_identity)
 
    def testEndRun(self):
        """test function"""
        self.startstart()
 
        # TODO: do I want to test which run was ended?
        # end run trigger
        self.sendsend(self.input_socket, "l", identity=self.input_identityinput_identity)
        self.assertHasOutputFileassertHasOutputFile("endrun_called", timeout=2)
        self.assertIsMsgTypeassertIsMsgType(self.output_socket, "l", router=True)
        self.sendsend(self.output_socket, "c", identity=self.output_identityoutput_identity)
 
        # Also the second one should give us an end run
        self.sendsend(self.input_socket, "l", identity=self.input_identityinput_identity)
        self.assertHasOutputFileassertHasOutputFile("endrun_called", timeout=1)
        self.assertIsMsgTypeassertIsMsgType(self.output_socket, "l", router=True)
        self.sendsend(self.output_socket, "c", identity=self.output_identityoutput_identity)
 
        # Sneak in an event in between -> should give beginRun
        self.sendsend(self.input_socket, "u", first_data=self.first_run_event_datafirst_run_event_data[0], identity=self.input_identityinput_identity)
        self.assertHasOutputFileassertHasOutputFile("beginrun_called", timeout=1)
        self.assertIsMsgTypeassertIsMsgType(self.output_socket, "w", router=True)
        self.sendsend(self.output_socket, "c", identity=self.output_identityoutput_identity)
 
        # And end the run again
        self.sendsend(self.input_socket, "l", identity=self.input_identityinput_identity)
        self.assertHasOutputFileassertHasOutputFile("endrun_called", timeout=1)
        self.assertIsMsgTypeassertIsMsgType(self.output_socket, "l", router=True)
        self.sendsend(self.output_socket, "c", identity=self.output_identityoutput_identity)
 
        # A second time...
        self.sendsend(self.input_socket, "l", identity=self.input_identityinput_identity)
        self.assertHasOutputFileassertHasOutputFile("endrun_called", timeout=1)
        self.assertIsMsgTypeassertIsMsgType(self.output_socket, "l", router=True)
        self.sendsend(self.output_socket, "c", identity=self.output_identityoutput_identity)
 
        # Sneak in a second event in between -> should give beginRun (as it is a new run)
        self.sendsend(self.input_socket, "u", first_data=self.second_run_event_datasecond_run_event_data[0], identity=self.input_identityinput_identity)
        self.assertHasOutputFileassertHasOutputFile("beginrun_called", timeout=1)
        self.assertIsMsgTypeassertIsMsgType(self.output_socket, "w", router=True)
        self.sendsend(self.output_socket, "c", identity=self.output_identityoutput_identity)
 
        # And end the run again
        self.sendsend(self.input_socket, "l", identity=self.input_identityinput_identity)
        self.assertHasOutputFileassertHasOutputFile("endrun_called", timeout=1)
        self.assertIsMsgTypeassertIsMsgType(self.output_socket, "l", router=True)
        self.sendsend(self.output_socket, "c", identity=self.output_identityoutput_identity)
 
        # Termination should also work
        self.sendsend(self.input_socket, "x", identity=self.input_identityinput_identity)
        # Attention: terminate is called in the different order
        self.assertIsMsgTypeassertIsMsgType(self.output_socket, "x", router=True)
        self.sendsend(self.output_socket, "c", identity=self.output_identityoutput_identity)
        self.assertHasOutputFileassertHasOutputFile("terminate_called", timeout=1)
 
        # And the termination should cause the process to go down
        self.assertIsDownassertIsDown("worker", timeout=200)
 
 
class DyingWorkerTestCase(WorkerTestCase):
    """Test case for dying workers"""
 
    
    extra_arguments = ["--exit", "--prefix", "dying_"]
 
    def testUnregistration(self):
        """test function"""
        self.startstart()
 
        # lets send some events
        for _ in range(10):
            self.sendsend(self.input_socket, "u", first_data=self.event_dataevent_data, identity=self.input_identityinput_identity)
            self.assertIsMsgTypeassertIsMsgType(self.output_socket, "w", router=True)
            self.sendsend(self.output_socket, "c", identity=self.output_identityoutput_identity)
 
        self.assertHasOutputFileassertHasOutputFile("dying_beginrun_called", timeout=1)
 
        # Now we kill one of the workers
        Path("dying_exit_request").touch()
 
        self.sendsend(self.input_socket, "u", first_data=self.event_dataevent_data, identity=self.input_identityinput_identity)
        self.assertHasOutputFileassertHasOutputFile("dying_exit_called", timeout=1)
        msg = self.assertIsMsgTypeassertIsMsgType(self.output_socket, "d", router=True)
        # the message content should be the worker that has died
        self.assertEqual(msg[2].decode(), self.output_identityoutput_identity)
        self.sendsend(self.output_socket, "c", identity=msg[0].decode())
 
        self.assertIsDownassertIsDown("worker", timeout=10)
 
 
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)