devplacepy/tests/conftest.py

import os
import sys
import tempfile
import subprocess
import time
import asyncio
import threading
from pathlib import Path
import pytest

PROJECT_ROOT = Path(__file__).resolve().parent.parent
SCREENSHOT_DIR = Path("/tmp/devplace_test_screenshots")


PORT = int(os.environ.get("DEVPLACE_TEST_PORT", "10501"))
BASE_URL = f"http://127.0.0.1:{PORT}"
# Older than any user a fixture can create, so the seeded admin stays primary administrator.
PRIMARY_ADMIN_CREATED_AT = "2000-01-01T00:00:00"
# The cross-worker cache-version read is 1s; give the server a beat to see the new admin set.
CACHE_VERSION_PROPAGATION_SECONDS = 1.5

_TEST_DB = tempfile.NamedTemporaryFile(suffix=f"_{PORT}.db", delete=False)
_TEST_DB.close()
_TEST_DATA_DIR = Path(tempfile.mkdtemp(suffix=f"_data_{PORT}"))
os.environ["DEVPLACE_DATABASE_URL"] = f"sqlite:///{_TEST_DB.name}"
os.environ["DEVPLACE_DATA_DIR"] = str(_TEST_DATA_DIR)
os.environ["SECRET_KEY"] = "test-secret-key"
os.environ["DEVPLACE_DISABLE_SERVICES"] = "1"
os.environ["DEVPLACE_RATE_LIMIT"] = "1000000"
# Pin a single web worker so the per-worker rate-limit divisor is 1 regardless of
# any DEVPLACE_WEB_WORKERS the host (or a server .env) exports.
os.environ["DEVPLACE_WEB_WORKERS"] = "1"
# Disable per-IP rate limiting outright for the suite: the middleware short-circuits
# when this is set, so request-dense tests (e.g. test_auth_matrix) are never throttled
# regardless of host environment. test_ratelimit.py re-enables it per test to exercise
# the limiter directly.
os.environ["DEVPLACE_DISABLE_RATE_LIMIT"] = "1"
os.environ["DEVPLACE_SITEMAP_TTL"] = "0"
os.environ["DEVPLACE_HOME_CACHE_TTL"] = "0"
os.environ["DEVPLACE_RANKING_TTL"] = "0"
os.environ["DEVPLACE_MARKET_SATURATION_TTL"] = "0"


_ASYNC_LOOP = None
_ASYNC_LOOP_LOCK = threading.Lock()


def _async_loop():
    # One shared background loop thread for the whole session, not one thread
    # per run_async() call: dataset's NullPool caches one SQLite connection per
    # OS thread ID forever (see the root CLAUDE.md NullPool note), so a fresh
    # throwaway thread per call - this helper is invoked 270+ times across the
    # suite - permanently claimed one file descriptor apiece for the rest of
    # the session and could exhaust the process's open-file limit on a long run.
    global _ASYNC_LOOP
    with _ASYNC_LOOP_LOCK:
        if _ASYNC_LOOP is None:
            loop = asyncio.new_event_loop()
            threading.Thread(
                target=loop.run_forever, name="run-async-loop", daemon=True
            ).start()
            _ASYNC_LOOP = loop
        return _ASYNC_LOOP


def run_async(coro):
    """Run a coroutine from sync test code on the shared background loop."""

    async def _wrapped():
        from devplacepy.database import refresh_snapshot

        try:
            return await coro
        finally:
            refresh_snapshot()

    future = asyncio.run_coroutine_threadsafe(_wrapped(), _async_loop())
    try:
        return future.result(timeout=30)
    except TimeoutError:
        raise TimeoutError(
            "run_async timed out waiting for coroutine to complete"
        ) from None
    finally:
        from devplacepy.database import refresh_snapshot

        refresh_snapshot()


@pytest.fixture(autouse=True)
def _fresh_db_snapshot():
    from devplacepy.database import refresh_snapshot

    refresh_snapshot()
    yield


def save_failure_screenshot(page, test_name):
    SCREENSHOT_DIR.mkdir(parents=True, exist_ok=True)
    try:
        if page and not page.is_closed():
            path = str(SCREENSHOT_DIR / f"{test_name}.png")
            page.screenshot(path=path, full_page=True)
    except Exception:
        pass


