Fix bug with https proxy acquired cleanup #1340 (#1450)

* Add functional tests for raw_host in proxy request

* Add tests of proxy acquired cleanup

* Fix bug with https proxy acquired cleanup

* More acquired tests. Fix test session close.

* Do not release waiter on detach

* Update CHANGES

* Close proxy connect transport

* Update CHANGES

* Clean client session test traceback

CHANGES.rst

@@ -52,7 +52,7 @@ CHANGES
 
 -
 
--
+- Fix bug with https proxy acquired cleanup #1340
 
 -
 

aiohttp/connector.py

@@ -91,6 +91,8 @@ def release(self):
             self._transport = None
 
     def detach(self):
+        if self._transport is not None:
+            self._connector._release_acquired(self._key, self._transport)
         self._transport = None
 
     @property
@@ -351,7 +353,7 @@ def _release_waiter(self, key):
                 waiter.set_result(None)
                 break
 
-    def _release(self, key, req, transport, protocol, *, should_close=False):
+    def _release_acquired(self, key, transport):
         if self._closed:
             # acquired connection is already released on connector closing
             return
@@ -362,9 +364,19 @@ def _release(self, key, req, transport, protocol, *, should_close=False):
         except KeyError:  # pragma: no cover
             # this may be result of undetermenistic order of objects
             # finalization due garbage collection.
-            pass
-        else:
-            if self._limit is not None and len(acquired) < self._limit:
+            return None
+
+        return acquired
+
+    def _release(self, key, req, transport, protocol, *, should_close=False):
+        if self._closed:
+            # acquired connection is already released on connector closing
+            return
+
+        acquired = self._release_acquired(key, transport)
+
+        if self._limit is not None and acquired is not None:
+            if len(acquired) < self._limit:
                 self._release_waiter(key)
 
         resp = req.response
@@ -677,13 +689,19 @@ def _create_proxy_connection(self, req):
                 raise
             else:
                 conn.detach()
-                if resp.status != 200:
-                    raise HttpProxyError(code=resp.status, message=resp.reason)
-                rawsock = transport.get_extra_info('socket', default=None)
-                if rawsock is None:
-                    raise RuntimeError(
-                        "Transport does not expose socket instance")
-                transport.pause_reading()
+                try:
+                    if resp.status != 200:
+                        raise HttpProxyError(code=resp.status,
+                                             message=resp.reason)
+                    rawsock = transport.get_extra_info('socket', default=None)
+                    if rawsock is None:
+                        raise RuntimeError(
+                            "Transport does not expose socket instance")
+                    # Duplicate the socket, so now we can close proxy transport
+                    rawsock = rawsock.dup()
+                finally:
+                    transport.close()
+
                 transport, proto = yield from self._loop.create_connection(
                     self._factory, ssl=self.ssl_context, sock=rawsock,
                     server_hostname=req.host)

tests/test_client_connection.py

@@ -95,5 +95,16 @@ def test_detach(connector, key, request, transport, protocol, loop):
     assert not conn.closed
     conn.detach()
     assert conn._transport is None
+    assert connector._release_acquired.called
     assert not connector._release.called
     assert conn.closed
+
+
+def test_detach_closed(connector, key, request, transport, protocol, loop):
+    conn = Connection(connector, key, request,
+                      transport, protocol, loop)
+    conn.release()
+    conn.detach()
+
+    assert not connector._release_acquired.called
+    assert conn._transport is None

tests/test_client_session.py

@@ -371,6 +371,7 @@ def test_request_ctx_manager_props(loop):
         assert isinstance(ctx_mgr.gi_frame, types.FrameType)
         assert not ctx_mgr.gi_running
         assert isinstance(ctx_mgr.gi_code, types.CodeType)
+        yield from asyncio.sleep(0.1, loop=loop)
 
 
 @asyncio.coroutine

tests/test_connector.py

@@ -149,11 +149,43 @@ def test_get_expired(loop):
     conn.close()
 
 
