ad/tests/conftest.py

# retoor <retoor@molodetz.nl>
import os
import shutil
import subprocess
import sys
import time
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parent.parent
TEST_DATA_DIR = REPO_ROOT / "data" / f".tmp-pytest-{os.getpid()}"
shutil.rmtree(TEST_DATA_DIR, ignore_errors=True)
TEST_DATA_DIR.mkdir(parents=True, exist_ok=True)
TEST_PORT = int(os.environ.get("MOLODETZ_TEST_PORT", "8099"))
ADMIN_USERNAME = "retoor"
ADMIN_PASSWORD = "test-admin-pass-4f9c2a"
MEMBER_USERNAME = "lezer"
MEMBER_PASSWORD = "test-member-pass-77d1"

TEST_ENV = {
    "MOLODETZ_DATA_DIR": str(TEST_DATA_DIR),
    "MOLODETZ_DATABASE_URL": f"sqlite:///{TEST_DATA_DIR / 'molodetz.db'}",
    "MOLODETZ_ADMIN_USERNAME": ADMIN_USERNAME,
    "MOLODETZ_ADMIN_EMAIL": "retoor@molodetz.nl",
    "MOLODETZ_ADMIN_PASSWORD": ADMIN_PASSWORD,
    "MOLODETZ_DISABLE_SERVICES": "1",
    "MOLODETZ_DISABLE_RATE_LIMIT": "1",
    "MOLODETZ_PORT": str(TEST_PORT),
    "MOLODETZ_INTERNAL_BASE_URL": f"http://127.0.0.1:{TEST_PORT}",
    "MOLODETZ_SITEMAP_TTL": "0",
    "MOLODETZ_LANDING_TTL": "0",
    "MOLODETZ_UNREAD_TTL": "0",
    "MOLODETZ_STATIC_VERSION": "test",
    "SECRET_KEY": "test-secret-key-not-for-production",
}
os.environ.update(TEST_ENV)

BASE_URL = f"http://127.0.0.1:{TEST_PORT}"


def _wait_ready(process, timeout=30.0):
    import httpx

    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if process.poll() is not None:
            raise RuntimeError("test server exited during startup")
        try:
            if httpx.get(f"{BASE_URL}/health", timeout=1.0).status_code == 200:
                return
        except httpx.HTTPError:
            pass
        time.sleep(0.2)
    raise RuntimeError("test server did not become ready")


@pytest.fixture(scope="session")
def app_server():
    env = {**os.environ, **TEST_ENV}
    log = open(TEST_DATA_DIR / "server.log", "w")
    process = subprocess.Popen(
        [sys.executable, "-m", "uvicorn", "molodetz.main:app", "--host", "127.0.0.1", "--port", str(TEST_PORT)],
        cwd=REPO_ROOT,
        env=env,
        stdout=log,
        stderr=subprocess.STDOUT,
    )
    try:
        _wait_ready(process)
        _ensure_member()
        yield BASE_URL
    finally:
        process.terminate()
        try:
            process.wait(timeout=10)
        except subprocess.TimeoutExpired:
            process.kill()
        log.close()


def _ensure_member():
    from molodetz.database import create_user, get_user_by_username
    from molodetz.utils.auth import hash_password

    if get_user_by_username(MEMBER_USERNAME) is None:
        create_user(MEMBER_USERNAME, "lezer@example.invalid", hash_password(MEMBER_PASSWORD), role="Member")


def _client(base_url):
    import httpx

    return httpx.Client(base_url=base_url, timeout=15.0, follow_redirects=False)


def _login(client, username, password):
    response = client.post("/auth/login", data={"username": username, "password": password})
    assert response.status_code in (302, 303), response.text[:300]
    return client


@pytest.fixture()
def anon(app_server):
    with _client(app_server) as client:
        yield client


@pytest.fixture()
def member(app_server):
    with _client(app_server) as client:
        yield _login(client, MEMBER_USERNAME, MEMBER_PASSWORD)


@pytest.fixture()
def admin(app_server):
    with _client(app_server) as client:
        yield _login(client, ADMIN_USERNAME, ADMIN_PASSWORD)


@pytest.fixture(scope="session")
def browser(app_server):
    from playwright.sync_api import sync_playwright

    headless = os.environ.get("PLAYWRIGHT_HEADLESS", "1") == "1"
    slow_mo = int(os.environ.get("PLAYWRIGHT_SLOW_MO", "0") or 0)
    with sync_playwright() as playwright:
        browser = playwright.chromium.launch(headless=headless, slow_mo=slow_mo)
        yield browser
        browser.close()


def _record_console_error(page, message):
    if message.type == "error" and "ERR_NETWORK_CHANGED" not in message.text:
        page.console_errors.append(message.text)


@pytest.fixture()
def page(browser):
    context = browser.new_context(viewport={"width": 1280, "height": 900}, locale="nl-NL")
    page = context.new_page()
    page.set_default_timeout(60000)
    page.console_errors = []
    page.on("pageerror", lambda error: page.console_errors.append(str(error)))
    page.on("console", lambda message: _record_console_error(page, message))
    yield page
    context.close()


def goto_ready(page, url):
    from playwright.sync_api import TimeoutError as PlaywrightTimeoutError

    page.goto(url, wait_until="domcontentloaded")
    try:
        page.wait_for_selector("html[data-app=ready]")
    except PlaywrightTimeoutError:
        page.reload(wait_until="domcontentloaded")
        page.wait_for_selector("html[data-app=ready]")


def pytest_sessionfinish(session, exitstatus):
    shutil.rmtree(TEST_DATA_DIR, ignore_errors=True)


def login_admin(page, base_url):
    page.goto(base_url + "/auth/login", wait_until="domcontentloaded")
    page.fill("input[name=username]", ADMIN_USERNAME)
    page.fill("input[name=password]", ADMIN_PASSWORD)
    page.click("button[type=submit]")
    page.wait_for_url("**/admin", wait_until="domcontentloaded")


def assert_no_horizontal_overflow(page):
    width, viewport = page.evaluate("[document.documentElement.scrollWidth, window.innerWidth]")
    assert width <= viewport, f"horizontal overflow: {width} > {viewport}"