Merge remote-tracking branch 'origin/dev-3.x-download' into dev-3.x

tests/test_api/test_client/test_session/test_aiohttp_session.py

@@ -1,4 +1,4 @@
-from typing import AsyncContextManager
+from typing import AsyncContextManager, AsyncGenerator
 
 import aiohttp
 import pytest
@@ -107,6 +107,26 @@ class TestAiohttpSession:
 
             assert patched_raise_for_status.called_once()
 
+    @pytest.mark.asyncio
+    async def test_stream_content(self, aresponses: ResponsesMockServer):
+        aresponses.add(
+            aresponses.ANY, aresponses.ANY, "get", aresponses.Response(status=200, body=b"\f" * 10)
+        )
+
+        session = AiohttpSession()
+        stream = session.stream_content(
+            "https://www.python.org/static/img/python-logo.png", timeout=5, chunk_size=1
+        )
+        assert isinstance(stream, AsyncGenerator)
+
+        size = 0
+        async for chunk in stream:
+            assert isinstance(chunk, bytes)
+            chunk_size = len(chunk)
+            assert chunk_size == 1
+            size += chunk_size
+        assert size == 10
+
     @pytest.mark.asyncio
     async def test_context_manager(self):
         session = AiohttpSession()

tests/test_api/test_client/test_session/test_base_session.py

@@ -1,5 +1,5 @@
 import datetime
-from typing import AsyncContextManager
+from typing import AsyncContextManager, AsyncGenerator
 
 import pytest
 
@@ -22,6 +22,14 @@ class CustomSession(BaseSession):
         assert isinstance(token, str)
         assert isinstance(method, TelegramMethod)
 
+    async def stream_content(
+        self, url: str, timeout: int, chunk_size: int
+    ) -> AsyncGenerator[bytes, None]:  # pragma: no cover
+        assert isinstance(url, str)
+        assert isinstance(timeout, int)
+        assert isinstance(chunk_size, int)
+        yield b"\f" * 10
+
 
 class TestBaseSession(DataMixin):
     def test_init_api(self):
@@ -100,6 +108,17 @@ class TestBaseSession(DataMixin):
 
         assert await session.make_request("42:TEST", GetMe()) is None
 
+    @pytest.mark.asyncio
+    async def test_stream_content(self):
+        session = CustomSession()
+        stream = session.stream_content(
+            "https://www.python.org/static/img/python-logo.png", timeout=5, chunk_size=65536
+        )
+        assert isinstance(stream, AsyncGenerator)
+
+        async for chunk in stream:
+            assert isinstance(chunk, bytes)
+
     @pytest.mark.asyncio
     async def test_context_manager(self):
         session = CustomSession()