131 lines
3.7 KiB
Python
131 lines
3.7 KiB
Python
import os
|
|
import sys
|
|
|
|
import pytest
|
|
from aiogram.exceptions import TelegramForbiddenError
|
|
from sqlalchemy import delete, select
|
|
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
import src.core.models # noqa: F401 - register SQLAlchemy models before init_db
|
|
from src.core.activity_service import ActivityService
|
|
from src.core.database import async_session_maker, init_db
|
|
from src.core.delivery import DeliveryService
|
|
from src.core.models import BlockedUser
|
|
|
|
|
|
class FakeTelegramMessage:
|
|
message_id = 123
|
|
|
|
|
|
async def _cleanup_blocked_user(telegram_id: int):
|
|
async with async_session_maker() as session:
|
|
await session.execute(
|
|
delete(BlockedUser).where(BlockedUser.telegram_id == telegram_id)
|
|
)
|
|
await session.commit()
|
|
|
|
|
|
async def _get_blocked_user(telegram_id: int):
|
|
async with async_session_maker() as session:
|
|
result = await session.execute(
|
|
select(BlockedUser).where(BlockedUser.telegram_id == telegram_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delivery_guard_skips_known_blocked_user():
|
|
await init_db()
|
|
telegram_id = 880000001
|
|
await _cleanup_blocked_user(telegram_id)
|
|
|
|
async with async_session_maker() as session:
|
|
session.add(
|
|
BlockedUser(
|
|
telegram_id=telegram_id,
|
|
error_type="blocked_bot",
|
|
error_message="Forbidden: bot was blocked by the user",
|
|
is_active=True,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
called = False
|
|
|
|
async def send_call():
|
|
nonlocal called
|
|
called = True
|
|
return FakeTelegramMessage()
|
|
|
|
result = await DeliveryService().send_with_guard(telegram_id, send_call)
|
|
|
|
assert result.success is False
|
|
assert result.skipped is True
|
|
assert result.status == "blocked_bot"
|
|
assert called is False
|
|
|
|
await _cleanup_blocked_user(telegram_id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delivery_guard_marks_user_blocked_on_forbidden():
|
|
await init_db()
|
|
telegram_id = 880000002
|
|
await _cleanup_blocked_user(telegram_id)
|
|
|
|
async def send_call():
|
|
raise TelegramForbiddenError(
|
|
method=None,
|
|
message="Forbidden: bot was blocked by the user",
|
|
)
|
|
|
|
result = await DeliveryService().send_with_guard(telegram_id, send_call)
|
|
blocked_user = await _get_blocked_user(telegram_id)
|
|
|
|
assert result.success is False
|
|
assert result.status == "blocked_bot"
|
|
assert blocked_user is not None
|
|
assert blocked_user.is_active is True
|
|
assert blocked_user.error_type == "blocked_bot"
|
|
|
|
await _cleanup_blocked_user(telegram_id)
|
|
|
|
|
|
def test_delivery_guard_does_not_classify_parse_errors_as_blocked():
|
|
service = DeliveryService()
|
|
|
|
assert service.classify_bad_request(Exception("Bad Request: can't parse entities")) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_activity_reactivates_delivery_after_user_interaction():
|
|
await init_db()
|
|
telegram_id = 880000003
|
|
await _cleanup_blocked_user(telegram_id)
|
|
|
|
async with async_session_maker() as session:
|
|
session.add(
|
|
BlockedUser(
|
|
telegram_id=telegram_id,
|
|
error_type="blocked_bot",
|
|
error_message="Forbidden: bot was blocked by the user",
|
|
is_active=True,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
async with async_session_maker() as session:
|
|
reactivated = await ActivityService.update_activity_and_reactivate(
|
|
session,
|
|
telegram_id,
|
|
)
|
|
|
|
blocked_user = await _get_blocked_user(telegram_id)
|
|
|
|
assert reactivated is True
|
|
assert blocked_user is not None
|
|
assert blocked_user.is_active is False
|
|
|
|
await _cleanup_blocked_user(telegram_id)
|