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

Merge branch 'dev-3.x' into dev-3.x

Browse source
This commit is contained in:
Aleksandr 2021-05-25 10:33:08 +03:00 • committed by GitHub
commit 4b4c9055ff
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
37 changed files with 1112 additions and 839 deletions
Download patch file Download diff file Expand all files Collapse all files

30
tests/mocked_bot.py
View file

@ -4,17 +4,17 @@ from typing import TYPE_CHECKING, AsyncGenerator, Deque, Optional, Type
from aiogram import Bot
from aiogram.client.session.base import BaseSession
from aiogram.methods import TelegramMethod
from aiogram.methods.base import Request, Response, T
from aiogram.types import UNSET
from aiogram.methods.base import Request, Response, TelegramType
from aiogram.types import UNSET, ResponseParameters
class MockedSession(BaseSession):
def __init__(self):
super(MockedSession, self).__init__()
self.responses: Deque[Response[T]] = deque()
self.responses: Deque[Response[TelegramType]] = deque()
self.requests: Deque[Request] = deque()
def add_result(self, response: Response[T]) -> Response[T]:
def add_result(self, response: Response[TelegramType]) -> Response[TelegramType]:
self.responses.append(response)
return response
@ -25,11 +25,13 @@ class MockedSession(BaseSession):
pass
async def make_request(
self, bot: Bot, method: TelegramMethod[T], timeout: Optional[int] = UNSET
) -> T:
self, bot: Bot, method: TelegramMethod[TelegramType], timeout: Optional[int] = UNSET
) -> TelegramType:
self.requests.append(method.build_request(bot))
response: Response[T] = self.responses.pop()
self.raise_for_status(response)
response: Response[TelegramType] = self.responses.pop()
self.check_response(
method=method, status_code=response.error_code, content=response.json()
)
return response.result # type: ignore
async def stream_content(
@ -47,21 +49,23 @@ class MockedBot(Bot):
def add_result_for(
self,
method: Type[TelegramMethod[T]],
method: Type[TelegramMethod[TelegramType]],
ok: bool,
result: T = None,
result: TelegramType = None,
description: Optional[str] = None,
error_code: Optional[int] = None,
migrate_to_chat_id: Optional[int] = None,
retry_after: Optional[int] = None,
) -> Response[T]:
) -> Response[TelegramType]:
response = Response[method.__returning__]( # type: ignore
ok=ok,
result=result,
description=description,
error_code=error_code,
migrate_to_chat_id=migrate_to_chat_id,
retry_after=retry_after,
parameters=ResponseParameters(
migrate_to_chat_id=migrate_to_chat_id,
retry_after=retry_after,
),
)
self.session.add_result(response)
return response

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

@ -172,14 +172,10 @@ class TestAiohttpSession:
return Request(method="method", data={})
call = TestMethod()
with patch(
"aiogram.client.session.base.BaseSession.raise_for_status"
) as patched_raise_for_status:
result = await session.make_request(bot, call)
assert isinstance(result, int)
assert result == 42
assert patched_raise_for_status.called_once()
result = await session.make_request(bot, call)
assert isinstance(result, int)
assert result == 42
@pytest.mark.asyncio
async def test_stream_content(self, aresponses: ResponsesMockServer):

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

