python/aiogram
1
0
Fork 0
mirror of https://github.com/aiogram/aiogram.git synced 2025-12-12 10:11:52 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Remove filters factory, introduce docs translation (#978)

Browse source 
* Rewrite filters

* Update README.rst

* Fixed tests

* Small optimization of the Text filter (TY to @bomzheg)

* Remove dataclass slots argument in due to the only Python 3.10 has an slots argument

* Fixed mypy

* Update tests

* Disable Python 3.11

* Fixed #1013: Empty mention should be None instead of empty string.

* Added #990 to the changelog

* Added #942 to the changelog

* Fixed coverage

* Update poetry and dependencies

* Fixed mypy

* Remove deprecated code

* Added more tests, update pyproject.toml

* Partial update docs

* Added initial Docs translation files

* Added more changes

* Added log message when connection is established in polling process

* Fixed action

* Disable lint for PyPy

* Added changelog for docs translation
This commit is contained in:
Alex Root Junior 2022-10-02 00:04:31 +03:00 • committed by GitHub
parent 94030903ec
commit f4251382e8
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
610 changed files with 61738 additions and 1687 deletions
Download patch file Download diff file Expand all files Collapse all files

6
tests/test_api/test_types/test_message_entity.py
View file

@ -1,13 +1,7 @@
from aiogram.types import MessageEntity
from tests.deprecated import check_deprecated
class TestMessageEntity:
def test_extract_from(self):
entity = MessageEntity(type="hashtag", length=4, offset=5)
assert entity.extract_from("#foo #bar #baz") == "#bar"
def test_extract(self):
entity = MessageEntity(type="hashtag", length=4, offset=5)
with check_deprecated("3.0b5", exception=AttributeError):
assert entity.extract("#foo #bar #baz") == "#bar"

13
tests/test_dispatcher/test_dispatcher.py
View file

@ -30,6 +30,7 @@ from aiogram.types import (
Update,
User,
)
from aiogram.types.error_event import ErrorEvent
from tests.mocked_bot import MockedBot
try:
@ -76,6 +77,12 @@ class TestDispatcher:
assert dp.update.handlers[0].callback == dp._listen_update
assert dp.update.outer_middleware
def test_init_args(self, bot: MockedBot):
with pytest.raises(TypeError):
Dispatcher(bot)
with pytest.raises(TypeError):
Dispatcher(storage=bot)
def test_data_bind(self):
dp = Dispatcher()
assert dp.get("foo") is None
@ -650,15 +657,15 @@ class TestDispatcher:
await dp.feed_update(bot, update)
@router.errors()
async def error_handler(event: Update, exception: Exception):
async def error_handler(event: ErrorEvent):
return "KABOOM"
response = await dp.feed_update(bot, update)
assert response == "KABOOM"
@dp.errors()
async def root_error_handler(event: Update, exception: Exception):
return exception
async def root_error_handler(event: ErrorEvent):
return event.exception
response = await dp.feed_update(bot, update)

16
tests/test_dispatcher/test_event/test_handler.py
View file

