shanxia2.0/backend/tests/test_bx_child_differential_save.py

"""Child identity and in-flight workflow guards, using isolated SQLite only."""
from datetime import date, datetime
from decimal import Decimal

import pytest
from fastapi import HTTPException
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.core.bx_auth import BxUserContext
from app.models.bx_models import (
    BxAdvanceApply, BxAttachment, BxInvoice, BxReimburseApply,
    BxReimburseDetail, BxReimburseInvoice, BxWorkflowDef, BxWorkflowNode,
)
from app.models.pl_models import PlVehicleRechargeApply
from app.schemas.bx.reimburse import ReimburseDetailCreate, ReimburseInvoiceLinkIn
from app.schemas.bx.workflow import WorkflowNodeCreate
from app.services.bx.reimburse import ReimburseService
from app.services.bx.workflow import WorkflowService

USER = BxUserContext('u', 'user', 'dept', 'super_admin', 'tenant')
OLD = datetime(2020, 1, 1)


@pytest.fixture(autouse=True)
def forbid_business_database(monkeypatch):
    def forbidden(*args, **kwargs):
        raise AssertionError('business database connections are forbidden')
    monkeypatch.setattr('pymysql.connect', forbidden)


@pytest.fixture
def db():
    engine = create_engine('sqlite://')
    for model in (BxWorkflowDef, BxWorkflowNode, BxReimburseApply, BxAdvanceApply,
                  BxReimburseDetail, BxInvoice, BxAttachment, BxReimburseInvoice,
                  PlVehicleRechargeApply):
        model.__table__.create(engine)
    with sessionmaker(bind=engine)() as session:
        yield session
    engine.dispose()


def apply(db, id='a', tenant='tenant', status='draft'):
    row = BxReimburseApply(id=id, tenant_no=tenant, apply_code=id, title='custom',
        applicant_id='u', applicant_name='user', expense_id='e', expense_name='Expense', amount=30, status=status,
        reimburse_date=date(2026, 1, 1), wf_id='wf', create_by='u')
    db.add(row)
    db.flush()
    return row


def detail(db, id, parent='a', tenant='tenant', amount=10):
    row = BxReimburseDetail(id=id, reimburse_id=parent, expense_id='e',
        expense_name='Expense', amount=amount, remark='reason', occur_date=date(2026, 1, 1),
        tenant_no=tenant, create_by='original', create_time=OLD, update_time=OLD)
    db.add(row)
    db.flush()
    return row


def data(id=None, amount=10):
    return dict(id=id, expense_id='e', expense_name='Expense', amount=amount,
                remark='reason', occur_date=date(2026, 1, 1))


def test_detail_diff_preserves_identity_metadata_and_fields(db):
    a = apply(db)
    old = detail(db, 'd1')
    removed = detail(db, 'd2', amount=20)
    new_line = {**data(amount=7), 'expense_id': 'new-expense'}
    ReimburseService.update_with_details(db, a.id, {}, [data('d1', 11), new_line], user=USER)
    live = db.query(BxReimburseDetail).filter_by(del_flag=0).all()
    assert old in live and old.id == 'd1' and old.create_time == OLD and old.create_by == 'original'
    assert old.amount == 11 and old.occur_date == date(2026, 1, 1) and old.remark == 'reason'
    assert removed.del_flag == 1 and len(live) == 2 and a.amount == Decimal('18')


def test_detail_omitted_and_empty_have_distinct_meanings(db):
    a = apply(db)
    row = detail(db, 'd1')
    ReimburseService.update_with_details(db, a.id, {}, None, user=USER)
    assert row.del_flag == 0
    ReimburseService.update_with_details(db, a.id, {}, [], user=USER)
    db.refresh(row)
    assert row.del_flag == 1 and a.amount == 0


