223 lines
12 KiB
Python
223 lines
12 KiB
Python
import io
|
||
import re
|
||
import secrets
|
||
from datetime import date, timedelta
|
||
|
||
import openpyxl
|
||
import pytest
|
||
|
||
from conftest import post
|
||
from test_workflow import action, approve, new, order, pending, save
|
||
|
||
|
||
def approved(app, clients):
|
||
oid = pending(app, clients)
|
||
approve(app, clients, oid)
|
||
return oid
|
||
|
||
|
||
def follow(app, clients, oid, actor='maker', **values):
|
||
fields = {'needs_prepayment': '1', 'invoice_status': 'received', 'delivery_status': 'partial',
|
||
'fulfillment_note': '已收第一批货'}
|
||
fields.update(values)
|
||
return action(app, clients[actor], oid, 'follow-up', **fields)
|
||
|
||
|
||
def pay(app, clients, oid, actor='approver', **values):
|
||
fields = {'payment_status': 'paid', 'payment_date': date.today().isoformat(), 'payment_note': '银行转账'}
|
||
fields.update(values)
|
||
return action(app, clients[actor], oid, 'payment', **fields)
|
||
|
||
|
||
def contains(client, key, number):
|
||
response = client.get('/orders?tracking=' + key)
|
||
assert response.status_code == 200
|
||
return number in response.get_data(as_text=True)
|
||
|
||
|
||
def test_follow_payment_dashboard_and_archived_pdf(app, clients):
|
||
oid = approved(app, clients)
|
||
pdf = clients['maker'].get('/orders/%s/pdf' % oid).data
|
||
assert follow(app, clients, oid).status_code == 302
|
||
with app.Session() as db:
|
||
number = db.get(app.Order, oid).order_no
|
||
for key in ('prepayment', 'invoice_due', 'partial_delivery', 'unpaid'):
|
||
assert contains(clients['maker'], key, number)
|
||
assert not contains(clients['other'], 'invoice_due', number)
|
||
assert pay(app, clients, oid, payment_status='partial').status_code == 302
|
||
assert contains(clients['admin'], 'partial_payment', number)
|
||
assert contains(clients['admin'], 'invoice_due', number)
|
||
assert pay(app, clients, oid, actor='admin').status_code == 302
|
||
assert not contains(clients['maker'], 'invoice_due', number)
|
||
assert not contains(clients['maker'], 'prepayment', number)
|
||
assert contains(clients['maker'], 'paid_undelivered', number)
|
||
assert follow(app, clients, oid, delivery_status='delivered').status_code == 302
|
||
assert contains(clients['approver'], 'completed', number)
|
||
assert not contains(clients['approver'], 'paid_undelivered', number)
|
||
assert clients['maker'].get('/orders/%s/pdf' % oid).data == pdf
|
||
dashboard = clients['admin'].get('/').get_data(as_text=True)
|
||
assert re.search(r'data-tracking="closed"[^>]*>查看已全部完成（1）', dashboard)
|
||
other_dashboard = clients['other'].get('/').get_data(as_text=True)
|
||
assert number not in other_dashboard
|
||
with app.Session() as db:
|
||
logs = db.get(app.Order, oid).logs
|
||
assert any(lg.user_name == 'maker' and '部分交货' in (lg.comment or '') for lg in logs)
|
||
assert any(lg.user_name == 'admin' and '部分付款 → 已付款' in (lg.comment or '') for lg in logs)
|
||
workbook = openpyxl.load_workbook(io.BytesIO(clients['admin'].get('/export/ledger').data))
|
||
headers, values = list(workbook.active.values)
|
||
row = dict(zip(headers, values))
|
||
assert row['付款状态'] == '已付款' and row['交货状态'] == '已交货' and row['收票状态'] == '已收票'
|
||
workbook.close()
|
||
|
||
|
||
@pytest.mark.parametrize('actor,endpoint,status', [
|
||
('maker', 'payment', 403), ('other', 'payment', 404),
|
||
('other', 'follow-up', 404), ('approver', 'follow-up', 403),
|
||
])
|
||
def test_tracking_role_and_order_scope(app, clients, actor, endpoint, status):
|
||
oid = approved(app, clients)
|
||
function = pay if endpoint == 'payment' else follow
|
||
assert function(app, clients, oid, actor=actor).status_code == status
|
||
with app.Session() as db:
|
||
row = db.get(app.Order, oid)
|
||
assert row.payment_status == 'unpaid' and row.invoice_status == 'pending'
|
||
|
||
|
||
def test_tracking_rejects_forged_fields_csrf_and_stale_page(app, clients):
|
||
oid = approved(app, clients)
|
||
version = order(app, oid)['version']
|
||
assert follow(app, clients, oid, payment_status='paid').status_code == 403
|
||
assert pay(app, clients, oid, delivery_status='delivered').status_code == 403
|
||
assert clients['approver'].post('/orders/%s/payment' % oid, data={'payment_status': 'paid'}).status_code == 400
|
||
assert follow(app, clients, oid).status_code == 302
|
||
assert post(clients['approver'], '/orders/%s/payment' % oid,
|
||
{'row_version': str(version), 'payment_status': 'paid', 'payment_date': date.today().isoformat()}).status_code == 409
|
||
with app.Session() as db:
|
||
assert db.get(app.Order, oid).payment_status == 'unpaid'
|
||
|
||
|
||
def test_tracking_requires_approval_and_valid_input(app, clients):
|
||
oid = new(app, clients['maker'])
|
||
assert follow(app, clients, oid).status_code == 403
|
||
assert pay(app, clients, oid).status_code == 403
|
||
assert save(app, clients['maker'], oid, intent='submit').status_code == 302
|
||
assert pay(app, clients, oid).status_code == 403
|
||
approve(app, clients, oid)
|
||
assert pay(app, clients, oid, payment_date='').status_code == 400
|
||
assert pay(app, clients, oid, payment_date=(date.today()+timedelta(days=1)).isoformat()).status_code == 400
|
||
assert pay(app, clients, oid, payment_status='invalid').status_code == 400
|
||
assert follow(app, clients, oid, delivery_status='invalid').status_code == 400
|
||
assert follow(app, clients, oid, fulfillment_note='x'*501).status_code == 400
|
||
assert action(app, clients['admin'], oid, 'void', comment='取消采购').status_code == 302
|
||
assert pay(app, clients, oid).status_code == 403
|
||
assert follow(app, clients, oid).status_code == 403
|
||
|
||
|
||
def test_corrections_require_reason_and_keep_audit(app, clients):
|
||
oid = approved(app, clients)
|
||
assert pay(app, clients, oid).status_code == 302
|
||
assert pay(app, clients, oid, payment_status='unpaid').status_code == 400
|
||
assert pay(app, clients, oid, payment_status='unpaid', reason='误登记，实际尚未转账').status_code == 302
|
||
assert follow(app, clients, oid, delivery_status='delivered').status_code == 302
|
||
assert follow(app, clients, oid, invoice_status='pending').status_code == 400
|
||
assert follow(app, clients, oid, invoice_status='pending', delivery_status='pending', reason='收票和收货录入有误').status_code == 302
|
||
with app.Session() as db:
|
||
row = db.get(app.Order, oid)
|
||
assert row.payment_status == 'unpaid' and row.payment_date == ''
|
||
assert row.delivery_status == row.invoice_status == 'pending'
|
||
assert any('误登记，实际尚未转账' in (log.comment or '') for log in row.logs)
|
||
|
||
|
||
def test_revision_requires_each_role_to_recheck_and_copy_starts_fresh(app, clients):
|
||
oid = approved(app, clients)
|
||
follow(app, clients, oid, delivery_status='delivered')
|
||
pay(app, clients, oid)
|
||
with app.Session() as db:
|
||
number = db.get(app.Order, oid).order_no
|
||
assert action(app, clients['maker'], oid, 'revise', comment='调整数量').status_code == 302
|
||
assert save(app, clients['maker'], oid, qty='5', intent='submit').status_code == 302
|
||
approve(app, clients, oid)
|
||
assert contains(clients['admin'], 'recheck', number)
|
||
assert not contains(clients['admin'], 'completed', number)
|
||
assert '订单内容已变更' in clients['maker'].get('/orders/%s/view' % oid).get_data(as_text=True)
|
||
assert follow(app, clients, oid, delivery_status='delivered').status_code == 302
|
||
assert contains(clients['admin'], 'recheck', number)
|
||
assert pay(app, clients, oid).status_code == 302
|
||
assert not contains(clients['admin'], 'recheck', number)
|
||
assert contains(clients['admin'], 'completed', number)
|
||
copied = action(app, clients['maker'], oid, 'copy', request_key=secrets.token_hex(16))
|
||
assert copied.status_code == 302
|
||
copy_id = int(re.search(r'/orders/(\d+)', copied.location)[1])
|
||
with app.Session() as db:
|
||
clone = db.get(app.Order, copy_id)
|
||
assert not clone.needs_prepayment
|
||
assert clone.invoice_status == clone.delivery_status == 'pending'
|
||
assert clone.payment_status == 'unpaid' and clone.payment_date == ''
|
||
|
||
|
||
def test_admin_can_follow_and_pay_and_invalid_filter_is_rejected(app, clients):
|
||
oid = approved(app, clients)
|
||
assert follow(app, clients, oid, actor='admin').status_code == 302
|
||
assert pay(app, clients, oid, actor='admin').status_code == 302
|
||
assert clients['admin'].get('/orders?tracking=invalid').status_code == 400
|
||
|
||
|
||
def test_action_queue_keeps_old_open_orders_and_unreceived_invoices(app, clients):
|
||
with app.Session() as db:
|
||
maker = db.query(app.User).filter_by(username='maker').one()
|
||
for i in range(16):
|
||
db.add(app.Order(order_no='QUEUE-%02d' % i, created_by=maker.id,
|
||
supplier_name='待办测试', status='approved', revision=1,
|
||
payment_status='paid', delivery_status='delivered',
|
||
invoice_status='received', fulfillment_revision=1, payment_revision=1))
|
||
db.flush()
|
||
rows = db.query(app.Order).order_by(app.Order.id).all()
|
||
rows[0].payment_status = 'unpaid'
|
||
rows[0].needs_prepayment = True
|
||
rows[1].invoice_status = 'pending'
|
||
rows[2].payment_revision = 0
|
||
rows[3].payment_status = 'unknown'
|
||
db.commit()
|
||
html = clients['maker'].get('/').get_data(as_text=True)
|
||
board = html.split('<table class="grid tracking-table order-cards">')[1].split('</table>')[0]
|
||
assert all('QUEUE-%02d' % i in board for i in range(4))
|
||
assert 'QUEUE-15' not in board
|
||
assert board.index('QUEUE-02') < board.index('QUEUE-00')
|
||
assert '跟进收票' in board and '变更前记录 · 待核对' in board
|
||
assert 'QUEUE-15' not in clients['maker'].get('/orders').get_data(as_text=True)
|
||
assert 'QUEUE-15' in clients['maker'].get('/orders?show_completed=1').get_data(as_text=True)
|
||
assert 'QUEUE-15' in clients['maker'].get('/orders?tracking=closed').get_data(as_text=True)
|
||
assert 'QUEUE-00' not in clients['other'].get('/').get_data(as_text=True)
|
||
assert clients['admin'].get('/?tracking=invalid').status_code == 400
|
||
|
||
|
||
@pytest.mark.parametrize('actor,kind,status', [
|
||
('maker', 'payment', 403), ('approver', 'follow-up', 403),
|
||
('other', 'follow-up', 404), ('maker', 'follow-up', 200),
|
||
('approver', 'payment', 200), ('admin', 'follow-up', 200),
|
||
])
|
||
def test_quick_editor_checks_permissions(app, clients, actor, kind, status):
|
||
oid = approved(app, clients)
|
||
for suffix in ('', '?fragment=1'):
|
||
assert clients[actor].get('/orders/%s/tracking/%s%s' % (oid, kind, suffix)).status_code == status
|
||
|
||
|
||
def test_quick_payment_preserves_concurrency_archive_and_latest_actor(app, clients):
|
||
oid = approved(app, clients)
|
||
pdf = clients['maker'].get('/orders/%s/pdf' % oid).data
|
||
version = order(app, oid)['version']
|
||
values = {'row_version': str(version), 'payment_status': 'partial',
|
||
'payment_date': date.today().isoformat(), 'payment_note': '已付首款，尾款待交齐'}
|
||
response = post(clients['approver'], '/orders/%s/payment' % oid, values,
|
||
headers={'X-Tracking-Modal': '1'})
|
||
assert response.status_code == 200 and response.json == {'ok': True}
|
||
stale = post(clients['approver'], '/orders/%s/payment' % oid, values,
|
||
headers={'X-Tracking-Modal': '1'})
|
||
assert stale.status_code == 409
|
||
assert follow(app, clients, oid).status_code == 302
|
||
html = clients['maker'].get('/orders/%s/view' % oid).get_data(as_text=True)
|
||
assert '最近登记：' in html and '最近跟进：' in html
|
||
assert '已付首款，尾款待交齐' in html and '已收第一批货' in html
|
||
assert html.index('付款与交货跟进') < html.index('单据信息')
|
||
assert clients['maker'].get('/orders/%s/pdf' % oid).data == pdf
|