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

refactor(types): unify InputFile with str in media types (dev-3.x)

Browse source 
Refactored various InputMedia types to support Union[InputFile, str] for
media properties, enhancing flexibility and unifying input data types.

- Modified media_group.py for AUDIO, PHOTO, VIDEO, and DOCUMENT.
- Updated input_media_document.py and input_media_audio.py
  to use Union[InputFile, str].
- Adjusted bot.py to utilize DateTime for close_date.
- Refactored aiohttp session for better file handle and serialization.
- Enhanced set_chat_photo.py to accept Union[InputFile, str].
- Re-organized input_file.py with modern BaseModel features.
- Added new client/form module for splitting file extraction logic.
- Adapted test cases to validate the new structure and behavior.
This commit is contained in:
zemf4you 2024-05-20 09:33:16 +07:00
parent 6f4452f4e0
commit 1ea41076cd
27 changed files with 346 additions and 305 deletions
Download patch file Download diff file Expand all files Collapse all files

50
tests/test_api/test_client/test_session/test_aiohttp_session.py
View file

@ -21,7 +21,13 @@ from aiogram.client.session import aiohttp
from aiogram.client.session.aiohttp import AiohttpSession
from aiogram.exceptions import TelegramNetworkError
from aiogram.methods import TelegramMethod
from aiogram.types import UNSET_PARSE_MODE, InputFile
from aiogram.types import (
InputFile,
InputMediaAudio,
InputMediaDocument,
InputMediaPhoto,
InputMediaVideo,
)
from tests.mocked_bot import MockedBot
@ -139,27 +145,53 @@ class TestAiohttpSession:
assert all(isinstance(field[2], str) for field in fields)
assert "null_" not in [item[0]["name"] for item in fields]
def test_build_form_data_with_file(self, bot: Bot):
class TestMethod(TelegramMethod[bool]):
__api_method__ = "test"
__returning__ = bool
document: Union[InputMediaAudio, InputMediaDocument, InputMediaPhoto, InputMediaVideo]
session = AiohttpSession()
form = session.build_form_data(
bot,
TestMethod(document=InputMediaDocument(media=BareInputFile(filename="file.txt"))),
)
fields = form._fields
assert len(fields) == 2
assert fields[0][0]["name"] == "document"
assert fields[0][2].count("attach://") == 1
assert fields[1][0]["filename"] == "file.txt"
assert isinstance(fields[1][2], AsyncIterable)
def test_build_form_data_with_files(self, bot: Bot):
class TestMethod(TelegramMethod[bool]):
__api_method__ = "test"
__returning__ = bool
key: str
document: InputFile
group: List[
Union[InputMediaAudio, InputMediaDocument, InputMediaPhoto, InputMediaVideo]
]
session = AiohttpSession()
form = session.build_form_data(
bot,
TestMethod(key="value", document=BareInputFile(filename="file.txt")),
TestMethod(
group=[
InputMediaDocument(media=BareInputFile(filename="file.txt")),
InputMediaDocument(media=BareInputFile(filename="file2.txt")),
]
),
)
fields = form._fields
assert len(fields) == 3
assert fields[1][0]["name"] == "document"
assert fields[1][2].startswith("attach://")
assert fields[2][0]["name"] == fields[1][2][9:]
assert fields[2][0]["filename"] == "file.txt"
assert fields[0][0]["name"] == "group"
assert fields[0][2].count("attach://") == 2
assert fields[1][0]["filename"] == "file.txt"
assert fields[2][0]["filename"] == "file2.txt"
assert isinstance(fields[1][2], AsyncIterable)
assert isinstance(fields[2][2], AsyncIterable)
async def test_make_request(self, bot: MockedBot, aresponses: ResponsesMockServer):

97
tests/test_api/test_client/test_session/test_base_session.py
View file