@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
    outcome = yield
    report = outcome.get_result()
    if report.when == "call" and report.failed:
        for name in ("page", "alice", "bob"):
            if name in item.funcargs:
                try:
                    obj = item.funcargs[name]
                    if name in ("alice", "bob"):
                        obj = obj[0]
                    save_failure_screenshot(
                        obj, item.nodeid.replace("::", "_").replace("/", "_")
                    )
                except Exception:
                    pass


@pytest.fixture(scope="session")
def test_db_path():
    yield _TEST_DB.name
    try:
        os.unlink(_TEST_DB.name)
    except OSError:
        pass
    import shutil

    shutil.rmtree(_TEST_DATA_DIR, ignore_errors=True)


def _port_in_use(port):
    import socket

    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.settimeout(0.5)
        return sock.connect_ex(("127.0.0.1", port)) == 0


# Set once app_server has successfully started, so pytest_runtest_setup can
# detect a mid-session crash (e.g. resource exhaustion on a constrained CI
# runner) and fail every subsequent HTTP-dependent test with one clear
# diagnostic instead of a wall of opaque connection-refused errors.
_APP_SERVER_STATE = {"proc": None, "log_path": None}


@pytest.fixture(scope="session")
def app_server(test_db_path):
    if _port_in_use(PORT):
        raise RuntimeError(
            f"Port {PORT} is already in use before app_server starts. A stale test "
            f"server is running with a different database; kill it (the readiness "
            f"check would otherwise bind tests to it and produce phantom failures)."
        )
    env = os.environ.copy()
    env["PYTHONUNBUFFERED"] = "1"
    log_file = tempfile.NamedTemporaryFile(suffix=f"_server_{PORT}.log", delete=False)
    proc = subprocess.Popen(
        [
            sys.executable,
            "-m",
            "uvicorn",
            "devplacepy.main:app",
            "--host",
            "127.0.0.1",
            "--port",
            str(PORT),
            "--log-level",
            "warning",
        ],
        cwd=str(PROJECT_ROOT),
        env=env,
        stdout=log_file,
        stderr=subprocess.STDOUT,
    )

    def server_log():
        try:
            with open(log_file.name, "r", errors="replace") as f:
                return f.read()[-5000:]
        except OSError:
            return ""

    deadline = time.time() + 30
    while time.time() < deadline:
        try:
            import urllib.request

            urllib.request.urlopen(f"{BASE_URL}/", timeout=2)
            break
        except Exception:
            poll = proc.poll()
            if poll is not None:
                proc.wait()
                raise RuntimeError(
                    f"Server process died (exit code {poll}). Log:\n{server_log()}"
                )
            time.sleep(0.5)
    else:
        proc.terminate()
        proc.wait(timeout=5)
        raise RuntimeError(f"Server did not start after 30s. Log:\n{server_log()}")

    from devplacepy.database import set_setting as _set_setting

    # Use set_setting (upsert + cache-version bump) rather than a raw insert: the
    # server process caches settings and only reloads when the version bumps, so a
    # bare insert can leave the worker on a stale/default rate limit and throttle
    # request-dense tests (e.g. test_auth_matrix) with spurious 429s. Pin the limit
    # high regardless of any pre-existing row or the host's DEVPLACE_* environment.
    _set_setting("rate_limit_per_minute", "1000000")
    _set_setting("rate_limit_window_seconds", "60")

    _APP_SERVER_STATE["proc"] = proc
    _APP_SERVER_STATE["log_path"] = log_file.name

    yield proc
    _APP_SERVER_STATE["proc"] = None
    try:
        proc.terminate()
        proc.wait(timeout=10)
    except Exception:
        proc.kill()
        proc.wait(timeout=5)
    finally:
        try:
            log_file.close()
            os.unlink(log_file.name)
        except OSError:
            pass


