python/aiogram
1
0
Fork 0
mirror of https://github.com/aiogram/aiogram.git synced 2026-04-08 16:37:47 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

[2.x] Don't save error as a file (#813)

Browse source 
* fix: don't save error as file

Raise an aiohttp.ClientResponseError if the response status is 400 or higher

#799

* fix tests

Co-authored-by: darksidecat <58224121+darksidecat@users.noreply.github.com>
This commit is contained in:
Oleg A 2022-02-08 03:29:53 +03:00 • committed by GitHub
parent 3731a62885
commit b39672f9b6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 60 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

35
tests/test_bot/test_bot_download_file.py
View file

@ -1,11 +1,14 @@
import os
from io import BytesIO
from pathlib import Path
from unittest.mock import AsyncMock
import pytest
from aiohttp import ClientResponseError
from aiogram import Bot
from aiogram.types import File
from aiogram.utils.json import json
from tests import TOKEN
from tests.types.dataset import FILE
@ -14,12 +17,9 @@ pytestmark = pytest.mark.asyncio
@pytest.fixture(name='bot')
async def bot_fixture():
async def get_file():
return File(**FILE)
""" Bot fixture """
_bot = Bot(TOKEN)
_bot.get_file = get_file
_bot.get_file = AsyncMock(return_value=File(**FILE))
yield _bot
session = await _bot.get_session()
await session.close()
@ -37,43 +37,54 @@ def tmppath(tmpdir, request):
os.chdir(request.config.invocation_dir)
@pytest.fixture()
def get_file_response(aresponses):
aresponses.add(response=aresponses.Response(body=json.dumps(FILE)))
class TestBotDownload:
async def test_download_file(self, tmppath, bot, file):
async def test_download_file(self, tmppath, bot, file, get_file_response):
f = await bot.download_file(file_path=file.file_path)
assert len(f.read()) != 0
async def test_download_file_destination(self, tmppath, bot, file):
async def test_download_file_destination(self, tmppath, bot, file, get_file_response):
await bot.download_file(file_path=file.file_path, destination="test.file")
assert os.path.isfile(tmppath.joinpath('test.file'))
async def test_download_file_destination_with_dir(self, tmppath, bot, file):
async def test_download_file_destination_with_dir(self, tmppath, bot, file, get_file_response):
await bot.download_file(file_path=file.file_path,
destination=os.path.join('dir_name', 'file_name'))
assert os.path.isfile(tmppath.joinpath('dir_name', 'file_name'))
async def test_download_file_destination_raise_file_not_found(self, tmppath, bot, file):
async def test_download_file_destination_raise_file_not_found(self, tmppath, bot, file, get_file_response):
with pytest.raises(FileNotFoundError):
await bot.download_file(file_path=file.file_path,
destination=os.path.join('dir_name', 'file_name'),
make_dirs=False)
async def test_download_file_destination_io_bytes(self, tmppath, bot, file):
async def test_download_file_destination_io_bytes(self, tmppath, bot, file, get_file_response):
f = BytesIO()
await bot.download_file(file_path=file.file_path,
destination=f)
assert len(f.read()) != 0
async def test_download_file_raise_value_error(self, tmppath, bot, file):
async def test_download_file_raise_value_error(self, tmppath, bot, file, get_file_response):
with pytest.raises(ValueError):
await bot.download_file(file_path=file.file_path, destination="a", destination_dir="b")
async def test_download_file_destination_dir(self, tmppath, bot, file):
async def test_download_file_destination_dir(self, tmppath, bot, file, get_file_response):
await bot.download_file(file_path=file.file_path, destination_dir='test_dir')
assert os.path.isfile(tmppath.joinpath('test_dir', file.file_path))
async def test_download_file_destination_dir_raise_file_not_found(self, tmppath, bot, file):
async def test_download_file_destination_dir_raise_file_not_found(self, tmppath, bot, file, get_file_response):
with pytest.raises(FileNotFoundError):
await bot.download_file(file_path=file.file_path,
destination_dir='test_dir',
make_dirs=False)
assert os.path.isfile(tmppath.joinpath('test_dir', file.file_path))
async def test_download_file_404(self, tmppath, bot, file):
with pytest.raises(ClientResponseError) as exc_info:
await bot.download_file(file_path=file.file_path)
assert exc_info.value.status == 404

48
tests/types/test_mixins.py
View file