@ -4,9 +4,9 @@ from typing import AsyncContextManager, AsyncGenerator, Optional
import pytest
from aiogram.client.session.base import BaseSession, T
from aiogram.client.session.base import BaseSession, TelegramType
from aiogram.client.telegram import PRODUCTION, TelegramAPIServer
from aiogram.methods import GetMe, Response, TelegramMethod
from aiogram.methods import DeleteMessage, GetMe, Response, TelegramMethod
from aiogram.types import UNSET
try:
@ -20,7 +20,7 @@ class CustomSession(BaseSession):
async def close(self):
pass
async def make_request(self, token: str, method: TelegramMethod[T], timeout: Optional[int] = UNSET) -> None: # type: ignore
async def make_request(self, token: str, method: TelegramMethod[TelegramType], timeout: Optional[int] = UNSET) -> None: # type: ignore
assert isinstance(token, str)
assert isinstance(method, TelegramMethod)
@ -135,12 +135,20 @@ class TestBaseSession:
assert session.clean_json(42) == 42
def test_raise_for_status(self):
def check_response(self):
session = CustomSession()
session.raise_for_status(Response[bool](ok=True, result=True))
session.check_response(
method=DeleteMessage(chat_id=42, message_id=42),
status_code=200,
content='{"ok":true,"result":true}',
)
with pytest.raises(Exception):
session.raise_for_status(Response[bool](ok=False, description="Error", error_code=400))
session.check_response(
method=DeleteMessage(chat_id=42, message_id=42),
status_code=400,
content='{"ok":false,"description":"test"}',
)
@pytest.mark.asyncio
async def test_make_request(self):

28
tests/test_api/test_types/test_message.py
View file

