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

Merge branch 'aiogram:dev-3.x' into add-feature-fsm

Browse source
This commit is contained in:
Вадим Христенко 2025-04-14 22:17:08 +03:00 • committed by GitHub
commit 6fc59321f0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
234 changed files with 9777 additions and 393 deletions
Download patch file Download diff file Expand all files Collapse all files

13
tests/test_api/test_methods/test_convert_gift_to_stars.py Normal file
View file

@ -0,0 +1,13 @@
from aiogram.methods import ConvertGiftToStars
from tests.mocked_bot import MockedBot
class TestConvertGiftToStars:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(ConvertGiftToStars, ok=True, result=True)
response: bool = await bot.convert_gift_to_stars(
business_connection_id="test_connection_id", owned_gift_id="test_gift_id"
)
request = bot.get_request()
assert response == prepare_result.result

13
tests/test_api/test_methods/test_delete_business_messages.py Normal file
View file

@ -0,0 +1,13 @@
from aiogram.methods import DeleteBusinessMessages
from tests.mocked_bot import MockedBot
class TestDeleteBusinessMessages:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(DeleteBusinessMessages, ok=True, result=True)
response: bool = await bot.delete_business_messages(
business_connection_id="test_connection_id", message_ids=[1, 2, 3]
)
request = bot.get_request()
assert response == prepare_result.result

13
tests/test_api/test_methods/test_delete_story.py Normal file
View file

@ -0,0 +1,13 @@
from aiogram.methods import DeleteStory
from tests.mocked_bot import MockedBot
class TestDeleteStory:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(DeleteStory, ok=True, result=True)
response: bool = await bot.delete_story(
business_connection_id="test_connection_id", story_id=42
)
request = bot.get_request()
assert response == prepare_result.result

26
tests/test_api/test_methods/test_edit_story.py Normal file
View file

@ -0,0 +1,26 @@
import datetime
from aiogram.methods import EditStory
from aiogram.types import Chat, InputStoryContentPhoto, Story
from tests.mocked_bot import MockedBot
class TestEditStory:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(
EditStory,
ok=True,
result=Story(
id=42,
chat=Chat(id=42, type="private"),
),
)
response: Story = await bot.edit_story(
business_connection_id="test_connection_id",
story_id=42,
content=InputStoryContentPhoto(type="photo", photo="test_photo"),
caption="Test caption",
)
request = bot.get_request()
assert response == prepare_result.result

41
tests/test_api/test_methods/test_get_business_account_gifts.py Normal file
View file

@ -0,0 +1,41 @@
import datetime
from aiogram.methods import GetBusinessAccountGifts
from aiogram.types import Gift, OwnedGiftRegular, OwnedGifts, Sticker
from tests.mocked_bot import MockedBot
class TestGetBusinessAccountGifts:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(
GetBusinessAccountGifts,
ok=True,
result=OwnedGifts(
total_count=1,
gifts=[
OwnedGiftRegular(
gift=Gift(
id="test_gift_id",
sticker=Sticker(
file_id="test_file_id",
file_unique_id="test_file_unique_id",
type="regular",
width=512,
height=512,
is_animated=False,
is_video=False,
),
star_count=100,
),
send_date=int(datetime.datetime.now().timestamp()),
)
],
),
)
response: OwnedGifts = await bot.get_business_account_gifts(
business_connection_id="test_connection_id",
limit=10,
)
request = bot.get_request()
assert response == prepare_result.result

21
tests/test_api/test_methods/test_get_business_account_star_balance.py Normal file
View file

@ -0,0 +1,21 @@
from aiogram.methods import GetBusinessAccountStarBalance
from aiogram.types import StarAmount
from tests.mocked_bot import MockedBot
class TestGetBusinessAccountStarBalance:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(
GetBusinessAccountStarBalance,
ok=True,
result=StarAmount(
amount=100,
nanostar_amount=500000000,
),
)
response: StarAmount = await bot.get_business_account_star_balance(
business_connection_id="test_connection_id",
)
request = bot.get_request()
assert response == prepare_result.result

8
tests/test_api/test_methods/test_get_chat.py
View file

@ -1,5 +1,5 @@
from aiogram.methods import GetChat
from aiogram.types import ChatFullInfo
from aiogram.types import AcceptedGiftTypes, ChatFullInfo
from tests.mocked_bot import MockedBot
@ -14,6 +14,12 @@ class TestGetChat:
title="chat",
accent_color_id=0,
max_reaction_count=0,
accepted_gift_types=AcceptedGiftTypes(
unlimited_gifts=True,
limited_gifts=True,
unique_gifts=True,
premium_subscription=True,
),
),
)