@ -1,6 +1,6 @@
import datetime
import json
from typing import Any, AsyncContextManager, AsyncGenerator, Dict, Optional
from typing import Any, AsyncContextManager, AsyncGenerator, Dict, Optional, Union
from unittest.mock import AsyncMock, patch
import pytest
@ -8,6 +8,7 @@ from pytz import utc
from aiogram import Bot
from aiogram.client.default import Default, DefaultBotProperties
from aiogram.client.form import form_serialize
from aiogram.client.session.base import BaseSession, TelegramType
from aiogram.client.telegram import PRODUCTION, TelegramAPIServer
from aiogram.enums import ChatType, ParseMode, TopicIconColor
@ -26,8 +27,18 @@ from aiogram.exceptions import (
TelegramUnauthorizedError,
)
from aiogram.methods import DeleteMessage, GetMe, TelegramMethod
from aiogram.types import UNSET_PARSE_MODE, User, LinkPreviewOptions
from aiogram.types.base import UNSET_DISABLE_WEB_PAGE_PREVIEW, UNSET_PROTECT_CONTENT
from aiogram.types import (
UNSET_PARSE_MODE,
DateTime,
InputFile,
LinkPreviewOptions,
User,
)
from aiogram.types.base import (
UNSET_DISABLE_WEB_PAGE_PREVIEW,
UNSET_PROTECT_CONTENT,
TelegramObject,
)
from tests.mocked_bot import MockedBot
@ -39,7 +50,7 @@ class CustomSession(BaseSession):
self,
token: str,
method: TelegramMethod[TelegramType],
timeout: Optional[int] = UNSET_PARSE_MODE,
timeout: Optional[int] = None,
) -> None: # type: ignore
assert isinstance(token, str)
assert isinstance(method, TelegramMethod)
@ -94,41 +105,30 @@ class TestBaseSession:
@pytest.mark.parametrize(
"value,result",
[
[None, None],
[None, ...],
["text", "text"],
[ChatType.PRIVATE, "private"],
[TopicIconColor.RED, "16478047"],
[42, "42"],
[True, "true"],
[["test"], '["test"]'],
[["test", ["test"]], '["test", ["test"]]'],
[[{"test": "pass", "spam": None}], '[{"test": "pass"}]'],
[{"test": "pass", "number": 42, "spam": None}, '{"test": "pass", "number": 42}'],
[{"foo": {"test": "pass", "spam": None}}, '{"foo": {"test": "pass"}}'],
[["test", ["test"]], '["test",["test"]]'],
[[{"test": "pass"}], '[{"test":"pass"}]'],
[{"test": "pass", "number": 42}, '{"test":"pass","number":42}'],
[{"foo": {"test": "pass"}}, '{"foo":{"test":"pass"}}'],
[
datetime.datetime(
year=2017, month=5, day=17, hour=4, minute=11, second=42, tzinfo=utc
),
"1494994302",
],
[
{"link_preview": LinkPreviewOptions(is_disabled=True)},
'{"link_preview": {"is_disabled": true}}',
],
[LinkPreviewOptions(is_disabled=True, url=None), '{"is_disabled":true}'],
[Default("parse_mode"), "HTML"],
[Default("protect_content"), "true"],
[Default("link_preview_is_disabled"), "true"],
],
)
def test_prepare_value(self, value: Any, result: str, bot: MockedBot):
session = CustomSession()
assert session.prepare_value(value, bot=bot, files={}) == result
def test_prepare_value_timedelta(self, bot: MockedBot):
session = CustomSession()
value = session.prepare_value(datetime.timedelta(minutes=2), bot=bot, files={})
assert isinstance(value, str)
def test_prepare_value_defaults_replace(self):
def test_form_serialize(self, value: Any, result: str):
bot = MockedBot(
default=DefaultBotProperties(
parse_mode=ParseMode.HTML,
@ -136,18 +136,43 @@ class TestBaseSession:
link_preview_is_disabled=True,
)
)
assert bot.session.prepare_value(Default("parse_mode"), bot=bot, files={}) == "HTML"
assert (
bot.session.prepare_value(Default("link_preview_is_disabled"), bot=bot, files={})
== "true"
)
assert bot.session.prepare_value(Default("protect_content"), bot=bot, files={}) == "true"
def test_prepare_value_defaults_unset(self):
field_type = type(value)
if issubclass(field_type, (datetime.datetime, datetime.timedelta)):
field_type = DateTime
elif issubclass(field_type, InputFile):
field_type = Union[InputFile, str]
elif issubclass(field_type, Default):
field_type = Optional[Union[Any, Default]]
class TestObject(TelegramObject):
field: field_type
obj = TestObject.model_validate({"field": value}, context={"bot": bot})
serialized_obj = obj.model_dump(mode="json", exclude_none=True)
if value is None:
assert "field" not in serialized_obj
else:
value = serialized_obj["field"]
assert form_serialize(value) == result
@pytest.mark.parametrize(
"default",
[
UNSET_PARSE_MODE,
UNSET_DISABLE_WEB_PAGE_PREVIEW,
UNSET_PROTECT_CONTENT,
],
)
def test_default_unset(self, default: Default):
bot = MockedBot()
assert bot.session.prepare_value(UNSET_PARSE_MODE, bot=bot, files={}) is None
assert bot.session.prepare_value(UNSET_DISABLE_WEB_PAGE_PREVIEW, bot=bot, files={}) is None
assert bot.session.prepare_value(UNSET_PROTECT_CONTENT, bot=bot, files={}) is None
class TestObject(TelegramObject):
field: Optional[Union[Any, Default]]
obj = TestObject.model_validate({"field": default}, context={"bot": bot})
serialized_obj = obj.model_dump(mode="json")
assert serialized_obj["field"] is None
@pytest.mark.parametrize(
"status_code,content,error",
@ -205,7 +230,7 @@ class TestBaseSession:
bot = MockedBot()
method = DeleteMessage(chat_id=42, message_id=42)
with pytest.raises(ClientDecodeError, match="JSONDecodeError"):
with pytest.raises(ClientDecodeError, match="Invalid JSON"):
session.check_response(
bot=bot,
method=method,

20
tests/test_api/test_types/test_input_file.py
View file

@ -1,5 +1,3 @@
from typing import AsyncIterable
from aresponses import ResponsesMockServer
from aiogram import Bot
@ -83,3 +81,21 @@ class TestInputFile:
assert chunk_size == 1
size += chunk_size
assert size == 10
async def test_url_input_file_with_default_bot(self, aresponses: ResponsesMockServer):
aresponses.add(
aresponses.ANY,
aresponses.ANY,
"get",
aresponses.Response(status=200, body=b"\f" * 10),
)
async with Bot(token="42:TEST").context() as bot:
file = URLInputFile("https://test.org/", chunk_size=1, bot=bot)
size = 0
async for chunk in file.read():
assert chunk == b"\f"
chunk_size = len(chunk)
assert chunk_size == 1
size += chunk_size
assert size == 10