"""Headless Edge checks against a temporary local server and temporary data only.""" import io import json import os import re import sys import tempfile import threading from pathlib import Path ROOT = Path(__file__).resolve().parents[1] OUT = ROOT / 'tools' / 'out' OUT.mkdir(exist_ok=True) RUN = Path(tempfile.mkdtemp(prefix='ui-check-', dir=OUT)) os.environ['PO_DATA_DIR'] = str(RUN) sys.path.insert(0, str(ROOT / 'po_app')) from app import app, Session from waitress.server import create_server from playwright.sync_api import sync_playwright import pymupdf import openpyxl from datetime import datetime, timedelta INITIAL = 'UI-test-initial-2026' PERSONAL = 'UI-test-personal-2026' def run(): server = create_server(app, host='127.0.0.1', port=0) thread = threading.Thread(target=server.run, daemon=True) thread.start() base = 'http://127.0.0.1:%s' % server.effective_port errors = [] summary = [] try: with sync_playwright() as p: browser = p.chromium.launch(channel='msedge', headless=True) def page(mobile=False): context = browser.new_context(viewport={'width':390,'height':844} if mobile else {'width':1440,'height':1080}, is_mobile=mobile, has_touch=mobile) tab = context.new_page() tab.on('pageerror', lambda e: errors.append(str(e))) tab.on('dialog', lambda d: d.accept()) return tab admin = page() admin.goto(base + '/setup') admin.locator('[name=setup_token]').fill((RUN / 'instance' / 'setup.token').read_text().strip()) admin.locator('[name=username]').fill('admin') admin.locator('[name=name]').fill('管理员') admin.locator('[name=password]').fill(PERSONAL) admin.get_by_role('button',name='创建管理员').click() admin.wait_for_url(base + '/') for username, role in [('maker','maker'),('approver','approver')]: admin.goto(base + '/users') admin.locator('[name=username]').fill(username) admin.locator('[name=name]').fill({'maker':'采购员','approver':'审批员'}[username]) admin.locator('[name=role]').select_option(role) admin.locator('[name=password]').fill(INITIAL) admin.get_by_role('button',name='保存用户').click() admin.wait_for_url(base + '/users') admin.screenshot(path=str(RUN / 'users.png'),full_page=True) def sign_in(username, destination): tab=page(mobile=username != 'maker'); tab.goto(destination) assert '/login' in tab.url if username != 'maker': tab.screenshot(path=str(RUN / (username+'-mobile-login.png'))) tab.locator('[name=username]').fill(username) tab.locator('[name=password]').fill(INITIAL) tab.get_by_role('button',name='登 录').click() tab.wait_for_url(re.compile(r'/account/password(?:\?|$)')) tab.locator('[name=old_password]').fill(INITIAL) tab.locator('[name=password]').fill(PERSONAL) tab.locator('[name=confirm_password]').fill(PERSONAL) tab.get_by_role('button',name='更新密码').click() tab.wait_for_url(destination) return tab accounts = {'maker': sign_in('maker', base + '/')} # A fake bot URL only: importing the app does not start the sender. admin.goto(base + '/settings/notifications') admin.locator('[name=base_url]').fill('http://192.168.1.10:8080') for channel in ('approval','result'): admin.locator('[name='+channel+'_webhook]').fill( 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=TEST-ONLY-NOT-A-REAL-KEY-'+channel) admin.locator('[name=enabled]').check() admin.get_by_role('button',name='保存通知配置').click() admin.wait_for_load_state('networkidle') assert admin.locator('[name=approval_webhook]').input_value() == '' admin.get_by_role('button',name='测试审批通知群',exact=True).click() admin.wait_for_load_state('networkidle') admin.screenshot(path=str(RUN/'notifications.png'), full_page=True) maker=accounts['maker'] for kind,count in [('purchase',2),('outsourcing',30)]: maker.goto(base + '/orders/new') maker.locator('[name=doc_type]').select_option(kind) maker.get_by_role('button',name='创建并填写明细').click() maker.wait_for_url('**/edit') edit_url=maker.url maker.locator('[name=supplier_name]').fill('测试供应商') maker.locator('[name=payment_terms]').fill('月结30天') for i in range(count): maker.get_by_role('button',name='+ 添加一行').click() maker.evaluate('''count => { document.querySelectorAll('#itemTable tr').forEach((tr,i) => { if (!i) return; const vals={code:'BROWSER-'+i,spec:'测试物料 A&B <规格>',unit:'件',qty:'1',price:'0.334',date:'2026-10-01',remark:''}; for (const [k,v] of Object.entries(vals)) tr.querySelector('[name=item_'+k+']').value=v; }); calc(); }''',count) assert maker.locator('#totalAmt').inner_text() == ('0.66' if count==2 else '9.90') maker.screenshot(path=str(RUN / (kind+'-edit.png')),full_page=True) maker.get_by_role('button',name='保存并提交审批').click() maker.wait_for_url('**/view') view_url=maker.url assert maker.locator('h1').inner_text().endswith('待审批') if 'approver' not in accounts: accounts['approver'] = sign_in('approver', view_url) approver=accounts['approver'];approver.goto(view_url) assert approver.evaluate('document.documentElement.scrollWidth <= innerWidth') approver.get_by_role('button',name='加盖采购专用章',exact=True).click() approver.locator('#apprConfirmBtn').scroll_into_view_if_needed() approver.screenshot(path=str(RUN / (kind+'-mobile-approve.png'))) approver.get_by_role('button',name='确认审批通过',exact=True).click() approver.wait_for_load_state('networkidle') assert '审批通过' in approver.locator('h1').inner_text() approver.screenshot(path=str(RUN / (kind+'-approved.png')),full_page=True) response=approver.request.get(view_url.replace('/view','/pdf')) assert response.status==200 data=response.body();(RUN/(kind+'.pdf')).write_bytes(data) document=pymupdf.open(stream=data,filetype='pdf') text='\n'.join(p.get_text() for p in document) assert '0.334' in text assert '待审批' not in text assert text.count('采购专用章') == 1 assert '审核专用章' not in text and '批准专用章' not in text for i,pdf_page in enumerate(document): pdf_page.get_pixmap(dpi=110).save(RUN/('%s-%s.png'%(kind,i+1))) for width in (320,390,768): approver.set_viewport_size({'width':width, 'height':844}) assert approver.evaluate('document.documentElement.scrollWidth <= innerWidth'), width for route in ('/orders', '/'): approver.goto(base + route) assert approver.evaluate('document.documentElement.scrollWidth <= innerWidth'), (width, route) approver.goto(view_url) approver.set_viewport_size({'width':390,'height':844}) summary.append({'kind':kind,'rows':count,'pdf_pages':len(document),'mobile_approval':True}) document.close() # Native Excel date is visible in the browser date input after upload. maker.goto(base + '/orders/new');maker.get_by_role('button',name='创建并填写明细').click() maker.wait_for_url('**/edit') workbook=openpyxl.Workbook();workbook.active.append(['物料代码','物料规格','单位','数量','单价','交期','备注']) workbook.active.append(['DATE','日期测试','件',1,2,datetime(2026,10,1),'']) out=io.BytesIO();workbook.save(out) maker.locator('[name=file]').set_input_files({'name':'采购明细.xlsx','mimeType':'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet','buffer':out.getvalue()}) maker.get_by_role('button',name='导入',exact=True).click() maker.wait_for_load_state('networkidle') assert maker.locator('[name=item_date]').input_value()=='2026-10-01' from notifications import dispatch_once from models import Notification delivered=[] for i in range(15): dispatch_once(Session, lambda *args: delivered.append(args), datetime.utcnow()+timedelta(seconds=i*5)) with Session() as db: assert db.query(Notification).filter_by(status='sent').count()==3 # Test + two results. assert db.query(Notification).filter_by(status='cancelled').count()==2 # Superseded tasks. assert len(delivered)==3 assert not errors, errors browser.close() (RUN/'summary.json').write_text(json.dumps(summary,ensure_ascii=False,indent=2),encoding='utf-8') print(json.dumps({'result':'passed','scenarios':summary,'output':str(RUN)},ensure_ascii=False)) finally: server.close() if __name__ == '__main__': run()