@ -2,12 +2,14 @@ import functools
from typing import Any, Dict, Union
import pytest
from magic_filter import F as A
from aiogram import F
from aiogram.dispatcher.event.handler import CallableMixin, FilterObject, HandlerObject
from aiogram.filters import BaseFilter
from aiogram.filters import Filter
from aiogram.handlers import BaseHandler
from aiogram.types import Update
from aiogram.utils.warnings import Recommendation
pytestmark = pytest.mark.asyncio
@ -28,7 +30,7 @@ async def callback4(foo: int, *, bar: int, baz: int):
return locals()
class Filter(BaseFilter):
class TestFilter(Filter):
async def __call__(self, foo: int, bar: int, baz: int) -> Union[bool, Dict[str, Any]]:
return locals()
@ -39,7 +41,7 @@ class SyncCallable:
class TestCallableMixin:
@pytest.mark.parametrize("callback", [callback2, Filter()])
@pytest.mark.parametrize("callback", [callback2, TestFilter()])
def test_init_awaitable(self, callback):
obj = CallableMixin(callback)
assert obj.awaitable
@ -57,7 +59,7 @@ class TestCallableMixin:
pytest.param(callback1, {"foo", "bar", "baz"}),
pytest.param(callback2, {"foo", "bar", "baz"}),
pytest.param(callback3, {"foo"}),
pytest.param(Filter(), {"self", "foo", "bar", "baz"}),
pytest.param(TestFilter(), {"self", "foo", "bar", "baz"}),
pytest.param(SyncCallable(), {"self", "foo", "bar", "baz"}),
],
)
@ -117,7 +119,7 @@ class TestCallableMixin:
{"foo": 42, "baz": "fuz", "bar": "test"},
),
pytest.param(
Filter(), {"foo": 42, "spam": True, "baz": "fuz"}, {"foo": 42, "baz": "fuz"}
TestFilter(), {"foo": 42, "spam": True, "baz": "fuz"}, {"foo": 42, "baz": "fuz"}
),
pytest.param(
SyncCallable(), {"foo": 42, "spam": True, "baz": "fuz"}, {"foo": 42, "baz": "fuz"}
@ -209,3 +211,7 @@ class TestHandlerObject:
assert len(handler.filters) == 1
result = await handler.call(Update(update_id=42))
assert result == 42
def test_warn_another_magic(self):
with pytest.warns(Recommendation):
FilterObject(callback=A.test.is_(True))

177
tests/test_dispatcher/test_event/test_telegram.py
View file

@ -1,17 +1,16 @@
import datetime
import functools
from typing import Any, Awaitable, Callable, Dict, NoReturn, Optional, Union
from typing import Any, Dict, NoReturn, Optional, Union
import pytest
from pydantic import BaseModel
from aiogram.dispatcher.event.bases import REJECTED, SkipHandler
from aiogram.dispatcher.event.handler import HandlerObject
from aiogram.dispatcher.event.telegram import TelegramEventObserver
from aiogram.dispatcher.router import Router
from aiogram.exceptions import FiltersResolveError
from aiogram.filters import BaseFilter, Command
from aiogram.filters import Filter
from aiogram.types import Chat, Message, User
from tests.deprecated import check_deprecated
pytestmark = pytest.mark.asyncio
@ -31,7 +30,7 @@ async def pipe_handler(*args, **kwargs):
return args, kwargs
class MyFilter1(BaseFilter):
class MyFilter1(Filter, BaseModel):
test: str
async def __call__(self, *args: Any, **kwargs: Any) -> Union[bool, Dict[str, Any]]:
@ -46,14 +45,14 @@ class MyFilter3(MyFilter1):
pass
class OptionalFilter(BaseFilter):
class OptionalFilter(Filter, BaseModel):
optional: Optional[str]
async def __call__(self, *args: Any, **kwargs: Any) -> Union[bool, Dict[str, Any]]:
return True
class DefaultFilter(BaseFilter):
class DefaultFilter(Filter, BaseModel):
default: str = "Default"
async def __call__(self, *args: Any, **kwargs: Any) -> Union[bool, Dict[str, Any]]:
@ -61,144 +60,9 @@ class DefaultFilter(BaseFilter):
class TestTelegramEventObserver:
def test_bind_filter(self):
event_observer = TelegramEventObserver(Router(), "test")
with pytest.raises(TypeError):
event_observer.bind_filter(object) # type: ignore
class MyFilter(BaseFilter):
async def __call__(
self, *args: Any, **kwargs: Any
) -> Callable[[Any], Awaitable[Union[bool, Dict[str, Any]]]]:
pass
event_observer.bind_filter(MyFilter)
assert event_observer.filters
assert MyFilter in event_observer.filters
def test_resolve_filters_chain(self):
router1 = Router(use_builtin_filters=False)
router2 = Router(use_builtin_filters=False)
router3 = Router(use_builtin_filters=False)
router1.include_router(router2)
router2.include_router(router3)
router1.message.bind_filter(MyFilter1)
router1.message.bind_filter(MyFilter2)
router2.message.bind_filter(MyFilter2)
router3.message.bind_filter(MyFilter3)
filters_chain1 = list(router1.message._resolve_filters_chain())
filters_chain2 = list(router2.message._resolve_filters_chain())
filters_chain3 = list(router3.message._resolve_filters_chain())
assert MyFilter1 in filters_chain1
assert MyFilter1 in filters_chain2
assert MyFilter1 in filters_chain3
assert MyFilter2 in filters_chain1
assert MyFilter2 in filters_chain2
assert MyFilter2 in filters_chain3
assert MyFilter3 in filters_chain3
assert MyFilter3 not in filters_chain1
async def test_resolve_filters_data_from_parent_router(self):
class FilterSet(BaseFilter):
set_filter: bool
async def __call__(self, message: Message) -> dict:
return {"test": "hello world"}
class FilterGet(BaseFilter):
get_filter: bool
async def __call__(self, message: Message, **data) -> bool:
assert "test" in data
return True
router1 = Router(use_builtin_filters=False)
router2 = Router(use_builtin_filters=False)
router1.include_router(router2)
router1.message.bind_filter(FilterSet)
router2.message.bind_filter(FilterGet)
@router2.message(set_filter=True, get_filter=True)
def handler_test(msg: Message, test: str):
assert test == "hello world"
await router1.propagate_event(
"message",
Message(message_id=1, date=datetime.datetime.now(), chat=Chat(id=1, type="private")),
)
def test_resolve_filters(self):
router = Router(use_builtin_filters=False)
observer = router.message
observer.bind_filter(MyFilter1)
resolved = observer.resolve_filters((), {"test": "PASS"})
assert isinstance(resolved, list)
assert any(isinstance(item, MyFilter1) for item in resolved)
# Unknown filter
with pytest.raises(FiltersResolveError, match="Unknown keyword filters: {'@bad'}"):
assert observer.resolve_filters((), {"@bad": "very"})
# Unknown filter
with pytest.raises(FiltersResolveError, match="Unknown keyword filters: {'@bad'}"):
assert observer.resolve_filters((), {"test": "ok", "@bad": "very"})
# Bad argument type
with pytest.raises(FiltersResolveError, match="Unknown keyword filters: {'test'}"):
assert observer.resolve_filters((), {"test": ...})
# Disallow same filter using
with pytest.raises(FiltersResolveError, match="Unknown keyword filters: {'test'}"):
observer.resolve_filters((MyFilter1(test="test"),), {"test": ...})
def test_dont_autoresolve_optional_filters_for_router(self):
router = Router(use_builtin_filters=False)
observer = router.message
observer.bind_filter(MyFilter1)
observer.bind_filter(OptionalFilter)
observer.bind_filter(DefaultFilter)
observer.filter(test="test")
assert len(observer._handler.filters) == 1
def test_register_autoresolve_optional_filters(self):
router = Router(use_builtin_filters=False)
observer = router.message
observer.bind_filter(MyFilter1)
observer.bind_filter(OptionalFilter)
observer.bind_filter(DefaultFilter)
assert observer.register(my_handler) == my_handler
assert isinstance(observer.handlers[0], HandlerObject)
assert isinstance(observer.handlers[0].filters[0].callback, OptionalFilter)
assert len(observer.handlers[0].filters) == 2
assert isinstance(observer.handlers[0].filters[0].callback, OptionalFilter)
assert isinstance(observer.handlers[0].filters[1].callback, DefaultFilter)
observer.register(my_handler, test="ok")
assert isinstance(observer.handlers[1], HandlerObject)
assert len(observer.handlers[1].filters) == 3
assert isinstance(observer.handlers[1].filters[0].callback, MyFilter1)
assert isinstance(observer.handlers[1].filters[1].callback, OptionalFilter)
assert isinstance(observer.handlers[1].filters[2].callback, DefaultFilter)
observer.register(my_handler, test="ok", optional="ok")
assert isinstance(observer.handlers[2], HandlerObject)
assert len(observer.handlers[2].filters) == 3
assert isinstance(observer.handlers[2].filters[0].callback, MyFilter1)
assert isinstance(observer.handlers[2].filters[1].callback, OptionalFilter)
assert isinstance(observer.handlers[2].filters[2].callback, DefaultFilter)
def test_register(self):
router = Router(use_builtin_filters=False)
router = Router()
observer = router.message
observer.bind_filter(MyFilter1)
assert observer.register(my_handler) == my_handler
assert isinstance(observer.handlers[0], HandlerObject)
@ -210,19 +74,19 @@ class TestTelegramEventObserver:
assert len(observer.handlers[1].filters) == 1
assert observer.handlers[1].filters[0].callback == f
observer.register(my_handler, test="PASS")
observer.register(my_handler, MyFilter1(test="PASS"))
assert isinstance(observer.handlers[2], HandlerObject)
assert any(isinstance(item.callback, MyFilter1) for item in observer.handlers[2].filters)
f2 = MyFilter2(test="ok")
observer.register(my_handler, f2, test="PASS")
observer.register(my_handler, f2, MyFilter1(test="PASS"))
assert isinstance(observer.handlers[3], HandlerObject)
callbacks = [filter_.callback for filter_ in observer.handlers[3].filters]
assert f2 in callbacks
assert MyFilter1(test="PASS") in callbacks
def test_register_decorator(self):
router = Router(use_builtin_filters=False)
router = Router()
observer = router.message
@observer()
@ -233,10 +97,9 @@ class TestTelegramEventObserver:
assert observer.handlers[0].callback == my_handler
async def test_trigger(self):
router = Router(use_builtin_filters=False)
router = Router()
observer = router.message
observer.bind_filter(MyFilter1)
observer.register(my_handler, test="ok")
observer.register(my_handler, MyFilter1(test="ok"))
message = Message(
message_id=42,
@ -258,7 +121,7 @@ class TestTelegramEventObserver:
),
)
def test_register_filters_via_decorator(self, count, handler, filters):
router = Router(use_builtin_filters=False)
router = Router()
observer = router.message
for index in range(count):
@ -272,7 +135,7 @@ class TestTelegramEventObserver:
assert len(registered_handler.filters) == len(filters)
async def test_trigger_right_context_in_handlers(self):
router = Router(use_builtin_filters=False)
router = Router()
observer = router.message
async def mix_unnecessary_data(event):
@ -328,7 +191,7 @@ class TestTelegramEventObserver:
assert list(middlewares) == [my_middleware1, my_middleware2, my_middleware3]
def test_register_global_filters(self):
router = Router(use_builtin_filters=False)
router = Router()
assert isinstance(router.message._handler.filters, list)
assert not router.message._handler.filters
@ -369,13 +232,3 @@ class TestTelegramEventObserver:
r2.message.register(handler)
assert await r1.message.trigger(None) is REJECTED
def test_deprecated_bind_filter(self):
router = Router()
with check_deprecated("3.0b5", exception=AttributeError):
router.message.bind_filter(MyFilter1)
def test_deprecated_resolve_filters(self):
router = Router()
with check_deprecated("3.0b5", exception=AttributeError):
router.message.resolve_filters([Command], full_config={"commands": ["test"]})

