kerrad/telegram-helper-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge 2

Browse Source
This commit is contained in:
Andrey
2026-02-02 00:12:20 +03:00
parent 2a09971628
commit 5f66c86d99
8 changed files with 106 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
tests/test_audio_file_service.py
View File

@@ -183,42 +183,46 @@ class TestDownloadAndSaveAudio:
@pytest.mark.asyncio
async def test_download_and_save_audio_no_message(self, audio_service, mock_bot):
"""Тест скачивания когда сообщение отсутствует"""
with pytest.raises(FileOperationError) as exc_info:
await audio_service.download_and_save_audio(mock_bot, None, "test_audio")
"""Тест скачивания когда сообщение отсутствует."""
with patch('helper_bot.handlers.voice.services.asyncio.sleep', new_callable=AsyncMock):
with pytest.raises(FileOperationError) as exc_info:
await audio_service.download_and_save_audio(mock_bot, None, "test_audio")
assert "Сообщение или голосовое сообщение не найдено" in str(exc_info.value)
@pytest.mark.asyncio
async def test_download_and_save_audio_no_voice(self, audio_service, mock_bot):
"""Тест скачивания когда у сообщения нет voice атрибута"""
"""Тест скачивания когда у сообщения нет voice атрибута."""
message = Mock()
message.voice = None
with pytest.raises(FileOperationError) as exc_info:
await audio_service.download_and_save_audio(mock_bot, message, "test_audio")
with patch('helper_bot.handlers.voice.services.asyncio.sleep', new_callable=AsyncMock):
with pytest.raises(FileOperationError) as exc_info:
await audio_service.download_and_save_audio(mock_bot, message, "test_audio")
assert "Сообщение или голосовое сообщение не найдено" in str(exc_info.value)
@pytest.mark.asyncio
async def test_download_and_save_audio_download_failed(self, audio_service, mock_bot, mock_message, mock_file_info):
"""Тест скачивания когда загрузка не удалась"""
"""Тест скачивания когда загрузка не удалась."""
mock_bot.get_file.return_value = mock_file_info
mock_bot.download_file.return_value = None
with pytest.raises(FileOperationError) as exc_info:
await audio_service.download_and_save_audio(mock_bot, mock_message, "test_audio")
with patch('helper_bot.handlers.voice.services.asyncio.sleep', new_callable=AsyncMock):
with pytest.raises(FileOperationError) as exc_info:
await audio_service.download_and_save_audio(mock_bot, mock_message, "test_audio")
assert "Не удалось скачать файл" in str(exc_info.value)
@pytest.mark.asyncio
async def test_download_and_save_audio_exception_handling(self, audio_service, mock_bot, mock_message):
"""Тест обработки исключений при скачивании"""
"""Тест обработки исключений при скачивании."""
mock_bot.get_file.side_effect = Exception("Network error")
with pytest.raises(FileOperationError) as exc_info:
await audio_service.download_and_save_audio(mock_bot, mock_message, "test_audio")
with patch('helper_bot.handlers.voice.services.asyncio.sleep', new_callable=AsyncMock):
with pytest.raises(FileOperationError) as exc_info:
await audio_service.download_and_save_audio(mock_bot, mock_message, "test_audio")
assert "Не удалось скачать и сохранить аудио" in str(exc_info.value)

82
tests/test_rate_limiter.py
View File

