fieldwitness/tests/e2e/conftest.py

"""
Playwright e2e test fixtures for the FieldWitness Flask web UI.

Isolation strategy
------------------
- Each test session uses a fresh tmp_path via the `live_server` fixture.
- `FIELDWITNESS_DATA_DIR` is set in the OS environment before the Flask app
  factory runs, so paths.py picks it up through its lazy `__getattr__`.
- The app is never imported at module level here — it is imported inside the
  fixture function *after* the env var is set.
- The Flask dev server is run in a daemon thread with use_reloader=False so
  the test process remains the controller.
"""

from __future__ import annotations

import os
import socket
import threading
import time
from pathlib import Path
from typing import Generator

import pytest
from playwright.sync_api import Page


# ---------------------------------------------------------------------------
# Constants shared across fixtures
# ---------------------------------------------------------------------------

TEST_ADMIN_USER = "testadmin"
TEST_ADMIN_PASS = "Fieldwitness-e2e-2024!"


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _find_free_port() -> int:
    """Return a free TCP port on localhost."""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


def _wait_for_server(base_url: str, timeout: float = 20.0) -> None:
    """Poll until the server responds or timeout is reached."""
    import urllib.request
    import urllib.error

    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        try:
            urllib.request.urlopen(f"{base_url}/login", timeout=2)
            return
        except urllib.error.HTTPError:
            # Server is up but returned an error — that's fine, it responded.
            return
        except (urllib.error.URLError, ConnectionError, OSError):
            time.sleep(0.2)
    raise RuntimeError(f"Server at {base_url} did not start within {timeout}s")


# ---------------------------------------------------------------------------
# Session-scoped fixtures
# ---------------------------------------------------------------------------


@pytest.fixture(scope="session")
def e2e_data_dir(tmp_path_factory: pytest.TempPathFactory) -> Path:
    """A single data directory for the entire test session.

    Using session scope means one Flask app instance serves all tests,
    matching how Playwright typically works (one browser, many pages).
    """
    return tmp_path_factory.mktemp("fieldwitness_e2e")


@pytest.fixture(scope="session")
def live_server(e2e_data_dir: Path) -> Generator[str, None, None]:
    """Start the Flask app on a random port, yield the base URL.

    The server runs in a daemon thread so it dies when the test process exits.

    Path isolation strategy
    -----------------------
    fieldwitness.paths.BASE_DIR is a module-level Path computed once at import
    time from FIELDWITNESS_DATA_DIR. By the time conftest runs, fieldwitness is
    already imported (pytest imports it for coverage), so setting the env var
    is too late. Instead we directly patch `fieldwitness.paths.BASE_DIR` for
    the duration of the test session; the module-level __getattr__ then resolves
    every derived path (IDENTITY_DIR, AUTH_DB, …) from that patched BASE_DIR.
    """
    data_dir = e2e_data_dir / ".fwmetadata"
    data_dir.mkdir(parents=True, exist_ok=True)

    # Ensure frontends/web/ is on sys.path so its local imports (temp_storage,
    # subprocess_stego, etc.) resolve correctly when the app is created here.
    import sys

    web_dir = str(Path(__file__).resolve().parents[2] / "frontends" / "web")
    if web_dir not in sys.path:
        sys.path.insert(0, web_dir)

    # Patch BASE_DIR so all lazy path resolution uses our temp directory
    import fieldwitness.paths as _paths

    original_base_dir = _paths.BASE_DIR
    _paths.BASE_DIR = data_dir

    # Also set the env var so any sub-process or re-import gets the right dir
    os.environ["FIELDWITNESS_DATA_DIR"] = str(data_dir)

    port = _find_free_port()

    from fieldwitness.config import FieldWitnessConfig
    from frontends.web.app import create_app

    config = FieldWitnessConfig(
        https_enabled=False,
        auth_enabled=True,
        deadman_enabled=False,
        killswitch_enabled=False,
        chain_enabled=False,
        chain_auto_wrap=False,
        max_upload_mb=16,
        session_timeout_minutes=60,
        login_lockout_attempts=10,
        login_lockout_minutes=1,
    )

    flask_app = create_app(config=config)
    flask_app.config["TESTING"] = True
    flask_app.config["WTF_CSRF_ENABLED"] = False  # Flask-WTF checks this per request
    flask_app.config["SECRET_KEY"] = "e2e-test-secret-key-not-for-production"

    def _run():
        from werkzeug.serving import make_server

        srv = make_server("127.0.0.1", port, flask_app)
        srv.serve_forever()

    thread = threading.Thread(target=_run, daemon=True)
    thread.start()

    base_url = f"http://127.0.0.1:{port}"
    _wait_for_server(base_url)

    yield base_url

    # Restore original BASE_DIR when the session ends
    _paths.BASE_DIR = original_base_dir


