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('')[1].split('
')[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