5
tests/test_api/test_methods/test_get_star_transactions.py
View file

@ -1,5 +1,6 @@
from datetime import datetime
from aiogram.enums import TransactionPartnerUserTransactionTypeEnum
from aiogram.methods import GetStarTransactions
from aiogram.types import (
File,
@ -26,6 +27,7 @@ class TestGetStarTransactions:
date=datetime.now(),
source=TransactionPartnerUser(
user=user,
transaction_type=TransactionPartnerUserTransactionTypeEnum.GIFT_PURCHASE,
),
),
StarTransaction(
@ -35,12 +37,13 @@ class TestGetStarTransactions:
date=datetime.now(),
receiver=TransactionPartnerUser(
user=user,
transaction_type=TransactionPartnerUserTransactionTypeEnum.GIFT_PURCHASE,
),
),
]
),
)
response: File = await bot.get_star_transactions(limit=10, offset=0)
response: StarTransactions = await bot.get_star_transactions(limit=10, offset=0)
request = bot.get_request()
assert response == prepare_result.result

16
tests/test_api/test_methods/test_gift_premium_subscription.py Normal file
View file

@ -0,0 +1,16 @@
from aiogram.methods import GiftPremiumSubscription
from tests.mocked_bot import MockedBot
class TestGiftPremiumSubscription:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(GiftPremiumSubscription, ok=True, result=True)
response: bool = await bot.gift_premium_subscription(
user_id=123456789,
month_count=3,
star_count=1000,
text="Enjoy your premium subscription!",
)
request = bot.get_request()
assert response == prepare_result.result

26
tests/test_api/test_methods/test_post_story.py Normal file
View file

@ -0,0 +1,26 @@
import datetime
from aiogram.methods import PostStory
from aiogram.types import Chat, InputStoryContentPhoto, Story
from tests.mocked_bot import MockedBot
class TestPostStory:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(
PostStory,
ok=True,
result=Story(
id=42,
chat=Chat(id=42, type="private"),
),
)
response: Story = await bot.post_story(
business_connection_id="test_connection_id",
content=InputStoryContentPhoto(type="photo", photo="test_photo"),
active_period=6 * 3600, # 6 hours
caption="Test story caption",
)
request = bot.get_request()
assert response == prepare_result.result

13
tests/test_api/test_methods/test_read_business_message.py Normal file
View file

@ -0,0 +1,13 @@
from aiogram.methods import ReadBusinessMessage
from tests.mocked_bot import MockedBot
class TestReadBusinessMessage:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(ReadBusinessMessage, ok=True, result=True)
response: bool = await bot.read_business_message(
business_connection_id="test_connection_id", chat_id=123456789, message_id=42
)
request = bot.get_request()
assert response == prepare_result.result

15
tests/test_api/test_methods/test_remove_business_account_profile_photo.py Normal file
View file

@ -0,0 +1,15 @@
from aiogram.methods import RemoveBusinessAccountProfilePhoto
from tests.mocked_bot import MockedBot
class TestRemoveBusinessAccountProfilePhoto:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(
RemoveBusinessAccountProfilePhoto, ok=True, result=True
)
response: bool = await bot.remove_business_account_profile_photo(
business_connection_id="test_connection_id", is_public=True
)
request = bot.get_request()
assert response == prepare_result.result

14
tests/test_api/test_methods/test_set_business_account_bio.py Normal file
View file

@ -0,0 +1,14 @@
from aiogram.methods import SetBusinessAccountBio
from tests.mocked_bot import MockedBot
class TestSetBusinessAccountBio:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(SetBusinessAccountBio, ok=True, result=True)
response: bool = await bot.set_business_account_bio(
business_connection_id="test_connection_id",
bio="This is a test bio for the business account",
)
request = bot.get_request()
assert response == prepare_result.result

21
tests/test_api/test_methods/test_set_business_account_gift_settings.py Normal file
View file

@ -0,0 +1,21 @@
from aiogram.methods import SetBusinessAccountGiftSettings
from aiogram.types import AcceptedGiftTypes
from tests.mocked_bot import MockedBot
class TestSetBusinessAccountGiftSettings:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(SetBusinessAccountGiftSettings, ok=True, result=True)
response: bool = await bot.set_business_account_gift_settings(
business_connection_id="test_connection_id",
show_gift_button=True,
accepted_gift_types=AcceptedGiftTypes(
unlimited_gifts=True,
limited_gifts=True,
unique_gifts=True,
premium_subscription=True,
),
)
request = bot.get_request()
assert response == prepare_result.result

