You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

296 lines
13 KiB

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" / str(os.getpid())
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").filter(has_text="Personal").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.get_by_role("dialog").wait_for(state="detached")
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)
page.locator(".drive-item strong", has_text="playwright-folder").first.dblclick()
page.locator(".directory-title-row", has_text="playwright-folder").wait_for()
page.screenshot(path=str(OUTPUT_DIR / "files-folder-open.png"), full_page=True)
page.get_by_label("Back to parent folder").click()
page.locator(".files-header").filter(has_text="Personal").first.wait_for()
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(".detail-panel h3", has_text="playwright-upload.txt").wait_for()
page.locator(".drive-item", has_text="playwright-upload.txt").get_by_role("button", name="More").click()
with page.expect_download() as download_info:
page.get_by_role("menu").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.locator(".drive-item", has_text="playwright-preview.png").locator(".replica-ratio", has_text="1/1").wait_for()
page.screenshot(path=str(OUTPUT_DIR / "files-preview-selected.png"), full_page=True)
page.locator(".detail-panel").get_by_role("button", name="More Info").click()
page.locator(".detail-drawer", has_text="Replica Info").wait_for()
page.locator(".detail-drawer", has_text="Desired Backends").wait_for()
page.locator(".detail-drawer", has_text="Replicas").wait_for()
if page.locator(".detail-drawer").get_by_role("button", name="Reconcile File").count():
failures.append("detail drawer should be read-only; found Reconcile File action")
page.screenshot(path=str(OUTPUT_DIR / "files-storage-selected.png"), full_page=True)
page.locator(".detail-drawer").get_by_role("button", name="Close").click()
page.locator(".files-command-bar").get_by_role("button", name="Grid").click()
page.locator(".drive-grid .drive-item", has_text="playwright-preview.png").get_by_role("button", name="More").click()
page.get_by_role("menu").get_by_role("button", name="Download").wait_for()
page.get_by_role("menu").get_by_role("button", name="Reconcile").wait_for()
page.screenshot(path=str(OUTPUT_DIR / "files-grid-menu.png"), full_page=True)
page.keyboard.press("Escape")
page.locator(".drive-grid .drive-item", has_text="playwright-preview.png").get_by_role("button", name="More").click()
page.get_by_role("menu").get_by_role("button", name="Delete").click()
delete_button = page.get_by_role("dialog").get_by_role("button", name="Confirm")
delete_button.wait_for()
delete_styles = delete_button.evaluate(
"(node) => { const style = getComputedStyle(node); return { color: style.color, backgroundColor: style.backgroundColor, backgroundImage: style.backgroundImage }; }"
)
has_visible_background = delete_styles["backgroundColor"] not in {
"rgba(0, 0, 0, 0)",
"transparent",
} or delete_styles["backgroundImage"] != "none"
if delete_styles["color"] != "rgb(255, 255, 255)" or not has_visible_background:
failures.append(f"delete confirm button has unreadable styles: {delete_styles}")
page.screenshot(path=str(OUTPUT_DIR / "files-delete-dialog.png"), full_page=True)
page.get_by_role("dialog").get_by_role("button", name="Cancel").click()
page.get_by_role("dialog").wait_for(state="detached")
page.get_by_label("Search").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/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.locator(".jobs-table-row").first.click()
page.locator(".jobs-detail-panel", has_text="Job Detail").wait_for()
page.locator(".payload-preview").wait_for()
page.screenshot(path=str(OUTPUT_DIR / "jobs-page.png"), full_page=True)
mobile_context = browser.new_context(viewport={"width": 390, "height": 844})
mobile_page = mobile_context.new_page()
mobile_page.on("response", track_response)
mobile_page.on("requestfailed", track_request_failed)
mobile_page.on("pageerror", lambda error: failures.append(f"mobile page error: {error}"))
mobile_page.goto(f"{base_url}/app/login", wait_until="networkidle")
mobile_page.get_by_label("Username").fill("admin")
mobile_page.get_by_label("Password").fill("changeme-iron")
mobile_page.get_by_role("button", name="Sign in").click()
mobile_page.wait_for_url(f"{base_url}/app/files")
mobile_page.locator(".files-header").filter(has_text="Personal").first.wait_for()
mobile_page.screenshot(path=str(OUTPUT_DIR / "mobile-files.png"), full_page=True)
mobile_context.close()
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())