172 lines
6.9 KiB
Python
172 lines
6.9 KiB
Python
"""再编辑并重新提审后，历史审批记录不得阻塞当前轮审批。"""
|
|
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
|