@ -5,6 +5,9 @@ import pytest
from aiogram.methods import (
CopyMessage,
DeleteMessage,
EditMessageCaption,
EditMessageText,
SendAnimation,
SendAudio,
SendContact,
@ -549,3 +552,28 @@ class TestMessage:
if method:
assert isinstance(method, expected_method)
# TODO: Check additional fields
def test_edit_text(self):
message = Message(
message_id=42, chat=Chat(id=42, type="private"), date=datetime.datetime.now()
)
method = message.edit_text(text="test")
assert isinstance(method, EditMessageText)
assert method.chat_id == message.chat.id
def test_edit_caption(self):
message = Message(
message_id=42, chat=Chat(id=42, type="private"), date=datetime.datetime.now()
)
method = message.edit_caption(caption="test")
assert isinstance(method, EditMessageCaption)
assert method.chat_id == message.chat.id
def test_delete(self):
message = Message(
message_id=42, chat=Chat(id=42, type="private"), date=datetime.datetime.now()
)
method = message.delete()
assert isinstance(method, DeleteMessage)
assert method.chat_id == message.chat.id
assert method.message_id == message.message_id

177
tests/test_dispatcher/test_filters/test_callback_data.py Normal file
View file

@ -0,0 +1,177 @@
from decimal import Decimal
from enum import Enum, auto
from fractions import Fraction
from typing import Optional
from uuid import UUID
import pytest
from magic_filter import MagicFilter
from pydantic import ValidationError
from aiogram import F
from aiogram.dispatcher.filters.callback_data import CallbackData
from aiogram.types import CallbackQuery, User
class MyIntEnum(Enum):
FOO = auto()
class MyStringEnum(str, Enum):
FOO = "FOO"
class MyCallback(CallbackData, prefix="test"):
foo: str
bar: int
class TestCallbackData:
def test_init_subclass_prefix_required(self):
assert MyCallback.prefix == "test"
with pytest.raises(ValueError, match="prefix required.+"):
class MyInvalidCallback(CallbackData):
pass
def test_init_subclass_sep_validation(self):
assert MyCallback.sep == ":"
class MyCallback2(CallbackData, prefix="test2", sep="@"):
pass
assert MyCallback2.sep == "@"
with pytest.raises(ValueError, match="Separator symbol '@' .+ 'sp@m'"):
class MyInvalidCallback(CallbackData, prefix="sp@m", sep="@"):
pass
@pytest.mark.parametrize(
"value,success,expected",
[
[None, True, ""],
[42, True, "42"],
["test", True, "test"],
[9.99, True, "9.99"],
[Decimal("9.99"), True, "9.99"],
[Fraction("3/2"), True, "3/2"],
[
UUID("123e4567-e89b-12d3-a456-426655440000"),
True,
"123e4567-e89b-12d3-a456-426655440000",
],
[MyIntEnum.FOO, True, "1"],
[MyStringEnum.FOO, True, "FOO"],
[..., False, "..."],
[object, False, "..."],
[object(), False, "..."],
[User(id=42, is_bot=False, first_name="test"), False, "..."],
],
)
def test_encode_value(self, value, success, expected):
callback = MyCallback(foo="test", bar=42)
if success:
assert callback._encode_value("test", value) == expected
else:
with pytest.raises(ValueError):
assert callback._encode_value("test", value) == expected
def test_pack(self):
with pytest.raises(ValueError, match="Separator symbol .+"):
assert MyCallback(foo="te:st", bar=42).pack()
with pytest.raises(ValueError, match=".+is too long.+"):
assert MyCallback(foo="test" * 32, bar=42).pack()
assert MyCallback(foo="test", bar=42).pack() == "test:test:42"
def test_pack_optional(self):
class MyCallback1(CallbackData, prefix="test1"):
foo: str
bar: Optional[int] = None
assert MyCallback1(foo="spam").pack() == "test1:spam:"
assert MyCallback1(foo="spam", bar=42).pack() == "test1:spam:42"
class MyCallback2(CallbackData, prefix="test2"):
foo: Optional[str] = None
bar: int
assert MyCallback2(bar=42).pack() == "test2::42"
assert MyCallback2(foo="spam", bar=42).pack() == "test2:spam:42"
class MyCallback3(CallbackData, prefix="test3"):
foo: Optional[str] = "experiment"
bar: int
assert MyCallback3(bar=42).pack() == "test3:experiment:42"
assert MyCallback3(foo="spam", bar=42).pack() == "test3:spam:42"
def test_unpack(self):
with pytest.raises(TypeError, match=".+ takes 2 arguments but 3 were given"):
MyCallback.unpack("test:test:test:test")
with pytest.raises(ValueError, match="Bad prefix .+"):
MyCallback.unpack("spam:test:test")
assert MyCallback.unpack("test:test:42") == MyCallback(foo="test", bar=42)
def test_unpack_optional(self):
with pytest.raises(ValidationError):
assert MyCallback.unpack("test:test:")
class MyCallback1(CallbackData, prefix="test1"):
foo: str
bar: Optional[int] = None
assert MyCallback1.unpack("test1:spam:") == MyCallback1(foo="spam")
assert MyCallback1.unpack("test1:spam:42") == MyCallback1(foo="spam", bar=42)
class MyCallback2(CallbackData, prefix="test2"):
foo: Optional[str] = None
bar: int
assert MyCallback2.unpack("test2::42") == MyCallback2(bar=42)
assert MyCallback2.unpack("test2:spam:42") == MyCallback2(foo="spam", bar=42)
class MyCallback3(CallbackData, prefix="test3"):
foo: Optional[str] = "experiment"
bar: int
assert MyCallback3.unpack("test3:experiment:42") == MyCallback3(bar=42)
assert MyCallback3.unpack("test3:spam:42") == MyCallback3(foo="spam", bar=42)
def test_build_filter(self):
filter_object = MyCallback.filter(F.foo == "test")
assert isinstance(filter_object.rule, MagicFilter)
assert filter_object.callback_data is MyCallback
class TestCallbackDataFilter:
@pytest.mark.parametrize(
"query,rule,result",
[
["test", F.foo == "test", False],
["test:spam:42", F.foo == "test", False],
["test:test:42", F.foo == "test", {"callback_data": MyCallback(foo="test", bar=42)}],
["test:test:", F.foo == "test", False],
],
)
@pytest.mark.asyncio
async def test_call(self, query, rule, result):
callback_query = CallbackQuery(
id="1",
from_user=User(id=42, is_bot=False, first_name="test"),
data=query,
chat_instance="test",
)
filter_object = MyCallback.filter(rule)
assert await filter_object(callback_query) == result
@pytest.mark.asyncio
async def test_invalid_call(self):
filter_object = MyCallback.filter(F.test)
assert not await filter_object(User(id=42, is_bot=False, first_name="test"))