212 lines
11 KiB
Python
212 lines
11 KiB
Python
import sqlite3
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
SCHEMA = Path(__file__).parent / 'fixtures' / 'legacy_schema.sql'
|
|
|
|
|
|
def test_v4_supplier_alias_upgrade_preserves_orders_and_creates_one_backup(tmp_path):
|
|
from models import Order, OrderVersion, Supplier, init_db
|
|
path = tmp_path / 'po.db'
|
|
engine, factory = init_db(str(path))
|
|
archived = b'%PDF-1.4 immutable before alias update'
|
|
with factory() as db:
|
|
db.add(Supplier(name='旧供应商', code='S01', contact='原联系人'))
|
|
db.add(Order(id=1, order_no='V4-ORDER', status='approved', revision=2,
|
|
payment_status='partial', invoice_status='received', delivery_status='delivered'))
|
|
db.flush()
|
|
db.add(OrderVersion(order_id=1, revision=2, content_json='{}', pdf_data=archived,
|
|
pdf_sha256='existing-digest', template_version='po-v4'))
|
|
db.commit()
|
|
engine.dispose()
|
|
with sqlite3.connect(path) as con:
|
|
con.execute('ALTER TABLE suppliers DROP COLUMN short_name')
|
|
con.execute('PRAGMA user_version=4')
|
|
old_orders = con.execute('SELECT * FROM orders').fetchall()
|
|
old_archives = con.execute('SELECT * FROM order_versions').fetchall()
|
|
engine, factory = init_db(str(path))
|
|
with factory() as db:
|
|
supplier = db.query(Supplier).one()
|
|
assert (supplier.name, supplier.code, supplier.contact, supplier.short_name) == ('旧供应商', 'S01', '原联系人', '')
|
|
engine.dispose()
|
|
with sqlite3.connect(path) as con:
|
|
assert con.execute('PRAGMA user_version').fetchone()[0] == 5
|
|
assert con.execute('SELECT * FROM orders').fetchall() == old_orders
|
|
assert con.execute('SELECT * FROM order_versions').fetchall() == old_archives
|
|
backups = list(tmp_path.glob('*.bak'))
|
|
assert len(backups) == 1
|
|
with sqlite3.connect(backups[0]) as con:
|
|
assert con.execute('PRAGMA user_version').fetchone()[0] == 4
|
|
assert 'short_name' not in {row[1] for row in con.execute('PRAGMA table_info(suppliers)')}
|
|
engine, _ = init_db(str(path))
|
|
engine.dispose()
|
|
assert len(list(tmp_path.glob('*.bak'))) == 1
|
|
|
|
|
|
def test_v3_tracking_upgrade_preserves_existing_state_and_archives(tmp_path):
|
|
import hashlib
|
|
from models import Order, OrderVersion, User, init_db
|
|
path = tmp_path / 'po.db'
|
|
engine, factory = init_db(str(path))
|
|
old_pdf = b'%PDF-1.4 immutable v3 archive'
|
|
digest = hashlib.sha256(old_pdf).hexdigest()
|
|
with factory() as db:
|
|
db.add(User(id=1, username='owner', name='Owner', role='admin', password_hash='hash',
|
|
session_version=7, must_change_password=False))
|
|
db.add(Order(id=1, order_no='V3-APPROVED', status='approved', revision=1, row_version=1))
|
|
db.add(Order(id=2, order_no='V3-PENDING', status='pending_approval', revision=1))
|
|
db.flush()
|
|
db.add(OrderVersion(order_id=1, revision=1, content_json='{}', pdf_data=old_pdf,
|
|
pdf_sha256=digest, template_version='po-v3'))
|
|
db.commit()
|
|
engine.dispose()
|
|
with sqlite3.connect(path) as con:
|
|
for field in ('needs_prepayment', 'invoice_status', 'delivery_status', 'payment_status',
|
|
'payment_date', 'fulfillment_note', 'payment_note', 'fulfillment_revision', 'payment_revision'):
|
|
con.execute('ALTER TABLE orders DROP COLUMN ' + field)
|
|
con.execute('PRAGMA user_version=3')
|
|
engine, factory = init_db(str(path))
|
|
with factory() as db:
|
|
row = db.get(Order, 1)
|
|
assert row.status == 'approved' and row.revision == row.row_version == 1
|
|
assert row.invoice_status == row.delivery_status == row.payment_status == 'unknown'
|
|
assert row.fulfillment_revision == row.payment_revision == -1
|
|
assert not row.fulfillment_stale and not row.payment_stale
|
|
pending_order = db.get(Order, 2)
|
|
assert pending_order.status == 'pending_approval' and pending_order.row_version == 1
|
|
assert db.get(User, 1).session_version == 7
|
|
archive = db.query(OrderVersion).one()
|
|
assert (archive.pdf_data, archive.pdf_sha256, archive.template_version) == (old_pdf, digest, 'po-v3')
|
|
engine.dispose()
|
|
backups = list(tmp_path.glob('*.bak'))
|
|
assert len(backups) == 1
|
|
with sqlite3.connect(backups[0]) as con:
|
|
assert con.execute('PRAGMA user_version').fetchone()[0] == 3
|
|
assert 'payment_status' not in {row[1] for row in con.execute('PRAGMA table_info(orders)')}
|
|
with sqlite3.connect(path) as con:
|
|
assert con.execute('PRAGMA user_version').fetchone()[0] == 5
|
|
assert con.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
|
|
engine, _ = init_db(str(path))
|
|
engine.dispose()
|
|
assert len(list(tmp_path.glob('*.bak'))) == 1
|
|
|
|
|
|
def legacy(path, duplicate=False):
|
|
with sqlite3.connect(path) as con:
|
|
con.executescript(SCHEMA.read_text(encoding='utf-8'))
|
|
con.execute("INSERT INTO users(id,username,password_hash,name,role,active) VALUES(1,'owner','old-hash','Owner','maker',1)")
|
|
con.execute("INSERT INTO orders(id,order_no,doc_type,status,total_amount,created_by) VALUES(1,'PO202609100001','purchase','approved',1,1)")
|
|
con.execute("INSERT INTO order_items(order_id,seq,qty,price,amount) VALUES(1,1,3,0.3333,1)")
|
|
con.execute("INSERT INTO materials(code,price) VALUES('MAT',12.3456)")
|
|
con.execute("INSERT INTO approval_logs(order_id,action,user_name) VALUES(1,'approved','Previous Manager')")
|
|
if duplicate:
|
|
con.execute("INSERT INTO orders(order_no,doc_type) VALUES('PO202609100001','purchase')")
|
|
|
|
|
|
def test_legacy_upgrade_preserves_history_and_fixed_amounts(tmp_path):
|
|
from models import Material, Order, User, init_db
|
|
path = tmp_path / 'po.db'
|
|
legacy(path)
|
|
engine, factory = init_db(str(path))
|
|
with factory() as db:
|
|
order = db.get(Order,1)
|
|
assert order.status == 'approved' and order.order_no == 'PO202609100001'
|
|
assert order.total_amount == Decimal('1')
|
|
assert order.items[0].price == Decimal('0.3333')
|
|
assert order.logs[0].user_name == 'Previous Manager'
|
|
assert db.get(Material,1).price == Decimal('12.3456')
|
|
assert db.get(User,1).must_change_password
|
|
engine.dispose()
|
|
backups = list(tmp_path.glob('*.bak'))
|
|
assert len(backups) == 1
|
|
with sqlite3.connect(backups[0]) as con:
|
|
assert con.execute('SELECT price FROM order_items').fetchone()[0] == 0.3333
|
|
assert 'price_units' not in {x[1] for x in con.execute('PRAGMA table_info(order_items)')}
|
|
again,_ = init_db(str(path))
|
|
again.dispose()
|
|
assert len(list(tmp_path.glob('*.bak'))) == 1
|
|
|
|
|
|
def test_duplicates_stop_migration_without_renumbering(tmp_path):
|
|
from models import init_db
|
|
path=tmp_path/'po.db'
|
|
legacy(path,duplicate=True)
|
|
with pytest.raises(RuntimeError,match='重复单号'):
|
|
init_db(str(path))
|
|
with sqlite3.connect(path) as con:
|
|
assert con.execute('SELECT COUNT(*) FROM orders').fetchone()[0] == 2
|
|
assert 'row_version' not in {x[1] for x in con.execute('PRAGMA table_info(orders)')}
|
|
|
|
|
|
def test_invalid_amount_rolls_back_whole_migration(tmp_path):
|
|
from models import init_db
|
|
path=tmp_path/'po.db'
|
|
legacy(path)
|
|
with sqlite3.connect(path) as con:
|
|
con.execute("UPDATE order_items SET price='Infinity'")
|
|
with pytest.raises(RuntimeError,match='无法迁移'):
|
|
init_db(str(path))
|
|
with sqlite3.connect(path) as con:
|
|
assert 'row_version' not in {x[1] for x in con.execute('PRAGMA table_info(orders)')}
|
|
assert con.execute('SELECT price FROM order_items').fetchone()[0] == 'Infinity'
|
|
|
|
|
|
@pytest.mark.parametrize('manager_hook', ['', 'manager-hook'])
|
|
def test_v2_merge_preserves_archives_and_migrates_pending_work(tmp_path, manager_hook):
|
|
import hashlib
|
|
from datetime import datetime
|
|
from models import (ApprovalLog, Notification, NotificationSettings, Order,
|
|
OrderVersion, User, init_db)
|
|
path = tmp_path / 'po.db'
|
|
engine, factory = init_db(str(path))
|
|
old_pdf = b'%PDF-1.4 old archived two-stamp file'
|
|
digest = hashlib.sha256(old_pdf).hexdigest()
|
|
with factory() as db:
|
|
db.add(User(id=1, username='r', password_hash='hash', name='Old Reviewer', role='reviewer',
|
|
session_version=4, must_change_password=False))
|
|
db.add(NotificationSettings(id=1, enabled=True, base_url='http://192.168.1.2:8080',
|
|
review_webhook='review-hook', approval_webhook=manager_hook))
|
|
for oid, status in [(1, 'pending_review'), (2, 'pending_approval'), (3, 'approved')]:
|
|
db.add(Order(id=oid, order_no='LEGACY-%s' % oid, status=status, revision=1,
|
|
reviewed_by='Old Reviewer' if oid > 1 else None,
|
|
reviewed_at=datetime(2026, 9, 16) if oid > 1 else None))
|
|
db.flush()
|
|
db.add(ApprovalLog(order_id=2, action='主管审核通过', user_name='Old Reviewer', revision=1,
|
|
from_status='pending_review', to_status='pending_approval'))
|
|
db.add(OrderVersion(order_id=3, revision=1, content_json='{}', pdf_data=old_pdf,
|
|
pdf_sha256=digest, template_version='po-v2'))
|
|
for oid, channel, status in [(1, 'review', 'pending_review'), (2, 'review', 'pending_review'),
|
|
(2, 'approval', 'pending_approval'), (3, 'approval', 'pending_approval')]:
|
|
db.add(Notification(order_id=oid, revision=1, channel=channel, event='old', expected_status=status))
|
|
db.add(Notification(order_id=3, revision=1, channel='review', event='old', status='sent'))
|
|
db.commit()
|
|
with engine.begin() as con:
|
|
con.exec_driver_sql('PRAGMA user_version=2')
|
|
engine.dispose()
|
|
engine, factory = init_db(str(path))
|
|
with factory() as db:
|
|
user = db.get(User, 1)
|
|
assert (user.role, user.session_version, user.must_change_password) == ('approver', 5, False)
|
|
assert db.get(Order, 1).status == db.get(Order, 2).status == 'pending_approval'
|
|
assert db.get(Order, 1).row_version == db.get(Order, 2).row_version == 2
|
|
assert db.get(Order, 2).reviewed_by == 'Old Reviewer'
|
|
assert db.query(ApprovalLog).one().action == '主管审核通过'
|
|
archive = db.query(OrderVersion).one()
|
|
assert (archive.pdf_data, archive.pdf_sha256, archive.template_version) == (old_pdf, digest, 'po-v2')
|
|
rows = db.query(Notification).order_by(Notification.id).all()
|
|
assert [n.status for n in rows] == ['pending', 'cancelled', 'pending', 'cancelled', 'sent']
|
|
assert rows[0].channel == 'approval' and rows[0].expected_status == 'pending_approval'
|
|
assert rows[4].channel == 'review' # Historical delivery remains untouched.
|
|
config = db.get(NotificationSettings, 1)
|
|
assert config.approval_webhook == (manager_hook or 'review-hook') and config.review_webhook == ''
|
|
engine.dispose()
|
|
assert len(list(tmp_path.glob('*.bak'))) == 1
|
|
again, factory = init_db(str(path))
|
|
with factory() as db:
|
|
assert db.get(User, 1).session_version == 5
|
|
assert db.get(Order, 1).row_version == 2
|
|
again.dispose()
|
|
assert len(list(tmp_path.glob('*.bak'))) == 1
|