@pytest.mark.parametrize('bad', ['foreign_parent', 'foreign_tenant', 'missing', 'duplicate', 'deleted'])
def test_detail_bad_id_rejected_before_any_mutation(db, bad):
    a = apply(db)
    row = detail(db, 'd1')
    detail(db, 'foreign_parent', parent='other')
    detail(db, 'foreign_tenant', tenant='other')
    detail(db, 'deleted').del_flag = 1
    db.flush()
    payload = [data('d1'), data('d1' if bad == 'duplicate' else bad)]
    with pytest.raises(HTTPException):
        ReimburseService.update_with_details(db, a.id, {'remark': 'must not change'}, payload, user=USER)
    assert row.del_flag == 0 and a.remark is None
    assert db.query(BxReimburseDetail).count() == 4


def test_legacy_detail_same_payload_preserves_identity_and_creation(db):
    a = apply(db)
    row = detail(db, 'd1')
    for _ in range(2):
        ReimburseService.update_with_details(db, a.id, {}, [data()], user=USER)
    assert row.del_flag == 0 and row.create_by == 'original' and row.create_time == OLD
    assert db.query(BxReimburseDetail).count() == 1


@pytest.mark.parametrize('old_count,incoming_count', [(2, 1), (2, 2), (1, 2)])
def test_legacy_repeated_expense_is_rejected_before_any_write(db, old_count, incoming_count):
    a = apply(db)
    for index in range(old_count):
        detail(db, f'd{index}', amount=10 + index)
    payload = [data(amount=10 + index) for index in range(incoming_count)]
    with pytest.raises(HTTPException) as error:
        ReimburseService.update_with_details(db, a.id, {'remark': 'must not change'}, payload, user=USER)
    assert error.value.status_code == 400
    assert a.remark is None and not db.dirty and not db.new


@pytest.mark.parametrize('anonymous_first', [True, False])
def test_mixed_explicit_and_legacy_ids_match_remaining_unique_expense(db, anonymous_first):
    a = apply(db)
    first, second = detail(db, 'd1'), detail(db, 'd2', amount=20)
    payload = [data('d1', 11), data(amount=21)]
    if anonymous_first:
        payload.reverse()
    ReimburseService.update_with_details(db, a.id, {}, payload, user=USER)
    assert first.amount == 11 and second.amount == 21
    assert first.del_flag == second.del_flag == 0
    assert db.query(BxReimburseDetail).count() == 2


def test_explicit_existing_lines_allow_new_repeated_expense(db):
    a = apply(db)
    old = detail(db, 'd1')
    ReimburseService.update_with_details(db, a.id, {}, [data('d1'), data(amount=7)], user=USER)
    assert old.del_flag == 0
    assert db.query(BxReimburseDetail).filter_by(del_flag=0).count() == 2


@pytest.mark.parametrize('explicit_null', [False, True])
def test_detail_api_preserves_omitted_fields_but_applies_explicit_null(db, monkeypatch, explicit_null):
    from types import SimpleNamespace
    from app.api.bx.reimburse import _build_details_data
    a = apply(db)
    old = detail(db, 'd1')
    monkeypatch.setattr('app.api.bx.reimburse.ExpenseItemService.get_by_id',
                        lambda *args, **kwargs: SimpleNamespace(item_name='Expense'))
    fields = {'remark': None, 'occurDate': None} if explicit_null else {}
    body = ReimburseDetailCreate(id='d1', expenseItemId='e', amount=11, **fields)
    payload = _build_details_data(db, 'tenant', [body], user=USER)
    ReimburseService.update_with_details(db, a.id, {}, payload, user=USER)
    assert old.amount == 11 and old.create_time == OLD
    assert old.remark == (None if explicit_null else 'reason')
    assert old.occur_date == (None if explicit_null else date(2026, 1, 1))


def test_legacy_direct_patch_preserves_omitted_fields(db):
    a = apply(db)
    old = detail(db, 'd1')
    payload = dict(expense_id='e', expense_name='Expense', amount=11)
    ReimburseService.update_with_details(db, a.id, {}, [payload], user=USER)
    assert old.del_flag == 0 and old.amount == 11
    assert old.remark == 'reason' and old.occur_date == date(2026, 1, 1)


