[3.x] Check status code when downloading file (#1079)

* Check status code when downloading file and raise an error if someting bad happends

* Style fixes

* Add doc

* Use "towncrier create <issue>.<type>" for creating file
This commit is contained in:
Dmitry Anfimov 2023-02-12 06:56:11 +06:00 committed by GitHub
parent 94e11ce8e9
commit 184ee1fbf8
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 45 additions and 7 deletions

View file

@ -37,7 +37,11 @@ class MockedSession(BaseSession):
return response.result # type: ignore
async def stream_content(
self, url: str, timeout: int, chunk_size: int
self,
url: str,
timeout: int,
chunk_size: int,
raise_for_status: bool,
) -> AsyncGenerator[bytes, None]: # pragma: no cover
yield b""

View file

@ -190,7 +190,10 @@ class TestAiohttpSession:
session = AiohttpSession()
stream = session.stream_content(
"https://www.python.org/static/img/python-logo.png", timeout=5, chunk_size=1
"https://www.python.org/static/img/python-logo.png",
timeout=5,
chunk_size=1,
raise_for_status=True,
)
assert isinstance(stream, AsyncGenerator)
@ -202,6 +205,28 @@ class TestAiohttpSession:
size += chunk_size
assert size == 10
async def test_stream_content_404(self, aresponses: ResponsesMockServer):
aresponses.add(
aresponses.ANY,
aresponses.ANY,
"get",
aresponses.Response(
status=404,
body=b"File not found",
),
)
session = AiohttpSession()
stream = session.stream_content(
"https://www.python.org/static/img/python-logo.png",
timeout=5,
chunk_size=1,
raise_for_status=True,
)
with pytest.raises(ClientError):
async for _ in stream:
...
async def test_context_manager(self):
session = AiohttpSession()
assert isinstance(session, AsyncContextManager)