10
tests/test_dispatcher/test_router.py
View file

@ -2,7 +2,6 @@ import pytest
from aiogram.dispatcher.event.bases import UNHANDLED, SkipHandler, skip
from aiogram.dispatcher.router import Router
from aiogram.utils.warnings import CodeHasNoEffect
pytestmark = pytest.mark.asyncio
@ -36,15 +35,6 @@ class TestRouter:
assert router3.parent_router is router2
assert router3.sub_routers == []
def test_include_router_code_has_no_effect(self):
router1 = Router()
router2 = Router(use_builtin_filters=False)
assert router1.use_builtin_filters
assert not router2.use_builtin_filters
with pytest.warns(CodeHasNoEffect):
assert router1.include_router(router2)
def test_include_router_by_string_bad_type(self):
router = Router()
with pytest.raises(ValueError, match=r"router should be instance of Router"):

26
tests/test_filters/test_base.py
View file

@ -2,7 +2,8 @@ from typing import Awaitable
import pytest
from aiogram.filters import BaseFilter
from aiogram.filters import Filter
from aiogram.filters.base import _InvertFilter
try:
from asynctest import CoroutineMock, patch
@ -13,16 +14,14 @@ except ImportError:
pytestmark = pytest.mark.asyncio
class MyFilter(BaseFilter):
foo: str
class MyFilter(Filter):
async def __call__(self, event: str):
return
class TestBaseFilter:
async def test_awaitable(self):
my_filter = MyFilter(foo="bar")
my_filter = MyFilter()
assert isinstance(my_filter, Awaitable)
@ -33,3 +32,20 @@ class TestBaseFilter:
call = my_filter(event="test")
await call
mocked_call.assert_awaited_with(event="test")
async def test_invert(self):
my_filter = MyFilter()
my_inverted_filter = ~my_filter
assert str(my_inverted_filter) == f"~{str(my_filter)}"
assert isinstance(my_inverted_filter, _InvertFilter)
with patch(
"tests.test_filters.test_base.MyFilter.__call__",
new_callable=CoroutineMock,
) as mocked_call:
call = my_inverted_filter(event="test")
result = await call
mocked_call.assert_awaited_with(event="test")
assert not result

