calminer/tests/e2e/conftest.py

import os
import subprocess
import time
from typing import Dict, Generator

import pytest

# type: ignore[import]
from playwright.sync_api import Browser, Page, Playwright, sync_playwright

import httpx
from sqlalchemy.engine import make_url

# Use a different port for the test server to avoid conflicts
TEST_PORT = 8001
BASE_URL = f"http://localhost:{TEST_PORT}"


@pytest.fixture(scope="session", autouse=True)
def live_server() -> Generator[str, None, None]:
    """Launch a live test server in a separate process."""
    env = _prepare_database_environment(os.environ.copy())

    process = subprocess.Popen(
        [
            "uvicorn",
            "main:app",
            "--host",
            "127.0.0.1",
            f"--port={TEST_PORT}",
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        env=env,
    )

    deadline = time.perf_counter() + 30
    last_error: Exception | None = None
    while time.perf_counter() < deadline:
        if process.poll() is not None:
            raise RuntimeError("uvicorn server exited before becoming ready")
        try:
            response = httpx.get(BASE_URL, timeout=1.0, trust_env=False)
            if response.status_code < 500:
                break
        except Exception as exc:  # noqa: BLE001
            last_error = exc
        time.sleep(0.5)
    else:
        process.terminate()
        process.wait(timeout=5)
        raise TimeoutError(
            "Timed out waiting for uvicorn test server to start"
        ) from last_error

    try:
        yield BASE_URL
    finally:
        if process.poll() is None:
            process.terminate()
            try:
                process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                process.kill()
                process.wait(timeout=5)


@pytest.fixture(scope="session", autouse=True)
def seed_default_currencies(live_server: str) -> None:
    """Ensure a baseline set of currencies exists for UI flows."""

    seeds = [
        {"code": "EUR", "name": "Euro", "symbol": "EUR", "is_active": True},
        {
            "code": "CLP",
            "name": "Chilean Peso",
            "symbol": "CLP$",
            "is_active": True,
        },
    ]

    with httpx.Client(
        base_url=live_server, timeout=5.0, trust_env=False
    ) as client:
        try:
            response = client.get("/api/currencies/?include_inactive=true")
            response.raise_for_status()
            existing_codes = {
                str(item.get("code"))
                for item in response.json()
                if isinstance(item, dict) and item.get("code")
            }
        except httpx.HTTPError as exc:  # noqa: BLE001
            raise RuntimeError("Failed to read existing currencies") from exc

        for payload in seeds:
            if payload["code"] in existing_codes:
                continue
            try:
                create_response = client.post("/api/currencies/", json=payload)
            except httpx.HTTPError as exc:  # noqa: BLE001
                raise RuntimeError("Failed to seed currencies") from exc

            if create_response.status_code == 409:
                continue
            create_response.raise_for_status()


@pytest.fixture(scope="session")
def playwright_instance() -> Generator[Playwright, None, None]:
    """Provide a Playwright instance for the test session."""
    with sync_playwright() as p:
        yield p


@pytest.fixture(scope="session")
def browser(
    playwright_instance: Playwright,
) -> Generator[Browser, None, None]:
    """Provide a browser instance for the test session."""
    browser = playwright_instance.chromium.launch()
    yield browser
    browser.close()


@pytest.fixture()
def page(browser: Browser, live_server: str) -> Generator[Page, None, None]:
    """Provide a new page for each test."""
    page = browser.new_page(base_url=live_server)
    page.goto("/")
    page.wait_for_load_state("networkidle")
    yield page
    page.close()


def _prepare_database_environment(env: Dict[str, str]) -> Dict[str, str]:
    """Ensure granular database env vars are available for the app under test."""

    required = (
        "DATABASE_HOST",
        "DATABASE_USER",
        "DATABASE_NAME",
        "DATABASE_PASSWORD",
    )
    if all(env.get(key) for key in required):
        return env

    legacy_url = env.get("DATABASE_URL")
    if not legacy_url:
        return env

    url = make_url(legacy_url)
    env.setdefault("DATABASE_DRIVER", url.drivername)
    if url.host:
        env.setdefault("DATABASE_HOST", url.host)
    if url.port:
        env.setdefault("DATABASE_PORT", str(url.port))
    if url.username:
        env.setdefault("DATABASE_USER", url.username)
    if url.password:
        env.setdefault("DATABASE_PASSWORD", url.password)
    if url.database:
        env.setdefault("DATABASE_NAME", url.database)

    query_options = dict(url.query) if url.query else {}
    options = query_options.get("options")
    if isinstance(options, str) and "search_path=" in options:
        env.setdefault("DATABASE_SCHEMA", options.split("search_path=")[-1])

    return env