mirror of
https://github.com/aiogram/aiogram.git
synced 2026-04-08 16:37:47 +00:00
7c6d78cf30
Bump .pre-commit-config.yaml Bump `mongo` feature deps Bump `proxy` feature dep Bump `test` feature deps Bump `dev` feature deps Set `aiohttp` max version `<3.14` Fix `test_isolation.py` tests Fix `test_storages.py` tests Add Py version limit `<3.15` (breaking changes possible) Add new `uvloop` starter to `Dispatcher.run_polling` Remove old `uvloop` `set_event_loop_policy` Remove `pytest-lazy-fixture`
61 lines
2.1 KiB
Python
61 lines
2.1 KiB
Python
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from aiogram.fsm.storage.base import BaseEventIsolation, StorageKey
|
|
from aiogram.fsm.storage.redis import RedisEventIsolation, RedisStorage
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"isolation",
|
|
["redis_isolation", "lock_isolation", "disabled_isolation"],
|
|
indirect=True,
|
|
)
|
|
class TestIsolations:
|
|
async def test_lock(
|
|
self,
|
|
isolation: BaseEventIsolation,
|
|
storage_key: StorageKey,
|
|
):
|
|
async with isolation.lock(key=storage_key):
|
|
assert True, "Are you kidding me?"
|
|
|
|
|
|
class TestRedisEventIsolation:
|
|
def test_create_isolation(self):
|
|
fake_redis = object()
|
|
storage = RedisStorage(redis=fake_redis)
|
|
isolation = storage.create_isolation()
|
|
assert isinstance(isolation, RedisEventIsolation)
|
|
assert isolation.redis is fake_redis
|
|
assert isolation.key_builder is storage.key_builder
|
|
|
|
def test_init_without_key_builder(self):
|
|
redis = AsyncMock()
|
|
isolation = RedisEventIsolation(redis=redis)
|
|
assert isolation.redis is redis
|
|
|
|
assert isolation.key_builder is not None
|
|
|
|
def test_init_with_key_builder(self):
|
|
redis = AsyncMock()
|
|
key_builder = AsyncMock()
|
|
isolation = RedisEventIsolation(redis=redis, key_builder=key_builder)
|
|
assert isolation.redis is redis
|
|
assert isolation.key_builder is key_builder
|
|
|
|
def test_create_from_url(self):
|
|
with patch("redis.asyncio.connection.ConnectionPool.from_url") as pool:
|
|
isolation = RedisEventIsolation.from_url("redis://localhost:6379/0")
|
|
assert isinstance(isolation, RedisEventIsolation)
|
|
assert isolation.redis is not None
|
|
assert isolation.key_builder is not None
|
|
|
|
pool.assert_called_once_with("redis://localhost:6379/0")
|
|
|
|
async def test_close(self):
|
|
isolation = RedisEventIsolation(redis=AsyncMock())
|
|
await isolation.close()
|
|
|
|
# close is not called because connection should be closed from the storage
|
|
# assert isolation.redis.close.called_once()
|