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

Cover FSM key builder (business_connection_id

Browse source
This commit is contained in:
JRoot Junior 2024-04-22 13:26:19 +03:00
parent 924fb755cb
commit 15d01c2244
No known key found for this signature in database
GPG key ID: 738964250D5FF6E2
2 changed files with 54 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

11
aiogram/fsm/storage/redis.py
View file

@ -44,7 +44,10 @@ class DefaultKeyBuilder(KeyBuilder):
Simple Redis key builder with default prefix.
Generates a colon-joined string with prefix, chat_id, user_id,
optional bot_id and optional destiny.
optional bot_id, business_connection_id and destiny.
Format:
:code:`<prefix>:<bot_id?>:<business_connection_id?>:<chat_id>:<user_id>:<destiny?>:<field>`
"""
def __init__(
@ -61,7 +64,7 @@ class DefaultKeyBuilder(KeyBuilder):
:param separator: separator
:param with_bot_id: include Bot id in the key
:param with_business_connection_id: include business connection id
:param with_destiny: include destiny key
:param with_destiny: include a destiny key
"""
self.prefix = prefix
self.separator = separator
@ -73,12 +76,12 @@ class DefaultKeyBuilder(KeyBuilder):
parts = [self.prefix]
if self.with_bot_id:
parts.append(str(key.bot_id))
if self.with_business_connection_id and key.business_connection_id:
parts.append(str(key.business_connection_id))
parts.append(str(key.chat_id))
if key.thread_id:
parts.append(str(key.thread_id))
parts.append(str(key.user_id))
if self.with_business_connection_id and key.business_connection_id:
parts.append(str(key.business_connection_id))
if self.with_destiny:
parts.append(key.destiny)
elif key.destiny != DEFAULT_DESTINY:

58
tests/test_fsm/storage/test_redis.py
View file

@ -12,26 +12,62 @@ BOT_ID = 42
CHAT_ID = -1
USER_ID = 2
THREAD_ID = 3
BUSINESS_CONNECTION_ID = "4"
FIELD = "data"
class TestRedisDefaultKeyBuilder:
@pytest.mark.parametrize(
"with_bot_id,with_destiny,result",
"key_builder,result",
[
[False, False, f"{PREFIX}:{CHAT_ID}:{USER_ID}:{FIELD}"],
[True, False, f"{PREFIX}:{BOT_ID}:{CHAT_ID}:{USER_ID}:{FIELD}"],
[True, True, f"{PREFIX}:{BOT_ID}:{CHAT_ID}:{USER_ID}:{DEFAULT_DESTINY}:{FIELD}"],
[False, True, f"{PREFIX}:{CHAT_ID}:{USER_ID}:{DEFAULT_DESTINY}:{FIELD}"],
[
DefaultKeyBuilder(
prefix=PREFIX,
with_bot_id=True,
with_destiny=True,
with_business_connection_id=True,
),
f"{PREFIX}:{BOT_ID}:{BUSINESS_CONNECTION_ID}:{CHAT_ID}:{USER_ID}:{DEFAULT_DESTINY}:{FIELD}",
],
[
DefaultKeyBuilder(prefix=PREFIX, with_bot_id=True, with_destiny=True),
f"{PREFIX}:{BOT_ID}:{CHAT_ID}:{USER_ID}:{DEFAULT_DESTINY}:{FIELD}",
],
[
DefaultKeyBuilder(
prefix=PREFIX, with_bot_id=True, with_business_connection_id=True
),
f"{PREFIX}:{BOT_ID}:{BUSINESS_CONNECTION_ID}:{CHAT_ID}:{USER_ID}:{FIELD}",
],
[
DefaultKeyBuilder(prefix=PREFIX, with_bot_id=True),
f"{PREFIX}:{BOT_ID}:{CHAT_ID}:{USER_ID}:{FIELD}",
],
[
DefaultKeyBuilder(
prefix=PREFIX, with_destiny=True, with_business_connection_id=True
),
f"{PREFIX}:{BUSINESS_CONNECTION_ID}:{CHAT_ID}:{USER_ID}:{DEFAULT_DESTINY}:{FIELD}",
],
[
DefaultKeyBuilder(prefix=PREFIX, with_destiny=True),
f"{PREFIX}:{CHAT_ID}:{USER_ID}:{DEFAULT_DESTINY}:{FIELD}",
],
[
DefaultKeyBuilder(prefix=PREFIX, with_business_connection_id=True),
f"{PREFIX}:{BUSINESS_CONNECTION_ID}:{CHAT_ID}:{USER_ID}:{FIELD}",
],
[DefaultKeyBuilder(prefix=PREFIX), f"{PREFIX}:{CHAT_ID}:{USER_ID}:{FIELD}"],
],
)
async def test_generate_key(self, with_bot_id: bool, with_destiny: bool, result: str):
key_builder = DefaultKeyBuilder(
prefix=PREFIX,
with_bot_id=with_bot_id,
with_destiny=with_destiny,
async def test_generate_key(self, key_builder: DefaultKeyBuilder, result: str):
key = StorageKey(
chat_id=CHAT_ID,
user_id=USER_ID,
bot_id=BOT_ID,
business_connection_id=BUSINESS_CONNECTION_ID,
destiny=DEFAULT_DESTINY,
)
key = StorageKey(chat_id=CHAT_ID, user_id=USER_ID, bot_id=BOT_ID, destiny=DEFAULT_DESTINY)
assert key_builder.build(key, FIELD) == result
async def test_destiny_check(self):