121 lines
6.1 KiB
Python
121 lines
6.1 KiB
Python
"""Exercise the real application with isolated data; never send notifications."""
|
|
import argparse
|
|
from datetime import date
|
|
import io
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import secrets
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import urllib.request
|
|
|
|
|
|
def run(source, artifacts=None):
|
|
import openpyxl
|
|
from PIL import Image
|
|
from waitress import create_server
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
|
|
password = 'Win7-Test-' + secrets.token_hex(10)
|
|
hashed = generate_password_hash(password)
|
|
assert check_password_hash(hashed, password), 'Password hashing failed'
|
|
# Also test the scrypt parameters generated by newer Windows deployments.
|
|
assert check_password_hash(generate_password_hash(password, method='scrypt:32768:8:1'), password)
|
|
Image.new('RGB', (3, 3)).save(io.BytesIO(), format='PNG')
|
|
|
|
with tempfile.TemporaryDirectory(prefix='po-win7-check-') as temporary:
|
|
os.environ['PO_DATA_DIR'] = temporary
|
|
os.environ.pop('PO_SECRET_KEY', None)
|
|
os.environ.pop('PO_HTTPS', None)
|
|
sys.path.insert(0, str(source.resolve()))
|
|
import app as module
|
|
from models import User
|
|
module.app.config['TESTING'] = True
|
|
assert (Path(temporary)/'instance/setup.token').is_file()
|
|
with module.Session() as session:
|
|
for name, role in [('maker', 'maker'), ('admin', 'admin')]:
|
|
session.add(User(username=name, name=name, role=role, active=True,
|
|
must_change_password=False, session_version=1, password_hash=hashed))
|
|
session.commit()
|
|
|
|
def post(client, path, data):
|
|
client.get('/login')
|
|
with client.session_transaction() as session:
|
|
token = session['csrf_token']
|
|
return client.post(path, data=dict(data, csrf_token=token))
|
|
|
|
clients = {}
|
|
for name in ('maker', 'admin'):
|
|
client = module.app.test_client()
|
|
assert post(client, '/login', {'username': name, 'password': password}).status_code == 302
|
|
clients[name] = client
|
|
|
|
def version(oid):
|
|
with module.Session() as session:
|
|
return session.get(module.Order, oid).row_version
|
|
|
|
def action(actor, oid, endpoint, **fields):
|
|
fields['row_version'] = str(version(oid))
|
|
return post(clients[actor], '/orders/%d/%s' % (oid, endpoint), fields)
|
|
|
|
try:
|
|
for kind in ('purchase', 'outsourcing'):
|
|
response = post(clients['maker'], '/orders/new',
|
|
{'doc_type': kind, 'request_key': secrets.token_hex(16)})
|
|
assert response.status_code == 302
|
|
oid = int(re.search(r'/orders/(\d+)', response.location).group(1))
|
|
response = action('maker', oid, 'edit', supplier_name='Win7 test supplier',
|
|
our_company='苏州博特维智能科技有限公司', payment_terms='月结',
|
|
delivery_place='苏州', order_date=date.today().isoformat(),
|
|
item_code='TEST-001', item_spec='测试物料', item_unit='件',
|
|
item_qty='2', item_price='12.3456', item_date=date.today().isoformat(),
|
|
item_remark='', intent='submit')
|
|
assert response.status_code == 302, response.get_data(as_text=True)
|
|
assert action('maker', oid, 'approve', stamped='1').status_code == 403
|
|
assert action('admin', oid, 'approve', stamped='1').status_code == 302
|
|
pdf = clients['maker'].get('/orders/%d/pdf' % oid).data
|
|
assert pdf.startswith(b'%PDF-') and b'%%EOF' in pdf
|
|
if artifacts:
|
|
artifacts.mkdir(parents=True, exist_ok=True)
|
|
(artifacts/(kind + '.pdf')).write_bytes(pdf)
|
|
for payment, delivery in [('partial', 'partial'), ('paid', 'delivered')]:
|
|
assert action('maker', oid, 'follow-up', needs_prepayment='1',
|
|
invoice_status='received', delivery_status=delivery,
|
|
fulfillment_note='离线环境自检', reason='').status_code == 302
|
|
assert action('admin', oid, 'payment', payment_status=payment,
|
|
payment_date=date.today().isoformat(), payment_note='', reason='').status_code == 302
|
|
assert clients['maker'].get('/orders/%d/pdf' % oid).data == pdf
|
|
print('PASS:', kind, 'login, submit, approval, PDF, partial/full follow-up, archive')
|
|
workbook = openpyxl.load_workbook(io.BytesIO(clients['admin'].get('/export/ledger').data))
|
|
assert workbook.active.max_row == 3
|
|
workbook.close()
|
|
assert clients['admin'].get('/orders?tracking=completed').status_code == 200
|
|
assert clients['admin'].get('/settings/notifications').status_code == 200
|
|
# Test a real HTTP socket using the exact Waitress compatibility wheel.
|
|
server = create_server(module.app, host='127.0.0.1', port=0, threads=2)
|
|
thread = threading.Thread(target=server.run, daemon=True)
|
|
thread.start()
|
|
try:
|
|
http = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
|
with http.open('http://127.0.0.1:%s/login' % server.effective_port, timeout=10) as response:
|
|
assert response.status == 200
|
|
response.read()
|
|
finally:
|
|
server.task_dispatcher.shutdown()
|
|
server.close()
|
|
thread.join(timeout=10)
|
|
print('PASS: Excel, dashboard, notification settings, Waitress HTTP')
|
|
finally:
|
|
module.engine.dispose()
|
|
print('SELFTEST PASSED. Only temporary data was used. No group messages sent.')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('source', type=Path)
|
|
parser.add_argument('--artifacts', type=Path)
|
|
args = parser.parse_args()
|
|
run(args.source, args.artifacts)
|