shanxia2.0/backend/tests/test_crm_approval_reedit.py

"""再编辑并重新提审后，历史审批记录不得阻塞当前轮审批。"""
from unittest.mock import MagicMock, patch

from app.crud.crm.quote_approval import can_user_approve as quote_can_user_approve
from app.crud.crm.quote_approval import invalidate_all_approvals as quote_invalidate_all
from app.crud.crm.contract_approval import can_user_approve as contract_can_user_approve
from app.crud.crm.contract_approval import invalidate_all_approvals as contract_invalidate_all


def _approval(*, approver_id: str, result: str, step: int, is_current: int = 1, create_time=None):
    from datetime import datetime

    row = MagicMock()
    row.approver_id = approver_id
    row.approval_result = result
    row.step = step
    row.is_current = is_current
    row.del_flag = 0
    # 历史记录默认更早；pending 默认较新，便于按「当前轮」划分
    if create_time is not None:
        row.create_time = create_time
    elif result == "pending":
        row.create_time = datetime(2026, 8, 10, 12, 0, 0)
    else:
        row.create_time = datetime(2026, 8, 1, 12, 0, 0)
    return row


def test_quote_can_approve_prefers_current_pending_over_historical_approved():
    """再提审后同审批人同时存在历史已通过 + 新 pending 时，应允许审批。"""
    old_approved = _approval(approver_id="u1", result="approved", step=1)
    new_pending = _approval(approver_id="u1", result="pending", step=1)
    db = MagicMock()
    db.query.return_value.filter.return_value.order_by.return_value.all.return_value = [
        old_approved,
        new_pending,
    ]
    assert quote_can_user_approve(db, "quote-1", "u1") is True


def test_quote_can_approve_false_when_only_historical_approved():
    old_approved = _approval(approver_id="u1", result="approved", step=1)
    db = MagicMock()
    db.query.return_value.filter.return_value.order_by.return_value.all.return_value = [
        old_approved
    ]
    assert quote_can_user_approve(db, "quote-1", "u1") is False


def test_contract_can_approve_prefers_current_pending_over_historical_approved():
    old_approved = _approval(approver_id="u1", result="approved", step=1, is_current=0)
    new_pending = _approval(approver_id="u1", result="pending", step=1, is_current=1)
    db = MagicMock()
    db.query.return_value.filter.return_value.order_by.return_value.all.return_value = [
        old_approved,
        new_pending,
    ]
    assert contract_can_user_approve(db, "contract-1", "u1") is True


def test_quote_invalidate_all_soft_deletes_approved_and_pending():
    approved = _approval(approver_id="u1", result="approved", step=1)
    pending = _approval(approver_id="u2", result="pending", step=2)
    db = MagicMock()
    db.query.return_value.filter.return_value.all.return_value = [approved, pending]

    count = quote_invalidate_all(db, "quote-1")

    assert count == 2
    assert approved.del_flag == 1
    assert pending.del_flag == 1
    db.flush.assert_called_once()


def test_contract_invalidate_all_soft_deletes_approved_and_pending():
    approved = _approval(approver_id="u1", result="approved", step=1, is_current=0)
    pending = _approval(approver_id="u2", result="pending", step=2, is_current=1)
    db = MagicMock()
    db.query.return_value.filter.return_value.all.return_value = [approved, pending]

    count = contract_invalidate_all(db, "contract-1")

    assert count == 2
    assert approved.del_flag == 1
    assert pending.del_flag == 1
    db.flush.assert_called_once()


def test_quote_update_invalidates_all_approvals_on_reedit():
    """编辑保存回草稿时须作废全部历史审批（含已通过），不能只删 pending。"""
    from app.crud.crm import quote as quote_crud

    old = MagicMock()
    old.status = "accepted"
    old.quote_type = "sales"
    old.customer_id = "c1"
    old.supplier_id = "s1"
    old.quote_date = None
    old.validity_days = None
    old.quote_no = "Q1"
    old.quote_title = "t"
    old.customer_name = "客户"
    old.id = "quote-1"

    with patch.object(quote_crud, "_get_quote_in_scope", return_value=old), patch.object(
        quote_crud, "invalidate_all_approvals"
    ) as inv, patch.object(quote_crud, "assert_products_allowed_for_supplier"), patch.object(
        quote_crud, "calculate_quote_amount"
    ), patch.object(quote_crud, "list_quote_products", return_value=[]), patch.object(
        quote_crud, "_enrich_contact_fields"
    ), patch.object(quote_crud, "_enrich_supplier_name"), patch.object(
        quote_crud, "get_quote_detail", return_value={"id": "quote-1"}
    ), patch.object(quote_crud, "delete_pending_approvals") as del_pending:
        db = MagicMock()
        quote_crud.update_quote(
            db,
            "quote-1",
            tenant_no="T1",
            data={"quoteTitle": "new"},
            user_id="u1",
        )
        inv.assert_called_once_with(db, "quote-1")
        del_pending.assert_not_called()
        assert old.status == "draft"


def test_contract_update_invalidates_all_approvals_on_reedit():
    from app.crud.crm import contract as contract_crud

    old = MagicMock()
    old.status = "approved"
    old.contract_type = "sale"
    old.supplier_id = "s1"
    old.customer_id = "c1"
    old.customer_name = "客户"
    old.party_a = "甲方"
    old.party_b = "乙方"
    old.contract_name = "旧合同名"
    old.sign_date = None
    old.contract_no = "C1"
    old.id = "contract-1"

    with patch.object(contract_crud, "_get_contract_row_in_scope", return_value=old), patch.object(
        contract_crud, "invalidate_all_approvals"
    ) as inv, patch.object(contract_crud, "_validate_contract_type_and_products"), patch.object(
        contract_crud, "_normalize_framework_contract_data"
    ), patch.object(contract_crud, "_validate_product_prices_by_contract_type"), patch.object(
        contract_crud, "_validate_sale_purchase_contract_amount"
    ), patch.object(contract_crud, "_apply_contract_entity_fields"), patch.object(
        contract_crud, "assert_products_allowed_for_supplier"
    ), patch.object(contract_crud, "list_contract_products", return_value=[]), patch.object(
        contract_crud, "align_contract_no_to_sign_date", return_value="C1"
    ), patch.object(
        contract_crud, "_build_contract_name_from_parties", return_value="新合同名"
    ), patch.object(contract_crud, "get_contract_detail", return_value={"id": "contract-1"}), patch.object(
        contract_crud, "delete_pending_approvals"
    ) as del_pending, patch.object(contract_crud, "delete_products_by_contract_id"), patch.object(
        contract_crud, "batch_insert_contract_products"
    ):
        db = MagicMock()
        contract_crud.update_contract(
            db,
            "contract-1",
            tenant_no="T1",
            data={"contractName": "new"},
            user_id="u1",
        )
        inv.assert_called_once_with(db, "contract-1")
        del_pending.assert_not_called()
        assert old.status == "draft"
        assert old.approval_status is None