← Back to branch summary

~ubuntu-branches/ubuntu/wily/pyzmq/wily

« back to all changes in this revision

Viewing changes to zmq/tests/test_log.py

  • Committer: Package Import Robot
  • Author(s): Julian Taylor
  • Date: 2013-02-24 19:23:15 UTC
  • mfrom: (1.2.1) (9 sid)
  • mto: This revision was merged to the branch mainline in revision 10.
  • Revision ID: package-import@ubuntu.com-20130224192315-qhmwp3m3ymk8r60d
Tags: 2.2.0.1-1
* New upstream release
* relicense debian packaging to LGPL-3
* update watch file to use github directly
  thanks to Bart Martens for the file
* add autopkgtests
* drop obsolete DM-Upload-Allowed
* bump standard to 3.9.4, no changes required

Show diffs side-by-side

added added

removed removed

Lines of Context:
zmq/tests/test_log.py
1
 
#
2
 
#    Copyright (c) 2010 Brian E. Granger
3
 
#
4
 
#    This file is part of pyzmq.
5
 
#
6
 
#    pyzmq is free software; you can redistribute it and/or modify it under
7
 
#    the terms of the Lesser GNU General Public License as published by
8
 
#    the Free Software Foundation; either version 3 of the License, or
9
 
#    (at your option) any later version.
10
 
#
11
 
#    pyzmq is distributed in the hope that it will be useful,
12
 
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
13
 
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14
 
#    Lesser GNU General Public License for more details.
15
 
#
16
 
#    You should have received a copy of the Lesser GNU General Public License
17
 
#    along with this program.  If not, see <http://www.gnu.org/licenses/>.
18
 
#
 
1
#-----------------------------------------------------------------------------
 
2
#  Copyright (c) 2010-2012 Brian Granger, Min Ragan-Kelley
 
3
#
 
4
#  This file is part of pyzmq
 
5
#
 
6
#  Distributed under the terms of the New BSD License.  The full license is in
 
7
#  the file COPYING.BSD, distributed as part of this software.
 
8
#-----------------------------------------------------------------------------
19
9
 
20
10
#-----------------------------------------------------------------------------
21
11
# Imports
22
12
#-----------------------------------------------------------------------------
23
13
 
 
14
import logging
24
15
import time
25
16
from unittest import TestCase
26
17
 
27
18
import zmq
 
19
from zmq.log import handlers
 
20
from zmq.utils.strtypes import b
28
21
from zmq.tests import BaseZMQTestCase
29
22
 
30
 
from zmq.log import handlers
31
 
import logging
32
 
 
33
23
#-----------------------------------------------------------------------------
34
24
# Tests
35
25
#-----------------------------------------------------------------------------
37
27
class TestPubLog(BaseZMQTestCase):
38
28
    
39
29
    iface = 'inproc://zmqlog'
40
 
    topic='zmq'.encode()
 
30
    topic= b'zmq'
41
31
    
42
32
    @property
43
33
    def logger(self):
60
50
    def test_init_iface(self):
61
51
        logger = self.logger
62
52
        ctx = self.context
63
 
        handler = handlers.PUBHandler(self.iface, ctx)
64
 
        self.assertTrue(handler.ctx is ctx)
65
 
        logger.removeHandler(handler)
66
 
        handler.socket.close()
67
 
        
 
53
        handler = handlers.PUBHandler(self.iface)
 
54
        self.assertFalse(handler.ctx is ctx)
 
55
        self.sockets.append(handler.socket)
 
56
        # handler.ctx.term()
68
57
        handler = handlers.PUBHandler(self.iface, self.context)
 
58
        self.sockets.append(handler.socket)
69
59
        self.assertTrue(handler.ctx is ctx)
70
 
        
71
60
        handler.setLevel(logging.DEBUG)
72
61
        handler.root_topic = self.topic
73
62
        logger.addHandler(handler)
74
 
        
75
63
        sub = ctx.socket(zmq.SUB)
 
64
        self.sockets.append(sub)
 
65
        sub.setsockopt(zmq.SUBSCRIBE, self.topic)
76
66
        sub.connect(self.iface)
77
 
        sub.setsockopt(zmq.SUBSCRIBE, self.topic)
78
 
        import time; time.sleep(0.1)
 
67
        import time; time.sleep(0.25)
79
68
        msg1 = 'message'
80
69
        logger.info(msg1)
81
70
        
82
71
        (topic, msg2) = sub.recv_multipart()
83
 
        self.assertEquals(topic, 'zmq.INFO'.encode())
84
 
        self.assertEquals(msg2, (msg1+'\n').encode())
 
72
        self.assertEquals(topic, b'zmq.INFO')
 
73
        self.assertEquals(msg2, b(msg1)+b'\n')
85
74
        logger.removeHandler(handler)
86
 
        handler.socket.close()
87
75
    
88
76
    def test_init_socket(self):
89
77
        pub,sub = self.create_bound_pair(zmq.PUB, zmq.SUB)
96
84
        self.assertTrue(handler.socket is pub)
97
85
        self.assertTrue(handler.ctx is pub.context)
98
86
        self.assertTrue(handler.ctx is self.context)
99
 
        # handler.socket.close()
100
87
        sub.setsockopt(zmq.SUBSCRIBE, self.topic)
101
88
        import time; time.sleep(0.1)
102
89
        msg1 = 'message'
103
90
        logger.info(msg1)
104
91
        
105
92
        (topic, msg2) = sub.recv_multipart()
106
 
        self.assertEquals(topic, 'zmq.INFO'.encode())
107
 
        self.assertEquals(msg2, (msg1+'\n').encode())
 
93
        self.assertEquals(topic, b'zmq.INFO')
 
94
        self.assertEquals(msg2, b(msg1)+b'\n')
108
95
        logger.removeHandler(handler)
109
 
        handler.socket.close()
110
96
    
111
97
    def test_root_topic(self):
112
98
        logger, handler, sub = self.connect_handler()
113
99
        handler.socket.bind(self.iface)
114
100
        sub2 = sub.context.socket(zmq.SUB)
 
101
        self.sockets.append(sub2)
115
102
        sub2.connect(self.iface)
116
 
        sub2.setsockopt(zmq.SUBSCRIBE, ''.encode())
117
 
        handler.root_topic = 'twoonly'.encode()
 
103
        sub2.setsockopt(zmq.SUBSCRIBE, b'')
 
104
        handler.root_topic = b'twoonly'
118
105
        msg1 = 'ignored'
119
106
        logger.info(msg1)
120
107
        self.assertRaisesErrno(zmq.EAGAIN, sub.recv, zmq.NOBLOCK)
121
108
        topic,msg2 = sub2.recv_multipart()
122
 
        self.assertEquals(topic, 'twoonly.INFO'.encode())
123
 
        self.assertEquals(msg2, (msg1+'\n').encode())
 
109
        self.assertEquals(topic, b'twoonly.INFO')
 
110
        self.assertEquals(msg2, b(msg1)+b'\n')
124
111
        
125
112
        
126
113