services/api/tests/test_admin_logs.py - master - Mihaham/Table-Time

services/api/tests/test_admin_logs.py
117 строк5 KB
MihahamYT
fix: ruff lint for CI and GitVerse Pages deploy workflow
14 июн 2026, 10:47
14 июн 2026, 10:47b35fc39
import os
import uuid
from unittest.mock import AsyncMock, patch

import pytest

os.environ["ADMIN_TELEGRAM_ID"] = "999888777"

from app.config import get_settings
from app.modules.admin.service import AdminService
from app.modules.auth.schemas import AuthContext
from app.modules.game.service import GameService
from app.modules.session.service import SessionService
from tests.conftest import MockEventPublisher, auth_header


@pytest.mark.asyncio
async def test_timeline_includes_started_and_action(session_factory):
    async with session_factory() as db_session:
        events = MockEventPublisher()
        sessions = SessionService(db_session, events)
        host = AuthContext(guest_id=f"g_adm_{uuid.uuid4().hex[:6]}", display_name="Host", is_guest=True)
        created = await sessions.create_session(host, "Host", 4)
        player = AuthContext(guest_id=f"g_adm2_{uuid.uuid4().hex[:6]}", display_name="P2", is_guest=True)
        await sessions.join_session(player, created.invite_code, "web", "P2", None)
        game_svc = GameService(db_session, events)
        started = await game_svc.start_game(created.id, "dice_board", host)
        game_id = uuid.UUID(started["game_id"])
        await game_svc.submit_action(game_id, uuid.uuid4(), "roll_dice", {}, host, "web")

        admin_svc = AdminService(db_session)
        log = await admin_svc.get_game_log(game_id)
        assert log is not None
        kinds = [e["kind"] for e in log["timeline"]]
        assert "event" in kinds
        assert "action" in kinds
        event_types = [e.get("event_type") for e in log["timeline"] if e["kind"] == "event"]
        assert "game.started" in event_types


@pytest.mark.asyncio
async def test_rejected_action_logged(session_factory):
    async with session_factory() as db_session:
        events = MockEventPublisher()
        sessions = SessionService(db_session, events)
        host = AuthContext(guest_id=f"g_rej_{uuid.uuid4().hex[:6]}", display_name="Host", is_guest=True)
        created = await sessions.create_session(host, "Host", 4)
        player = AuthContext(guest_id=f"g_rej2_{uuid.uuid4().hex[:6]}", display_name="P2", is_guest=True)
        await sessions.join_session(player, created.invite_code, "web", "P2", None)
        game_svc = GameService(db_session, events)
        started = await game_svc.start_game(created.id, "dice_board", host)
        game_id = uuid.UUID(started["game_id"])

        with pytest.raises(ValueError):
            await game_svc.submit_action(game_id, uuid.uuid4(), "roll_dice", {}, player, "telegram")

        admin_svc = AdminService(db_session)
        log = await admin_svc.get_game_log(game_id)
        rejected = [e for e in log["timeline"] if e.get("event_type") == "action.rejected"]
        assert len(rejected) == 1
        assert rejected[0]["payload"]["error_code"] == "NOT_YOUR_TURN"


def test_game_audit_requires_admin(client):
    session_id, token1, _, _ = _setup_two_player_session(client)
    started = client.post(
        f"/api/v1/sessions/{session_id}/games",
        json={"plugin_id": "dice_board"},
        headers=auth_header(token1),
    )
    game_id = started.json()["game_id"]

    r = client.get(f"/api/v1/games/{game_id}/audit")
    assert r.status_code == 403


def test_game_audit_with_admin_token(client):
    from tests.test_admin_auth import FakeRedis

    get_settings.cache_clear()
    fake_redis = FakeRedis()
    with patch("app.modules.admin.auth.get_redis", new_callable=AsyncMock) as mock_redis:
        mock_redis.return_value = fake_redis
        req = client.post("/api/v1/auth/admin/otp/request", json={"telegram_id": 999888777})
        code = req.json()["code"]
        verify = client.post(
            "/api/v1/auth/admin/otp/verify",
            json={"username": "Mihaham", "code": code},
        )
        admin_token = verify.json()["access_token"]

    session_id, token1, _, _ = _setup_two_player_session(client)
    started = client.post(
        f"/api/v1/sessions/{session_id}/games",
        json={"plugin_id": "dice_board"},
        headers=auth_header(token1),
    )
    game_id = started.json()["game_id"]
    client.post(
        f"/api/v1/games/{game_id}/actions",
        json={"action_id": str(uuid.uuid4()), "action_type": "roll_dice", "payload": {}},
        headers={**auth_header(token1), "X-Channel": "web"},
    )

    r = client.get(
        f"/api/v1/games/{game_id}/audit",
        headers={"Authorization": f"Bearer {admin_token}"},
    )
    assert r.status_code == 200
    assert len(r.json()["actions"]) == 1
    assert r.json()["actions"][0]["result_state"] is not None


def _setup_two_player_session(client):
    from tests.test_game import _setup_two_player_session as setup

    return setup(client)