1b2732e9dSimarom# -*- coding: utf8 -*-
2b2732e9dSimarom# Copyright (C) PyZMQ Developers
3b2732e9dSimarom# Distributed under the terms of the Modified BSD License.
4b2732e9dSimarom
5b2732e9dSimarom
6b2732e9dSimaromimport sys
7b2732e9dSimaromimport time
8b2732e9dSimarom
9b2732e9dSimaromfrom unittest import TestCase
10b2732e9dSimarom
11b2732e9dSimaromimport zmq
12b2732e9dSimaromfrom zmq.eventloop import ioloop, zmqstream
13b2732e9dSimarom
14b2732e9dSimaromclass TestZMQStream(TestCase):
15b2732e9dSimarom
16b2732e9dSimarom    def setUp(self):
17b2732e9dSimarom        self.context = zmq.Context()
18b2732e9dSimarom        self.socket = self.context.socket(zmq.REP)
19b2732e9dSimarom        self.loop = ioloop.IOLoop.instance()
20b2732e9dSimarom        self.stream = zmqstream.ZMQStream(self.socket)
21b2732e9dSimarom
22b2732e9dSimarom    def tearDown(self):
23b2732e9dSimarom        self.socket.close()
24b2732e9dSimarom        self.context.term()
25b2732e9dSimarom
26b2732e9dSimarom    def test_callable_check(self):
27b2732e9dSimarom        """Ensure callable check works (py3k)."""
28b2732e9dSimarom
29b2732e9dSimarom        self.stream.on_send(lambda *args: None)
30b2732e9dSimarom        self.stream.on_recv(lambda *args: None)
31b2732e9dSimarom        self.assertRaises(AssertionError, self.stream.on_recv, 1)
32b2732e9dSimarom        self.assertRaises(AssertionError, self.stream.on_send, 1)
33b2732e9dSimarom        self.assertRaises(AssertionError, self.stream.on_recv, zmq)
34b2732e9dSimarom
35