Merge branch 'worktree-visual-test-overhaul'
# Conflicts: # web/tests/test_e2e_flows.py
This commit is contained in:
@@ -237,3 +237,86 @@ def sign_payload(payload_bytes: bytes, secret: str = "whsec_test_secret") -> str
|
||||
data = f"{ts}:{payload_bytes.decode()}".encode()
|
||||
h1 = hmac.new(secret.encode(), data, hashlib.sha256).hexdigest()
|
||||
return f"ts={ts};h1={h1}"
|
||||
|
||||
|
||||
# ── Visual test fixtures (Playwright) ────────────────────────
|
||||
# Session-scoped: one server + one browser for all visual tests.
|
||||
|
||||
_VISUAL_PORT = 5111
|
||||
|
||||
|
||||
def _run_visual_server(ready_event):
|
||||
"""Run a Quart dev server in a subprocess for visual/E2E tests.
|
||||
|
||||
Forces RESEND_API_KEY="" so no real emails are sent.
|
||||
Runs migrations in-process to build the full schema (including FTS tables).
|
||||
"""
|
||||
import asyncio
|
||||
import os
|
||||
|
||||
os.environ["RESEND_API_KEY"] = ""
|
||||
|
||||
async def _serve():
|
||||
# Build schema DDL in-process (FTS5 virtual tables need this)
|
||||
tmp_db = str(Path(tempfile.mkdtemp()) / "schema.db")
|
||||
migrate(tmp_db)
|
||||
tmp_conn = sqlite3.connect(tmp_db)
|
||||
rows = tmp_conn.execute(
|
||||
"SELECT sql FROM sqlite_master"
|
||||
" WHERE sql IS NOT NULL"
|
||||
" AND name NOT LIKE 'sqlite_%'"
|
||||
" AND name NOT LIKE '%_fts_%'"
|
||||
" AND name != '_migrations'"
|
||||
" ORDER BY rowid"
|
||||
).fetchall()
|
||||
tmp_conn.close()
|
||||
schema_ddl = ";\n".join(r[0] for r in rows) + ";"
|
||||
|
||||
conn = await aiosqlite.connect(":memory:")
|
||||
conn.row_factory = aiosqlite.Row
|
||||
await conn.execute("PRAGMA foreign_keys=ON")
|
||||
await conn.executescript(schema_ddl)
|
||||
await conn.commit()
|
||||
core._db = conn
|
||||
|
||||
# Patch init_db/close_db where they're USED (app.py imports them
|
||||
# locally via `from .core import init_db` — patching core.init_db
|
||||
# alone doesn't affect the local binding in app.py).
|
||||
# Patches must stay active through run_task() because before_serving
|
||||
# hooks call init_db() which would replace our in-memory DB.
|
||||
with patch("padelnomics.app.init_db", new_callable=AsyncMock), \
|
||||
patch("padelnomics.app.close_db", new_callable=AsyncMock):
|
||||
app = create_app()
|
||||
app.config["TESTING"] = True
|
||||
ready_event.set()
|
||||
await app.run_task(host="127.0.0.1", port=_VISUAL_PORT)
|
||||
|
||||
asyncio.run(_serve())
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def live_server():
|
||||
"""Start a live Quart server on port 5111 for all visual/E2E tests."""
|
||||
import multiprocessing
|
||||
|
||||
ready = multiprocessing.Event()
|
||||
proc = multiprocessing.Process(
|
||||
target=_run_visual_server, args=(ready,), daemon=True
|
||||
)
|
||||
proc.start()
|
||||
ready.wait(timeout=10)
|
||||
time.sleep(1)
|
||||
yield f"http://127.0.0.1:{_VISUAL_PORT}"
|
||||
proc.terminate()
|
||||
proc.join(timeout=5)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def browser():
|
||||
"""Launch a headless Chromium browser (once per test session)."""
|
||||
from playwright.sync_api import sync_playwright
|
||||
|
||||
with sync_playwright() as p:
|
||||
b = p.chromium.launch(headless=True)
|
||||
yield b
|
||||
b.close()
|
||||
|
||||
Reference in New Issue
Block a user