From dab741a80699f86e86c91718872a052cca9bbb25 Mon Sep 17 00:00:00 2001
From: Dan Klein <danklei@cisco.com>
Date: Mon, 24 Aug 2015 13:22:48 +0300
Subject: Fixed dependencies of Control Plane to use external_lib sources
---
external_libs/python/zmq/tests/test_log.py | 116 -----------------------------
1 file changed, 116 deletions(-)
delete mode 100644 external_libs/python/zmq/tests/test_log.py
(limited to 'external_libs/python/zmq/tests/test_log.py')
diff --git a/external_libs/python/zmq/tests/test_log.py b/external_libs/python/zmq/tests/test_log.py
deleted file mode 100644
index 9206f095..00000000
--- a/external_libs/python/zmq/tests/test_log.py
+++ /dev/null
@@ -1,116 +0,0 @@
-# encoding: utf-8
-
-# Copyright (C) PyZMQ Developers
-# Distributed under the terms of the Modified BSD License.
-
-
-import logging
-import time
-from unittest import TestCase
-
-import zmq
-from zmq.log import handlers
-from zmq.utils.strtypes import b, u
-from zmq.tests import BaseZMQTestCase
-
-
-class TestPubLog(BaseZMQTestCase):
-
- iface = 'inproc://zmqlog'
- topic= 'zmq'
-
- @property
- def logger(self):
- # print dir(self)
- logger = logging.getLogger('zmqtest')
- logger.setLevel(logging.DEBUG)
- return logger
-
- def connect_handler(self, topic=None):
- topic = self.topic if topic is None else topic
- logger = self.logger
- pub,sub = self.create_bound_pair(zmq.PUB, zmq.SUB)
- handler = handlers.PUBHandler(pub)
- handler.setLevel(logging.DEBUG)
- handler.root_topic = topic
- logger.addHandler(handler)
- sub.setsockopt(zmq.SUBSCRIBE, b(topic))
- time.sleep(0.1)
- return logger, handler, sub
-
- 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)
-
- def test_init_socket(self):
- pub,sub = self.create_bound_pair(zmq.PUB, zmq.SUB)
- logger = self.logger
- handler = handlers.PUBHandler(pub)
- handler.setLevel(logging.DEBUG)
- handler.root_topic = self.topic
- logger.addHandler(handler)
-
- self.assertTrue(handler.socket is pub)
- self.assertTrue(handler.ctx is pub.context)
- self.assertTrue(handler.ctx is self.context)
- sub.setsockopt(zmq.SUBSCRIBE, b(self.topic))
- import time; time.sleep(0.1)
- 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)
-
- def test_root_topic(self):
- logger, handler, sub = self.connect_handler()
- handler.socket.bind(self.iface)
- sub2 = sub.context.socket(zmq.SUB)
- self.sockets.append(sub2)
- sub2.connect(self.iface)
- sub2.setsockopt(zmq.SUBSCRIBE, b'')
- handler.root_topic = b'twoonly'
- msg1 = 'ignored'
- logger.info(msg1)
- self.assertRaisesErrno(zmq.EAGAIN, sub.recv, zmq.NOBLOCK)
- topic,msg2 = sub2.recv_multipart()
- self.assertEqual(topic, b'twoonly.INFO')
- self.assertEqual(msg2, b(msg1)+b'\n')
-
- logger.removeHandler(handler)
-
- def test_unicode_message(self):
- logger, handler, sub = self.connect_handler()
- base_topic = b(self.topic + '.INFO')
- for msg, expected in [
- (u('hello'), [base_topic, b('hello\n')]),
- (u('héllo'), [base_topic, b('héllo\n')]),
- (u('tøpic::héllo'), [base_topic + b('.tøpic'), b('héllo\n')]),
- ]:
- logger.info(msg)
- received = sub.recv_multipart()
- self.assertEqual(received, expected)
-
--
cgit