15
tests/test_api/test_methods/test_set_business_account_name.py Normal file
View file

@ -0,0 +1,15 @@
from aiogram.methods import SetBusinessAccountName
from tests.mocked_bot import MockedBot
class TestSetBusinessAccountName:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(SetBusinessAccountName, ok=True, result=True)
response: bool = await bot.set_business_account_name(
business_connection_id="test_connection_id",
first_name="Test Business",
last_name="Account Name",
)
request = bot.get_request()
assert response == prepare_result.result

16
tests/test_api/test_methods/test_set_business_account_profile_photo.py Normal file
View file

@ -0,0 +1,16 @@
from aiogram.methods import SetBusinessAccountProfilePhoto
from aiogram.types import InputProfilePhotoStatic
from tests.mocked_bot import MockedBot
class TestSetBusinessAccountProfilePhoto:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(SetBusinessAccountProfilePhoto, ok=True, result=True)
response: bool = await bot.set_business_account_profile_photo(
business_connection_id="test_connection_id",
photo=InputProfilePhotoStatic(photo="test_photo_file_id"),
is_public=True,
)
request = bot.get_request()
assert response == prepare_result.result

13
tests/test_api/test_methods/test_set_business_account_username.py Normal file
View file

@ -0,0 +1,13 @@
from aiogram.methods import SetBusinessAccountUsername
from tests.mocked_bot import MockedBot
class TestSetBusinessAccountUsername:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(SetBusinessAccountUsername, ok=True, result=True)
response: bool = await bot.set_business_account_username(
business_connection_id="test_connection_id", username="test_business_username"
)
request = bot.get_request()
assert response == prepare_result.result

13
tests/test_api/test_methods/test_transfer_business_account_stars.py Normal file
View file

@ -0,0 +1,13 @@
from aiogram.methods import TransferBusinessAccountStars
from tests.mocked_bot import MockedBot
class TestTransferBusinessAccountStars:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(TransferBusinessAccountStars, ok=True, result=True)
response: bool = await bot.transfer_business_account_stars(
business_connection_id="test_connection_id", star_count=100
)
request = bot.get_request()
assert response == prepare_result.result

16
tests/test_api/test_methods/test_transfer_gift.py Normal file
View file

@ -0,0 +1,16 @@
from aiogram.methods import TransferGift
from tests.mocked_bot import MockedBot
class TestTransferGift:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(TransferGift, ok=True, result=True)
response: bool = await bot.transfer_gift(
business_connection_id="test_connection_id",
owned_gift_id="test_gift_id",
new_owner_chat_id=123456789,
star_count=50,
)
request = bot.get_request()
assert response == prepare_result.result

16
tests/test_api/test_methods/test_upgrade_gift.py Normal file
View file

@ -0,0 +1,16 @@
from aiogram.methods import UpgradeGift
from tests.mocked_bot import MockedBot
class TestUpgradeGift:
async def test_bot_method(self, bot: MockedBot):
prepare_result = bot.add_result_for(UpgradeGift, ok=True, result=True)
response: bool = await bot.upgrade_gift(
business_connection_id="test_connection_id",
owned_gift_id="test_gift_id",
keep_original_details=True,
star_count=100,
)
request = bot.get_request()
assert response == prepare_result.result

101
tests/test_api/test_types/test_message.py
View file