def test_detail_schema_retains_id():
    assert ReimburseDetailCreate(id='d1', expenseItemId='e', amount=10).model_dump()['id'] == 'd1'


def test_invoice_natural_key_diff_preserves_unchanged_links(db):
    apply(db)
    for id in ('i1', 'i2', 'i3'):
        db.add(BxInvoice(id=id, tenant_no='tenant', owner_user_id='u', owner_user_name='user', amount=10))
    old = BxReimburseInvoice(id='link1', reimburse_id='a', invoice_id='i1', tenant_no='tenant', create_time=OLD, create_by='original')
    removed = BxReimburseInvoice(id='link2', reimburse_id='a', invoice_id='i2', tenant_no='tenant')
    db.add_all([old, removed]); db.flush()
    ReimburseService.sync_reimburse_invoices(db, 'a', ['i1', 'i3'], 'tenant', USER, applicant_id='u')
    db.refresh(old); db.refresh(removed)
    assert old.del_flag == 0 and old.create_time == OLD and old.create_by == 'original'
    assert removed.del_flag == 1
    assert db.query(BxReimburseInvoice).count() == 3
    ReimburseService.sync_reimburse_invoices(db, 'a', ['i1', 'i3'], 'tenant', USER, applicant_id='u')
    assert db.query(BxReimburseInvoice).count() == 3


def workflow(db):
    db.add(BxWorkflowDef(id='wf', biz_type='reimburse', wf_name='Workflow', tenant_no='tenant'))
    row = BxWorkflowNode(id='n1', wf_id='wf', tenant_no='tenant', node_name='Review',
        node_order=1, approver_id='u', approver_name='User', action_type='approve',
        create_by='original', create_time=OLD, update_time=OLD)
    db.add(row); db.flush()
    return row


def node(id='n1', name='Review'):
    return dict(id=id, node_name=name, node_order=1, approver_id='u', approver_name='User', action_type='approve')


@pytest.mark.parametrize('with_id', [True, False])
def test_workflow_noop_preserves_node_even_with_pending(db, with_id):
    row = workflow(db); apply(db, status='pending')
    payload = node()
    if not with_id: payload.pop('id')
    saved = WorkflowService.save_nodes(db, 'wf', [payload], user=USER)
    assert saved[0].id == row.id == 'n1' and saved[0].create_time == OLD
    assert db.query(BxWorkflowNode).count() == 1


@pytest.mark.parametrize('kind', ['reimburse', 'advance'])
def test_workflow_change_rejected_for_pending_business(db, kind):
    row = workflow(db)
    if kind == 'reimburse': apply(db, status='pending')
    else:
        db.add(BxAdvanceApply(id='advance', apply_code='advance', applicant_id='u', applicant_name='user',
            amount=1, reason='test', status='pending', wf_id='wf', tenant_no='tenant'))
        db.flush()
    with pytest.raises(ValueError, match='审批'):
        WorkflowService.save_nodes(db, 'wf', [node(name='Changed')], user=USER)
    assert row.node_name == 'Review' and row.del_flag == 0


def test_workflow_diff_updates_existing_soft_deletes_removed(db):
    row = workflow(db)
    WorkflowService.save_nodes(db, 'wf', [node(name='Changed')], user=USER)
    assert row.node_name == 'Changed' and row.create_time == OLD
    WorkflowService.save_nodes(db, 'wf', [], user=USER)
    assert row.del_flag == 1 and db.query(BxWorkflowNode).count() == 1


@pytest.mark.parametrize('payload', [[node('missing')], [node(), node()], [node(), node(None)]])
def test_workflow_rejects_unknown_duplicate_identity_before_changes(db, payload):
    row = workflow(db)
    with pytest.raises(ValueError): WorkflowService.save_nodes(db, 'wf', payload, user=USER)
    assert row.del_flag == 0 and db.query(BxWorkflowNode).count() == 1


def test_workflow_input_schema_retains_node_id():
    assert WorkflowNodeCreate(**node()).model_dump()['id'] == 'n1'


