test_monqueue.py 文件源码

def test_mq_check_prefix(self):
        ins = self.context.socket(zmq.ROUTER)
        outs = self.context.socket(zmq.DEALER)
        mons = self.context.socket(zmq.PUB)
        self.sockets.extend([ins, outs, mons])

        ins = unicode('in')
        outs = unicode('out')
        self.assertRaises(TypeError, devices.monitoredqueue, ins, outs, mons)