@@ -57,42 +57,37 @@ class TestChatRateLimiter:
@pytest.mark.asyncio
async def test_wait_if_needed_with_wait(self):
"""Тест что ждет если нужно"""
"""Тест что ждет если нужно (sleep патчится, проверяем вызов с нужной длительностью)."""
config = RateLimitConfig(messages_per_second=0.5, burst_limit=10) # 1 сообщение в 2 секунды
limiter = ChatRateLimiter(config)
# Первый вызов не должен ждать
start_time = time.time()
await limiter.wait_if_needed()
first_call_time = time.time() - start_time
# Второй вызов должен ждать
start_time = time.time()
await limiter.wait_if_needed()
second_call_time = time.time() - start_time
assert first_call_time < 0.1
assert second_call_time >= 1.8 # Должно ждать около 2 секунд
with patch('helper_bot.utils.rate_limiter.asyncio.sleep', new_callable=AsyncMock) as mock_sleep:
await limiter.wait_if_needed()
mock_sleep.assert_not_called()
await limiter.wait_if_needed()
mock_sleep.assert_called_once()
# min_interval = 2.0, ждём ~2 сек
call_arg = mock_sleep.call_args[0][0]
assert 1.8 <= call_arg <= 2.2
@pytest.mark.asyncio
async def test_burst_limit(self):
"""Тест ограничения burst"""
"""Тест ограничения burst (sleep патчится, проверяем вызов на 3-м вызове)."""
config = RateLimitConfig(messages_per_second=10.0, burst_limit=2)
limiter = ChatRateLimiter(config)
# Первые два вызова не должны ждать
start_time = time.time()
await limiter.wait_if_needed()
await limiter.wait_if_needed()
first_two_calls_time = time.time() - start_time
# Третий вызов должен ждать
start_time = time.time()
await limiter.wait_if_needed()
third_call_time = time.time() - start_time
assert first_two_calls_time < 0.2 # Более мягкое ограничение
assert third_call_time >= 0.8 # Должно ждать около 1 секунды (с учетом погрешности)
with patch('helper_bot.utils.rate_limiter.asyncio.sleep', new_callable=AsyncMock) as mock_sleep:
await limiter.wait_if_needed()
await limiter.wait_if_needed()
mock_sleep.reset_mock()
await limiter.wait_if_needed()
# Третий вызов: сначала sleep по burst (~1.0 с), затем по min_interval (~0.1 с)
assert mock_sleep.call_count >= 1
args = [c[0][0] for c in mock_sleep.call_args_list]
burst_waits = [a for a in args if 0.8 <= a <= 1.2]
assert len(burst_waits) >= 1, f"Ожидался вызов sleep(~1.0) по burst, получены: {args}"
class TestGlobalRateLimiter:
@@ -145,33 +140,28 @@ class TestRetryHandler:
@pytest.mark.asyncio
async def test_execute_with_retry_retry_after(self):
"""Тест retry после RetryAfter ошибки"""
"""Тест retry после RetryAfter ошибки (sleep патчится, проверяем вызов)."""
from aiogram.exceptions import TelegramRetryAfter
config = RateLimitConfig(retry_after_multiplier=1.0, max_retry_delay=1.0)
handler = RetryHandler(config)
mock_func = AsyncMock()
# Создаем мок для TelegramRetryAfter
from unittest.mock import MagicMock
retry_after_error = TelegramRetryAfter(
method=MagicMock(),
message="Flood control exceeded",
retry_after=1 # 1 секунда
retry_after=1
)
mock_func.side_effect = [
retry_after_error, # Первый вызов - ошибка
"success" # Второй вызов - успех
]
start_time = time.time()
result = await handler.execute_with_retry(mock_func, 123, max_retries=1)
end_time = time.time()
mock_func.side_effect = [retry_after_error, "success"]
with patch('helper_bot.utils.rate_limiter.asyncio.sleep', new_callable=AsyncMock) as mock_sleep:
result = await handler.execute_with_retry(mock_func, 123, max_retries=1)
assert result == "success"
assert mock_func.call_count == 2
assert end_time - start_time >= 0.1 # Должно ждать
mock_sleep.assert_called_once()
assert mock_sleep.call_args[0][0] == 1.0 # retry_after
class TestTelegramRateLimiter:

34
tests/test_voice_bot_architecture.py
View File

@@ -123,36 +123,32 @@ class TestVoiceBotService:
@pytest.mark.asyncio
async def test_send_welcome_messages_success(self, voice_service, mock_bot_db, mock_settings):
"""Тест успешной отправки приветственных сообщений"""
"""Тест успешной отправки приветственных сообщений."""
mock_message = Mock()
mock_message.from_user.id = 123
mock_message.answer = AsyncMock()
mock_message.answer.return_value = Mock()
mock_message.answer_sticker = AsyncMock()
with patch.object(voice_service, 'get_welcome_sticker') as mock_sticker:
mock_sticker.return_value = 'test_sticker.tgs'
await voice_service.send_welcome_messages(mock_message, '😊')
# Проверяем, что сообщения отправлены
assert mock_message.answer.call_count >= 1
with patch.object(voice_service, 'get_welcome_sticker', new_callable=AsyncMock, return_value='test_sticker.tgs'):
with patch('helper_bot.handlers.voice.services.asyncio.sleep', new_callable=AsyncMock):
await voice_service.send_welcome_messages(mock_message, '😊')
assert mock_message.answer.call_count >= 1
@pytest.mark.asyncio
async def test_send_welcome_messages_no_sticker(self, voice_service, mock_bot_db, mock_settings):
"""Тест отправки приветственных сообщений без стикера"""
"""Тест отправки приветственных сообщений без стикера."""
mock_message = Mock()
mock_message.from_user.id = 123
mock_message.answer = AsyncMock()
mock_message.answer.return_value = Mock()
with patch.object(voice_service, 'get_welcome_sticker') as mock_sticker:
mock_sticker.return_value = None
await voice_service.send_welcome_messages(mock_message, '😊')
# Проверяем, что сообщения отправлены
assert mock_message.answer.call_count >= 1
with patch.object(voice_service, 'get_welcome_sticker', new_callable=AsyncMock, return_value=None):
with patch('helper_bot.handlers.voice.services.asyncio.sleep', new_callable=AsyncMock):
await voice_service.send_welcome_messages(mock_message, '😊')
assert mock_message.answer.call_count >= 1
class TestVoiceHandlers:

