user-engine/scripts/browser_journeys.py

#!/usr/bin/env python3
"""Run Chromium against an isolated loopback-only portal with synthetic identities."""
from pathlib import Path
import os
import shutil
import subprocess
import sys
import tempfile
from threading import Thread
import time
from wsgiref.simple_server import make_server, WSGIRequestHandler

ROOT=Path(__file__).resolve().parents[1]
sys.path[:0]=[str(ROOT/'src'),str(ROOT/'tests')]
from test_journey_roles import JourneyFixture
from test_factor_recovery_journey import FactorRecoveryJourney

chrome=os.environ.get('JOURNEY_CHROME') or shutil.which('chromium') or shutil.which('google-chrome')
if not chrome:
    matches=sorted((Path.home()/'.cache/ms-playwright').glob('chromium-*/chrome-linux64/chrome'))
    chrome=str(matches[-1]) if matches else None
if not chrome or not shutil.which('node'):
    raise SystemExit('Chromium and Node are required; set JOURNEY_CHROME to the Chromium executable. Browser tests were not run.')
fixture=FactorRecoveryJourney();fixture.setUp();fixture.member(email='actual.login@example.test')
from user_engine.domain import OutboxEvent
fixture.app.service.store.append_outbox(OutboxEvent(event_id="p05-browser", event_type="family_member.invited", aggregate_id="fixture", tenant="tenant:trial:demo-company", correlation_id="p05-browser", payload={"primary_email":"fixture@example.test", "invitation_id":"fixture"}))
class Delivery:
    attempts = 0
    def __call__(self, event):
        self.attempts += 1
        if self.attempts == 1: raise RuntimeError("synthetic outage")
    def mail_delivery_status(self, event_id):
        return {"state":"failed" if self.attempts == 1 else "provider_accepted" if self.attempts else "not_found"}
fixture.app.outbox_delivery = Delivery()
class Quiet(WSGIRequestHandler):
    def log_message(self,*args):pass
server=make_server('127.0.0.1',0,fixture.app,handler_class=Quiet)
worker=Thread(target=server.serve_forever,daemon=True);worker.start()
try:
    with tempfile.TemporaryDirectory(prefix='user-engine-browser-') as profile:
        process=subprocess.Popen([chrome,'--headless','--no-sandbox','--disable-gpu','--remote-debugging-address=127.0.0.1',
            '--remote-debugging-port=0','--user-data-dir='+profile,'about:blank'],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
        try:
            port_file=Path(profile)/'DevToolsActivePort'
            for _ in range(100):
                if port_file.exists():break
                if process.poll() is not None:raise RuntimeError('Chromium exited before test connection')
                time.sleep(.1)
            if not port_file.exists():raise RuntimeError('Chromium did not expose its test connection')
            port=port_file.read_text().splitlines()[0]
            subprocess.run(['node',str(ROOT/'scripts/browser_journeys.mjs'),'http://127.0.0.1:'+port,
                            'http://127.0.0.1:'+str(server.server_port)],check=True,timeout=55)
        finally:
            process.terminate()
            try:process.wait(timeout=5)
            except subprocess.TimeoutExpired:process.kill();process.wait()
finally:
    server.shutdown();server.server_close();worker.join(timeout=5)