forked from HSE_team/BetterCallPraskovia
194 lines
6.8 KiB
Python
194 lines
6.8 KiB
Python
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from datetime import datetime, timedelta
|
|
from tg_bot.domain.services.user_service import UserService
|
|
from tg_bot.infrastructure.database.models import UserModel
|
|
|
|
|
|
class TestUserService:
|
|
|
|
@pytest.fixture
|
|
def mock_session(self):
|
|
session = AsyncMock()
|
|
session.execute = AsyncMock()
|
|
session.add = MagicMock()
|
|
session.commit = AsyncMock()
|
|
session.rollback = AsyncMock()
|
|
return session
|
|
|
|
@pytest.fixture
|
|
def user_service(self, mock_session):
|
|
return UserService(mock_session)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_by_telegram_id_success(self, user_service, mock_session):
|
|
telegram_id = 123456789
|
|
mock_user = UserModel(
|
|
telegram_id=str(telegram_id),
|
|
username="test_user",
|
|
first_name="Test",
|
|
last_name="User"
|
|
)
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none = MagicMock(return_value=mock_user)
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
result = await user_service.get_user_by_telegram_id(telegram_id)
|
|
|
|
assert result == mock_user
|
|
assert result.telegram_id == str(telegram_id)
|
|
mock_session.execute.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_by_telegram_id_not_found(self, user_service, mock_session):
|
|
telegram_id = 999999999
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none = MagicMock(return_value=None)
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
result = await user_service.get_user_by_telegram_id(telegram_id)
|
|
|
|
assert result is None
|
|
mock_session.execute.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_or_create_user_new_user(self, user_service, mock_session):
|
|
telegram_id = 123456789
|
|
username = "new_user"
|
|
first_name = "New"
|
|
last_name = "User"
|
|
|
|
mock_result_not_found = MagicMock()
|
|
mock_result_not_found.scalar_one_or_none = MagicMock(return_value=None)
|
|
|
|
mock_result_found = MagicMock()
|
|
created_user = UserModel(
|
|
telegram_id=str(telegram_id),
|
|
username=username,
|
|
first_name=first_name,
|
|
last_name=last_name
|
|
)
|
|
mock_result_found.scalar_one_or_none = MagicMock(return_value=created_user)
|
|
|
|
mock_session.execute.side_effect = [mock_result_not_found, mock_result_found]
|
|
|
|
result = await user_service.get_or_create_user(telegram_id, username, first_name, last_name)
|
|
|
|
assert result is not None
|
|
assert result.telegram_id == str(telegram_id)
|
|
assert result.username == username
|
|
mock_session.add.assert_called_once()
|
|
mock_session.commit.assert_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_or_create_user_existing_user(self, user_service, mock_session):
|
|
telegram_id = 123456789
|
|
existing_user = UserModel(
|
|
telegram_id=str(telegram_id),
|
|
username="old_username",
|
|
first_name="Old",
|
|
last_name="Name"
|
|
)
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none = MagicMock(return_value=existing_user)
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
result = await user_service.get_or_create_user(
|
|
telegram_id, "new_username", "New", "Name"
|
|
)
|
|
|
|
assert result == existing_user
|
|
assert result.username == "new_username"
|
|
assert result.first_name == "New"
|
|
assert result.last_name == "Name"
|
|
mock_session.commit.assert_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_user_questions_success(self, user_service, mock_session):
|
|
telegram_id = 123456789
|
|
user = UserModel(
|
|
telegram_id=str(telegram_id),
|
|
questions_used=5
|
|
)
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none = MagicMock(return_value=user)
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
result = await user_service.update_user_questions(telegram_id)
|
|
|
|
assert result is True
|
|
assert user.questions_used == 6
|
|
mock_session.commit.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_user_questions_user_not_found(self, user_service, mock_session):
|
|
telegram_id = 999999999
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none = MagicMock(return_value=None)
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
result = await user_service.update_user_questions(telegram_id)
|
|
|
|
assert result is False
|
|
mock_session.commit.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_activate_premium_success(self, user_service, mock_session):
|
|
telegram_id = 123456789
|
|
user = UserModel(
|
|
telegram_id=str(telegram_id),
|
|
is_premium=False,
|
|
premium_until=None
|
|
)
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none = MagicMock(return_value=user)
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
result = await user_service.activate_premium(telegram_id)
|
|
|
|
assert result is True
|
|
assert user.is_premium is True
|
|
assert user.premium_until is not None
|
|
assert user.premium_until > datetime.now()
|
|
mock_session.commit.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_activate_premium_extend_existing(self, user_service, mock_session):
|
|
telegram_id = 123456789
|
|
existing_premium_until = datetime.now() + timedelta(days=10)
|
|
user = UserModel(
|
|
telegram_id=str(telegram_id),
|
|
is_premium=True,
|
|
premium_until=existing_premium_until
|
|
)
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none = MagicMock(return_value=user)
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
result = await user_service.activate_premium(telegram_id)
|
|
|
|
assert result is True
|
|
assert user.is_premium is True
|
|
assert user.premium_until > existing_premium_until
|
|
mock_session.commit.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_activate_premium_user_not_found(self, user_service, mock_session):
|
|
telegram_id = 999999999
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none = MagicMock(return_value=None)
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
result = await user_service.activate_premium(telegram_id)
|
|
|
|
assert result is False
|
|
mock_session.commit.assert_not_called()
|