52
tests/test_voice_services.py
View File

@@ -161,53 +161,47 @@ class TestVoiceBotService:
@pytest.mark.asyncio
async def test_send_welcome_messages_success(self, voice_service, mock_bot_db, mock_settings):
"""Тест успешной отправки приветственных сообщений"""
"""Тест успешной отправки приветственных сообщений."""
mock_message = Mock()
mock_message.from_user.id = 123
mock_message.answer = AsyncMock()
mock_message.answer.return_value = Mock()
mock_message.answer_sticker = AsyncMock()
with patch.object(voice_service, 'get_welcome_sticker') as mock_sticker:
mock_sticker.return_value = 'test_sticker.tgs'
await voice_service.send_welcome_messages(mock_message, '😊')
# Проверяем, что сообщения отправлены
assert mock_message.answer.call_count >= 1
with patch.object(voice_service, 'get_welcome_sticker', new_callable=AsyncMock, return_value='test_sticker.tgs'):
with patch('helper_bot.handlers.voice.services.asyncio.sleep', new_callable=AsyncMock):
await voice_service.send_welcome_messages(mock_message, '😊')
assert mock_message.answer.call_count >= 1
@pytest.mark.asyncio
async def test_send_welcome_messages_no_sticker(self, voice_service, mock_bot_db, mock_settings):
"""Тест отправки приветственных сообщений без стикера"""
"""Тест отправки приветственных сообщений без стикера."""
mock_message = Mock()
mock_message.from_user.id = 123
mock_message.answer = AsyncMock()
mock_message.answer.return_value = Mock()
with patch.object(voice_service, 'get_welcome_sticker') as mock_sticker:
mock_sticker.return_value = None
await voice_service.send_welcome_messages(mock_message, '😊')
# Проверяем, что сообщения отправлены
assert mock_message.answer.call_count >= 1
with patch.object(voice_service, 'get_welcome_sticker', new_callable=AsyncMock, return_value=None):
with patch('helper_bot.handlers.voice.services.asyncio.sleep', new_callable=AsyncMock):
await voice_service.send_welcome_messages(mock_message, '😊')
assert mock_message.answer.call_count >= 1
@pytest.mark.asyncio
async def test_send_welcome_messages_with_sticker(self, voice_service, mock_bot_db, mock_settings):
"""Тест отправки приветственных сообщений со стикером"""
"""Тест отправки приветственных сообщений со стикером."""
mock_message = Mock()
mock_message.from_user.id = 123
mock_message.answer = AsyncMock()
mock_message.answer.return_value = Mock()
mock_message.answer_sticker = AsyncMock()
with patch.object(voice_service, 'get_welcome_sticker') as mock_sticker:
mock_sticker.return_value = 'test_sticker.tgs'
await voice_service.send_welcome_messages(mock_message, '😊')
# Проверяем, что сообщения отправлены
assert mock_message.answer.call_count >= 1
with patch.object(voice_service, 'get_welcome_sticker', new_callable=AsyncMock, return_value='test_sticker.tgs'):
with patch('helper_bot.handlers.voice.services.asyncio.sleep', new_callable=AsyncMock):
await voice_service.send_welcome_messages(mock_message, '😊')
assert mock_message.answer.call_count >= 1
@pytest.mark.asyncio
async def test_get_welcome_sticker_with_tgs_files(self, voice_service, mock_settings):