Fix regression with connection upgrade

CHANGES/7879.bugfix

@@ -0,0 +1 @@
+Fixed a regression where connection may get closed during upgrade. -- by :user:`Dreamsorcerer`

aiohttp/client_reqrep.py

@@ -940,19 +940,9 @@ def _response_eof(self) -> None:
         if self._closed:
             return
 
-        if self._connection is not None:
-            # websocket, protocol could be None because
-            # connection could be detached
-            if (
-                self._connection.protocol is not None
-                and self._connection.protocol.upgraded
-            ):
-                return
-
-            self._release_connection()
-
         self._closed = True
         self._cleanup_writer()
+        self._release_connection()
 
     @property
     def closed(self) -> bool:
@@ -978,7 +968,7 @@ def release(self) -> Any:
         self._closed = True
 
         self._cleanup_writer()
-        self._release_connection()
+        self._release_connection(skip_upgraded=False)
         return noop()
 
     @property
@@ -1003,8 +993,13 @@ def raise_for_status(self) -> None:
                 headers=self.headers,
             )
 
-    def _release_connection(self) -> None:
+    def _release_connection(self, skip_upgraded: bool = True) -> None:
         if self._connection is not None:
+            # protocol could be None because connection could be detached
+            protocol = self._connection.protocol
+            if skip_upgraded and protocol is not None and protocol.upgraded:
+                return
+
             if self._writer is None:
                 self._connection.release()
                 self._connection = None

tests/test_client_functional.py

@@ -174,6 +174,22 @@ async def handler(request):
     assert 1 == len(client._session.connector._conns)
 
 
+async def test_upgrade_connection_not_released_after_read(aiohttp_client) -> None:
+    async def handler(request):
+        body = await request.read()
+        assert b"" == body
+        return web.Response(status=101, headers={"Connection": "Upgrade"})
+
+    app = web.Application()
+    app.router.add_route("GET", "/", handler)
+
+    client = await aiohttp_client(app)
+
+    resp = await client.get("/")
+    await resp.read()
+    assert resp.connection is not None
+
+
 async def test_keepalive_server_force_close_connection(aiohttp_client: Any) -> None:
     async def handler(request):
         body = await request.read()