GmRelayBot/tests/e2e/dashboard/test_dashboard_auth_and_sessions.py

"""
Playwright E2E tests for the GmRelay Blazor/Web dashboard.

These tests use a mocked Telegram Mini App initData payload so they can run
locally against a real GmRelay.Web instance without talking to Telegram.

Prerequisites:
  pip install playwright
  playwright install chromium

Environment:
  GMRELAY_E2E_BASE_URL  - Web dashboard URL (default: http://localhost:8080)
  GMRELAY_E2E_BOT_TOKEN - Bot token matching the target Web instance
  GMRELAY_E2E_TELEGRAM_ID - Telegram ID to use for the mocked user
  GMRELAY_E2E_DATABASE_URL - PostgreSQL DSN for DB assertions (optional)
"""

from __future__ import annotations

import os
import time
import urllib.parse
import uuid
from typing import Optional

from playwright.sync_api import Page, expect, sync_playwright

import sys
import pathlib

sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))

from helpers.telegram_init_data import build_mini_app_init_data


def _base_url() -> str:
    return os.environ.get("GMRELAY_E2E_BASE_URL", "http://localhost:8080").rstrip("/")


def _bot_token() -> str:
    token = os.environ.get("GMRELAY_E2E_BOT_TOKEN")
    if not token:
        raise RuntimeError(
            "GMRELAY_E2E_BOT_TOKEN is required. It must match the bot token used by the target Web instance."
        )
    return token


def _telegram_id() -> int:
    return int(os.environ.get("GMRELAY_E2E_TELEGRAM_ID", "9000000001"))


def _database_url() -> Optional[str]:
    return os.environ.get("GMRELAY_E2E_DATABASE_URL")


def _seed_group_in_database(
    telegram_id: int,
    group_id: str,
    group_name: str,
    session_id: str,
    session_title: str,
    batch_id: Optional[str] = None,
) -> None:
    """
    Insert minimal test data via the shared Npgsql connection so the dashboard
    has something to render. This is intentionally thin: full state setup is
    performed by the console runner in issue #151.
    """
    import psycopg2

    dsn = _database_url()
    if not dsn:
        return  # rely on a pre-seeded dev database

    platform = "Telegram"
    external_user_id = str(telegram_id)
    now = time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime())
    scheduled_at = time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(time.time() + 86400))
    batch_id = batch_id or str(uuid.uuid4())

    with psycopg2.connect(dsn) as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO gmrelay.players (platform, external_user_id, display_name, created_at)
                VALUES (%s, %s, %s, %s)
                ON CONFLICT (platform, external_user_id)
                WHERE platform IS NOT NULL AND external_user_id IS NOT NULL
                DO UPDATE SET display_name = EXCLUDED.display_name
                RETURNING id
                """,
                (platform, external_user_id, "E2E Test GM", now),
            )
            player_id = cur.fetchone()[0]

            cur.execute(
                """
                INSERT INTO gmrelay.game_groups (id, platform, external_group_id, telegram_chat_id, name, created_at)
                VALUES (%s, %s, %s, %s, %s, %s)
                ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name
                RETURNING id
                """,
                (group_id, platform, "-1001234567890", -1001234567890, group_name, now),
            )

            cur.execute(
                """
                INSERT INTO gmrelay.group_managers (group_id, player_id, role, created_at)
                VALUES (%s, %s, %s, %s)
                ON CONFLICT (group_id, player_id) DO UPDATE SET role = EXCLUDED.role
                """,
                (group_id, player_id, "Owner", now),
            )

            cur.execute(
                """
                INSERT INTO gmrelay.sessions (
                    id, group_id, batch_id, title, scheduled_at, join_link, status, publication_mode,
                    notification_mode, max_players, format, location_address, system, created_at
                )
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                ON CONFLICT (id) DO UPDATE SET
                    title = EXCLUDED.title,
                    scheduled_at = EXCLUDED.scheduled_at,
                    join_link = EXCLUDED.join_link,
                    max_players = EXCLUDED.max_players,
                    publication_mode = EXCLUDED.publication_mode,
                    format = EXCLUDED.format,
                    location_address = EXCLUDED.location_address,
                    system = EXCLUDED.system
                """,
                (
                    session_id,
                    group_id,
                    batch_id,
                    session_title,
                    scheduled_at,
                    "https://example.com/join",
                    "Planned",
                    "None",
                    "GroupAndDirect",
                    5,
                    "Online",
                    "Discord",
                    "D&D 5e",
                    now,
                ),
            )

            cur.execute(
                """
                INSERT INTO gmrelay.session_participants (session_id, player_id, registration_status, rsvp_status, is_gm, created_at)
                VALUES (%s, %s, %s, %s, %s, %s)
                ON CONFLICT (session_id, player_id) DO NOTHING
                """,
                (session_id, player_id, "Active", "Pending", True, now),
            )

            conn.commit()


def _get_session_from_db(session_id: str):
    import psycopg2

    dsn = _database_url()
    if not dsn:
        return None

    with psycopg2.connect(dsn) as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT title, join_link, max_players, publication_mode, format, location_address, system, status
                FROM gmrelay.sessions
                WHERE id = %s
                """,
                (session_id,),
            )
            row = cur.fetchone()
            if row is None:
                return None
            return {
                "title": row[0],
                "join_link": row[1],
                "max_players": row[2],
                "publication_mode": row[3],
                "format": row[4],
                "location_address": row[5],
                "system": row[6],
                "status": row[7],
            }


