summaryrefslogtreecommitdiffstats
path: root/scripts/external_libs/zmq/tests/test_monitor.py
diff options
context:
space:
mode:
authorimarom <imarom@cisco.com>2015-12-27 06:37:18 -0500
committerimarom <imarom@cisco.com>2015-12-27 07:27:36 -0500
commitaec3c8f4a0fe4da9a964a051d86fae808f336a55 (patch)
treeb9cdd5fb06b55141a234d83c5be3f72e4a093c78 /scripts/external_libs/zmq/tests/test_monitor.py
parent9d1cd91825d48a97ca0ea21fa7bd34900f6c7450 (diff)
provide a CEL 5.9 a way to run trex-console
Diffstat (limited to 'scripts/external_libs/zmq/tests/test_monitor.py')
-rw-r--r--scripts/external_libs/zmq/tests/test_monitor.py71
1 files changed, 0 insertions, 71 deletions
diff --git a/scripts/external_libs/zmq/tests/test_monitor.py b/scripts/external_libs/zmq/tests/test_monitor.py
deleted file mode 100644
index 4f035388..00000000
--- a/scripts/external_libs/zmq/tests/test_monitor.py
+++ /dev/null
@@ -1,71 +0,0 @@
-# -*- coding: utf-8 -*-
-# Copyright (C) PyZMQ Developers
-# Distributed under the terms of the Modified BSD License.
-
-
-import sys
-import time
-import struct
-
-from unittest import TestCase
-
-import zmq
-from zmq.tests import BaseZMQTestCase, skip_if, skip_pypy
-from zmq.utils.monitor import recv_monitor_message
-
-skip_lt_4 = skip_if(zmq.zmq_version_info() < (4,), "requires zmq >= 4")
-
-class TestSocketMonitor(BaseZMQTestCase):
-
- @skip_lt_4
- def test_monitor(self):
- """Test monitoring interface for sockets."""
- s_rep = self.context.socket(zmq.REP)
- s_req = self.context.socket(zmq.REQ)
- self.sockets.extend([s_rep, s_req])
- s_req.bind("tcp://127.0.0.1:6666")
- # try monitoring the REP socket
-
- s_rep.monitor("inproc://monitor.rep", zmq.EVENT_ALL)
- # create listening socket for monitor
- s_event = self.context.socket(zmq.PAIR)
- self.sockets.append(s_event)
- s_event.connect("inproc://monitor.rep")
- s_event.linger = 0
- # test receive event for connect event
- s_rep.connect("tcp://127.0.0.1:6666")
- m = recv_monitor_message(s_event)
- if m['event'] == zmq.EVENT_CONNECT_DELAYED:
- self.assertEqual(m['endpoint'], b"tcp://127.0.0.1:6666")
- # test receive event for connected event
- m = recv_monitor_message(s_event)
- self.assertEqual(m['event'], zmq.EVENT_CONNECTED)
- self.assertEqual(m['endpoint'], b"tcp://127.0.0.1:6666")
-
- # test monitor can be disabled.
- s_rep.disable_monitor()
- m = recv_monitor_message(s_event)
- self.assertEqual(m['event'], zmq.EVENT_MONITOR_STOPPED)
-
-
- @skip_lt_4
- def test_monitor_connected(self):
- """Test connected monitoring socket."""
- s_rep = self.context.socket(zmq.REP)
- s_req = self.context.socket(zmq.REQ)
- self.sockets.extend([s_rep, s_req])
- s_req.bind("tcp://127.0.0.1:6667")
- # try monitoring the REP socket
- # create listening socket for monitor
- s_event = s_rep.get_monitor_socket()
- s_event.linger = 0
- self.sockets.append(s_event)
- # test receive event for connect event
- s_rep.connect("tcp://127.0.0.1:6667")
- m = recv_monitor_message(s_event)
- if m['event'] == zmq.EVENT_CONNECT_DELAYED:
- self.assertEqual(m['endpoint'], b"tcp://127.0.0.1:6667")
- # test receive event for connected event
- m = recv_monitor_message(s_event)
- self.assertEqual(m['event'], zmq.EVENT_CONNECTED)
- self.assertEqual(m['endpoint'], b"tcp://127.0.0.1:6667")