def test_workflow_metadata_change_and_delete_blocked_when_pending(db):
    workflow(db); apply(db, status='pending')
    with pytest.raises(ValueError, match='审批'):
        WorkflowService.update(db, 'wf', {'status': 0}, user=USER)
    with pytest.raises(ValueError, match='审批'):
        WorkflowService.soft_delete(db, 'wf', user=USER)


def test_workflow_change_rejected_for_in_approval_recharge(db):
    row = workflow(db)
    db.get(BxWorkflowDef, 'wf').biz_type = 'vehicle_recharge'
    db.add(PlVehicleRechargeApply(id='r', apply_no='r', license_plate='test',
        recharge_type='etc', apply_date=date(2026, 1, 1), applicant_id='u',
        status='in_approval', wf_id='wf', current_node='n1', tenant_no='tenant'))
    db.flush()
    with pytest.raises(ValueError, match='审批'):
        WorkflowService.save_nodes(db, 'wf', [node(name='Changed')], user=USER)
    assert row.node_name == 'Review'


def test_workflow_api_returns_business_error_for_pending_change(db):
    from app.api.bx.workflow import update_workflow
    from app.schemas.bx.workflow import WorkflowDefUpdate
    workflow(db); apply(db, status='pending'); db.commit()
    with pytest.raises(HTTPException) as exc:
        update_workflow('wf', WorkflowDefUpdate(status=0), db=db, user=USER)
    assert exc.value.status_code == 400
    assert db.get(BxWorkflowDef, 'wf').status == 1


def test_create_rejects_source_detail_id_without_inserting_parent(db):
    with pytest.raises(HTTPException):
        ReimburseService.create_with_details(db, {}, [data('source-id')], user=USER)
    assert db.query(BxReimburseApply).count() == 0


def test_advance_submit_locks_definition_before_application(db, monkeypatch):
    from app.services.bx import advance as svc
    workflow(db)
    db.add(BxAdvanceApply(id='advance', apply_code='advance', applicant_id='u', applicant_name='user',
        reason='test', amount=1, status='draft', wf_id='wf', tenant_no='tenant'))
    db.flush()
    locks = []
    original_lock = WorkflowService.lock_definition
    original_get = svc.AdvanceService.get_by_id
    def lock(*args, **kwargs):
        locks.append('definition')
        return original_lock(*args, **kwargs)
    def get(*args, **kwargs):
        if kwargs.get('for_update'): locks.append('application')
        return original_get(*args, **kwargs)
    monkeypatch.setattr(WorkflowService, 'lock_definition', lock)
    monkeypatch.setattr(svc.AdvanceService, 'get_by_id', get)
    monkeypatch.setattr(svc, 'sync_approval_notifications', lambda *args, **kwargs: None)
    result = svc.AdvanceService.submit(db, 'advance', USER)
    assert result.status == 'pending'
    assert locks[:2] == ['definition', 'application']


@pytest.mark.parametrize('kind', ['advance', 'reimburse'])
def test_submit_rejects_empty_configured_workflow_before_pending_or_invoice_debit(db, monkeypatch, kind):
    from app.services.bx import advance as advance_module
    from app.services.bx import reimburse as reimburse_module
    workflow(db).del_flag = 1
    if kind == 'advance':
        row = BxAdvanceApply(id='advance', apply_code='advance', applicant_id='u', applicant_name='user',
            reason='test', amount=1, status='draft', wf_id='wf', tenant_no='tenant')
        db.add(row); db.flush()
        service = advance_module.AdvanceService
        monkeypatch.setattr(advance_module, 'sync_approval_notifications', lambda *args, **kwargs: None)
    else:
        row = apply(db)
        service = ReimburseService
        monkeypatch.setattr(service, '_has_debit_reimburse', lambda *args, **kwargs: True)
        monkeypatch.setattr(reimburse_module, 'sync_approval_notifications', lambda *args, **kwargs: None)
    with pytest.raises(HTTPException, match='节点'):
        service.submit(db, row.id, USER)
    assert row.status == 'draft' and row.current_node is None