test_supvisorszmq.py 文件源码

def setUp(self):
        """ Create a dummy supvisors, ZMQ context and sockets. """
        from supvisors.supvisorszmq import RequestPusher, RequestPuller
        # the dummy Supvisors is used for addresses and ports
        self.supvisors = MockedSupvisors()
        # create pusher and puller
        self.pusher = RequestPusher(self.supvisors.logger)
        self.puller = RequestPuller()
        # socket configuration is meant to be blocking
        # however, a failure would block the unit test,
        # so a timeout is set for emission and reception
        self.puller.socket.setsockopt(zmq.SNDTIMEO, 1000)
        self.puller.socket.setsockopt(zmq.RCVTIMEO, 1000)