def pytest_runtest_setup(item):
    proc = _APP_SERVER_STATE["proc"]
    if proc is None or "tests/unit/" in item.nodeid:
        return
    exit_code = proc.poll()
    if exit_code is None:
        return
    log_path = _APP_SERVER_STATE["log_path"]
    try:
        with open(log_path, "r", errors="replace") as f:
            log_tail = f.read()[-5000:]
    except OSError:
        log_tail = "(log unavailable)"
    pytest.fail(
        f"app_server died mid-session (exit code {exit_code}). It was healthy "
        f"earlier in this run and has since crashed - this is a shared-server "
        f"failure, not a bug in {item.nodeid}. Server log tail:\n{log_tail}",
        pytrace=False,
    )


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

    with sync_playwright() as p:
        yield p


@pytest.fixture(scope="session")
def browser(playwright_instance):
    headless = os.environ.get("PLAYWRIGHT_HEADLESS", "1") == "1"
    b = playwright_instance.chromium.launch(
        headless=headless,
        slow_mo=int(os.environ.get("PLAYWRIGHT_SLOW_MO", "0")),
        args=["--window-size=1400,900"],
    )
    yield b
    b.close()


@pytest.fixture(scope="module")
def browser_context(browser):
    ctx = browser.new_context(
        viewport={"width": 1400, "height": 900},
        permissions=["clipboard-read", "clipboard-write"],
    )
    yield ctx
    ctx.close()


@pytest.fixture
def page(browser_context):
    browser_context.clear_cookies()
    p = browser_context.new_page()
    p.set_default_timeout(15000)
    p.bring_to_front()
    yield p
    p.close()


def signup_user(page, user):
    page.bring_to_front()
    page.goto(f"{BASE_URL}/auth/signup", wait_until="domcontentloaded")
    page.fill("#username", user["username"])
    page.fill("#email", user["email"])
    page.fill("#password", user["password"])
    page.fill("#confirm_password", user["password"])
    page.fill("#birth_date", "1990-01-01")
    page.check("#accept_terms")
    page.click("button:has-text('Create account')")
    page.wait_for_url("**/feed", timeout=10000, wait_until="domcontentloaded")


def login_user(page, user):
    page.bring_to_front()
    page.goto(f"{BASE_URL}/auth/login", wait_until="domcontentloaded")
    page.fill("#email", user["email"])
    page.fill("#password", user["password"])
    page.click("button:has-text('Sign in')")
    page.wait_for_url("**/feed", timeout=10000, wait_until="domcontentloaded")


def paste_image(page, selector, name="pasted.png"):
    import base64
    import io
    from PIL import Image

    buf = io.BytesIO()
    Image.new("RGB", (4, 4), (0, 128, 255)).save(buf, "PNG")
    page.eval_on_selector(
        selector,
        """(target, [data, filename]) => {
            const bytes = Uint8Array.from(atob(data), (ch) => ch.charCodeAt(0));
            const transfer = new DataTransfer();
            transfer.items.add(new File([bytes], filename, { type: "image/png" }));
            target.dispatchEvent(
                new ClipboardEvent("paste", {
                    clipboardData: transfer,
                    bubbles: true,
                    cancelable: true,
                })
            );
        }""",
        [base64.b64encode(buf.getvalue()).decode(), name],
    )


def create_post_with_files(page, content, files, expected_count=1):
    from playwright.sync_api import expect

    page.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded")
    page.locator(".feed-fab").first.wait_for(state="visible", timeout=10000)
    page.locator(".feed-fab").first.click()
    page.fill("#post-content", content)
    page.locator("#create-post-modal dp-upload .dp-upload-input").first.set_input_files(
        files
    )
    expect(
        page.locator("#create-post-modal dp-upload .dp-upload-count").first
    ).to_have_text(f"({expected_count})", timeout=15000)
    page.locator("#create-post-modal button.btn-primary:has-text('Post')").click()
    page.wait_for_url(f"{BASE_URL}/posts/*", wait_until="domcontentloaded")


def assert_share_copies(page, expected_fragment):
    from playwright.sync_api import expect

    share = page.locator("button[data-share]").first
    share.scroll_into_view_if_needed()
    assert expected_fragment in (share.get_attribute("data-share") or ""), (
        f"data-share={share.get_attribute('data-share')!r}"
    )
    share.click()
    expect(share).to_have_text("Copied!", timeout=3000)
    expect(share).not_to_have_text("Copied!", timeout=3000)
    try:
        clip = page.evaluate("navigator.clipboard.readText()")
    except Exception:
        clip = None
    if clip:
        assert clip.startswith("http") and expected_fragment in clip, (
            f"clipboard={clip!r}"
        )


