"""Chromium about:blank fixtures bridged to real ASGI endpoints; synthetic worker handoffs. This is not live systemd/Core/SSH qualification. Bootstrap may be a declared substitute. HTMX is deliberately not loaded when unavailable: status replacements are driven explicitly and labeled as synthetic lifecycle coverage, not HTMX proof. Never install fixture assets into production static/vendor. """ from pathlib import Path import argparse import json import re import sys import tempfile import time from urllib.parse import urlsplit sys.path.insert(0, str(Path(__file__).resolve().parents[1]/'src')) import pytest from fastapi.testclient import TestClient from playwright.sync_api import sync_playwright from aim_webgui.app import create_app from aim_webgui.credentials import wire from test_credential_ux import authz as auth_fixture, make_job, SYNTHETIC ROOT = Path(__file__).resolve().parents[1] def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--out', type=Path, required=True) parser.add_argument('--bootstrap', type=Path, default=Path('/opt/imagemagick/share/doc/ImageMagick-7/www/assets/bootstrap.min.css')) args = parser.parse_args(); args.out.mkdir(parents=True, exist_ok=True) cases = [] with tempfile.TemporaryDirectory(prefix='aim-credential-qa-') as tmp, pytest.MonkeyPatch.context() as monkey: auth, users = auth_fixture.__wrapped__(Path(tmp), monkey) fixture = (auth, users) packets = [] class Sock: def __enter__(self): return self def __exit__(self, *args): pass monkey.setattr(wire, 'connect', lambda *a, **k: Sock()) monkey.setattr(wire, 'send', lambda sock, packet, limit: packets.append(json.loads(json.dumps(packet)))) def claimed(*args): ident = packets[-1]['job'] with auth.store.transaction() as db: db.execute("UPDATE jobs SET credential_phase='claimed' WHERE id=?", (ident,)) return {'accepted': True} monkey.setattr(wire, 'receive', claimed) origin = auth.settings.public_url with TestClient(create_app(auth.settings), base_url=origin, follow_redirects=False) as server, sync_playwright() as pw: browser = pw.chromium.launch(executable_path='/usr/bin/chromium', args=['--no-sandbox', '--disable-dev-shm-usage']) try: for width, height in ((320,750),(390,844),(760,900),(1440,1000),(740,390)): for theme in ('light','dark'): # Independent viewport fixtures must not share the live five/minute bucket. # Production throttle remains enabled and is covered in unit tests. with auth.store.transaction() as db: db.execute('DELETE FROM rate_limits') token, session = auth.new_session(users['operator']) context = browser.new_context(viewport={'width':width,'height':height}, locale='en-GB', timezone_id='Europe/Berlin', reduced_motion='reduce') context.add_cookies([{'name':auth.settings.cookie_name, 'value':token, 'url':origin,'secure':True,'httpOnly':True,'sameSite':'Lax'}]) errors = [] mode = {'drop_post':False} def bridge(url, options): path = urlsplit(url).path method = options.get('method', 'GET') if mode['drop_post'] and method == 'POST' and path.endswith('/credentials'): mode['post_count'] = mode.get('post_count',0)+1 return {'network_error': True} headers = {**options.get('headers', {}), 'Origin': origin} response = server.request(method, path, headers=headers, content=options.get('body')) return {'status': response.status_code, 'body': response.text} server.cookies.set(auth.settings.cookie_name, token) page = context.new_page(); page.on('pageerror', lambda e: errors.append(str(e))) page.expose_function('_fixtureHTTP', bridge) def mount(path): page.goto('about:blank') content = server.get(path).text content = re.sub(r']+src="[^"]+"[^>]*>', '', content) content = re.sub(r']+rel="stylesheet"[^>]*>', '', content) css = args.bootstrap.read_text() + '\n' + '\n'.join((ROOT/'src/aim_webgui/static/css'/n).read_text() for n in ('tokens.css','bootstrap-overrides.css','aim.css','experience.css','credentials.css','evidence.css')) fixture_js = """ window.fetch = async function(url, options) { const r=await window._fixtureHTTP(url, options||{}); if(r.network_error) throw new TypeError('Synthetic lost response'); return new Response(r.body,{status:r.status,headers:{'Content-Type':'application/json'}}); }; window.EventSource=class { constructor(){} addEventListener(){} close(){} }; """ js='\n'.join((ROOT/'src/aim_webgui/static/js'/n).read_text() for n in ('theme.js','aim.js','experience.js','credentials.js','evidence.js')) content=content.replace('','').replace('','') page.set_content(content, wait_until='load') try: ident = make_job(fixture) mount('/jobs/'+ident) page.locator('[data-theme-option='+theme+']:visible').click() opener = page.locator('[data-credential-open]') opener.click() panel = page.locator('dialog [data-credential-panel]'); panel.wait_for() assert page.locator('dialog').evaluate('(e)=>e.open') assert page.locator('#credential-dialog-title').evaluate('(e)=>e===document.activeElement') assert page.locator('[data-key-source-choice]').is_visible() assert not page.locator('[name="ssh_key_passphrase"]').is_visible() assert not page.locator('dialog').evaluate('(e)=>e.scrollWidth>e.clientWidth+1') assert page.evaluate('document.documentElement.scrollWidth<=innerWidth') box=page.locator('dialog').bounding_box(); assert box['x']>=0 and box['y']>=0 and box['x']+box['width']<=width+1 and box['y']+box['height']<=height+1 # Native modal keeps background inert and cycles focus within dialog. for _ in range(12): page.keyboard.press('Tab') assert page.evaluate("document.querySelector('dialog').contains(document.activeElement)") page.locator('[name="vault_password"]').fill(SYNTHETIC) page.locator('button[aria-controls="credential-vault"]').click() assert page.locator('[name="vault_password"]').get_attribute('type')=='text' # Poll replacement cannot replace the modal or its typed/revealed input. page.evaluate("async (job)=>{const r=await fetch('/_partials/jobs/'+job);const html=await r.text();document.getElementById('job-status').outerHTML=html;document.dispatchEvent(new CustomEvent('htmx:afterSwap',{detail:{target:document.getElementById('job-status')}}));}", ident) assert page.locator('[name="vault_password"]').input_value()==SYNTHETIC page.locator('label[for="credential-key-separate"]').click() page.locator('[name="ssh_key_passphrase"]').fill('Synthetic separate key') page.locator('label[for="credential-key-vault"]').click() assert page.locator('[name="ssh_key_passphrase"]').input_value()=='' # Revealed input must be cleared on Escape, not just hidden visually. page.keyboard.press('Escape') assert not page.locator('dialog').evaluate('(e)=>e.open') assert page.locator('[data-credential-secret]').count()==0 assert page.locator('[data-credential-open]').evaluate('(e)=>e===document.activeElement') page.locator('[data-credential-open]').click(); panel.wait_for() assert page.locator('[name="vault_password"]').input_value()=='' if theme=='dark' and width in (390,1440): page.screenshot(path=str(args.out/f'credential-dialog-{width}.png'), full_page=False) start=len(packets) page.locator('[name="vault_password"]').fill(SYNTHETIC) page.locator('[data-credential-submit]').click() page.locator('[data-credential-feedback][data-tone="accepted"]').wait_for() assert len(packets)==start+1 and packets[-1]['credentials']['vault_password']==SYNTHETIC assert 'ssh_key_passphrase' not in packets[-1]['credentials'] assert page.locator('[name="vault_password"]').input_value()=='' assert 'accepted' in page.locator('[data-credential-feedback]').inner_text().lower() assert page.url=='about:blank' page.locator('[data-credential-dismiss]').click() assert not page.locator('dialog').evaluate('(e)=>e.open') # Attention action belongs to this viewer; choose only their credentials. ident2=make_job(fixture, requirements=['vault_password']) mount('/jobs') page.locator('#attention [data-job-id="'+ident2+'"]').click(); panel.wait_for() assert page.locator('[data-key-source-choice]').count()==0 # Lost acknowledgement: only one POST, then GET reconciliation; no replay. mode['drop_post']=True;mode['post_count']=0 page.locator('[name="vault_password"]').fill(SYNTHETIC) page.locator('[data-credential-submit]').click() page.wait_for_function("document.querySelector('[data-credential-feedback]').textContent.includes('not confirmed')") page.wait_for_timeout(3500) assert mode['post_count']==1 and page.locator('[name="vault_password"]').input_value()=='' page.locator('[data-credential-dismiss]').click() page.locator('#attention [data-job-id="'+ident2+'"]').click();panel.wait_for() page.wait_for_function("document.querySelector('[data-credential-feedback]').textContent.includes('not been confirmed')") assert page.locator('[data-credential-submit]').is_hidden() page.keyboard.press('Escape');mode['drop_post']=False # Server cancels while the form is open: input is cleared without POST. ident3=make_job(fixture) mount('/jobs/'+ident3);page.locator('[data-credential-open]').click();panel.wait_for() page.locator('[name="vault_password"]').fill(SYNTHETIC) with auth.store.transaction() as db: db.execute('UPDATE jobs SET cancel_requested=1 WHERE id=?',(ident3,)) page.wait_for_function("document.querySelector('[data-credential-feedback]').textContent.includes('cancellation')") assert page.locator('[name="vault_password"]').input_value()=='' page.keyboard.press('Escape') # Short visual viewport is internally scrollable; footer stays reachable. ident4=make_job(fixture) mount('/jobs/'+ident4);page.locator('[data-credential-open]').click();panel.wait_for() page.set_viewport_size({'width':width,'height':min(360,height)}) page.wait_for_timeout(100) assert page.locator('.credential-dialog-body').evaluate('(e)=>e.scrollHeight>e.clientHeight') page.locator('[data-credential-submit]').scroll_into_view_if_needed() box=page.locator('dialog').bounding_box();assert box['y']>=-1 and box['y']+box['height']<=min(360,height)+1 # Page hide clears revealed secrets too. BFCache style re-init never restores them. page.locator('[name="vault_password"]').fill(SYNTHETIC) page.locator('button[aria-controls="credential-vault"]').click() page.evaluate("window.dispatchEvent(new PageTransitionEvent('pagehide'))") assert page.locator('[data-credential-secret]').count()==0 assert not errors, errors cases.append({'width':width,'height':height,'theme':theme,'passed':True}) print('Credential QA passed',width,height,theme,flush=True) finally: context.close() # No-JS native markup and a plain HTTP form POST are verified separately. with auth.store.transaction() as db: db.execute('DELETE FROM rate_limits') token, session = auth.new_session(users['operator']) server.cookies.set(auth.settings.cookie_name, token) ident=make_job(fixture,requirements=['vault_password']) r=server.get('/jobs/'+ident+'/credentials') assert r.status_code==200 and 'method="post"' in r.text r=server.post('/jobs/'+ident+'/credentials', data={'_csrf':session['csrf'],'vault_password':SYNTHETIC},headers={'Origin':origin}) assert r.status_code==303 and r.headers['location']=='/jobs/'+ident cases.append({'plain_http_form_fallback':True,'passed':True}) finally: browser.close() report={'fixture_only':True,'cases':cases,'asgi_auth_csrf_endpoints':True,'browser_network':'about:blank fetch bridge to real TestClient; no real browser CSP/TLS/cookie transport qualification', 'bootstrap_source':str(args.bootstrap),'bootstrap_substitution':'5.3.6, production pin 5.3.8', 'htmx_loaded':False,'htmx_swap':'synthetic replacement + lifecycle event', 'real_worker_core_execution':False,'browser':'Chromium via Playwright'} (args.out/'results.json').write_text(json.dumps(report,indent=2)) print(len(cases),'browser scenario groups passed') if __name__=='__main__': main()