from __future__ import annotations import os import shutil import socket import subprocess import tempfile import time import base64 from pathlib import Path from alembic import command from alembic.config import Config from playwright.sync_api import sync_playwright ROOT = Path(__file__).resolve().parent.parent OUTPUT_DIR = ROOT / "output" / "playwright" RUNTIME_DIR = OUTPUT_DIR / "runtime" CHROME_PATH = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome" def main() -> int: OUTPUT_DIR.mkdir(parents=True, exist_ok=True) if RUNTIME_DIR.exists(): shutil.rmtree(RUNTIME_DIR) RUNTIME_DIR.mkdir(parents=True, exist_ok=True) ensure_frontend_built() port = find_free_port() base_url = f"http://127.0.0.1:{port}" runtime_db = RUNTIME_DIR / "ui-playwright.db" prepare_database(runtime_db) process = start_server(port, runtime_db) try: wait_for_server(base_url) run_browser_flow(base_url) finally: process.terminate() try: process.wait(timeout=10) except subprocess.TimeoutExpired: process.kill() print(f"Playwright smoke flow passed. Screenshots saved to {OUTPUT_DIR}") return 0 def ensure_frontend_built() -> None: dist_index = ROOT / "app" / "web" / "dist" / "index.html" if dist_index.exists(): return subprocess.run(["npm", "run", "build"], cwd=ROOT, check=True) def find_free_port() -> int: with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) return int(sock.getsockname()[1]) def prepare_database(runtime_db: Path) -> None: config = Config(str(ROOT / "alembic.ini")) config.set_main_option("sqlalchemy.url", f"sqlite+aiosqlite:///{runtime_db}") command.upgrade(config, "head") def start_server(port: int, runtime_db: Path) -> subprocess.Popen[str]: env = os.environ.copy() env["IRON_DATABASE_URL"] = f"sqlite+aiosqlite:///{runtime_db}" env["IRON_UPLOAD_TEMP_DIR"] = str(RUNTIME_DIR / "uploads") env["IRON_LOCAL_STORAGE_DIR"] = str(RUNTIME_DIR / "storage") env["IRON_CACHE_DIR"] = str(RUNTIME_DIR / "cache") env["IRON_BOOTSTRAP_USERNAME"] = "admin" env["IRON_BOOTSTRAP_PASSWORD"] = "changeme-iron" log_path = OUTPUT_DIR / "server.log" log_handle = open(log_path, "w", encoding="utf-8") return subprocess.Popen( [ str(ROOT / ".venv" / "bin" / "python"), "-m", "uvicorn", "app.main:app", "--host", "127.0.0.1", "--port", str(port), ], cwd=ROOT, env=env, stdout=log_handle, stderr=log_handle, text=True, ) def wait_for_server(base_url: str) -> None: deadline = time.time() + 20 while time.time() < deadline: try: import urllib.request with urllib.request.urlopen(f"{base_url}/api/system/health", timeout=1) as response: if response.status == 200: return except Exception: time.sleep(0.4) log_path = OUTPUT_DIR / "server.log" log_tail = log_path.read_text(encoding="utf-8") if log_path.exists() else "" raise RuntimeError(f"Server did not start in time.\n{log_tail}") def run_browser_flow(base_url: str) -> None: with sync_playwright() as playwright: browser = playwright.chromium.launch( executable_path=CHROME_PATH, headless=True, ) context = browser.new_context(viewport={"width": 1440, "height": 1024}) page = context.new_page() failures: list[str] = [] def track_response(response) -> None: # type: ignore[no-untyped-def] if "/api/" in response.url and response.status >= 400: failures.append(f"{response.status} {response.url}") def track_request_failed(request) -> None: # type: ignore[no-untyped-def] if "/api/" in request.url: failure = request.failure or "unknown request failure" failures.append(f"request failed {request.url}: {failure}") page.on("response", track_response) page.on("requestfailed", track_request_failed) page.on("pageerror", lambda error: failures.append(f"page error: {error}")) page.goto(f"{base_url}/app/login", wait_until="networkidle") page.screenshot(path=str(OUTPUT_DIR / "login.png"), full_page=True) page.get_by_label("Username").fill("admin") page.get_by_label("Password").fill("changeme-iron") page.get_by_role("button", name="Sign in").click() page.wait_for_url(f"{base_url}/app/files") page.locator(".files-header .crumb-button").filter(has_text="/").first.wait_for() page.screenshot(path=str(OUTPUT_DIR / "files-initial.png"), full_page=True) page.locator(".panel-toolbar").get_by_role("button", name="New").click() page.get_by_role("dialog").get_by_label("Name").fill("playwright-folder") page.get_by_role("dialog").get_by_role("button", name="Confirm").click() page.locator(".drive-item strong", has_text="playwright-folder").first.wait_for() page.screenshot(path=str(OUTPUT_DIR / "files-after-create-folder.png"), full_page=True) temp_upload = create_temp_upload_file() preview_upload = create_temp_preview_file() try: page.locator(".files-command-bar").get_by_role("button", name="Upload", exact=True).click() page.locator("input[type='file']").set_input_files(str(temp_upload)) page.locator(".drive-item strong", has_text="playwright-upload.txt").first.wait_for(timeout=15000) page.locator(".files-command-bar").get_by_role("button", name="Upload", exact=True).click() page.locator("input[type='file']").set_input_files(str(preview_upload)) page.locator(".drive-item strong", has_text="playwright-preview.png").first.wait_for(timeout=15000) finally: temp_upload.unlink(missing_ok=True) preview_upload.unlink(missing_ok=True) page.screenshot(path=str(OUTPUT_DIR / "files-after-upload.png"), full_page=True) page.locator(".drive-item strong", has_text="playwright-upload.txt").first.click() page.locator(".inspector-hero strong", has_text="playwright-upload.txt").wait_for() with page.expect_download() as download_info: page.locator(".detail-panel").get_by_role("button", name="Download").click() download = download_info.value if download.suggested_filename != "playwright-upload.txt": failures.append(f"unexpected download filename: {download.suggested_filename}") page.screenshot(path=str(OUTPUT_DIR / "files-detail-selected.png"), full_page=True) page.locator(".drive-item strong", has_text="playwright-preview.png").first.click() page.locator("img.preview-frame").wait_for(timeout=15000) page.screenshot(path=str(OUTPUT_DIR / "files-preview-selected.png"), full_page=True) page.get_by_role("button", name="Storage").click() page.get_by_role("button", name="Reconcile File").wait_for() page.screenshot(path=str(OUTPUT_DIR / "files-storage-selected.png"), full_page=True) page.get_by_label("Search current directory").fill("playwright") page.locator(".drive-item strong", has_text="playwright-folder").first.wait_for() page.screenshot(path=str(OUTPUT_DIR / "files-search.png"), full_page=True) page.goto(f"{base_url}/app/uploads", wait_until="networkidle") page.screenshot(path=str(OUTPUT_DIR / "uploads-page.png"), full_page=True) page.goto(f"{base_url}/app/recycle-bin", wait_until="networkidle") page.screenshot(path=str(OUTPUT_DIR / "recycle-bin-page.png"), full_page=True) page.goto(f"{base_url}/app/storage", wait_until="networkidle") page.get_by_role("button", name="Run All Checks").click() page.locator(".toast-item").first.wait_for() page.get_by_role("button", name="Edit").first.click() page.get_by_role("dialog").get_by_text("Stored secrets").wait_for() page.screenshot(path=str(OUTPUT_DIR / "storage-edit-dialog.png"), full_page=True) page.get_by_role("dialog").get_by_role("button", name="Cancel").click() page.screenshot(path=str(OUTPUT_DIR / "storage-page.png"), full_page=True) page.goto(f"{base_url}/app/policies", wait_until="networkidle") page.get_by_role("button", name="Save Policy").wait_for() page.get_by_role("button", name="Save Policy").click() page.locator(".toast-item").first.wait_for() page.screenshot(path=str(OUTPUT_DIR / "policies-page.png"), full_page=True) page.goto(f"{base_url}/app/jobs", wait_until="networkidle") page.screenshot(path=str(OUTPUT_DIR / "jobs-page.png"), full_page=True) context.close() browser.close() if failures: raise AssertionError("Playwright E2E failures:\n" + "\n".join(failures)) def create_temp_upload_file() -> Path: handle = tempfile.NamedTemporaryFile(delete=False, suffix="-playwright-upload.txt") path = Path(handle.name) handle.write(b"playwright upload validation\n") handle.close() target = path.with_name("playwright-upload.txt") path.replace(target) return target def create_temp_preview_file() -> Path: png_data = base64.b64decode( "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+yWZ0AAAAASUVORK5CYII=" ) handle = tempfile.NamedTemporaryFile(delete=False, suffix="-playwright-preview.png") path = Path(handle.name) handle.write(png_data) handle.close() target = path.with_name("playwright-preview.png") path.replace(target) return target if __name__ == "__main__": raise SystemExit(main())