4
tests/test_filters/test_callback_data.py
View file

@ -179,3 +179,7 @@ class TestCallbackDataFilter:
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"))
def test_str(self):
filter_object = MyCallback.filter(F.test)
assert str(filter_object).startswith("CallbackQueryFilter(callback_data=")

4
tests/test_filters/test_chat_member_updated.py
View file

@ -343,3 +343,7 @@ class TestChatMemberUpdatedStatusFilter:
)
assert await updated_filter(event) is result
def test_str(self):
updated_filter = ChatMemberUpdatedFilter(member_status_changed=JOIN_TRANSITION)
assert str(updated_filter).startswith("ChatMemberUpdatedFilter(member_status_changed=")

88
tests/test_filters/test_command.py
View file

@ -6,44 +6,67 @@ import pytest
from aiogram import F
from aiogram.filters import Command, CommandObject
from aiogram.filters.command import CommandStart
from aiogram.types import Chat, Message, User
from aiogram.types import BotCommand, Chat, Message, User
from tests.mocked_bot import MockedBot
pytestmark = pytest.mark.asyncio
class TestCommandFilter:
def test_commands_not_iterable(self):
with pytest.raises(ValueError):
Command(commands=1)
def test_bad_type(self):
with pytest.raises(ValueError):
Command(1)
def test_without_args(self):
with pytest.raises(ValueError):
Command()
def test_resolve_bot_command(self):
command = Command(BotCommand(command="test", description="Test"))
assert isinstance(command.commands[0], str)
assert command.commands[0] == "test"
def test_convert_to_list(self):
cmd = Command(commands="start")
assert cmd.commands
assert isinstance(cmd.commands, list)
assert isinstance(cmd.commands, tuple)
assert cmd.commands[0] == "start"
assert cmd == Command(commands=["start"])
# assert cmd == Command(commands=["start"])
@pytest.mark.parametrize(
"text,command,result",
[
["/test@tbot", Command(commands=["test"], commands_prefix="/"), True],
["!test", Command(commands=["test"], commands_prefix="/"), False],
["/test@mention", Command(commands=["test"], commands_prefix="/"), False],
["/tests", Command(commands=["test"], commands_prefix="/"), False],
["/", Command(commands=["test"], commands_prefix="/"), False],
["/ test", Command(commands=["test"], commands_prefix="/"), False],
["", Command(commands=["test"], commands_prefix="/"), False],
[" ", Command(commands=["test"], commands_prefix="/"), False],
["test", Command(commands=["test"], commands_prefix="/"), False],
[" test", Command(commands=["test"], commands_prefix="/"), False],
["a", Command(commands=["test"], commands_prefix="/"), False],
["/test@tbot", Command(commands=["test"], prefix="/"), True],
["/test@tbot", Command("test", prefix="/"), True],
[
"/test@tbot",
Command(BotCommand(command="test", description="description"), prefix="/"),
True,
],
["!test", Command(commands=["test"], prefix="/"), False],
["/test@mention", Command(commands=["test"], prefix="/"), False],
["/tests", Command(commands=["test"], prefix="/"), False],
["/", Command(commands=["test"], prefix="/"), False],
["/ test", Command(commands=["test"], prefix="/"), False],
["", Command(commands=["test"], prefix="/"), False],
[" ", Command(commands=["test"], prefix="/"), False],
["test", Command(commands=["test"], prefix="/"), False],
[" test", Command(commands=["test"], prefix="/"), False],
["a", Command(commands=["test"], prefix="/"), False],
["/test@tbot some args", Command(commands=["test"]), True],
["/test42@tbot some args", Command(commands=[re.compile(r"test(\d+)")]), True],
[
"/test42@tbot some args",
Command(commands=[re.compile(r"test(\d+)")], command_magic=F.args == "some args"),
Command(commands=[re.compile(r"test(\d+)")], magic=F.args == "some args"),
True,
],
[
"/test42@tbot some args",
Command(commands=[re.compile(r"test(\d+)")], command_magic=F.args == "test"),
Command(commands=[re.compile(r"test(\d+)")], magic=F.args == "test"),
False,
],
["/start test", CommandStart(), True],
@ -86,6 +109,7 @@ class TestCommandFilter:
),
True,
],
[None, False],
],
)
async def test_call(self, message: Message, result: bool, bot: MockedBot):
@ -99,11 +123,41 @@ class TestCommandFilter:
chat=Chat(id=42, type="private"),
date=datetime.datetime.now(),
)
command = Command(commands=["test"], command_magic=(F.args.as_("args")))
command = Command(commands=["test"], magic=(F.args.as_("args")))
result = await command(message=message, bot=bot)
assert "args" in result
assert result["args"] == "42"
async def test_empty_mention_is_none(self, bot: MockedBot):
# Fixed https://github.com/aiogram/aiogram/issues/1013:
# Empty mention should be None instead of empty string.
message = Message(
message_id=0,
text="/test",
chat=Chat(id=42, type="private"),
date=datetime.datetime.now(),
)
command = Command("test")
result = await command(message=message, bot=bot)
assert "command" in result
command_obj: CommandObject = result["command"]
assert command_obj.mention is None
def test_str(self):
cmd = Command(commands=["start"])
assert str(cmd) == "Command('start', prefix='/', ignore_case=False, ignore_mention=False)"
class TestCommandStart:
def test_str(self):
cmd = CommandStart()
assert (
str(cmd)
== "CommandStart(ignore_case=False, ignore_mention=False, deep_link=False, deep_link_encoded=False)"
)
class TestCommandObject:
@pytest.mark.parametrize(

52
tests/test_filters/test_content_types.py
View file

@ -1,52 +0,0 @@
from dataclasses import dataclass
from typing import cast
import pytest
from pydantic import ValidationError
from aiogram.filters import ContentTypesFilter
from aiogram.types import ContentType, Message
pytestmark = pytest.mark.asyncio
@dataclass
class MinimalMessage:
content_type: str
class TestContentTypesFilter:
def test_validator_empty_list(self):
filter_ = ContentTypesFilter(content_types=[])
assert filter_.content_types == []
def test_convert_to_list(self):
filter_ = ContentTypesFilter(content_types="text")
assert filter_.content_types
assert isinstance(filter_.content_types, list)
assert filter_.content_types[0] == "text"
assert filter_ == ContentTypesFilter(content_types=["text"])
@pytest.mark.parametrize("values", [["text", "photo"], ["sticker"]])
def test_validator_with_values(self, values):
filter_ = ContentTypesFilter(content_types=values)
assert filter_.content_types == values
@pytest.mark.parametrize("values", [["test"], ["text", "test"], ["TEXT"]])
def test_validator_with_bad_values(self, values):
with pytest.raises(ValidationError):
ContentTypesFilter(content_types=values)
@pytest.mark.parametrize(
"values,content_type,result",
[
[[ContentType.TEXT], ContentType.TEXT, True],
[[ContentType.PHOTO], ContentType.TEXT, False],
[[ContentType.ANY], ContentType.TEXT, True],
[[ContentType.TEXT, ContentType.PHOTO, ContentType.DOCUMENT], ContentType.TEXT, True],
[[ContentType.ANY, ContentType.PHOTO, ContentType.DOCUMENT], ContentType.TEXT, True],
],
)
async def test_call(self, values, content_type, result):
filter_ = ContentTypesFilter(content_types=values)
assert await filter_(cast(Message, MinimalMessage(content_type=content_type))) == result

19
tests/test_filters/test_exception.py
View file

@ -5,6 +5,7 @@ import pytest
from aiogram import Dispatcher
from aiogram.filters import ExceptionMessageFilter, ExceptionTypeFilter
from aiogram.types import Update
from aiogram.types.error_event import ErrorEvent
pytestmark = pytest.mark.asyncio
@ -18,13 +19,17 @@ class TestExceptionMessageFilter:
async def test_match(self):
obj = ExceptionMessageFilter(pattern="KABOOM")
result = await obj(Update(update_id=0), exception=Exception())
result = await obj(ErrorEvent(update=Update(update_id=0), exception=Exception()))
assert not result
result = await obj(Update(update_id=0), exception=Exception("KABOOM"))
result = await obj(ErrorEvent(update=Update(update_id=0), exception=Exception("KABOOM")))
assert isinstance(result, dict)
assert "match_exception" in result
async def test_str(self):
obj = ExceptionMessageFilter(pattern="KABOOM")
assert str(obj) == "ExceptionMessageFilter(pattern=re.compile('KABOOM'))"
class MyException(Exception):
pass
@ -46,12 +51,16 @@ class TestExceptionTypeFilter:
],
)
async def test_check(self, exception: Exception, value: bool):
obj = ExceptionTypeFilter(exception=MyException)
obj = ExceptionTypeFilter(MyException)
result = await obj(Update(update_id=0), exception=exception)
result = await obj(ErrorEvent(update=Update(update_id=0), exception=exception))
assert result == value
def test_without_arguments(self):
with pytest.raises(ValueError):
ExceptionTypeFilter()
class TestDispatchException:
async def test_handle_exception(self, bot):
@ -62,7 +71,7 @@ class TestDispatchException:
raise ValueError("KABOOM")
@dp.errors(ExceptionMessageFilter(pattern="KABOOM"))
async def handler0(update, exception):
async def handler0(event):
return "Handled"
assert await dp.feed_update(bot, Update(update_id=0)) == "Handled"

