280 lines
17 KiB
Python
280 lines
17 KiB
Python
"""Headless Edge checks against a temporary local server and temporary data only."""
|
||
import argparse
|
||
import hashlib
|
||
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 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(executable=None):
|
||
if executable:
|
||
from packaged_server import PackagedServer
|
||
server = PackagedServer(executable, RUN)
|
||
base = server.base
|
||
else:
|
||
from app import app, Session
|
||
from waitress.server import create_server
|
||
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 = []
|
||
archives = []
|
||
try:
|
||
if executable:
|
||
server.start()
|
||
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)
|
||
admin.goto(base + '/suppliers')
|
||
supplier_form = admin.locator('.card').first.locator('form')
|
||
supplier_form.locator('[name=name]').fill('测试供应商')
|
||
supplier_form.locator('[name=short_name]').fill('测试简称')
|
||
supplier_form.get_by_role('button', name='新增', exact=True).click()
|
||
admin.wait_for_url(base + '/suppliers')
|
||
admin.get_by_role('button', name='编辑', exact=True).click()
|
||
assert admin.locator('#edit_short_name').input_value() == '测试简称'
|
||
admin.locator('#edit_short_name').fill('测试供应商简称')
|
||
admin.locator('#editModal').get_by_role('button', name='保存', exact=True).click()
|
||
admin.wait_for_url(base + '/suppliers')
|
||
admin.screenshot(path=str(RUN / 'supplier-short-name.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 + '/')}
|
||
# The EXE starts a real sender. Keep notifications disabled there.
|
||
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)
|
||
if not executable:
|
||
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() == ''
|
||
if not executable:
|
||
admin.get_by_role('button',name='测试审批通知群',exact=True).click()
|
||
admin.wait_for_load_state('networkidle')
|
||
else:
|
||
assert not admin.locator('[name=enabled]').is_checked()
|
||
admin.screenshot(path=str(RUN/'notifications.png'), full_page=True)
|
||
for path, marker in [('/static/money.js', b'lineCents'),
|
||
('/static/style.css', b'seal-company'),
|
||
('/materials/import_template.xlsx', b'PK'),
|
||
('/suppliers/import_template.xlsx', b'PK'),
|
||
('/import_template.xlsx', b'PK')]:
|
||
response = admin.request.get(base + path)
|
||
assert response.status == 200 and marker in response.body(), path
|
||
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
|
||
from werkzeug.http import parse_options_header
|
||
filename = parse_options_header(response.headers['content-disposition'])[1]['filename']
|
||
assert filename.endswith('_测试供应商简称.pdf')
|
||
data=response.body();(RUN/(kind+'.pdf')).write_bytes(data)
|
||
archives.append((view_url.replace('/view','/pdf'), hashlib.sha256(data).hexdigest()))
|
||
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()
|
||
# Follow-up is independent of approval and the frozen PDF.
|
||
maker.goto(view_url)
|
||
assert maker.locator('#payment-form').count() == 0
|
||
maker.locator('[name=needs_prepayment]').select_option('1' if kind == 'purchase' else '0')
|
||
maker.locator('[name=invoice_status]').select_option('received')
|
||
maker.locator('[name=delivery_status]').select_option('partial' if kind == 'purchase' else 'pending')
|
||
maker.locator('[name=fulfillment_note]').fill('已收票；首批到货，跟进剩余交期' if kind == 'purchase' else '供应商备货中')
|
||
maker.get_by_role('button', name='保存采购跟进', exact=True).click()
|
||
maker.wait_for_load_state('networkidle')
|
||
approver.goto(view_url)
|
||
assert approver.locator('#follow-up-form').count() == 0
|
||
approver.locator('[name=payment_status]').select_option('partial' if kind == 'purchase' else 'paid')
|
||
approver.locator('[name=payment_date]').fill(datetime.now().strftime('%Y-%m-%d'))
|
||
approver.locator('[name=payment_note]').fill('已付预付款' if kind == 'purchase' else '款项全部付清')
|
||
approver.get_by_role('button', name='保存付款状态', exact=True).click()
|
||
approver.wait_for_load_state('networkidle')
|
||
approver.locator('#order-tracking').scroll_into_view_if_needed()
|
||
approver.screenshot(path=str(RUN / (kind+'-mobile-tracking.png')))
|
||
assert hashlib.sha256(approver.request.get(view_url.replace('/view','/pdf')).body()).hexdigest() == archives[-1][1]
|
||
admin.goto(base + '/')
|
||
for key in ('prepayment', 'invoice_due', 'paid_undelivered'):
|
||
assert admin.locator('[data-tracking="'+key+'"] .v').inner_text() == '1', key
|
||
assert admin.locator('#board-filter option[value="partial_delivery"]').inner_text() == '部分交货（1）'
|
||
admin.screenshot(path=str(RUN / 'tracking-dashboard.png'), full_page=True)
|
||
admin.locator('[data-tracking="paid_undelivered"]').click()
|
||
admin.wait_for_load_state('networkidle')
|
||
assert admin.locator('.tracking-table tbody tr').count() == 1
|
||
assert 'WO' in admin.locator('.tracking-table tbody tr').first.inner_text()
|
||
admin.screenshot(path=str(RUN / 'tracking-filter.png'), full_page=True)
|
||
for width in (320, 390, 768):
|
||
approver.set_viewport_size({'width':width, 'height':844})
|
||
for path in (base + '/', view_url + '#order-tracking'):
|
||
approver.goto(path)
|
||
assert approver.evaluate('document.documentElement.scrollWidth <= innerWidth'), (width, path)
|
||
approver.set_viewport_size({'width':390,'height':844})
|
||
approver.goto(base + '/')
|
||
approver.screenshot(path=str(RUN / 'tracking-dashboard-mobile.png'), full_page=True)
|
||
# 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'
|
||
response = admin.request.get(base + '/export/ledger')
|
||
assert response.status == 200
|
||
ledger = openpyxl.load_workbook(io.BytesIO(response.body()))
|
||
assert ledger.active.max_row == 4
|
||
ledger.close()
|
||
if executable:
|
||
import sqlite3
|
||
server.restart()
|
||
for url, expected in archives:
|
||
response = maker.request.get(url)
|
||
assert response.status == 200
|
||
assert hashlib.sha256(response.body()).hexdigest() == expected
|
||
with sqlite3.connect(RUN / 'instance' / 'po.db') as db:
|
||
assert db.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
|
||
assert db.execute('PRAGMA user_version').fetchone()[0] == 5
|
||
assert db.execute('SELECT count(*) FROM orders WHERE status = ?', ('approved',)).fetchone()[0] == 2
|
||
assert db.execute('SELECT count(*) FROM notifications').fetchone()[0] == 0
|
||
assert db.execute("SELECT count(*) FROM orders WHERE payment_status='paid' AND delivery_status='pending'").fetchone()[0] == 1
|
||
assert not (server.cwd / 'instance').exists()
|
||
else:
|
||
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')
|
||
result = {'result':'passed','scenarios':summary,'output':str(RUN),
|
||
'runtime': 'exe' if executable else 'source', 'tracking_dashboard': True}
|
||
if executable:
|
||
result.update(exe_sha256=hashlib.sha256(Path(executable).read_bytes()).hexdigest(),
|
||
restart_preserves_pdf=True, python_on_path=False, notifications_sent=0)
|
||
(RUN/'result.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf-8')
|
||
print(json.dumps(result,ensure_ascii=False))
|
||
finally:
|
||
server.close()
|
||
|
||
|
||
if __name__ == '__main__':
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument('--exe', type=Path, help='Test an isolated copy of this single-file EXE')
|
||
run(parser.parse_args().exe)
|