test_error.py revision 781d71db
1# -*- coding: utf8 -*-
2# Copyright (C) PyZMQ Developers
3# Distributed under the terms of the Modified BSD License.
4
5import sys
6import time
7
8import zmq
9from zmq import ZMQError, strerror, Again, ContextTerminated
10from zmq.tests import BaseZMQTestCase
11
12if sys.version_info[0] >= 3:
13    long = int
14
15class TestZMQError(BaseZMQTestCase):
16
17    def test_strerror(self):
18        """test that strerror gets the right type."""
19        for i in range(10):
20            e = strerror(i)
21            self.assertTrue(isinstance(e, str))
22
23    def test_zmqerror(self):
24        for errno in range(10):
25            e = ZMQError(errno)
26            self.assertEqual(e.errno, errno)
27            self.assertEqual(str(e), strerror(errno))
28
29    def test_again(self):
30        s = self.context.socket(zmq.REP)
31        self.assertRaises(Again, s.recv, zmq.NOBLOCK)
32        self.assertRaisesErrno(zmq.EAGAIN, s.recv, zmq.NOBLOCK)
33        s.close()
34
35    def atest_ctxterm(self):
36        s = self.context.socket(zmq.REP)
37        t = Thread(target=self.context.term)
38        t.start()
39        self.assertRaises(ContextTerminated, s.recv, zmq.NOBLOCK)
40        self.assertRaisesErrno(zmq.TERM, s.recv, zmq.NOBLOCK)
41        s.close()
42        t.join()
43
44