Cover I18n module

.coveragerc

@@ -2,3 +2,4 @@
 exclude_lines =
     pragma: no cover
     if TYPE_CHECKING:
+    @abstractmethod

aiogram/dispatcher/filters/state.py

@@ -1,5 +1,5 @@
 from inspect import isclass
-from typing import Any, Dict, Optional, Sequence, Type, Union, cast
+from typing import Any, Dict, Optional, Sequence, Type, Union, cast, no_type_check
 
 from pydantic import validator
 
@@ -21,6 +21,7 @@ class StateFilter(BaseFilter):
         arbitrary_types_allowed = True
 
     @validator("state", always=True)
+    @no_type_check  # issubclass breaks things
     def _validate_state(cls, v: Union[StateType, Sequence[StateType]]) -> Sequence[StateType]:
         if (
             isinstance(v, (str, State, StatesGroup))

aiogram/utils/i18n/__init__.py

@@ -1,5 +1,5 @@
-from .babel import I18n
 from .context import get_i18n, gettext, lazy_gettext, lazy_ngettext, ngettext
+from .core import I18n
 from .middleware import (
     ConstI18nMiddleware,
     FSMI18nMiddleware,

aiogram/utils/i18n/context.py

@@ -1,7 +1,7 @@
 from contextvars import ContextVar
 from typing import Any, Optional
 
-from aiogram.utils.i18n.babel import I18n
+from aiogram.utils.i18n.core import I18n
 from aiogram.utils.i18n.lazy_proxy import LazyProxy
 
 ctx_i18n: ContextVar[Optional[I18n]] = ContextVar("aiogram_ctx_i18n", default=None)
@@ -18,12 +18,8 @@ def gettext(*args: Any, **kwargs: Any) -> str:
     return get_i18n().gettext(*args, **kwargs)
 
 
-def _lazy_lazy_gettext(*args: Any, **kwargs: Any) -> str:
-    return str(get_i18n().lazy_gettext(*args, **kwargs))
-
-
 def lazy_gettext(*args: Any, **kwargs: Any) -> LazyProxy:
-    return LazyProxy(_lazy_lazy_gettext, *args, **kwargs)
+    return LazyProxy(gettext, *args, **kwargs)
 
 
 ngettext = gettext

aiogram/utils/i18n/core.py

@@ -44,9 +44,9 @@ class I18n:
             mo_path = os.path.join(self.path, name, "LC_MESSAGES", self.domain + ".mo")
 
             if os.path.exists(mo_path):
-                with open(mo_path, "r") as fp:
-                    translations[name] = gettext.GNUTranslations(fp)
-            elif os.path.exists(mo_path[:-2] + "po"):
+                with open(mo_path, "rb") as fp:
+                    translations[name] = gettext.GNUTranslations(fp)  # type: ignore
+            elif os.path.exists(mo_path[:-2] + "po"):  # pragma: no cover
                 raise RuntimeError(f"Found locale '{name}' but this language is not compiled!")
 
         return translations

aiogram/utils/i18n/middleware.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import Any, Awaitable, Callable, Dict, Optional, Set, TypeVar, cast
+from typing import Any, Awaitable, Callable, Dict, Optional, Set, cast
 
 try:
     from babel import Locale
@@ -9,30 +9,29 @@ except ImportError:  # pragma: no cover
 from aiogram import BaseMiddleware, Router
 from aiogram.dispatcher.fsm.context import FSMContext
 from aiogram.types import TelegramObject, User
-from aiogram.utils.i18n.babel import I18n
 from aiogram.utils.i18n.context import ctx_i18n
-
-T = TypeVar("T")
+from aiogram.utils.i18n.core import I18n
 
 
 class I18nMiddleware(BaseMiddleware, ABC):
     def __init__(
         self,
         i18n: I18n,
-        gettext_key: Optional[str] = "gettext",
+        i18n_key: Optional[str] = "i18n",
         middleware_key: str = "i18n_middleware",
     ) -> None:
         self.i18n = i18n
-        self.gettext_key = gettext_key
+        self.i18n_key = i18n_key
         self.middleware_key = middleware_key
 
     def setup(
         self: BaseMiddleware, router: Router, exclude: Optional[Set[str]] = None
     ) -> BaseMiddleware:
         if exclude is None:
-            exclude = {"update"}
+            exclude = set()
+        exclude_events = {"update", "error", *exclude}
         for event_name, observer in router.observers.items():
-            if event_name in exclude:
+            if event_name in exclude_events:
                 continue
             observer.outer_middleware(self)
         return self
@@ -45,8 +44,8 @@ class I18nMiddleware(BaseMiddleware, ABC):
     ) -> Any:
         self.i18n.current_locale = await self.get_locale(event=event, data=data)
 
-        if self.gettext_key:
-            data[self.gettext_key] = self.i18n
+        if self.i18n_key:
+            data[self.i18n_key] = self.i18n
         if self.middleware_key:
             data[self.middleware_key] = self
         token = ctx_i18n.set(self.i18n)
@@ -64,7 +63,7 @@ class SimpleI18nMiddleware(I18nMiddleware):
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
 
-        if Locale is None:
+        if Locale is None:  # pragma: no cover
             raise RuntimeError(
                 f"{type(self).__name__} can be used only when Babel installed\n"
                 "Just install Babel (`pip install Babel`) "
@@ -72,7 +71,7 @@ class SimpleI18nMiddleware(I18nMiddleware):
             )
 
     async def get_locale(self, event: TelegramObject, data: Dict[str, Any]) -> str:
-        if Locale is None:
+        if Locale is None:  # pragma: no cover
             raise RuntimeError(
                 f"{type(self).__name__} can be used only when Babel installed\n"
                 "Just install Babel (`pip install Babel`) "

tests/conftest.py

@@ -1,3 +1,5 @@
+from pathlib import Path
+
 import pytest
 from _pytest.config import UsageError
 from aioredis.connection import parse_url as parse_redis_url
@@ -7,6 +9,8 @@ from aiogram.dispatcher.fsm.storage.memory import MemoryStorage
 from aiogram.dispatcher.fsm.storage.redis import RedisStorage
 from tests.mocked_bot import MockedBot
 
+DATA_DIR = Path(__file__).parent / "data"
+
 
 def pytest_addoption(parser):
     parser.addoption("--redis", default=None, help="run tests which require redis connection")

tests/data/locales/en/LC_MESSAGES/messages.po

@@ -0,0 +1,2 @@
+msgid "test"
+msgstr ""

tests/data/locales/messages.pot

@@ -0,0 +1,2 @@
+msgid "test"
+msgstr ""

tests/data/locales/uk/LC_MESSAGES/messages.po

@@ -0,0 +1,2 @@
+msgid "test"
+msgstr "тест"

tests/test_utils/test_i18n.py

@@ -0,0 +1,139 @@
+from typing import Any, Dict
+
+import pytest
+
+from aiogram import Dispatcher
+from aiogram.dispatcher.fsm.context import FSMContext
+from aiogram.dispatcher.fsm.storage.memory import MemoryStorage
+from aiogram.types import Update, User
+from aiogram.utils.i18n import ConstI18nMiddleware, FSMI18nMiddleware, I18n, SimpleI18nMiddleware
+from aiogram.utils.i18n.context import ctx_i18n, get_i18n, gettext, lazy_gettext
+from tests.conftest import DATA_DIR
+from tests.mocked_bot import MockedBot
+
+
+@pytest.fixture(name="i18n")
+def i18n_fixture() -> I18n:
+    return I18n(path=DATA_DIR / "locales")
+
+
+class TestI18nCore:
+    def test_init(self, i18n: I18n):
+        assert set(i18n.available_locales) == {"en", "uk"}
+
+    def test_reload(self, i18n: I18n):
+        i18n.reload()
+        assert set(i18n.available_locales) == {"en", "uk"}
+
+    def test_current_locale(self, i18n: I18n):
+        assert i18n.current_locale == "en"
+        i18n.current_locale = "uk"
+        assert i18n.current_locale == "uk"
+        assert i18n.ctx_locale.get() == "uk"
+
+    def test_get_i18n(self, i18n: I18n):
+        with pytest.raises(LookupError):
+            get_i18n()
+
+        token = ctx_i18n.set(i18n)
+        assert get_i18n() == i18n
+        ctx_i18n.reset(token)
+
+    @pytest.mark.parametrize(
+        "locale,case,result",
+        [
+            [None, dict(singular="test"), "test"],
+            [None, dict(singular="test", locale="uk"), "тест"],
+            ["en", dict(singular="test", locale="uk"), "тест"],
+            ["uk", dict(singular="test", locale="uk"), "тест"],
+            ["uk", dict(singular="test"), "тест"],
+            ["it", dict(singular="test"), "test"],
+            [None, dict(singular="test", n=2), "test"],
+            [None, dict(singular="test", n=2, locale="uk"), "тест"],
+            ["en", dict(singular="test", n=2, locale="uk"), "тест"],
+            ["uk", dict(singular="test", n=2, locale="uk"), "тест"],
+            ["uk", dict(singular="test", n=2), "тест"],
+            ["it", dict(singular="test", n=2), "test"],
+            [None, dict(singular="test", plural="test2", n=2), "test2"],
+            [None, dict(singular="test", plural="test2", n=2, locale="uk"), "test2"],
+            ["en", dict(singular="test", plural="test2", n=2, locale="uk"), "test2"],
+            ["uk", dict(singular="test", plural="test2", n=2, locale="uk"), "test2"],
+            ["uk", dict(singular="test", plural="test2", n=2), "test2"],
+            ["it", dict(singular="test", plural="test2", n=2), "test2"],
+        ],
+    )
+    def test_gettext(self, i18n: I18n, locale: str, case: Dict[str, Any], result: str):
+        if locale is not None:
+            i18n.current_locale = locale
+        token = ctx_i18n.set(i18n)
+        try:
+            assert i18n.gettext(**case) == result
+            assert str(i18n.lazy_gettext(**case)) == result
+            assert gettext(**case) == result
+            assert str(lazy_gettext(**case)) == result
+        finally:
+            ctx_i18n.reset(token)
+
+
+async def next_call(event, data):
+    assert "i18n" in data
+    assert "i18n_middleware" in data
+    return gettext("test")
+
+
+@pytest.mark.asyncio
+class TestSimpleI18nMiddleware:
+    @pytest.mark.parametrize(
+        "event_from_user,result",
+        [
+            [None, "test"],
+            [User(id=42, is_bot=False, language_code="uk", first_name="Test"), "тест"],
+            [User(id=42, is_bot=False, language_code="it", first_name="Test"), "test"],
+        ],
+    )
+    async def test_middleware(self, i18n: I18n, event_from_user, result):
+        middleware = SimpleI18nMiddleware(i18n=i18n)
+        result = await middleware(
+            next_call,
+            Update(update_id=42),
+            {"event_from_user": event_from_user},
+        )
+        assert result == result
+
+    async def test_setup(self, i18n: I18n):
+        dp = Dispatcher()
+        middleware = SimpleI18nMiddleware(i18n=i18n)
+        middleware.setup(router=dp)
+
+        assert middleware not in dp.update.outer_middlewares
+        assert middleware in dp.message.outer_middlewares
+
+
+@pytest.mark.asyncio
+class TestConstI18nMiddleware:
+    async def test_middleware(self, i18n: I18n):
+        middleware = ConstI18nMiddleware(i18n=i18n, locale="uk")
+        result = await middleware(
+            next_call,
+            Update(update_id=42),
+            {"event_from_user": User(id=42, is_bot=False, language_code="it", first_name="Test")},
+        )
+        assert result == "тест"
+
+
+@pytest.mark.asyncio
+class TestFSMI18nMiddleware:
+    async def test_middleware(self, i18n: I18n, bot: MockedBot):
+        middleware = FSMI18nMiddleware(i18n=i18n)
+        storage = MemoryStorage()
+        state = FSMContext(bot=bot, storage=storage, user_id=42, chat_id=42)
+        data = {
+            "event_from_user": User(id=42, is_bot=False, language_code="it", first_name="Test"),
+            "state": state,
+        }
+        result = await middleware(next_call, Update(update_id=42), data)
+        assert result == "test"
+        await middleware.set_locale(state, "uk")
+        assert i18n.current_locale == "uk"
+        result = await middleware(next_call, Update(update_id=42), data)
+        assert result == "тест"