improve test coverage

giampaolo · Feb 25, 2016 · 4e2a7d7 · 4e2a7d7
1 parent cf8add0
commit 4e2a7d7
Show file tree

Hide file tree

Showing 2 changed files with 54 additions and 0 deletions.
diff --git a/.coveragerc b/.coveragerc
@@ -17,3 +17,5 @@ exclude_lines =
     if hasattr(select, 'epoll'):
     if hasattr(select, 'kqueue'):
     if PY3:
+    if hasattr(select, 'devpoll')
+    if hasattr(select, 'kqueue')
diff --git a/pyftpdlib/test/test_ioloop.py b/pyftpdlib/test/test_ioloop.py
@@ -151,6 +151,10 @@ class DefaultIOLoopTestCase(unittest.TestCase, BaseIOLoopTestCase):
     ioloop_class = pyftpdlib.ioloop.IOLoop
 
 
+# ===================================================================
+# select()
+# ===================================================================
+
 class SelectIOLoopTestCase(unittest.TestCase, BaseIOLoopTestCase):
     ioloop_class = pyftpdlib.ioloop.Select
 
@@ -169,24 +173,72 @@ def test_select_eintr(self):
             self.assertRaises(select.error, s.poll, 0)
 
 
+# ===================================================================
+# poll()
+# ===================================================================
+
 @unittest.skipUnless(hasattr(pyftpdlib.ioloop, 'Poll'),
                      "poll() not available on this platform")
 class PollIOLoopTestCase(unittest.TestCase, BaseIOLoopTestCase):
     ioloop_class = getattr(pyftpdlib.ioloop, "Poll", None)
 
+    def test_poll_eintr(self, ):
+        # EINTR is supposed to be ignored
+        with mock.patch("pyftpdlib.ioloop.Poll._poller",
+                        return_vaue=mock.Mock()) as m_poll:
+            m_poll.return_value.poll.side_effect = select.error
+            m_poll.return_value.poll.side_effect.errno = errno.EINTR
+            s, rd, wr = self.test_register()
+            s.poll(0)
+        # ...but just that
+        with mock.patch("pyftpdlib.ioloop.Poll._poller",
+                        return_vaue=mock.Mock()) as m_poll:
+            m_poll.return_value.poll.side_effect = select.error
+            m_poll.return_value.poll.side_effect.errno = errno.EBADF
+            s, rd, wr = self.test_register()
+            self.assertRaises(select.error, s.poll, 0)
+
+
+# ===================================================================
+# epoll()
+# ===================================================================
 
 @unittest.skipUnless(hasattr(pyftpdlib.ioloop, 'Epoll'),
                      "epoll() not available on this platform (Linux only)")
 class EpollIOLoopTestCase(unittest.TestCase, BaseIOLoopTestCase):
     ioloop_class = getattr(pyftpdlib.ioloop, "Epoll", None)
 
+    def test_epoll_eintr(self):
+        # EINTR is supposed to be ignored
+        with mock.patch("pyftpdlib.ioloop.Epoll._poller",
+                        return_vaue=mock.Mock()) as m_poll:
+            m_poll.return_value.poll.side_effect = select.error
+            m_poll.return_value.poll.side_effect.errno = errno.EINTR
+            s, rd, wr = self.test_register()
+            s.poll(0)
+        # ...but just that
+        with mock.patch("pyftpdlib.ioloop.Epoll._poller",
+                        return_vaue=mock.Mock()) as m_poll:
+            m_poll.return_value.poll.side_effect = select.error
+            m_poll.return_value.poll.side_effect.errno = errno.EBADF
+            s, rd, wr = self.test_register()
+            self.assertRaises(select.error, s.poll, 0)
+
+
+# ===================================================================
+# /dev/poll
+# ===================================================================
 
 @unittest.skipUnless(hasattr(pyftpdlib.ioloop, 'DevPoll'),
                      "/dev/poll not available on this platform (Solaris only)")
 class DevPollIOLoopTestCase(unittest.TestCase, BaseIOLoopTestCase):
     ioloop_class = getattr(pyftpdlib.ioloop, "DevPoll", None)
 
 
+# ===================================================================
+# kqueue
+# ===================================================================
+
 @unittest.skipUnless(hasattr(pyftpdlib.ioloop, 'Kqueue'),
                      "/dev/poll not available on this platform (BSD only)")
 class KqueueIOLoopTestCase(unittest.TestCase, BaseIOLoopTestCase):