def _assert_session_deleted(session_id: str) -> None:
    import psycopg2

    dsn = _database_url()
    if not dsn:
        return

    with psycopg2.connect(dsn) as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT COUNT(*) FROM gmrelay.sessions WHERE id = %s", (session_id,))
            assert cur.fetchone()[0] == 0, f"Session {session_id} was not deleted"


def _delete_test_data(group_id: str, session_id: str) -> None:
    import psycopg2

    dsn = _database_url()
    if not dsn:
        return

    telegram_id = str(_telegram_id())
    with psycopg2.connect(dsn) as conn:
        with conn.cursor() as cur:
            cur.execute("DELETE FROM gmrelay.session_participants WHERE session_id = %s", (session_id,))
            cur.execute("DELETE FROM gmrelay.sessions WHERE id = %s", (session_id,))
            cur.execute("DELETE FROM gmrelay.group_managers WHERE group_id = %s", (group_id,))
            cur.execute("DELETE FROM gmrelay.game_groups WHERE id = %s", (group_id,))

            # Only remove the player if the test created it. If the Telegram ID
            # belongs to a real production user (e.g. the Toutsu account), leave
            # the row intact so we do not break that account.
            cur.execute(
                "SELECT id FROM gmrelay.players WHERE platform = 'Telegram' AND external_user_id = %s",
                (telegram_id,),
            )
            if cur.fetchone() is not None:
                # Determine whether this player has any other groups or sessions.
                cur.execute(
                    "SELECT COUNT(*) FROM gmrelay.group_managers m "
                    "JOIN gmrelay.players p ON p.id = m.player_id "
                    "WHERE p.platform = 'Telegram' AND p.external_user_id = %s",
                    (telegram_id,),
                )
                manager_count = cur.fetchone()[0]
                cur.execute(
                    "SELECT COUNT(*) FROM gmrelay.session_participants sp "
                    "JOIN gmrelay.players p ON p.id = sp.player_id "
                    "WHERE p.platform = 'Telegram' AND p.external_user_id = %s",
                    (telegram_id,),
                )
                participant_count = cur.fetchone()[0]
                if manager_count == 0 and participant_count == 0:
                    cur.execute(
                        "DELETE FROM gmrelay.players WHERE platform = 'Telegram' AND external_user_id = %s",
                        (telegram_id,),
                    )
            conn.commit()


def _authenticate_page(page: Page, base_url: str, bot_token: str, telegram_id: int) -> None:
    """
    Authenticate by calling /auth/telegram-webapp with a valid initData payload.
    The response sets the auth cookie; subsequent page navigations are authenticated.
    """
    init_data = build_mini_app_init_data(
        bot_token=bot_token,
        telegram_id=telegram_id,
        first_name="E2E",
        last_name="Test",
        username="e2e_test",
    )

    response = page.request.post(
        f"{base_url}/auth/telegram-webapp",
        data={"initData": init_data.init_data_raw},
    )
    expect(response).to_be_ok()
    page.goto(base_url)


def _wait_for_blazor(page: Page) -> None:
    """Wait until Blazor has finished the initial render."""
    page.wait_for_selector("text=Добро пожаловать", timeout=15000)