+def test_release_acquired(loop):
+    conn = aiohttp.BaseConnector(loop=loop, limit=5)
+    conn._release_waiter = unittest.mock.Mock()
+
+    key, tr = 1, unittest.mock.Mock()
+    conn._acquired[key].add(tr)
+    acquired = conn._release_acquired(key, tr)
+    assert 0 == len(conn._acquired[key])
+    assert acquired == conn._acquired[key]
+    assert not conn._release_waiter.called
+
+    acquired = conn._release_acquired(key, tr)
+    assert 0 == len(conn._acquired[key])
+    assert acquired is None
+
+    conn.close()
+
+
+def test_release_acquired_closed(loop):
+    conn = aiohttp.BaseConnector(loop=loop, limit=5)
+    conn._release_waiter = unittest.mock.Mock()
+
+    key, tr = 1, unittest.mock.Mock()
+    conn._acquired[key].add(tr)
+    conn._closed = True
+    conn._release_acquired(key, tr)
+    assert 1 == len(conn._acquired[key])
+    assert not conn._release_waiter.called
+    conn.close()
+
+
 def test_release(loop):
     loop.time = mock.Mock(return_value=10)
 
     conn = aiohttp.BaseConnector(loop=loop)
     conn._start_cleanup_task = unittest.mock.Mock()
+    conn._release_waiter = unittest.mock.Mock()
     req = unittest.mock.Mock()
     resp = req.response = unittest.mock.Mock()
     resp._should_close = False
@@ -162,11 +194,63 @@ def test_release(loop):
     key = 1
     conn._acquired[key].add(tr)
     conn._release(key, req, tr, proto)
+    assert conn._release_waiter.called
     assert conn._conns[1][0] == (tr, proto, 10)
     assert conn._start_cleanup_task.called
     conn.close()
 
 
+def test_release_already_closed(loop):
+    conn = aiohttp.BaseConnector(loop=loop)
+
+    tr, proto = unittest.mock.Mock(), unittest.mock.Mock()
+    key = 1
+    conn._acquired[key].add(tr)
+    conn.close()
+
+    conn._start_cleanup_task = unittest.mock.Mock()
+    conn._release_waiter = unittest.mock.Mock()
+    conn._release_acquired = unittest.mock.Mock()
+    req = unittest.mock.Mock()
+
+    conn._release(key, req, tr, proto)
+    assert not conn._release_waiter.called
+    assert not conn._start_cleanup_task.called
+    assert not conn._release_acquired.called
+
+
+def test_release_do_not_call_release_waiter(loop):
+    req = unittest.mock.Mock()
+    resp = req.response = unittest.mock.Mock()
+    resp._should_close = False
+    tr, proto = unittest.mock.Mock(), unittest.mock.Mock()
+    key = 1
+
+    # limit is None
+    conn = aiohttp.BaseConnector(limit=None, loop=loop)
+    conn._release_waiter = unittest.mock.Mock()
+    conn._acquired[key].add(tr)
+    conn._release(key, req, tr, proto)
+    assert not conn._release_waiter.called
+    conn.close()
+
+    # acquired key error
+    conn = aiohttp.BaseConnector(loop=loop)
+    conn._release_waiter = unittest.mock.Mock()
+    conn._release(key, req, tr, proto)
+    assert not conn._release_waiter.called
+    conn.close()
+
+    # acquired len >= limit
+    conn = aiohttp.BaseConnector(limit=1, loop=loop)
+    conn._release_waiter = unittest.mock.Mock()
+    conn._acquired[key].add(unittest.mock.Mock())
+    conn._acquired[key].add(tr)
+    conn._release(key, req, tr, proto)
+    assert not conn._release_waiter.called
+    conn.close()
+
+
 def test_release_close(loop):
     conn = aiohttp.BaseConnector(loop=loop)
     req = unittest.mock.Mock()

tests/test_proxy.py

@@ -228,7 +228,7 @@ def test_https_connect(self, ClientRequestMock):
         self.assertEqual(req.url.path, '/')
         self.assertEqual(proxy_req.method, 'CONNECT')
         self.assertEqual(proxy_req.url, URL('https://www.python.org'))
-        tr.pause_reading.assert_called_once_with()
+        tr.close.assert_called_once_with()
         tr.get_extra_info.assert_called_with('socket', default=None)
 
         self.loop.run_until_complete(proxy_req.close())
@@ -409,7 +409,7 @@ def test_https_connect_pass_ssl_context(self, ClientRequestMock):
         self.assertEqual(req.url.path, '/')
         self.assertEqual(proxy_req.method, 'CONNECT')
         self.assertEqual(proxy_req.url, URL('https://www.python.org'))
-        tr.pause_reading.assert_called_once_with()
+        tr.close.assert_called_once_with()
         tr.get_extra_info.assert_called_with('socket', default=None)
 
         self.loop.run_until_complete(proxy_req.close())