35
35
def test_basic(self):
36
36
s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP)
38
msg1 = 'message 1'.encode()
39
39
msg2 = self.ping_pong(s1, s2, msg1)
40
40
self.assertEquals(msg1, msg2)
43
43
s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP)
45
45
for i in range(10):
47
47
msg2 = self.ping_pong(s1, s2, msg1)
48
48
self.assertEquals(msg1, msg2)
50
50
def test_bad_send_recv(self):
51
51
s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP)
52
self.assertRaisesErrno(zmq.EFSM, s1.recv)
53
self.assertRaisesErrno(zmq.EFSM, s2.send, 'asdf')
52
for copy in (True,False):
53
self.assertRaisesErrno(zmq.EFSM, s1.recv, copy=copy)
54
self.assertRaisesErrno(zmq.EFSM, s2.send, 'asdf'.encode(), copy=copy)
55
56
# I have to have this or we die on an Abort trap.
57
msg1 = 'asdf'.encode()
57
58
msg2 = self.ping_pong(s1, s2, msg1)
58
59
self.assertEquals(msg1, msg2)
60
61
def test_json(self):
61
62
s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP)
62
o = dict(a=10,b=range(10))
63
o = dict(a=10,b=list(range(10)))
63
64
o2 = self.ping_pong_json(s1, s2, o)
65
66
def test_pyobj(self):
70
71
def test_large_msg(self):
71
72
s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP)
73
msg1 = 10000*'X'.encode()
74
75
for i in range(10):
75
76
msg2 = self.ping_pong(s1, s2, msg1)