[3.6] Reset the sock_read timeout when data arrives (#4142) (#4143)

The sock_read timeout should apply *per read operation*, but currently applies
cumulatively across all reads. Reschedule the timeout each time data is
received and add tests to validate that the timeout doesn't interfere with
overall reading and correctly catches reads that take too long.


Co-authored-by: Martijn Pieters <github.com@zopatista.com>

CHANGES/3808.bugfix

@@ -0,0 +1 @@
+Reset the ``sock_read`` timeout each time data is received for a ``aiohttp.client`` response.

CONTRIBUTORS.txt

@@ -160,6 +160,7 @@ Manuel Miranda
 Marat Sharafutdinov
 Marco Paolini
 Mariano Anaya
+Martijn Pieters
 Martin Melka
 Martin Richard
 Mathias Fröjdman

aiohttp/client_proto.py

@@ -176,6 +176,8 @@ def _on_read_timeout(self) -> None:
             self._payload.set_exception(exc)
 
     def data_received(self, data: bytes) -> None:
+        self._reschedule_timeout()
+
         if not data:
             return
 

tests/test_client_functional.py

@@ -598,6 +598,56 @@ async def handler(request):
         await client.get('/')
 
 
+async def test_read_timeout_between_chunks(aiohttp_client, mocker) -> None:
+    mocker.patch('aiohttp.helpers.ceil').side_effect = ceil
+
+    async def handler(request):
+        resp = aiohttp.web.StreamResponse()
+        await resp.prepare(request)
+        # write data 4 times, with pauses. Total time 0.4 seconds.
+        for _ in range(4):
+            await asyncio.sleep(0.1)
+            await resp.write(b'data\n')
+        return resp
+
+    app = web.Application()
+    app.add_routes([web.get('/', handler)])
+
+    # A timeout of 0.2 seconds should apply per read.
+    timeout = aiohttp.ClientTimeout(sock_read=0.2)
+    client = await aiohttp_client(app, timeout=timeout)
+
+    res = b''
+    async with await client.get('/') as resp:
+        res += await resp.read()
+
+    assert res == b'data\n' * 4
+
+
+async def test_read_timeout_on_reading_chunks(aiohttp_client, mocker) -> None:
+    mocker.patch('aiohttp.helpers.ceil').side_effect = ceil
+
+    async def handler(request):
+        resp = aiohttp.web.StreamResponse()
+        await resp.prepare(request)
+        await resp.write(b'data\n')
+        await asyncio.sleep(1)
+        await resp.write(b'data\n')
+        return resp
+
+    app = web.Application()
+    app.add_routes([web.get('/', handler)])
+
+    # A timeout of 0.2 seconds should apply per read.
+    timeout = aiohttp.ClientTimeout(sock_read=0.2)
+    client = await aiohttp_client(app, timeout=timeout)
+
+    async with await client.get('/') as resp:
+        assert (await resp.content.read(5)) == b'data\n'
+        with pytest.raises(asyncio.TimeoutError):
+            await resp.content.read()
+
+
 async def test_timeout_on_reading_data(aiohttp_client, mocker) -> None:
     loop = asyncio.get_event_loop()