← Back to branch summary

~ubuntu-branches/ubuntu/precise/pyzmq/precise

« back to all changes in this revision

Viewing changes to zmq/tests/test_multipart.py

  • Committer: Package Import Robot
  • Author(s): Debian Python Modules Team
  • Date: 2011-09-23 00:16:39 UTC
  • mfrom: (1.1.4 upstream)
  • Revision ID: package-import@ubuntu.com-20110923001639-girjqodpb7uv17yu
Tags: 2.1.9-1
http://bugs.debian.org/637777
* New upstream version
  - should build on kFreeBSD without patches (Closes: #637777).
* Build-depend on zeromq 2.1.9

Show diffs side-by-side

added added

removed removed

Lines of Context:
zmq/tests/test_multipart.py
1
1
#
2
 
#    Copyright (c) 2010 Brian E. Granger
 
2
#    Copyright (c) 2010-2011 Brian E. Granger & Min Ragan-Kelley
3
3
#
4
4
#    This file is part of pyzmq.
5
5
#
24
24
import zmq
25
25
from zmq.utils.strtypes import asbytes
26
26
 
27
 
from zmq.tests import BaseZMQTestCase
 
27
from zmq.tests import BaseZMQTestCase, SkipTest
28
28
 
29
29
#-----------------------------------------------------------------------------
30
30
# Tests
32
32
 
33
33
class TestMultipart(BaseZMQTestCase):
34
34
 
35
 
    def test_xrep_xreq(self):
36
 
        xrep, xreq = self.create_bound_pair(zmq.XREP, zmq.XREQ)
 
35
    def test_router_dealer(self):
 
36
        if zmq.zmq_version() >= '3.0.0':
 
37
            raise SkipTest("Known bug in libzmq 3.0.0, see https://zeromq.jira.com/browse/LIBZMQ-232")
 
38
        router, dealer = self.create_bound_pair(zmq.ROUTER, zmq.DEALER)
37
39
 
38
40
        msg1 = asbytes('message1')
39
 
        xreq.send(msg1)
40
 
        ident = xrep.recv()
41
 
        more = xrep.rcvmore()
 
41
        dealer.send(msg1)
 
42
        ident = router.recv()
 
43
        more = router.rcvmore()
42
44
        self.assertEquals(more, True)
43
 
        msg2 = xrep.recv()
 
45
        msg2 = router.recv()
44
46
        self.assertEquals(msg1, msg2)
45
 
        more = xrep.rcvmore()
 
47
        more = router.rcvmore()
46
48
        self.assertEquals(more, False)
 
49
    
 
50
    def test_basic_multipart(self):
 
51
        a,b = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
 
52
        msg = [ asbytes(s) for s in ['hi', 'there', 'b'] ]
 
53
        a.send_multipart(msg)
 
54
        recvd = b.recv_multipart()
 
55
        self.assertEquals(msg, recvd)
47
56