""" 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)