tgit/tests/access/test_service.py

from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import select

from app.access.service import AccessService, RedeemStatus, TgUser, hash_token
from app.config import AccessSettings
from app.db.models import Draft, FsmState, Invite, Role, User
from app.db.session import SessionMaker

ADMIN = 1
NEWBIE = TgUser(tg_id=500, full_name="Новичок", username="newbie")


class Clock:
    def __init__(self) -> None:
        self.now = datetime(2026, 10, 15, 12, 0, tzinfo=UTC)

    def __call__(self) -> datetime:
        return self.now


@pytest.fixture
def clock() -> Clock:
    return Clock()


@pytest.fixture
async def svc(sessions: SessionMaker, clock: Clock) -> AccessService:
    service = AccessService(sessions, AccessSettings(admin_ids=frozenset({ADMIN}), invite_ttl_hours=24), clock=clock)
    await service.bootstrap_admins()
    return service


async def test_bootstrap_is_idempotent_and_restores_admin(svc: AccessService, sessions: SessionMaker) -> None:
    async with sessions.begin() as s:
        admin = await s.get(User, ADMIN)
        assert admin is not None
        admin.is_active, admin.role = False, Role.MEMBER
    await svc.bootstrap_admins()
    await svc.bootstrap_admins()

    member = await svc.get_member(ADMIN)
    assert member is not None and member.role is Role.ADMIN


async def test_invite_stores_only_hash(svc: AccessService, sessions: SessionMaker, clock: Clock) -> None:
    invite = await svc.create_invite(ADMIN)

    assert len(invite.token) == 22 and invite.token.replace("-", "").replace("_", "").isalnum()
    assert invite.expires_at == clock.now + timedelta(hours=24)
    async with sessions() as s:
        stored = await s.scalar(select(Invite))
        assert stored is not None and stored.token_hash == hash_token(invite.token) != invite.token


async def test_redeem_once_then_rejected_for_others(svc: AccessService, sessions: SessionMaker) -> None:
    invite = await svc.create_invite(ADMIN)

    assert await svc.redeem(invite.token, NEWBIE) is RedeemStatus.OK
    assert await svc.redeem(invite.token, NEWBIE) is RedeemStatus.ALREADY_MEMBER
    assert await svc.redeem(invite.token, TgUser(tg_id=501, full_name="x")) is RedeemStatus.REJECTED

    member = await svc.get_member(NEWBIE.tg_id)
    assert member is not None and member.role is Role.MEMBER and member.invited_by == ADMIN
    async with sessions() as s:
        stored = await s.scalar(select(Invite))
        assert stored is not None and stored.used_by == NEWBIE.tg_id
        assert await s.get(User, 501) is None  # the loser's user row was rolled back


async def test_expired_invite_rejected(svc: AccessService, clock: Clock) -> None:
    invite = await svc.create_invite(ADMIN)
    clock.now += timedelta(hours=24, seconds=1)

    assert await svc.redeem(invite.token, NEWBIE) is RedeemStatus.REJECTED
    assert await svc.get_member(NEWBIE.tg_id) is None


async def test_brute_force_is_throttled_even_with_a_valid_token(svc: AccessService, clock: Clock) -> None:
    for i in range(5):
        assert await svc.redeem(f"guess-{i}", NEWBIE) is RedeemStatus.REJECTED
    invite = await svc.create_invite(ADMIN)

    assert await svc.redeem(invite.token, NEWBIE) is RedeemStatus.THROTTLED
    clock.now += timedelta(hours=1, seconds=1)
    assert await svc.redeem(invite.token, NEWBIE) is RedeemStatus.OK


async def test_revoke_drops_access_draft_and_dialog(svc: AccessService, sessions: SessionMaker, clock: Clock) -> None:
    await svc.redeem((await svc.create_invite(ADMIN)).token, NEWBIE)
    async with sessions.begin() as s:
        s.add(Draft(user_id=NEWBIE.tg_id, base_sha="0" * 40, expires_at=clock.now + timedelta(hours=1)))
        s.add(FsmState(key=f"42:{NEWBIE.tg_id}:{NEWBIE.tg_id}:::default", state="x", data={}))
        s.add(FsmState(key=f"42:{ADMIN}:{ADMIN}:::default", state="y", data={}))

    assert await svc.revoke(NEWBIE.tg_id)

    assert await svc.get_member(NEWBIE.tg_id) is None
    async with sessions() as s:
        assert await s.scalar(select(Draft)) is None
        assert [r.key for r in await s.scalars(select(FsmState))] == [f"42:{ADMIN}:{ADMIN}:::default"]


async def test_admin_cannot_be_revoked_and_unknown_is_noop(svc: AccessService) -> None:
    assert not await svc.revoke(ADMIN)
    assert not await svc.revoke(999)


async def test_revoked_member_can_be_invited_again(svc: AccessService) -> None:
    await svc.redeem((await svc.create_invite(ADMIN)).token, NEWBIE)
    await svc.revoke(NEWBIE.tg_id)

    assert await svc.redeem((await svc.create_invite(ADMIN)).token, NEWBIE) is RedeemStatus.OK
    assert await svc.get_member(NEWBIE.tg_id) is not None


async def test_members_listed_admins_first(svc: AccessService) -> None:
    await svc.redeem((await svc.create_invite(ADMIN)).token, NEWBIE)

    assert [m.tg_id for m in await svc.list_members()] == [ADMIN, NEWBIE.tg_id]


async def test_touch_is_throttled_but_tracks_renames(svc: AccessService, clock: Clock) -> None:
    member = await svc.get_member(ADMIN)
    assert member is not None
    await svc.touch(member, TgUser(ADMIN, "Админ", "boss"))
    first = await svc.get_member(ADMIN)
    assert first is not None and first.full_name == "Админ" and first.last_seen_at is not None

    clock.now += timedelta(minutes=1)
    await svc.touch(first, TgUser(ADMIN, "Админ", "boss"))
    same = await svc.get_member(ADMIN)
    assert same is not None and same.last_seen_at == first.last_seen_at  # no write within 5 minutes

    await svc.touch(same, TgUser(ADMIN, "Новое имя", "boss"))
    renamed = await svc.get_member(ADMIN)
    assert renamed is not None and renamed.full_name == "Новое имя"