22
22
#-----------------------------------------------------------------------------
24
26
from unittest import TestCase
31
from nose import SkipTest
33
class SkipTest(Exception):
28
36
#-----------------------------------------------------------------------------
30
38
#-----------------------------------------------------------------------------
33
41
class BaseZMQTestCase(TestCase):
36
45
self.context = zmq.Context()
50
sock = self.sockets.pop()
57
s = self.context.socket(zmq.PUB)
59
s.bind('epgm://:12345')
62
if e.errno == zmq.EPROTONOSUPPORT:
70
def skip_if_pgm(self):
71
major,minor = zmq.zmq_version().split('.',2)[:2]
72
if self.has_pgm() and major == '2' and minor == '0':
73
raise SkipTest("Known Failure on 0MQ 2.0.x+PGM")
38
75
def create_bound_pair(self, type1, type2, interface='tcp://127.0.0.1'):
39
76
"""Create a bound socket pair using a random port."""
40
77
s1 = zmq.Socket(self.context, type1)
41
78
port = s1.bind_to_random_port(interface)
42
79
s2 = zmq.Socket(self.context, type2)
43
80
s2.connect('%s:%s' % (interface, port))
81
self.sockets.extend([s1,s2])
46
84
def ping_pong(self, s1, s2, msg):
64
102
o3 = s1.recv_pyobj()
67
def assertRaisesErrno(self, errno, func, *args):
105
def assertRaisesErrno(self, errno, func, *args, **kwargs):
70
except zmq.ZMQError, e:
107
func(*args, **kwargs)
109
e = sys.exc_info()[1]
71
110
self.assertEqual(e.errno, errno, "wrong error raised, expected '%s' \
72
111
got '%s'" % (zmq.ZMQError(errno), zmq.ZMQError(e.errno)))