159 lines
4.9 KiB
Python
159 lines
4.9 KiB
Python
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
|