def test_dashboard_authenticates_and_shows_groups() -> None:
    base_url = _base_url()
    bot_token = _bot_token()
    telegram_id = _telegram_id()

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        context = browser.new_context()
        page = context.new_page()

        _authenticate_page(page, base_url, bot_token, telegram_id)
        _wait_for_blazor(page)

        heading = page.locator("h2")
        expect(heading).to_contain_text("Добро пожаловать")

        browser.close()


def test_dashboard_session_edit_flow() -> None:
    base_url = _base_url()
    bot_token = _bot_token()
    telegram_id = _telegram_id()

    group_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa1"
    session_id = "bbbbbbbb-bbbb-bbbb-bbbb-aaaaaaaaaaa1"
    batch_id = "cccccccc-cccc-cccc-cccc-aaaaaaaaaaa1"
    original_title = "E2E Original Title"
    updated_title = "E2E Updated Title"
    updated_join_link = "https://example.com/updated-join"

    try:
        _seed_group_in_database(
            telegram_id=telegram_id,
            group_id=group_id,
            group_name="E2E Test Group",
            session_id=session_id,
            session_title=original_title,
            batch_id=batch_id,
        )

        with sync_playwright() as p:
            browser = p.chromium.launch(headless=True)
            context = browser.new_context()
            page = context.new_page()

            _authenticate_page(page, base_url, bot_token, telegram_id)
            _wait_for_blazor(page)

            page.goto(f"{base_url}/group/{group_id}")
            page.wait_for_selector(f"text={original_title}", timeout=15000)

            page.locator("a:has-text('Изменить')").first.click()
            page.wait_for_selector("text=Редактирование сессии", timeout=15000)

            title_input = page.get_by_label("Название игры")
            title_input.fill(updated_title)

            join_input = page.get_by_label("Ссылка для подключения")
            join_input.fill(updated_join_link)

            max_players_input = page.get_by_label("Лимит мест")
            max_players_input.fill("3")

            page.get_by_label("Режим публикации").select_option("Catalog")

            save_button = page.locator("button:has-text('Сохранить изменения')").first
            save_button.click()

            page.wait_for_selector(f"text={updated_title}", timeout=15000)
            expect(page.locator(f"text={updated_title}").first).to_be_visible()

            db_session = _get_session_from_db(session_id)
            if db_session is not None:
                assert db_session["title"] == updated_title
                assert db_session["join_link"] == updated_join_link
                assert db_session["max_players"] == 3
                assert db_session["publication_mode"] == "Catalog"

            browser.close()
    finally:
        _delete_test_data(group_id, session_id)


def test_dashboard_session_delete_flow() -> None:
    base_url = _base_url()
    bot_token = _bot_token()
    telegram_id = _telegram_id()

    group_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa2"
    session_id = "bbbbbbbb-bbbb-bbbb-bbbb-aaaaaaaaaaa2"
    batch_id = "cccccccc-cccc-cccc-cccc-aaaaaaaaaaa2"
    session_title = "E2E Delete Target"

    try:
        _seed_group_in_database(
            telegram_id=telegram_id,
            group_id=group_id,
            group_name="E2E Delete Group",
            session_id=session_id,
            session_title=session_title,
            batch_id=batch_id,
        )

        with sync_playwright() as p:
            browser = p.chromium.launch(headless=True)
            context = browser.new_context()
            page = context.new_page()

            _authenticate_page(page, base_url, bot_token, telegram_id)
            _wait_for_blazor(page)

            page.goto(f"{base_url}/group/{group_id}")
            page.wait_for_selector(f"text={session_title}", timeout=15000)

            page.on("dialog", lambda dialog: dialog.accept())

            delete_button = page.locator("button:has-text('Удалить')").first
            expect(delete_button).to_be_visible()
            delete_button.click()

            page.wait_for_selector("text=Сессия удалена.", timeout=15000)
            expect(page.locator(f"text={session_title}")).to_have_count(0)

            _assert_session_deleted(session_id)

            browser.close()
    finally:
        _delete_test_data(group_id, session_id)


if __name__ == "__main__":
    test_dashboard_authenticates_and_shows_groups()
    print("PASS test_dashboard_authenticates_and_shows_groups")
    test_dashboard_session_edit_flow()
    print("PASS test_dashboard_session_edit_flow")
    test_dashboard_session_delete_flow()
    print("PASS test_dashboard_session_delete_flow")