""" 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 = 10.0) -> None: """Poll until the server responds or timeout is reached.""" deadline = time.monotonic() + timeout while time.monotonic() < deadline: try: import urllib.request urllib.request.urlopen(f"{base_url}/health", timeout=1) return except Exception: time.sleep(0.1) 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 / ".fieldwitness" data_dir.mkdir(parents=True, exist_ok=True) # 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 # --------------------------------------------------------------------------- # Function-scoped fixtures # --------------------------------------------------------------------------- @pytest.fixture() def authenticated_page(live_server: str, 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)