@ -58,6 +58,8 @@ from aiogram.types import (
Game,
GeneralForumTopicHidden,
GeneralForumTopicUnhidden,
Gift,
GiftInfo,
Giveaway,
GiveawayCompleted,
GiveawayCreated,
@ -71,6 +73,7 @@ from aiogram.types import (
MessageEntity,
PaidMediaInfo,
PaidMediaPhoto,
PaidMessagePriceChanged,
PassportData,
PhotoSize,
Poll,
@ -82,6 +85,12 @@ from aiogram.types import (
Sticker,
Story,
SuccessfulPayment,
UniqueGift,
UniqueGiftBackdrop,
UniqueGiftBackdropColors,
UniqueGiftInfo,
UniqueGiftModel,
UniqueGiftSymbol,
User,
UserShared,
UsersShared,
@ -606,6 +615,92 @@ TEST_MESSAGE_UNKNOWN = Message(
chat=Chat(id=42, type="private"),
from_user=User(id=42, is_bot=False, first_name="Test"),
)
TEST_MESSAGE_GIFT = Message(
message_id=42,
date=datetime.datetime.now(),
chat=Chat(id=42, type="private"),
from_user=User(id=42, is_bot=False, first_name="Test"),
gift=GiftInfo(
gift=Gift(
id="test_gift_id",
sticker=Sticker(
file_id="test_file_id",
file_unique_id="test_file_unique_id",
type="regular",
width=512,
height=512,
is_animated=False,
is_video=False,
),
star_count=100,
),
owned_gift_id="test_owned_gift_id",
convert_star_count=50,
prepaid_upgrade_star_count=25,
can_be_upgraded=True,
text="Test gift message",
is_private=False,
),
)
TEST_MESSAGE_UNIQUE_GIFT = Message(
message_id=42,
date=datetime.datetime.now(),
chat=Chat(id=42, type="private"),
from_user=User(id=42, is_bot=False, first_name="Test"),
unique_gift=UniqueGiftInfo(
gift=UniqueGift(
base_name="test_gift",
name="test_unique_gift",
number=1,
model=UniqueGiftModel(
name="test_model",
sticker=Sticker(
file_id="test_file_id",
file_unique_id="test_file_unique_id",
type="regular",
width=512,
height=512,
is_animated=False,
is_video=False,
),
rarity_per_mille=100,
),
symbol=UniqueGiftSymbol(
name="test_symbol",
sticker=Sticker(
file_id="test_file_id",
file_unique_id="test_file_unique_id",
type="regular",
width=512,
height=512,
is_animated=False,
is_video=False,
),
rarity_per_mille=100,
),
backdrop=UniqueGiftBackdrop(
name="test_backdrop",
colors=UniqueGiftBackdropColors(
center_color=0xFFFFFF,
edge_color=0x000000,
symbol_color=0xFF0000,
text_color=0x0000FF,
),
rarity_per_mille=100,
),
),
origin="upgrade",
),
)
TEST_MESSAGE_PAID_MESSAGE_PRICE_CHANGED = Message(
message_id=42,
date=datetime.datetime.now(),
chat=Chat(id=42, type="private"),
from_user=User(id=42, is_bot=False, first_name="Test"),
paid_message_price_changed=PaidMessagePriceChanged(
paid_message_star_count=100,
),
)
MESSAGES_AND_CONTENT_TYPES = [
[TEST_MESSAGE_TEXT, ContentType.TEXT],
@ -670,6 +765,9 @@ MESSAGES_AND_CONTENT_TYPES = [
[TEST_MESSAGE_BOOST_ADDED, ContentType.BOOST_ADDED],
[TEST_CHAT_BACKGROUND_SET, ContentType.CHAT_BACKGROUND_SET],
[TEST_REFUND_PAYMENT, ContentType.REFUNDED_PAYMENT],
[TEST_MESSAGE_GIFT, ContentType.GIFT],
[TEST_MESSAGE_UNIQUE_GIFT, ContentType.UNIQUE_GIFT],
[TEST_MESSAGE_PAID_MESSAGE_PRICE_CHANGED, ContentType.PAID_MESSAGE_PRICE_CHANGED],
[TEST_MESSAGE_UNKNOWN, ContentType.UNKNOWN],
]
@ -731,6 +829,9 @@ MESSAGES_AND_COPY_METHODS = [
[TEST_MESSAGE_BOOST_ADDED, None],
[TEST_CHAT_BACKGROUND_SET, None],
[TEST_REFUND_PAYMENT, None],
[TEST_MESSAGE_GIFT, None],
[TEST_MESSAGE_UNIQUE_GIFT, None],
[TEST_MESSAGE_PAID_MESSAGE_PRICE_CHANGED, None],
[TEST_MESSAGE_UNKNOWN, None],
]

160
tests/test_dispatcher/test_dispatcher.py
View file

@ -5,7 +5,8 @@ import time
import warnings
from asyncio import Event
from collections import Counter
from typing import Any
from contextlib import suppress
from typing import Any, Optional
from unittest.mock import AsyncMock, patch
import pytest
@ -793,6 +794,163 @@ class TestDispatcher:
else:
mocked_process_update.assert_awaited()
@pytest.mark.parametrize(
"handle_as_tasks,tasks_concurrency_limit,should_create_semaphore",
[
(True, 10, True),
(True, None, False),
(False, 10, False),
(False, None, False),
],
)
async def test_polling_with_semaphore(
self,
bot: MockedBot,
handle_as_tasks: bool,
tasks_concurrency_limit: int,
should_create_semaphore: bool,
):
"""Test that semaphore is created only when handle_as_tasks=True and tasks_concurrency_limit is not None"""
dispatcher = Dispatcher()
async def _mock_updates(*_):
yield Update(update_id=42)
with (
patch(
"aiogram.dispatcher.dispatcher.Dispatcher._process_update", new_callable=AsyncMock
) as mocked_process_update,
patch(
"aiogram.dispatcher.dispatcher.Dispatcher._listen_updates"
) as patched_listen_updates,
patch("asyncio.Semaphore") as mocked_semaphore,
):
patched_listen_updates.return_value = _mock_updates()
# Set up the mock semaphore
if should_create_semaphore:
mock_semaphore_instance = AsyncMock()
mock_semaphore_instance.acquire.return_value = asyncio.Future()
mock_semaphore_instance.acquire.return_value.set_result(None)
mocked_semaphore.return_value = mock_semaphore_instance
await dispatcher._polling(
bot=bot,
handle_as_tasks=handle_as_tasks,
tasks_concurrency_limit=tasks_concurrency_limit,
)
if should_create_semaphore:
mocked_semaphore.assert_called_once_with(tasks_concurrency_limit)
else:
mocked_semaphore.assert_not_called()
async def test_process_with_semaphore(self):
"""Test that _process_with_semaphore correctly processes updates and releases the semaphore"""
dispatcher = Dispatcher()
# Create a real coroutine for handle_update
async def mock_handle_update():
return "test result"
# Create a mock for the semaphore
semaphore = AsyncMock()
semaphore.release = AsyncMock()
# Call the _process_with_semaphore method
await dispatcher._process_with_semaphore(mock_handle_update(), semaphore)
# Verify that semaphore.release was called, which indicates that the coroutine was awaited
semaphore.release.assert_called_once()
async def test_process_with_semaphore_exception(self):
"""Test that _process_with_semaphore releases the semaphore even if an exception occurs"""
dispatcher = Dispatcher()
# Create a real coroutine that raises an exception
async def mock_handle_update_with_exception():
raise Exception("Test exception")
# Create a mock for the semaphore
semaphore = AsyncMock()
semaphore.release = AsyncMock()
# Call the _process_with_semaphore method and expect an exception
with pytest.raises(Exception, match="Test exception"):
await dispatcher._process_with_semaphore(
mock_handle_update_with_exception(), semaphore
)
# Verify that semaphore.release was called even though an exception occurred
semaphore.release.assert_called_once()
async def test_concurrent_updates_limit(self, bot: MockedBot):
"""Test that concurrent updates are limited when using the semaphore"""
dispatcher = Dispatcher()
tasks_concurrency_limit = 2
# Create a real semaphore for this test
semaphore = asyncio.Semaphore(tasks_concurrency_limit)
# Create a list to track when updates are processed
processed_updates = []
async def mock_process_update(*args, **kwargs):
# Record that an update is being processed
update_id = len(processed_updates)
processed_updates.append(f"start_{update_id}")
# Simulate some processing time
await asyncio.sleep(0.1)
processed_updates.append(f"end_{update_id}")
return True
# Create mock updates
async def _mock_updates(*_):
for i in range(5): # Send 5 updates
yield Update(update_id=i)
with (
patch(
"aiogram.dispatcher.dispatcher.Dispatcher._process_update",
side_effect=mock_process_update,
) as mocked_process_update,
patch(
"aiogram.dispatcher.dispatcher.Dispatcher._listen_updates"
) as patched_listen_updates,
patch(
"asyncio.Semaphore",
return_value=semaphore,
),
):
patched_listen_updates.return_value = _mock_updates()
# Start polling with concurrent_updates_limit
polling_task = asyncio.create_task(
dispatcher._polling(
bot=bot, handle_as_tasks=True, tasks_concurrency_limit=tasks_concurrency_limit
)
)
# Wait longer for all updates to be processed
await asyncio.sleep(0.6)
# Cancel the polling task
polling_task.cancel()
with suppress(asyncio.CancelledError):
await polling_task
# Check that at most concurrent_updates_limit updates were being processed at the same time
# This is a bit tricky to test precisely, but we can check that we have the expected number
# of start/end pairs in the processed_updates list
starts = [item for item in processed_updates if item.startswith("start_")]
ends = [item for item in processed_updates if item.startswith("end_")]
# We should have an equal number of starts and ends
assert len(starts) == len(ends)
# The semaphore should have been acquired and released the same number of times
assert semaphore._value == tasks_concurrency_limit
async def test_exception_handler_catch_exceptions(self, bot: MockedBot):
dp = Dispatcher()
router = Router()