def test_init_iface(self):
logger = self.logger
ctx = self.context
handler = handlers.PUBHandler(self.iface)
self.assertFalse(handler.ctx is ctx)
self.sockets.append(handler.socket)
# handler.ctx.term()
handler = handlers.PUBHandler(self.iface, self.context)
self.sockets.append(handler.socket)
self.assertTrue(handler.ctx is ctx)
handler.setLevel(logging.DEBUG)
handler.root_topic = self.topic
logger.addHandler(handler)
sub = ctx.socket(zmq.SUB)
self.sockets.append(sub)
sub.setsockopt(zmq.SUBSCRIBE, b(self.topic))
sub.connect(self.iface)
import time; time.sleep(0.25)
msg1 = 'message'
logger.info(msg1)
(topic, msg2) = sub.recv_multipart()
self.assertEqual(topic, b'zmq.INFO')
self.assertEqual(msg2, b(msg1)+b'\n')
logger.removeHandler(handler)
评论列表
文章目录