diff options
Diffstat (limited to 'tests/test_unix_events.py')
| -rw-r--r-- | tests/test_unix_events.py | 130 |
1 files changed, 71 insertions, 59 deletions
diff --git a/tests/test_unix_events.py b/tests/test_unix_events.py index e397598..6468371 100644 --- a/tests/test_unix_events.py +++ b/tests/test_unix_events.py @@ -1,11 +1,12 @@ """Tests for unix_events.py.""" import collections -import gc +import contextlib +#import gc import errno import io import os -import pprint +#import pprint import signal import socket import stat @@ -13,22 +14,23 @@ import sys import tempfile import threading import unittest -from unittest import mock if sys.platform == 'win32': raise unittest.SkipTest('UNIX only') -import asyncio -from asyncio import log -from asyncio import test_utils -from asyncio import unix_events +import trollius as asyncio +from trollius import log +from trollius import test_utils +from trollius import unix_events +from trollius.py33_exceptions import BlockingIOError, ChildProcessError +from trollius.test_utils import mock MOCK_ANY = mock.ANY -@unittest.skipUnless(signal, 'Signals are not supported') +@test_utils.skipUnless(signal, 'Signals are not supported') class SelectorEventLoopSignalTests(test_utils.TestCase): def setUp(self): @@ -53,7 +55,7 @@ class SelectorEventLoopSignalTests(test_utils.TestCase): self.loop._handle_signal(signal.NSIG + 1) self.loop.remove_signal_handler.assert_called_with(signal.NSIG + 1) - @mock.patch('asyncio.unix_events.signal') + @mock.patch('trollius.unix_events.signal') def test_add_signal_handler_setup_error(self, m_signal): m_signal.NSIG = signal.NSIG m_signal.set_wakeup_fd.side_effect = ValueError @@ -63,7 +65,7 @@ class SelectorEventLoopSignalTests(test_utils.TestCase): self.loop.add_signal_handler, signal.SIGINT, lambda: True) - @mock.patch('asyncio.unix_events.signal') + @mock.patch('trollius.unix_events.signal') def test_add_signal_handler(self, m_signal): m_signal.NSIG = signal.NSIG @@ -73,7 +75,7 @@ class SelectorEventLoopSignalTests(test_utils.TestCase): self.assertIsInstance(h, asyncio.Handle) self.assertEqual(h._callback, cb) - @mock.patch('asyncio.unix_events.signal') + @mock.patch('trollius.unix_events.signal') def test_add_signal_handler_install_error(self, m_signal): m_signal.NSIG = signal.NSIG @@ -91,8 +93,8 @@ class SelectorEventLoopSignalTests(test_utils.TestCase): self.loop.add_signal_handler, signal.SIGINT, lambda: True) - @mock.patch('asyncio.unix_events.signal') - @mock.patch('asyncio.base_events.logger') + @mock.patch('trollius.unix_events.signal') + @mock.patch('trollius.base_events.logger') def test_add_signal_handler_install_error2(self, m_logging, m_signal): m_signal.NSIG = signal.NSIG @@ -108,8 +110,8 @@ class SelectorEventLoopSignalTests(test_utils.TestCase): self.assertFalse(m_logging.info.called) self.assertEqual(1, m_signal.set_wakeup_fd.call_count) - @mock.patch('asyncio.unix_events.signal') - @mock.patch('asyncio.base_events.logger') + @mock.patch('trollius.unix_events.signal') + @mock.patch('trollius.base_events.logger') def test_add_signal_handler_install_error3(self, m_logging, m_signal): class Err(OSError): errno = errno.EINVAL @@ -123,7 +125,7 @@ class SelectorEventLoopSignalTests(test_utils.TestCase): self.assertFalse(m_logging.info.called) self.assertEqual(2, m_signal.set_wakeup_fd.call_count) - @mock.patch('asyncio.unix_events.signal') + @mock.patch('trollius.unix_events.signal') def test_remove_signal_handler(self, m_signal): m_signal.NSIG = signal.NSIG @@ -136,7 +138,7 @@ class SelectorEventLoopSignalTests(test_utils.TestCase): self.assertEqual( (signal.SIGHUP, m_signal.SIG_DFL), m_signal.signal.call_args[0]) - @mock.patch('asyncio.unix_events.signal') + @mock.patch('trollius.unix_events.signal') def test_remove_signal_handler_2(self, m_signal): m_signal.NSIG = signal.NSIG m_signal.SIGINT = signal.SIGINT @@ -153,8 +155,8 @@ class SelectorEventLoopSignalTests(test_utils.TestCase): (signal.SIGINT, m_signal.default_int_handler), m_signal.signal.call_args[0]) - @mock.patch('asyncio.unix_events.signal') - @mock.patch('asyncio.base_events.logger') + @mock.patch('trollius.unix_events.signal') + @mock.patch('trollius.base_events.logger') def test_remove_signal_handler_cleanup_error(self, m_logging, m_signal): m_signal.NSIG = signal.NSIG self.loop.add_signal_handler(signal.SIGHUP, lambda: True) @@ -164,7 +166,7 @@ class SelectorEventLoopSignalTests(test_utils.TestCase): self.loop.remove_signal_handler(signal.SIGHUP) self.assertTrue(m_logging.info) - @mock.patch('asyncio.unix_events.signal') + @mock.patch('trollius.unix_events.signal') def test_remove_signal_handler_error(self, m_signal): m_signal.NSIG = signal.NSIG self.loop.add_signal_handler(signal.SIGHUP, lambda: True) @@ -174,7 +176,7 @@ class SelectorEventLoopSignalTests(test_utils.TestCase): self.assertRaises( OSError, self.loop.remove_signal_handler, signal.SIGHUP) - @mock.patch('asyncio.unix_events.signal') + @mock.patch('trollius.unix_events.signal') def test_remove_signal_handler_error2(self, m_signal): m_signal.NSIG = signal.NSIG self.loop.add_signal_handler(signal.SIGHUP, lambda: True) @@ -186,7 +188,7 @@ class SelectorEventLoopSignalTests(test_utils.TestCase): self.assertRaises( RuntimeError, self.loop.remove_signal_handler, signal.SIGHUP) - @mock.patch('asyncio.unix_events.signal') + @mock.patch('trollius.unix_events.signal') def test_close(self, m_signal): m_signal.NSIG = signal.NSIG @@ -203,8 +205,8 @@ class SelectorEventLoopSignalTests(test_utils.TestCase): m_signal.set_wakeup_fd.assert_called_once_with(-1) -@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), - 'UNIX Sockets are not supported') +@test_utils.skipUnless(hasattr(socket, 'AF_UNIX'), + 'UNIX Sockets are not supported') class SelectorEventLoopUnixSocketTests(test_utils.TestCase): def setUp(self): @@ -215,7 +217,7 @@ class SelectorEventLoopUnixSocketTests(test_utils.TestCase): with test_utils.unix_socket_path() as path: sock = socket.socket(socket.AF_UNIX) sock.bind(path) - with sock: + with contextlib.closing(sock): coro = self.loop.create_unix_server(lambda: None, path) with self.assertRaisesRegex(OSError, 'Address.*is already in use'): @@ -243,18 +245,19 @@ class SelectorEventLoopUnixSocketTests(test_utils.TestCase): def test_create_unix_server_path_inetsock(self): sock = socket.socket() - with sock: + with contextlib.closing(sock): coro = self.loop.create_unix_server(lambda: None, path=None, sock=sock) with self.assertRaisesRegex(ValueError, 'A UNIX Domain Socket was expected'): self.loop.run_until_complete(coro) - @mock.patch('asyncio.unix_events.socket') + @mock.patch('trollius.unix_events.socket') def test_create_unix_server_bind_error(self, m_socket): # Ensure that the socket is closed on any bind error sock = mock.Mock() m_socket.socket.return_value = sock + m_socket.error = socket.error sock.bind.side_effect = OSError coro = self.loop.create_unix_server(lambda: None, path="/test") @@ -306,7 +309,7 @@ class UnixReadPipeTransportTests(test_utils.TestCase): self.pipe = mock.Mock(spec_set=io.RawIOBase) self.pipe.fileno.return_value = 5 - blocking_patcher = mock.patch('asyncio.unix_events._set_nonblocking') + blocking_patcher = mock.patch('trollius.unix_events._set_nonblocking') blocking_patcher.start() self.addCleanup(blocking_patcher.stop) @@ -365,7 +368,7 @@ class UnixReadPipeTransportTests(test_utils.TestCase): test_utils.run_briefly(self.loop) self.assertFalse(self.protocol.data_received.called) - @mock.patch('asyncio.log.logger.error') + @mock.patch('trollius.log.logger.error') @mock.patch('os.read') def test__read_ready_error(self, m_read, m_logexc): tr = unix_events._UnixReadPipeTransport( @@ -469,7 +472,7 @@ class UnixWritePipeTransportTests(test_utils.TestCase): self.pipe = mock.Mock(spec_set=io.RawIOBase) self.pipe.fileno.return_value = 5 - blocking_patcher = mock.patch('asyncio.unix_events._set_nonblocking') + blocking_patcher = mock.patch('trollius.unix_events._set_nonblocking') blocking_patcher.start() self.addCleanup(blocking_patcher.stop) @@ -555,7 +558,7 @@ class UnixWritePipeTransportTests(test_utils.TestCase): self.loop.assert_writer(5, tr._write_ready) self.assertEqual([b'data'], tr._buffer) - @mock.patch('asyncio.unix_events.logger') + @mock.patch('trollius.unix_events.logger') @mock.patch('os.write') def test_write_err(self, m_write, m_log): tr = unix_events._UnixWritePipeTransport( @@ -655,7 +658,7 @@ class UnixWritePipeTransportTests(test_utils.TestCase): self.loop.assert_writer(5, tr._write_ready) self.assertEqual([b'data'], tr._buffer) - @mock.patch('asyncio.log.logger.error') + @mock.patch('trollius.log.logger.error') @mock.patch('os.write') def test__write_ready_err(self, m_write, m_logexc): tr = unix_events._UnixWritePipeTransport( @@ -777,7 +780,7 @@ class UnixWritePipeTransportTests(test_utils.TestCase): self.assertFalse(self.protocol.connection_lost.called) -class AbstractChildWatcherTests(unittest.TestCase): +class AbstractChildWatcherTests(test_utils.TestCase): def test_not_implemented(self): f = mock.Mock() @@ -796,7 +799,7 @@ class AbstractChildWatcherTests(unittest.TestCase): NotImplementedError, watcher.__exit__, f, f, f) -class BaseChildWatcherTests(unittest.TestCase): +class BaseChildWatcherTests(test_utils.TestCase): def test_not_implemented(self): f = mock.Mock() @@ -866,19 +869,27 @@ class ChildWatcherTestsMixin: def waitpid_mocks(func): def wrapped_func(self): + exit_stack = [] + def patch(target, wrapper): - return mock.patch(target, wraps=wrapper, - new_callable=mock.Mock) - - with patch('os.WTERMSIG', self.WTERMSIG) as m_WTERMSIG, \ - patch('os.WEXITSTATUS', self.WEXITSTATUS) as m_WEXITSTATUS, \ - patch('os.WIFSIGNALED', self.WIFSIGNALED) as m_WIFSIGNALED, \ - patch('os.WIFEXITED', self.WIFEXITED) as m_WIFEXITED, \ - patch('os.waitpid', self.waitpid) as m_waitpid: + m = mock.patch(target, wraps=wrapper) + exit_stack.append(m) + return m.__enter__() + + m_waitpid = patch('os.waitpid', self.waitpid) + m_WIFEXITED = patch('os.WIFEXITED', self.WIFEXITED) + m_WIFSIGNALED = patch('os.WIFSIGNALED', self.WIFSIGNALED) + m_WEXITSTATUS = patch('os.WEXITSTATUS', self.WEXITSTATUS) + m_WTERMSIG = patch('os.WTERMSIG', self.WTERMSIG) + try: func(self, WaitPidMocks(m_waitpid, m_WIFEXITED, m_WIFSIGNALED, m_WEXITSTATUS, m_WTERMSIG, )) + finally: + for obj in reversed(exit_stack): + obj.__exit__(None, None, None) + return wrapped_func @waitpid_mocks @@ -1351,17 +1362,18 @@ class ChildWatcherTestsMixin: callback1 = mock.Mock() callback2 = mock.Mock() - with self.ignore_warnings, self.watcher: - self.running = True - # child 1 terminates - self.add_zombie(591, 7) - # an unknown child terminates - self.add_zombie(593, 17) + with self.ignore_warnings: + with self.watcher: + self.running = True + # child 1 terminates + self.add_zombie(591, 7) + # an unknown child terminates + self.add_zombie(593, 17) - self.watcher._sig_chld() + self.watcher._sig_chld() - self.watcher.add_child_handler(591, callback1) - self.watcher.add_child_handler(592, callback2) + self.watcher.add_child_handler(591, callback1) + self.watcher.add_child_handler(592, callback2) callback1.assert_called_once_with(591, 7) self.assertFalse(callback2.called) @@ -1380,15 +1392,15 @@ class ChildWatcherTestsMixin: self.loop = self.new_test_loop() patch = mock.patch.object - with patch(old_loop, "remove_signal_handler") as m_old_remove, \ - patch(self.loop, "add_signal_handler") as m_new_add: + with patch(old_loop, "remove_signal_handler") as m_old_remove: + with patch(self.loop, "add_signal_handler") as m_new_add: - self.watcher.attach_loop(self.loop) + self.watcher.attach_loop(self.loop) - m_old_remove.assert_called_once_with( - signal.SIGCHLD) - m_new_add.assert_called_once_with( - signal.SIGCHLD, self.watcher._sig_chld) + m_old_remove.assert_called_once_with( + signal.SIGCHLD) + m_new_add.assert_called_once_with( + signal.SIGCHLD, self.watcher._sig_chld) # child terminates self.running = False @@ -1500,7 +1512,7 @@ class FastChildWatcherTests (ChildWatcherTestsMixin, test_utils.TestCase): return asyncio.FastChildWatcher() -class PolicyTests(unittest.TestCase): +class PolicyTests(test_utils.TestCase): def create_policy(self): return asyncio.DefaultEventLoopPolicy() |