@ -1,12 +1,15 @@
import os
from io import BytesIO
from pathlib import Path
from unittest.mock import AsyncMock
import pytest
from aiohttp import ClientResponseError
from aiogram import Bot
from aiogram.types import File
from aiogram.types.mixins import Downloadable
from aiogram.utils.json import json
from tests import TOKEN
from tests.types.dataset import FILE
@ -18,7 +21,8 @@ async def bot_fixture():
""" Bot fixture """
_bot = Bot(TOKEN)
yield _bot
await (await _bot.get_session()).close()
session = await _bot.get_session()
await session.close()
@pytest.fixture
@ -30,73 +34,81 @@ def tmppath(tmpdir, request):
@pytest.fixture
def downloadable(bot):
async def get_file():
return File(**FILE)
downloadable = Downloadable()
downloadable.get_file = get_file
downloadable.get_file = AsyncMock(return_value=File(**FILE))
downloadable.bot = bot
return downloadable
@pytest.fixture()
def get_file_response(aresponses):
aresponses.add(response=aresponses.Response(body=json.dumps(FILE)))
class TestDownloadable:
async def test_download_make_dirs_false_nodir(self, tmppath, downloadable):
async def test_download_make_dirs_false_nodir(self, tmppath, downloadable, get_file_response):
with pytest.raises(FileNotFoundError):
await downloadable.download(make_dirs=False)
async def test_download_make_dirs_false_mkdir(self, tmppath, downloadable):
async def test_download_make_dirs_false_mkdir(self, tmppath, downloadable, get_file_response):
os.mkdir('voice')
await downloadable.download(make_dirs=False)
assert os.path.isfile(tmppath.joinpath(FILE["file_path"]))
async def test_download_make_dirs_true(self, tmppath, downloadable):
async def test_download_make_dirs_true(self, tmppath, downloadable, get_file_response):
await downloadable.download(make_dirs=True)
assert os.path.isfile(tmppath.joinpath(FILE["file_path"]))
async def test_download_deprecation_warning(self, tmppath, downloadable):
async def test_download_deprecation_warning(self, tmppath, downloadable, get_file_response):
with pytest.deprecated_call():
await downloadable.download("test.file")
async def test_download_destination(self, tmppath, downloadable):
async def test_download_destination(self, tmppath, downloadable, get_file_response):
with pytest.deprecated_call():
await downloadable.download("test.file")
assert os.path.isfile(tmppath.joinpath('test.file'))
async def test_download_destination_dir_exist(self, tmppath, downloadable):
async def test_download_destination_dir_exist(self, tmppath, downloadable, get_file_response):
os.mkdir("test_folder")
with pytest.deprecated_call():
await downloadable.download("test_folder")
assert os.path.isfile(tmppath.joinpath('test_folder', FILE["file_path"]))
async def test_download_destination_with_dir(self, tmppath, downloadable):
async def test_download_destination_with_dir(self, tmppath, downloadable, get_file_response):
with pytest.deprecated_call():
await downloadable.download(os.path.join('dir_name', 'file_name'))
assert os.path.isfile(tmppath.joinpath('dir_name', 'file_name'))
async def test_download_destination_io_bytes(self, tmppath, downloadable):
async def test_download_destination_io_bytes(self, tmppath, downloadable, get_file_response):
file = BytesIO()
with pytest.deprecated_call():
await downloadable.download(file)
assert len(file.read()) != 0
async def test_download_raise_value_error(self, tmppath, downloadable):
async def test_download_raise_value_error(self, tmppath, downloadable, get_file_response):
with pytest.raises(ValueError):
await downloadable.download(destination_dir="a", destination_file="b")
async def test_download_destination_dir(self, tmppath, downloadable):
async def test_download_destination_dir(self, tmppath, downloadable, get_file_response):
await downloadable.download(destination_dir='test_dir')
assert os.path.isfile(tmppath.joinpath('test_dir', FILE["file_path"]))
async def test_download_destination_file(self, tmppath, downloadable):
async def test_download_destination_file(self, tmppath, downloadable, get_file_response):
await downloadable.download(destination_file='file_name')
assert os.path.isfile(tmppath.joinpath('file_name'))
async def test_download_destination_file_with_dir(self, tmppath, downloadable):
async def test_download_destination_file_with_dir(self, tmppath, downloadable, get_file_response):
await downloadable.download(destination_file=os.path.join('dir_name', 'file_name'))
assert os.path.isfile(tmppath.joinpath('dir_name', 'file_name'))
async def test_download_io_bytes(self, tmppath, downloadable):
async def test_download_io_bytes(self, tmppath, downloadable, get_file_response):
file = BytesIO()
await downloadable.download(destination_file=file)
assert len(file.read()) != 0
async def test_download_404(self, tmppath, downloadable):
with pytest.raises(ClientResponseError) as exc_info:
await downloadable.download(destination_file='file_name')
assert exc_info.value.status == 404