def assert_no_horizontal_overflow(page, layout):
    report = page.evaluate(
        """(selector) => {
            const root = document.documentElement;
            const container = document.querySelector(selector);
            const widest = [];
            document.querySelectorAll("body *").forEach((el) => {
                if (el.getBoundingClientRect().right <= root.clientWidth + 1) return;
                widest.push(
                    el.tagName.toLowerCase() + "." + (el.className || "").toString().trim()
                );
            });
            return {
                page: root.scrollWidth - root.clientWidth,
                layout: container ? container.scrollWidth - container.clientWidth : null,
                widest: widest.slice(0, 5),
            };
        }""",
        layout,
    )
    assert report["layout"] is not None, f"{layout} not found"
    assert report["page"] <= 1, (
        f"page scrolls horizontally by {report['page']}px: {report['widest']}"
    )
    assert report["layout"] <= 1, (
        f"{layout} overflows its own width by {report['layout']}px: {report['widest']}"
    )


@pytest.fixture(scope="session")
def seeded_db(app_server):
    """Create test users once at session level using requests directly."""
    import urllib.request
    import urllib.parse

    users = [
        ("alice_test", "alice@test.devplace", "secret123"),
        ("bob_test", "bob@test.devplace", "secret456"),
    ]
    for username, email, pw in users:
        data = urllib.parse.urlencode(
            {
                "username": username,
                "email": email,
                "password": pw,
                "confirm_password": pw,
                "birth_date": "1990-01-01",
                "accept_terms": "1",
            }
        ).encode()
        req = urllib.request.Request(
            f"{BASE_URL}/auth/signup",
            data=data,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
        )
        urllib.request.urlopen(req)

    from devplacepy.database import get_table as _get_table

    users_table = _get_table("users")
    for username, role in (("alice_test", "Admin"), ("bob_test", "Member")):
        row = users_table.find_one(username=username)
        if row and row.get("role") != role:
            users_table.update({"uid": row["uid"], "role": role}, ["uid"])

    # The primary administrator is the earliest-created Admin, and unit-tier fixtures
    # create their own admins in this same database before the api tier seeds alice.
    # Back-date her so she is unambiguously the platform's first administrator, which
    # is what every primary-admin test (dbapi, backup download, container isolation)
    # assumes. Without this, a combined unit+api run hands primary admin to a fixture
    # user and those tests 403.
    from devplacepy.database import invalidate_admins_cache as _invalidate_admins

    alice_row = users_table.find_one(username="alice_test")
    if alice_row:
        users_table.update(
            {"uid": alice_row["uid"], "created_at": PRIMARY_ADMIN_CREATED_AT}, ["uid"]
        )
        _invalidate_admins()
        time.sleep(CACHE_VERSION_PROPAGATION_SECONDS)
    return {
        "alice": {
            "username": "alice_test",
            "email": "alice@test.devplace",
            "password": "secret123",
        },
        "bob": {
            "username": "bob_test",
            "email": "bob@test.devplace",
            "password": "secret456",
        },
    }


@pytest.fixture
def alice(page, seeded_db):
    login_user(page, seeded_db["alice"])
    return page, seeded_db["alice"]


@pytest.fixture
def bob(browser, seeded_db):
    ctx = browser.new_context(
        viewport={"width": 1400, "height": 900},
        permissions=["clipboard-read", "clipboard-write"],
    )
    p = ctx.new_page()
    p.set_default_timeout(15000)
    p.bring_to_front()
    login_user(p, seeded_db["bob"])
    yield p, seeded_db["bob"]
    p.close()
    ctx.close()


@pytest.fixture
def mobile_page(browser, seeded_db):
    ctx = browser.new_context(
        viewport={"width": 412, "height": 840},
        has_touch=True,
        permissions=["clipboard-read", "clipboard-write"],
    )
    p = ctx.new_page()
    p.set_default_timeout(15000)
    p.bring_to_front()
    login_user(p, seeded_db["alice"])
    yield p, seeded_db["alice"]
    p.close()
    ctx.close()


@pytest.fixture(scope="session")
def local_db():
    from devplacepy.database import init_db, db

    init_db()
    return db