37
tests/test_filters/test_logic.py
View file

@ -1,37 +0,0 @@
import pytest
from aiogram.filters import Text, and_f, invert_f, or_f
from aiogram.filters.logic import _AndFilter, _InvertFilter, _OrFilter
class TestLogic:
@pytest.mark.parametrize(
"obj,case,result",
[
[True, and_f(lambda t: t is True, lambda t: t is True), True],
[True, and_f(lambda t: t is True, lambda t: t is False), False],
[True, and_f(lambda t: t is False, lambda t: t is False), False],
[True, and_f(lambda t: {"t": t}, lambda t: t is False), False],
[True, and_f(lambda t: {"t": t}, lambda t: t is True), {"t": True}],
[True, or_f(lambda t: t is True, lambda t: t is True), True],
[True, or_f(lambda t: t is True, lambda t: t is False), True],
[True, or_f(lambda t: t is False, lambda t: t is False), False],
[True, or_f(lambda t: t is False, lambda t: t is True), True],
[True, or_f(lambda t: t is False, lambda t: {"t": t}), {"t": True}],
[True, or_f(lambda t: {"t": t}, lambda t: {"a": 42}), {"t": True}],
[True, invert_f(lambda t: t is False), True],
],
)
async def test_logic(self, obj, case, result):
assert await case(obj) == result
@pytest.mark.parametrize(
"case,type_",
[
[Text(text="test") | Text(text="test"), _OrFilter],
[Text(text="test") & Text(text="test"), _AndFilter],
[~Text(text="test"), _InvertFilter],
],
)
def test_dunder_methods(self, case, type_):
assert isinstance(case, type_)

