order_system/tests/test_migration.py

import sqlite3
from decimal import Decimal
from pathlib import Path

import pytest

SCHEMA = Path(__file__).parent / 'fixtures' / 'legacy_schema.sql'


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] == 4
        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