diff --git a/.coveragerc b/.coveragerc deleted file mode 100644 index 9feee202..00000000 --- a/.coveragerc +++ /dev/null @@ -1,5 +0,0 @@ -[report] -exclude_lines = - pragma: no cover - if TYPE_CHECKING: - @abstractmethod diff --git a/aiogram/dispatcher/webhook/aiohttp_server.py b/aiogram/dispatcher/webhook/aiohttp_server.py index 9394664a..fefcf2b4 100644 --- a/aiogram/dispatcher/webhook/aiohttp_server.py +++ b/aiogram/dispatcher/webhook/aiohttp_server.py @@ -26,10 +26,10 @@ def setup_application(app: Application, dispatcher: Dispatcher, /, **kwargs: Any **kwargs, } - async def on_startup(*a: Any, **kw: Any) -> None: + async def on_startup(*a: Any, **kw: Any) -> None: # pragma: no cover await dispatcher.emit_startup(**workflow_data) - async def on_shutdown(*a: Any, **kw: Any) -> None: + async def on_shutdown(*a: Any, **kw: Any) -> None: # pragma: no cover await dispatcher.emit_shutdown(**workflow_data) app.on_startup.append(on_startup) diff --git a/pyproject.toml b/pyproject.toml index edb8df95..cf6e8959 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -77,6 +77,7 @@ pytest-lazy-fixture = "^0.6.3" pytest-mock = "^3.6.0" pytest-mypy = "^0.8.1" pytest-cov = "^2.11.1" +pytest-aiohttp = "^0.3.0" aresponses = "^2.1.4" asynctest = "^0.13.0" toml = "^0.10.2" diff --git a/tests/test_dispatcher/test_webhook/__init__.py b/tests/test_dispatcher/test_webhook/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/test_dispatcher/test_webhook/test_aiohtt_server.py b/tests/test_dispatcher/test_webhook/test_aiohtt_server.py new file mode 100644 index 00000000..c75ed518 --- /dev/null +++ b/tests/test_dispatcher/test_webhook/test_aiohtt_server.py @@ -0,0 +1,37 @@ +from typing import Any + +import pytest +from aiohttp.web_app import Application + +from aiogram import Dispatcher +from aiogram.dispatcher.webhook.aiohttp_server import ip_filter_middleware, setup_application +from aiogram.dispatcher.webhook.security import IPFilter +from aiogram.methods import Request + + +class TestAiohttpServer: + def test_setup_application(self): + app = Application() + + dp = Dispatcher() + setup_application(app, dp) + + assert len(app.router.routes()) == 0 + assert len(app.on_startup) == 2 + assert len(app.on_shutdown) == 1 + + async def test_middleware(self, aiohttp_client: Any): + app = Application() + app.middlewares.append(ip_filter_middleware(IPFilter.default())) + + async def handler1(request: Request): + pass + + async def handler2(request: Request): + pytest.fail() + + app.router.add_route("POST", "/good", handler1) + app.router.add_route("POST", "/bad", handler2) + client = await aiohttp_client(app) + resp = await client.get("/bad") + assert resp diff --git a/tests/test_dispatcher/test_webhook/test_security.py b/tests/test_dispatcher/test_webhook/test_security.py new file mode 100644 index 00000000..6a9b84a0 --- /dev/null +++ b/tests/test_dispatcher/test_webhook/test_security.py @@ -0,0 +1,57 @@ +from ipaddress import IPv4Address, IPv4Network + +import pytest + +from aiogram.dispatcher.webhook.security import IPFilter + + +class TestSecurity: + def test_empty_init(self): + ip_filter = IPFilter() + assert not ip_filter._allowed_ips + + @pytest.mark.parametrize( + "ip,result", + [ + ("127.0.0.1", True), + ("127.0.0.2", False), + (IPv4Address("127.0.0.1"), True), + (IPv4Address("127.0.0.2"), False), + (IPv4Address("192.168.0.32"), True), + ("192.168.0.33", False), + ("10.111.0.5", True), + ("10.111.0.100", True), + ("10.111.1.100", False), + ], + ) + def test_check_ip(self, ip, result): + ip_filter = IPFilter( + ips=["127.0.0.1", IPv4Address("192.168.0.32"), IPv4Network("10.111.0.0/24")] + ) + assert (ip in ip_filter) is result + + def test_default(self): + ip_filter = IPFilter.default() + assert isinstance(ip_filter, IPFilter) + assert len(ip_filter._allowed_ips) == 5116 + assert "91.108.4.50" in ip_filter + assert "149.154.160.20" in ip_filter + + @pytest.mark.parametrize( + "ip,ip_range", + [ + ["127.0.0.1", {IPv4Address("127.0.0.1")}], + ["91.108.4.0/22", set(IPv4Network("91.108.4.0/22").hosts())], + [IPv4Address("91.108.4.5"), {IPv4Address("91.108.4.5")}], + [IPv4Network("91.108.4.0/22"), set(IPv4Network("91.108.4.0/22").hosts())], + [42, set()], + ], + ) + def test_allow_ip(self, ip, ip_range): + ip_filter = IPFilter() + if not ip_range: + with pytest.raises(ValueError): + ip_filter.allow_ip(ip) + else: + ip_filter.allow_ip(ip) + assert ip_filter._allowed_ips == ip_range