149 lines
7.7 KiB
Python
149 lines
7.7 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
from sqlalchemy import func, select
|
|
|
|
from src.core.database import async_session_maker
|
|
from src.core.models import Account, Participation, User, Winner
|
|
from src.core.services import LotteryService
|
|
from src.handlers.account_services import AccountParticipationService
|
|
from src.handlers.cashier_handlers import CashierStates
|
|
from src.utils.account_input import parse_cashier_records
|
|
from test_operator_scenarios import ACCOUNT, ADMIN, CLIENT, assert_clean, output, seed, send
|
|
from test_text_inputs import context
|
|
|
|
NEW_ACCOUNT = "01-02-03-04-05-06-07"
|
|
|
|
|
|
@pytest.mark.parametrize("text", [ACCOUNT, NEW_ACCOUNT, f"0007 {NEW_ACCOUNT}\n{ACCOUNT}"])
|
|
async def test_cashier_requires_card_for_every_row_without_partial_writes(text, caplog):
|
|
lottery_id, _ = await seed(participants=False)
|
|
await send(900002, callback=f"cash_add:{lottery_id}")
|
|
calls = await send(900002, text=text)
|
|
assert "обязательна клубная карта" in output(calls)
|
|
assert await context(900002).get_state() == CashierStates.accounts.state
|
|
async with async_session_maker() as session:
|
|
assert await session.scalar(select(func.count(Participation.id))) == 0
|
|
assert await session.scalar(select(func.count(Account.id))) == 2
|
|
await send(900002, text=f"0007 {NEW_ACCOUNT}")
|
|
assert await context(900002).get_state() is None
|
|
async with async_session_maker() as session:
|
|
assert await session.scalar(select(func.count(Participation.id))) == 1
|
|
assert_clean(caplog)
|
|
|
|
|
|
async def test_strict_service_rejects_missing_card_before_any_writes():
|
|
lottery_id, _ = await seed(participants=False)
|
|
async with async_session_maker() as session:
|
|
result = await AccountParticipationService.add_account_to_lottery(session, lottery_id, ACCOUNT, require_card=True)
|
|
assert not result["success"]
|
|
result = await AccountParticipationService.add_accounts_bulk(
|
|
session, lottery_id, [f"0007 {NEW_ACCOUNT}", ACCOUNT], require_card=True
|
|
)
|
|
assert result["added"] == result["linked"] == 0 and result["errors"]
|
|
assert await session.scalar(select(func.count(Participation.id))) == 0
|
|
assert await session.scalar(select(func.count(Account.id))) == 2
|
|
|
|
|
|
@pytest.mark.parametrize("existing_account", [False, True])
|
|
async def test_cashier_restores_anonymous_participation_and_client_interaction(existing_account, caplog):
|
|
lottery_id, owner_id = await seed(participants=False)
|
|
account_number = ACCOUNT if existing_account else NEW_ACCOUNT
|
|
async with async_session_maker() as session:
|
|
ticket = Participation(lottery_id=lottery_id, account_number=account_number)
|
|
session.add(ticket)
|
|
await session.commit()
|
|
ticket_id, created_at = ticket.id, ticket.created_at
|
|
await send(900002, callback=f"cash_add:{lottery_id}")
|
|
calls = await send(900002, text=f"0007 {account_number}")
|
|
assert "Привязано существующих участий: 1" in output(calls)
|
|
assert account_number in output(await send(CLIENT, text="/my_accounts"))
|
|
async with async_session_maker() as session:
|
|
ticket = await session.get(Participation, ticket_id)
|
|
account = await session.scalar(select(Account).where(Account.account_number == account_number))
|
|
assert ticket.user_id == account.owner_id == owner_id and ticket.account_id == account.id
|
|
assert ticket.created_at.replace(tzinfo=None) == created_at.replace(tzinfo=None)
|
|
assert await session.scalar(select(func.count(Participation.id))) == 1
|
|
await send(900002, callback=f"cash_add:{lottery_id}")
|
|
calls = await send(900002, text=f"0007 {account_number}")
|
|
assert "уже участвует" in output(calls)
|
|
notifications = await send(ADMIN, callback=f"admin_conduct_confirmed_{lottery_id}")
|
|
assert any(call.__api_method__ == "sendMessage" and getattr(call, "chat_id", None) == CLIENT
|
|
and "Вы выиграли" in (getattr(call, "text", "") or "") for call in notifications)
|
|
await send(900002, text=f"/verify_winner TEST1234 {lottery_id}")
|
|
async with async_session_maker() as session:
|
|
winner = await session.scalar(select(Winner))
|
|
assert winner.is_claimed and winner.user_id == owner_id and winner.account_number == account_number
|
|
assert_clean(caplog)
|
|
|
|
|
|
async def test_card_cannot_replace_existing_participant_owner(caplog):
|
|
lottery_id, _ = await seed(participants=False)
|
|
async with async_session_maker() as session:
|
|
other = User(telegram_id=960005, club_card_number="0008")
|
|
session.add(other)
|
|
await session.flush()
|
|
other_id = other.id
|
|
session.add(Participation(lottery_id=lottery_id, account_number=NEW_ACCOUNT, user_id=other_id))
|
|
await session.commit()
|
|
await send(900002, callback=f"cash_add:{lottery_id}")
|
|
calls = await send(900002, text=f"0007 {NEW_ACCOUNT}")
|
|
assert "другим владельцем" in output(calls)
|
|
async with async_session_maker() as session:
|
|
assert await session.scalar(select(Participation.user_id)) == other_id
|
|
assert not await session.scalar(select(Account.id).where(Account.account_number == NEW_ACCOUNT))
|
|
assert_clean(caplog)
|
|
|
|
|
|
async def test_two_cashiers_repair_once_and_closed_draw_cannot_be_relinked():
|
|
lottery_id, owner_id = await seed(participants=False)
|
|
async with async_session_maker() as session:
|
|
session.add(Participation(lottery_id=lottery_id, account_number=NEW_ACCOUNT))
|
|
await session.commit()
|
|
async def link():
|
|
async with async_session_maker() as session:
|
|
return await AccountParticipationService.add_account_to_lottery(
|
|
session, lottery_id, f"0007 {NEW_ACCOUNT}", require_card=True
|
|
)
|
|
results = await asyncio.gather(link(), link())
|
|
assert sum(bool(item.get("linked")) for item in results) == 1
|
|
async with async_session_maker() as session:
|
|
assert await session.scalar(select(func.count(Participation.id))) == 1
|
|
assert await session.scalar(select(Participation.user_id)) == owner_id
|
|
await LotteryService.conduct_draw(session, lottery_id)
|
|
assert not (await link())["success"]
|
|
|
|
|
|
async def test_shared_add_account_callback_cannot_drop_card_from_state(caplog):
|
|
lottery_id, _ = await seed(participants=False)
|
|
await context(900002).update_data(accounts=[{"account_number": ACCOUNT}])
|
|
calls = await send(900002, callback=f"add_to_lottery_{lottery_id}")
|
|
assert "обязательна клубная карта" in output(calls)
|
|
async with async_session_maker() as session:
|
|
assert await session.scalar(select(func.count(Participation.id))) == 0
|
|
await context(900002).update_data(accounts=[{"club_card": "0007", "account_number": ACCOUNT}])
|
|
await send(900002, callback=f"add_to_lottery_{lottery_id}")
|
|
async with async_session_maker() as session:
|
|
ticket = await session.scalar(select(Participation))
|
|
assert ticket.user_id and ticket.account_id
|
|
assert_clean(caplog)
|
|
|
|
|
|
@pytest.mark.parametrize("card", ["0007", "01234567890123", "0" * 49 + "1"])
|
|
async def test_cashier_preserves_card_lengths_and_leading_zeroes(card, caplog):
|
|
lottery_id, owner_id = await seed(participants=False)
|
|
async with async_session_maker() as session:
|
|
user = await session.get(User, owner_id)
|
|
user.club_card_number = card
|
|
await session.commit()
|
|
await send(900002, callback=f"cash_add:{lottery_id}")
|
|
await send(900002, text=f"{card}\t{NEW_ACCOUNT}")
|
|
async with async_session_maker() as session:
|
|
assert await session.scalar(select(Participation.user_id)) == owner_id
|
|
assert_clean(caplog)
|
|
|
|
|
|
def test_cashier_parser_keeps_table_and_viposnova_card_pairs():
|
|
for text in (f"0007\t{ACCOUNT}", f"{ACCOUNT}\n0007", f"Viposnova\n{ACCOUNT}\n0007"):
|
|
assert parse_cashier_records(text) == ([f"0007 {ACCOUNT}"], [])
|