# ---------------------------------------------------------------------------
# Session-scoped admin setup — must run before any test that expects a user
# ---------------------------------------------------------------------------


@pytest.fixture(scope="session")
def admin_user(live_server: str) -> None:
    """Ensure the admin account exists in the session-scoped live server.

    Uses a direct Flask test client POST so no Playwright page is needed and
    no race with browser test ordering can occur.  The fixture is session-scoped
    so it runs once; subsequent calls are no-ops because create_admin_user()
    returns early when a user already exists.
    """
    import sys
    from pathlib import Path

    # frontends/web must already be on sys.path (live_server fixture adds it)
    web_dir = str(Path(__file__).resolve().parents[2] / "frontends" / "web")
    if web_dir not in sys.path:
        sys.path.insert(0, web_dir)

    import urllib.request
    import urllib.parse

    # POST to /setup to create the admin user.  CSRF is disabled in test config
    # so a plain POST with the form fields is sufficient.
    data = urllib.parse.urlencode(
        {
            "username": TEST_ADMIN_USER,
            "password": TEST_ADMIN_PASS,
            "password_confirm": TEST_ADMIN_PASS,
        }
    ).encode()

    req = urllib.request.Request(
        f"{live_server}/setup",
        data=data,
        method="POST",
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )
    try:
        urllib.request.urlopen(req, timeout=10)
    except urllib.error.HTTPError:
        # Any HTTP-level error (e.g. redirect to /login because user already
        # exists) is fine — the user exists either way.
        pass


# ---------------------------------------------------------------------------
# Function-scoped fixtures
# ---------------------------------------------------------------------------


@pytest.fixture()
def authenticated_page(live_server: str, admin_user: None, page: Page) -> Page:
    """Return a Playwright page that is authenticated as the test admin.

    Handles first-run setup (creating the admin user) on the first call.
    Subsequent calls log in directly.
    """
    # Check if we need first-run setup
    page.goto(f"{live_server}/")
    page.wait_for_load_state("networkidle")

    if "/setup" in page.url:
        # First-run: create admin account
        page.fill("input[name='username']", TEST_ADMIN_USER)
        page.fill("input[name='password']", TEST_ADMIN_PASS)
        page.fill("input[name='password_confirm']", TEST_ADMIN_PASS)
        page.click("button[type='submit']")
        page.wait_for_load_state("networkidle")
    elif "/login" in page.url:
        # Subsequent runs: log in
        page.fill("input[name='username']", TEST_ADMIN_USER)
        page.fill("input[name='password']", TEST_ADMIN_PASS)
        page.click("button[type='submit']")
        page.wait_for_load_state("networkidle")

    # Ensure we are at the index (not stuck on setup or login)
    if "/login" in page.url or "/setup" in page.url:
        raise RuntimeError(
            f"Failed to authenticate. Currently at: {page.url}\n"
            f"Page content: {page.content()[:500]}"
        )

    return page


@pytest.fixture()
def dropbox_token(live_server: str, authenticated_page: Page) -> str:
    """Create a drop box upload token via the admin UI and return the token URL.

    Returns the full upload URL so tests can navigate directly to the source
    upload page without needing to parse the admin UI.
    """
    page = authenticated_page
    page.goto(f"{live_server}/dropbox/admin")
    page.wait_for_load_state("networkidle")

    # Fill in token creation form
    label_input = page.locator("input[name='label']")
    if label_input.count():
        label_input.fill("e2e-test-source")

    page.fill("input[name='hours']", "24")
    page.fill("input[name='max_files']", "10")

    # Submit and capture the flash message containing the upload URL
    page.click("button[type='submit']")
    page.wait_for_load_state("networkidle")

    # Extract the upload URL from the success flash message
    flash_text = page.locator(".alert-success, .alert.success, [class*='alert']").first.inner_text()

    # The URL is embedded in the flash: "Share this URL with your source: http://..."
    import re

    match = re.search(r"http://\S+", flash_text)
    if not match:
        raise RuntimeError(f"Could not find upload URL in flash message: {flash_text!r}")

    return match.group(0)