tests/conftest.py

"""
Pytest configuration and shared fixtures for DemoPrep E2E tests.
"""
import os
import time
import pytest
import requests
from pathlib import Path
from dotenv import load_dotenv
from playwright.sync_api import Page, BrowserContext, Browser, expect

load_dotenv(os.path.join(os.path.dirname(__file__), '..', '.env'))

BASE_URL = "https://thoughtspot-dp-demoprep.hf.space"
TEST_USER = os.getenv("TEST_USER")
TEST_PASSWORD = os.getenv("TEST_PASSWORD")

if not TEST_USER or not TEST_PASSWORD:
    raise EnvironmentError("TEST_USER and TEST_PASSWORD must be set in .env")


def _wake_space(max_wait: int = 180) -> None:
    """
    Ping HF Spaces until it responds with HTTP 200.
    HF free-tier spaces go to sleep after inactivity and take 30-60s to wake.
    Raises if the space doesn't wake within max_wait seconds.
    """
    deadline = time.time() + max_wait
    last_status = None
    while time.time() < deadline:
        try:
            r = requests.get(BASE_URL, timeout=10, allow_redirects=True)
            last_status = r.status_code
            if r.status_code == 200:
                return
        except requests.RequestException:
            pass
        time.sleep(5)
    raise RuntimeError(
        f"HF Space did not wake within {max_wait}s (last HTTP status: {last_status})"
    )


def _do_login(page: Page) -> None:
    """Perform the login flow on a page."""
    page.goto(BASE_URL, timeout=90000)
    page.wait_for_selector('input[placeholder="Type here..."]', timeout=90000)
    page.fill('input[type=text]', TEST_USER)
    page.fill('input[type=password]', TEST_PASSWORD)
    page.click('button:has-text("Login")')
    page.wait_for_selector('.gradio-container', timeout=90000)
    page.wait_for_timeout(3000)  # allow Gradio JS to settle


def login(page: Page) -> None:
    """Log in to the app with test credentials (for tests that need a fresh login)."""
    _do_login(page)


@pytest.fixture(scope="session", autouse=True)
def wake_space(request):
    """Wake the HF Space before any tests run. Blocks until it's responsive."""
    e2e_files = {
        "e2e_chat.py",
        "e2e_settings.py",
        "e2e_smoke.py",
        "e2e_z_auth.py",
        "test_mcp_liveboard.py",
    }
    collected_files = {Path(str(item.fspath)).name for item in request.session.items}
    if not collected_files.intersection(e2e_files):
        return
    print("\n⏳ Waiting for HF Space to be ready...", flush=True)
    _wake_space(max_wait=180)
    print("✅ HF Space is up.", flush=True)


@pytest.fixture(scope="session")
def auth_context(browser: Browser, tmp_path_factory):
    """
    Session-scoped authenticated browser context.

    Logs in once, saves storage state, and yields a persistent context.
    All logged_in_page fixtures share this context, avoiding repeated expensive
    logins and connection setup against HF Spaces.
    """
    state_file = str(tmp_path_factory.mktemp("auth") / "state.json")

    # Log in on a temp page to capture auth state
    setup_ctx = browser.new_context(viewport={"width": 1280, "height": 900})
    setup_page = setup_ctx.new_page()
    _do_login(setup_page)
    setup_ctx.storage_state(path=state_file)
    setup_page.close()
    setup_ctx.close()

    # Create the long-lived context with saved auth state
    ctx = browser.new_context(
        storage_state=state_file,
        viewport={"width": 1280, "height": 900},
    )
    yield ctx
    ctx.close()


@pytest.fixture()
def logged_in_page(auth_context: BrowserContext):
    """
    Function-scoped: opens a fresh page in the shared authenticated context.
    Each test gets an isolated page (own URL, own Gradio session) without
    the overhead of a full login or new browser context.
    """
    page = auth_context.new_page()
    for attempt in range(2):
        try:
            page.goto(BASE_URL, timeout=120000)
            # Wait for tabs — the minimum signal that Gradio has initialized.
            # Individual tests wait for their own specific elements.
            page.wait_for_selector('button[role=tab]', timeout=120000)
            page.wait_for_timeout(3000)
            break
        except Exception:
            if attempt == 1:
                raise
            page.wait_for_timeout(10000)
    yield page
    page.close()


def get_chat_input(page: Page):
    """Return the main chat message input field."""
    return page.locator('input[placeholder*="Amazon.com"]')


# Alias for backwards compatibility
def get_chat_textarea(page: Page):
    return get_chat_input(page)


def wait_for_response(page: Page, timeout_ms: int = 90000) -> None:
    """Wait until the assistant has finished responding."""
    send_btn = page.locator('button:has-text("Send")')
    expect(send_btn).to_be_enabled(timeout=timeout_ms)
    page.wait_for_timeout(500)