[Python-checkins] cpython: asyncio: Avoid ResourceWarning. Fix issue 19580 by Vajrasky Kok.
guido.van.rossum
python-checkins at python.org
Thu Nov 14 19:06:32 CET 2013
http://hg.python.org/cpython/rev/df50f73f03ca
changeset: 87099:df50f73f03ca
user: Guido van Rossum <guido at dropbox.com>
date: Thu Nov 14 10:06:18 2013 -0800
summary:
asyncio: Avoid ResourceWarning. Fix issue 19580 by Vajrasky Kok.
files:
Lib/test/test_asyncio/test_base_events.py | 22 +++++++++--
1 files changed, 18 insertions(+), 4 deletions(-)
diff --git a/Lib/test/test_asyncio/test_base_events.py b/Lib/test/test_asyncio/test_base_events.py
--- a/Lib/test/test_asyncio/test_base_events.py
+++ b/Lib/test/test_asyncio/test_base_events.py
@@ -458,16 +458,26 @@
self.loop.sock_connect.return_value = ()
self.loop._make_ssl_transport = unittest.mock.Mock()
+ class _SelectorTransportMock:
+ _sock = None
+
+ def close(self):
+ self._sock.close()
+
def mock_make_ssl_transport(sock, protocol, sslcontext, waiter,
**kwds):
waiter.set_result(None)
+ transport = _SelectorTransportMock()
+ transport._sock = sock
+ return transport
self.loop._make_ssl_transport.side_effect = mock_make_ssl_transport
ANY = unittest.mock.ANY
# First try the default server_hostname.
self.loop._make_ssl_transport.reset_mock()
coro = self.loop.create_connection(MyProto, 'python.org', 80, ssl=True)
- self.loop.run_until_complete(coro)
+ transport, _ = self.loop.run_until_complete(coro)
+ transport.close()
self.loop._make_ssl_transport.assert_called_with(
ANY, ANY, ANY, ANY,
server_side=False,
@@ -476,7 +486,8 @@
self.loop._make_ssl_transport.reset_mock()
coro = self.loop.create_connection(MyProto, 'python.org', 80, ssl=True,
server_hostname='perl.com')
- self.loop.run_until_complete(coro)
+ transport, _ = self.loop.run_until_complete(coro)
+ transport.close()
self.loop._make_ssl_transport.assert_called_with(
ANY, ANY, ANY, ANY,
server_side=False,
@@ -485,7 +496,8 @@
self.loop._make_ssl_transport.reset_mock()
coro = self.loop.create_connection(MyProto, 'python.org', 80, ssl=True,
server_hostname='')
- self.loop.run_until_complete(coro)
+ transport, _ = self.loop.run_until_complete(coro)
+ transport.close()
self.loop._make_ssl_transport.assert_called_with(ANY, ANY, ANY, ANY,
server_side=False,
server_hostname='')
@@ -505,8 +517,10 @@
self.assertRaises(ValueError, self.loop.run_until_complete, coro)
coro = self.loop.create_connection(MyProto, None, 80, ssl=True)
self.assertRaises(ValueError, self.loop.run_until_complete, coro)
+ sock = socket.socket()
coro = self.loop.create_connection(MyProto, None, None,
- ssl=True, sock=socket.socket())
+ ssl=True, sock=sock)
+ self.addCleanup(sock.close)
self.assertRaises(ValueError, self.loop.run_until_complete, coro)
def test_create_server_empty_host(self):
--
Repository URL: http://hg.python.org/cpython
More information about the Python-checkins
mailing list