admin_rb
137 lines
11 KiB
Python
137 lines
11 KiB
Python
"""Chromium fixture evidence QA. ASGI GET with simulated EventSource, no native tasks.
|
|
|
|
Default CSS is the explicitly identified Bootstrap 5.3.6 fixture substitute.
|
|
No fixture stylesheet is copied into the release. HTMX is deliberately unavailable
|
|
unless the operator supplies the pinned asset; this is not proxy/HTTPS acceptance.
|
|
"""
|
|
from pathlib import Path
|
|
import argparse,json,os,socket,sys,tempfile,time,re
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'src'))
|
|
from dataclasses import replace
|
|
from fastapi.testclient import TestClient
|
|
from playwright.sync_api import sync_playwright
|
|
from aim_webgui.app import create_app
|
|
from aim_webgui.auth.service import Auth
|
|
from aim_webgui.config import Settings
|
|
from aim_webgui.journal import Journal,project
|
|
from aim_webgui.reports import persist, TITLES
|
|
from evidence_fixtures import declared,sample,plan,result,job,ev
|
|
|
|
|
|
def main():
|
|
p=argparse.ArgumentParser();p.add_argument('--out',type=Path,required=True);p.add_argument('--bootstrap',type=Path,default=Path('/opt/imagemagick/share/doc/ImageMagick-7/www/assets/bootstrap.min.css'));args=p.parse_args();args.out.mkdir(parents=True,exist_ok=True)
|
|
with tempfile.TemporaryDirectory(prefix='aim-evidence-browser-')as td:
|
|
state=Path(td)/'state'
|
|
sock=socket.socket();sock.bind(('127.0.0.1',0));port=sock.getsockname()[1];sock.close()
|
|
url=f'http://127.0.0.1:{port}'
|
|
s=Settings(state_dir=state,public_url=url,core_transport='stdio');a=Auth(s);a.bootstrap()
|
|
with a.store.transaction()as db:
|
|
db.execute('UPDATE users SET must_change_password=0');uid=db.execute('SELECT id FROM users').fetchone()[0]
|
|
paths={};jobs={}
|
|
for name in TITLES:
|
|
d=declared(name);data=sample(d['data_schema'])
|
|
if name=='host_capabilities_v1':data['is_unifi_controller']=True
|
|
if name=='filesystem_usage_v1':
|
|
data={'platform':'linux','filesystems':[{'name':'/dev/synthetic1','mount':'/','filesystem_type':'ext4','used_bytes':42949672960,'total_bytes':128849018880,'available_bytes':85899345920,'used_percent':33.3,'status':'available'},{'name':'/dev/synthetic2','mount':'/srv/archive','filesystem_type':None,'used_bytes':None,'total_bytes':None,'available_bytes':None,'used_percent':None,'status':'unavailable'}]}
|
|
ident=job(a,uid,plan(d),status='successful');r=result(d,report_data=data)
|
|
with a.store.transaction()as db:
|
|
small=persist(db,s,ident,plan(d),r)
|
|
db.execute('UPDATE jobs SET core_result=?,finished_at=? WHERE id=?',(json.dumps(small),time.time(),ident))
|
|
paths[name]='/jobs/'+ident+'/reports';jobs[name]=ident
|
|
running=job(a,uid,plan(declared()),status='running');j=Journal(s);j.begin(running)
|
|
def add(start,end):j.append(running,[(*project(ev(i),{'test01.example'}),time.time())for i in range(start,end+1)])
|
|
add(1,300)
|
|
paths['progress']='/jobs/'+running
|
|
client=TestClient(create_app(s),base_url=url)
|
|
token,_=a.new_session(uid);client.cookies.set(s.cookie_name,token);records=[]
|
|
static=Path(__file__).resolve().parents[1]/'src/aim_webgui/static'
|
|
def html(path):
|
|
response=client.get(path);assert response.status_code==200,(path,response.text[:400])
|
|
text=response.text
|
|
def css(m):
|
|
name=m.group(1).split('?')[0]
|
|
source=args.bootstrap if 'bootstrap.min.css' in name else static/name.removeprefix('/static/')
|
|
return '<style>'+source.read_text()+'</style>'
|
|
def js(m):
|
|
name=m.group(1).split('?')[0]
|
|
if 'htmx.min.js'in name:return ''
|
|
source=static/name.removeprefix('/static/')
|
|
return '<script>'+source.read_text()+'</script>'
|
|
text=re.sub(r'<link[^>]+href="(/static/[^"]+\.css(?:\?[^"]*)?)"[^>]*>',css,text)
|
|
text=re.sub(r'<script[^>]+src="(/static/[^"]+\.js(?:\?[^"]*)?)"[^>]*></script>',js,text)
|
|
return text
|
|
def bridge_read(source,path):
|
|
response=client.get(path)
|
|
return {'status':response.status_code,'text':response.text}
|
|
fixture_js="""
|
|
window.__sources=[];
|
|
window.fetch=async function(url,opts){const r=await window.fixtureRead(String(url));return {ok:r.status>=200&&r.status<300,status:r.status,json:async()=>JSON.parse(r.text),text:async()=>r.text};};
|
|
window.EventSource=class{constructor(url){this.url=url;this.listeners={};window.__sources.push(this);setTimeout(()=>this.emit('open',{}),0);}addEventListener(k,f){this.listeners[k]=f;}close(){this.closed=true;}emit(k,d){if(this.listeners[k]&&!this.closed)this.listeners[k]({data:JSON.stringify(d)});}};
|
|
window.__emit=function(kind,data){window.__sources.forEach(s=>s.emit(kind,data));};
|
|
"""
|
|
def mount(page,path):
|
|
content=html(path)
|
|
page.set_content(content.replace('<head>','<head><script>'+fixture_js+'</script>'),wait_until='domcontentloaded')
|
|
def emit(page,start):
|
|
data=j.snapshot(uid,running,after=start)
|
|
page.evaluate('(d)=>window.__emit("snapshot",d)',data)
|
|
for row in data['events']:page.evaluate('(r)=>window.__emit("line",{record:r,text:r.text})',row)
|
|
try:
|
|
with sync_playwright()as p:
|
|
browser=p.chromium.launch(executable_path=os.environ.get('AIM_BROWSER_EXECUTABLE','/usr/bin/chromium'),args=['--no-sandbox'])
|
|
for w,h in ((320,750),(390,844),(760,900),(1440,1000),(740,390)):
|
|
for theme in ('light','dark'):
|
|
ctx=browser.new_context(viewport={'width':w,'height':h},reduced_motion='reduce')
|
|
ctx.add_cookies([{'name':s.cookie_name,'value':token,'url':url}])
|
|
ctx.add_init_script("localStorage.setItem('aim-web-display-theme',"+json.dumps(theme)+");")
|
|
page=ctx.new_page();page.expose_binding("fixtureRead",bridge_read);errors=[];page.on('pageerror',lambda e:errors.append(str(e)))
|
|
for name,path in paths.items():
|
|
print('Browser',name,w,h,theme,flush=True)
|
|
mount(page,path);page.evaluate("(t)=>document.querySelectorAll('[data-theme-option]').forEach(b=>{if(b.dataset.themeOption===t)b.click()})",theme);page.wait_for_timeout(120)
|
|
assert page.locator('h1').count()==1,(name,page.content()[:300])
|
|
assert page.evaluate('document.documentElement.scrollWidth<=innerWidth+1'),(name,w,h,'horizontal overflow')
|
|
if name=='progress':
|
|
page.wait_for_function("document.querySelector('[data-journal-output]').textContent.includes('Synthetic task 300')")
|
|
box=page.locator('[data-journal-output]');height=box.evaluate('(e)=>e.clientHeight')
|
|
assert height<h and box.evaluate('(e)=>e.scrollHeight>e.clientHeight')
|
|
else:
|
|
box=page.locator('[data-report-json]');box.locator('summary').click();page.wait_for_function("document.querySelector('[data-json-status]').textContent.includes('Retained')")
|
|
assert page.locator('[data-json-output]').inner_text() not in ('','null'),name
|
|
if name=='checkmk_user_config_v1':assert 'full configuration sections were not retained' in page.locator('body').inner_text()
|
|
records.append({'page':name,'width':w,'height':h,'theme':theme,'passed':True})
|
|
if (w==1440 and theme=='dark' and name in ('host_capabilities_v1','filesystem_usage_v1')) or(w==390 and theme=='dark' and name in ('host_capabilities_v1','progress')):
|
|
page.evaluate('window.scrollTo(0,0)');page.wait_for_timeout(80);page.screenshot(path=str(args.out/f'{name}-{w}-{theme}.png'),full_page=True)
|
|
assert not errors,errors
|
|
ctx.close()
|
|
# Two independent browser sessions, HTTP snapshots and real EventSource.
|
|
ctx1=browser.new_context();ctx2=browser.new_context()
|
|
for ctx in (ctx1,ctx2):ctx.add_cookies([{'name':s.cookie_name,'value':token,'url':url}])
|
|
one=ctx1.new_page();two=ctx2.new_page()
|
|
for page in(one,two):page.expose_binding('fixtureRead',bridge_read)
|
|
for page in (one,two):mount(page,paths['progress'])
|
|
for page in (one,two):page.wait_for_function("document.querySelector('[data-journal-output]').textContent.includes('Synthetic task 300')")
|
|
add(301,310)
|
|
for page in(one,two):emit(page,300)
|
|
for page in(one,two):page.wait_for_function("document.querySelector('[data-journal-output]').textContent.includes('Synthetic task 310')")
|
|
mount(one,paths['progress']);one.wait_for_function("document.querySelector('[data-journal-output]').textContent.includes('Synthetic task 310')")
|
|
add(311,320)
|
|
for page in(one,two):emit(page,310)
|
|
for page in(one,two):page.wait_for_function("document.querySelector('[data-journal-output]').textContent.includes('Synthetic task 320')")
|
|
assert one.locator('[data-journal-output]').inner_text().count('Synthetic task 320')==1
|
|
# Manual upward scrolling pauses following without moving the page.
|
|
one.locator('[data-journal-output]').evaluate('(e)=>{e.scrollTop=0;e.dispatchEvent(new Event("scroll"));}')
|
|
assert not one.locator('[data-journal-follow]').is_checked()
|
|
position=one.evaluate('scrollY');add(321,321);emit(one,320);emit(two,320);one.wait_for_timeout(700)
|
|
assert one.evaluate('scrollY')==position
|
|
with a.store.transaction()as db:db.execute("UPDATE jobs SET status='successful',finished_at=? WHERE id=?",(time.time(),running))
|
|
for page in(one,two):page.evaluate('()=>window.__emit("end",{cursor:321,job_status:"successful"})')
|
|
for page in(one,two):page.wait_for_function("document.querySelector('[data-journal-state]').textContent.includes('Ended')")
|
|
mount(one,paths['progress']);one.wait_for_function("document.querySelector('[data-journal-output]').textContent.includes('Synthetic task 321')")
|
|
records.append({'scenario':'two fixture views remount replay final reread internal follow','passed':True})
|
|
ctx1.close();ctx2.close();browser.close()
|
|
finally:client.close()
|
|
(args.out/'results.json').write_text(json.dumps({'fixture':'ASGI-backed reads, synthetic EventSource/remount; Bootstrap5.3.6 substitute; no HTMX/HTTPS/native operations','results':records},indent=2))
|
|
print(len(records),'browser cases passed')
|
|
|
|
if __name__=='__main__':main()
|