def setUp(self):
self.loop = self.new_test_loop()
+ self.addCleanup(self.loop.close)
self.proactor = mock.Mock()
self.loop._proactor = self.proactor
self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
self.assertIsNone(self.loop._ssock)
self.assertIsNone(self.loop._csock)
+ # Don't call close(): _close_self_pipe() cannot be called twice
+ self.loop._closed = True
+
def test_close(self):
self.loop._close_self_pipe = mock.Mock()
self.loop.close()
class TestBaseSelectorEventLoop(BaseSelectorEventLoop):
+ def close(self):
+ # Don't call the close() method of the parent class, because the
+ # selector is mocked
+ self._closed = True
+
def _make_self_pipe(self):
self._ssock = mock.Mock()
self._csock = mock.Mock()
self.selector = mock.Mock()
self.selector.select.return_value = []
self.loop = TestBaseSelectorEventLoop(self.selector)
- self.set_event_loop(self.loop, cleanup=False)
+ self.set_event_loop(self.loop)
def test_make_socket_transport(self):
m = mock.Mock()
self.loop._make_ssl_transport(m, m, m, m)
def test_close(self):
+ class EventLoop(BaseSelectorEventLoop):
+ def _make_self_pipe(self):
+ self._ssock = mock.Mock()
+ self._csock = mock.Mock()
+ self._internal_fds += 1
+
+ self.loop = EventLoop(self.selector)
+ self.set_event_loop(self.loop)
+
ssock = self.loop._ssock
ssock.fileno.return_value = 7
csock = self.loop._csock