122 lines
5.4 KiB
Python
122 lines
5.4 KiB
Python
import asyncio
|
|
|
|
from aiogram import Bot
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from src.core.access import get_role
|
|
from src.core.database import async_session_maker
|
|
from src.core.models import User
|
|
from src.core.services import UserService
|
|
from src.core.staff_service import StaffError, change_role, get_member
|
|
from test_dispatcher import dispatch, event
|
|
|
|
|
|
async def create_user(telegram_id, **flags):
|
|
async with async_session_maker() as session:
|
|
user = await UserService.get_or_create_user(session, telegram_id, first_name="<Employee>")
|
|
for key, value in flags.items():
|
|
setattr(user, key, value)
|
|
await session.commit()
|
|
return user
|
|
|
|
|
|
async def test_roles_are_exclusive_and_removal_does_not_restore_old_cashier_privilege():
|
|
await create_user(810, is_cashier=True)
|
|
await change_role(900001, 810, "admin", (False, True))
|
|
member = await get_member(900001, 810)
|
|
assert member.role == "admin" and member.flags == (True, False)
|
|
await change_role(900001, 810, "cashier", member.flags)
|
|
assert await get_role(810) == "cashier"
|
|
await change_role(900001, 810, "user", (False, True))
|
|
assert await get_role(810) == "user"
|
|
|
|
|
|
@pytest.mark.parametrize("actor", [811, 900002])
|
|
async def test_non_system_staff_cannot_assign_roles_even_by_service(actor):
|
|
await create_user(actor, is_admin=actor == 811)
|
|
await create_user(812)
|
|
with pytest.raises(StaffError, match="системный администратор"):
|
|
await change_role(actor, 812, "admin", (False, False))
|
|
assert await get_role(812) == "user"
|
|
|
|
|
|
@pytest.mark.parametrize("target", [900001, 900002])
|
|
async def test_server_configured_staff_cannot_be_changed(target):
|
|
await create_user(target)
|
|
with pytest.raises(StaffError):
|
|
await change_role(900001, target, "user", (False, False))
|
|
|
|
|
|
async def test_two_system_admins_cannot_overwrite_a_changed_role(monkeypatch):
|
|
from src.core import staff_service
|
|
monkeypatch.setattr(staff_service, "ADMIN_IDS", [900001, 900003])
|
|
await create_user(813)
|
|
results = await asyncio.gather(
|
|
change_role(900001, 813, "admin", (False, False)),
|
|
change_role(900003, 813, "cashier", (False, False)), return_exceptions=True)
|
|
assert sum(isinstance(result, StaffError) for result in results) == 1
|
|
assert sum(result is None for result in results) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("flags", [{"is_admin": True}, {"is_cashier": True}])
|
|
async def test_cleanup_cannot_revoke_staff_access(flags):
|
|
user = await create_user(814, **flags)
|
|
async with async_session_maker() as session:
|
|
assert not await UserService.delete_user(session, user.id)
|
|
assert await session.scalar(select(User.id).where(User.id == user.id)) == user.id
|
|
|
|
|
|
def confirmation(calls):
|
|
return next(button.callback_data for call in calls if getattr(call, "reply_markup", None)
|
|
for row in call.reply_markup.inline_keyboard for button in row
|
|
if button.callback_data and button.callback_data.startswith("staff:confirm:"))
|
|
|
|
|
|
async def test_actual_assignment_requires_confirmation_and_old_button_cannot_change_new_target():
|
|
await create_user(815)
|
|
await create_user(816)
|
|
first = await dispatch(event(900001, text="/add_admin 815"))
|
|
old_button = confirmation(first)
|
|
assert await get_role(815) == "user"
|
|
assert any("<Employee>" in (getattr(call, "text", "") or "") for call in first)
|
|
second = await dispatch(event(900001, text="/add_cashier 816"))
|
|
new_button = confirmation(second)
|
|
await dispatch(event(900001, callback_data=old_button))
|
|
assert await get_role(815) == "user" and await get_role(816) == "user"
|
|
await dispatch(event(900001, callback_data=new_button))
|
|
assert await get_role(816) == "cashier"
|
|
await change_role(900001, 816, "user", (False, True))
|
|
await dispatch(event(900001, callback_data=new_button))
|
|
assert await get_role(816) == "user"
|
|
|
|
|
|
@pytest.mark.parametrize("actor", [817, 900002])
|
|
async def test_actual_ordinary_admin_and_cashier_cannot_forge_staff_callback(actor):
|
|
await create_user(actor, is_admin=actor == 817)
|
|
await create_user(818)
|
|
for update in (event(actor, text="/add_admin 818"),
|
|
event(actor, callback_data="staff:role:admin"),
|
|
event(actor, callback_data="admin_remove_select:900001")):
|
|
calls = await dispatch(update)
|
|
assert any("системный администратор" in (getattr(call, "text", "") or "") for call in calls)
|
|
assert await get_role(818) == "user"
|
|
|
|
|
|
async def test_revoked_cashier_cannot_continue_open_participant_dialog():
|
|
await create_user(819, is_cashier=True)
|
|
await dispatch(event(819, callback_data="cash_add:1"))
|
|
await change_role(900001, 819, "user", (False, True))
|
|
calls = await dispatch(event(819, text="11-22-33-44-55-66-77"))
|
|
assert any("Недостаточно прав" in (getattr(call, "text", "") or "") for call in calls)
|
|
|
|
|
|
async def test_guides_are_delivered_only_to_the_appropriate_roles():
|
|
from src.handlers.operator_help import DOCS, GUIDES
|
|
assert all((DOCS / spec[0]).is_file() for spec in GUIDES.values())
|
|
calls = await dispatch(event(900002, text="/cashier_help"))
|
|
assert any(call.__api_method__ == "sendDocument" for call in calls)
|
|
for update in (event(900002, text="/admin_help"), event(820, callback_data="guide:system")):
|
|
calls = await dispatch(update)
|
|
assert not any(call.__api_method__ == "sendDocument" for call in calls)
|