4
tests/test_filters/test_magic_data.py
View file

@ -28,3 +28,7 @@ class TestMagicDataFilter:
assert called
assert isinstance(result, dict)
assert result["test"]
def test_str(self):
f = MagicData(magic_data=F.event.text == "test")
assert str(f).startswith("MagicData(magic_data=")

28
tests/test_filters/test_state.py
View file

@ -16,13 +16,11 @@ class MyGroup(StatesGroup):
class TestStateFilter:
@pytest.mark.parametrize(
"state", [None, State("test"), MyGroup, MyGroup(), "state", ["state"]]
)
@pytest.mark.parametrize("state", [None, State("test"), MyGroup, MyGroup(), "state"])
def test_validator(self, state):
f = StateFilter(state=state)
assert isinstance(f.state, list)
value = f.state[0]
f = StateFilter(state)
assert isinstance(f.states, tuple)
value = f.states[0]
assert (
isinstance(value, (State, str, MyGroup))
or (isclass(value) and issubclass(value, StatesGroup))
@ -32,17 +30,11 @@ class TestStateFilter:
@pytest.mark.parametrize(
"state,current_state,result",
[
[State("state"), "@:state", True],
[[State("state")], "@:state", True],
[MyGroup, "MyGroup:state", True],
[[MyGroup], "MyGroup:state", True],
[MyGroup(), "MyGroup:state", True],
[[MyGroup()], "MyGroup:state", True],
["*", "state", True],
[None, None, True],
[["*"], "state", True],
[[None], None, True],
[None, "state", False],
[[], "state", False],
[[State("state"), "state"], "state", True],
[[MyGroup(), State("state")], "@:state", True],
[[MyGroup, State("state")], "state", False],
@ -50,9 +42,13 @@ class TestStateFilter:
)
@pytestmark
async def test_filter(self, state, current_state, result):
f = StateFilter(state=state)
f = StateFilter(*state)
assert bool(await f(obj=Update(update_id=42), raw_state=current_state)) is result
def test_empty_filter(self):
with pytest.raises(ValueError):
StateFilter()
@pytestmark
async def test_create_filter_from_state(self):
FilterObject(callback=State(state="state"))
@ -72,3 +68,7 @@ class TestStateFilter:
states = {SG.state: "OK"}
assert states.get(copy(SG.state)) == "OK"
def test_str(self):
f = StateFilter("test")
assert str(f) == "StateFilter('test')"

71
tests/test_filters/test_text.py
View file

@ -3,50 +3,38 @@ from itertools import permutations
from typing import Sequence, Type
import pytest
from pydantic import ValidationError
from aiogram.filters import BUILTIN_FILTERS, Text
from aiogram.filters import Text
from aiogram.types import CallbackQuery, Chat, InlineQuery, Message, Poll, PollOption, User
pytestmark = pytest.mark.asyncio
class TestText:
def test_default_for_observer(self):
registered_for = {
update_type for update_type, filters in BUILTIN_FILTERS.items() if Text in filters
}
assert registered_for == {
"message",
"edited_message",
"channel_post",
"edited_channel_post",
"inline_query",
"callback_query",
}
def test_validator_not_enough_arguments(self):
with pytest.raises(ValidationError):
Text()
with pytest.raises(ValidationError):
Text(text_ignore_case=True)
@pytest.mark.parametrize(
"first,last",
permutations(["text", "text_contains", "text_startswith", "text_endswith"], 2),
"kwargs",
[
{},
{"ignore_case": True},
{"ignore_case": False},
],
)
@pytest.mark.parametrize("ignore_case", [True, False])
def test_validator_too_few_arguments(self, first, last, ignore_case):
kwargs = {first: "test", last: "test"}
if ignore_case:
kwargs["text_ignore_case"] = True
with pytest.raises(ValidationError):
def test_not_enough_arguments(self, kwargs):
with pytest.raises(ValueError):
Text(**kwargs)
@pytest.mark.parametrize(
"argument", ["text", "text_contains", "text_startswith", "text_endswith"]
"first,last",
permutations(["text", "contains", "startswith", "endswith"], 2),
)
@pytest.mark.parametrize("ignore_case", [True, False])
def test_validator_too_few_arguments(self, first, last, ignore_case):
kwargs = {first: "test", last: "test", "ignore_case": ignore_case}
with pytest.raises(ValueError):
Text(**kwargs)
@pytest.mark.parametrize("argument", ["text", "contains", "startswith", "endswith"])
@pytest.mark.parametrize("input_type", [str, list, tuple])
def test_validator_convert_to_list(self, argument: str, input_type: Type):
text = Text(**{argument: input_type("test")})
@ -121,7 +109,7 @@ class TestText:
False,
],
[
"text_startswith",
"startswith",
False,
"test",
Message(
@ -134,7 +122,7 @@ class TestText:
True,
],
[
"text_endswith",
"endswith",
False,
"case",
Message(
@ -147,7 +135,7 @@ class TestText:
True,
],
[
"text_contains",
"contains",
False,
" ",
Message(
@ -160,7 +148,7 @@ class TestText:
True,
],
[
"text_startswith",
"startswith",
True,
"question",
Message(
@ -182,7 +170,7 @@ class TestText:
True,
],
[
"text_startswith",
"startswith",
True,
"callback:",
CallbackQuery(
@ -194,7 +182,7 @@ class TestText:
True,
],
[
"text_startswith",
"startswith",
True,
"query",
InlineQuery(
@ -242,5 +230,10 @@ class TestText:
],
)
async def test_check_text(self, argument, ignore_case, input_value, result, update_type):
text = Text(**{argument: input_value}, text_ignore_case=ignore_case)
assert await text(obj=update_type) is result
text = Text(**{argument: input_value}, ignore_case=ignore_case)
test = await text(update_type)
assert test is result
def test_str(self):
text = Text("test")
assert str(text) == "Text(text=['test'], ignore_case=False)"