304 lines
13 KiB
Python
304 lines
13 KiB
Python
"""
|
|
Tests for collection_protocol API/CRUD convention compliance.
|
|
|
|
Per AGENTS.md engineering standards:
|
|
1. _get_tenant_no() MUST NOT silently fall back to hardcoded "zkxw"
|
|
— must raise PermissionError when user_id missing or SysUser not found.
|
|
2. All list queries SHALL sort by order_num ASC, create_time DESC
|
|
(unless a business-specific sort is justified in the design doc).
|
|
3. list_audits, list_receipt_confirm_details, list_refund_confirm_details
|
|
currently only sort by create_time DESC — missing order_num ASC prefix.
|
|
"""
|
|
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch, call
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# H1: _get_tenant_no must raise PermissionError (not fall back to "zkxw")
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCollectionProtocolGetTenantNo(unittest.TestCase):
|
|
"""collection_protocol._get_tenant_no should match basic_info pattern."""
|
|
|
|
def test_raises_on_missing_user_id(self):
|
|
"""When user dict has no user_id, must raise PermissionError."""
|
|
from app.api.collection_protocol import _get_tenant_no
|
|
|
|
mock_db = MagicMock()
|
|
user = {} # no user_id key
|
|
with self.assertRaises(PermissionError):
|
|
_get_tenant_no(mock_db, user)
|
|
|
|
def test_raises_on_none_user_id(self):
|
|
"""When user_id is None, must raise PermissionError."""
|
|
from app.api.collection_protocol import _get_tenant_no
|
|
|
|
mock_db = MagicMock()
|
|
user = {"user_id": None}
|
|
with self.assertRaises(PermissionError):
|
|
_get_tenant_no(mock_db, user)
|
|
|
|
def test_raises_on_empty_user_id(self):
|
|
"""When user_id is empty string, must raise PermissionError."""
|
|
from app.api.collection_protocol import _get_tenant_no
|
|
|
|
mock_db = MagicMock()
|
|
user = {"user_id": ""}
|
|
with self.assertRaises(PermissionError):
|
|
_get_tenant_no(mock_db, user)
|
|
|
|
def test_raises_on_missing_sys_user(self):
|
|
"""When SysUser record not found in DB, must raise PermissionError."""
|
|
from app.api.collection_protocol import _get_tenant_no
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_db.query.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.first.return_value = None # SysUser not found
|
|
|
|
user = {"user_id": "some-id"}
|
|
with self.assertRaises(PermissionError):
|
|
_get_tenant_no(mock_db, user)
|
|
|
|
def test_raises_on_sys_user_no_tenant(self):
|
|
"""When SysUser exists but tenant_no is None, must raise PermissionError."""
|
|
from app.api.collection_protocol import _get_tenant_no
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_db.query.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_user = MagicMock()
|
|
mock_user.tenant_no = None
|
|
mock_query.first.return_value = mock_user
|
|
|
|
user = {"user_id": "some-id"}
|
|
with self.assertRaises(PermissionError):
|
|
_get_tenant_no(mock_db, user)
|
|
|
|
def test_returns_tenant_no_on_valid_user(self):
|
|
"""When SysUser has valid tenant_no, should return it."""
|
|
from app.api.collection_protocol import _get_tenant_no
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_db.query.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_user = MagicMock()
|
|
mock_user.tenant_no = "tenant-ABC"
|
|
mock_query.first.return_value = mock_user
|
|
|
|
result = _get_tenant_no(mock_db, {"user_id": "user-1"})
|
|
self.assertEqual(result, "tenant-ABC")
|
|
|
|
def test_does_not_return_hardcoded_zkxw(self):
|
|
"""Must never return the hardcoded fallback 'zkxw'."""
|
|
from app.api.collection_protocol import _get_tenant_no
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_db.query.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.first.return_value = None
|
|
|
|
user = {"user_id": "missing-user"}
|
|
with self.assertRaises(PermissionError) as ctx:
|
|
_get_tenant_no(mock_db, user)
|
|
# Ensure it raised, not returned "zkxw"
|
|
self.assertNotIsInstance(ctx.exception, type(None))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# M1: Sort order — list_audits should include order_num ASC
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestListAuditsSortOrder(unittest.TestCase):
|
|
"""list_audits should sort by order_num ASC, create_time DESC."""
|
|
|
|
@patch("app.crud.collection_protocol.BizProtocolAudit")
|
|
@patch("app.crud.collection_protocol.BizCollectionProtocol")
|
|
def test_list_audits_includes_order_num_in_sort(self, MockProtocol, MockAudit):
|
|
"""order_by should include order_num ASC before create_time DESC."""
|
|
from app.crud.collection_protocol import list_audits
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_query.join.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.order_by.return_value = mock_query
|
|
mock_query.with_entities.return_value = mock_query
|
|
mock_query.scalar.return_value = 0
|
|
mock_query.offset.return_value = mock_query
|
|
mock_query.limit.return_value = mock_query
|
|
mock_query.all.return_value = []
|
|
mock_db.query.return_value = mock_query
|
|
|
|
list_audits(mock_db, tenant_no="t1")
|
|
|
|
# Check that order_by was called at least once for the data query
|
|
# The sort call should include order_num ASC
|
|
order_by_calls = mock_query.order_by.call_args_list
|
|
self.assertGreaterEqual(len(order_by_calls), 1, "order_by should be called for data query")
|
|
|
|
# The data query's order_by should contain order_num.asc()
|
|
# We check the last order_by call (the data query, not the count query)
|
|
last_sort_call = order_by_calls[-1]
|
|
sort_args = last_sort_call[0] # positional args
|
|
# Should have at least 2 args: order_num ASC, create_time DESC
|
|
self.assertGreaterEqual(
|
|
len(sort_args), 2,
|
|
f"order_by should have at least 2 args (order_num ASC, create_time DESC), got: {sort_args}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# M1: Sort order — list_receipt_confirm_details should include order_num ASC
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestListReceiptConfirmSortOrder(unittest.TestCase):
|
|
"""list_receipt_confirm_details should sort by order_num ASC, create_time DESC."""
|
|
|
|
@patch("app.crud.collection_protocol.BizProtocolPaymentDetail")
|
|
@patch("app.crud.collection_protocol.BizProtocolPayment")
|
|
@patch("app.crud.collection_protocol.BizCollectionProtocol")
|
|
def test_receipt_confirm_includes_order_num_in_sort(self, MockProtocol, MockPayment, MockDetail):
|
|
"""order_by should include order_num ASC before create_time DESC."""
|
|
from app.crud.collection_protocol import list_receipt_confirm_details
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_query.join.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.order_by.return_value = mock_query
|
|
mock_query.with_entities.return_value = mock_query
|
|
mock_query.scalar.return_value = 0
|
|
mock_query.offset.return_value = mock_query
|
|
mock_query.limit.return_value = mock_query
|
|
mock_query.all.return_value = []
|
|
mock_db.query.return_value = mock_query
|
|
|
|
list_receipt_confirm_details(mock_db, tenant_no="t1")
|
|
|
|
order_by_calls = mock_query.order_by.call_args_list
|
|
self.assertGreaterEqual(len(order_by_calls), 1)
|
|
|
|
last_sort_call = order_by_calls[-1]
|
|
sort_args = last_sort_call[0]
|
|
self.assertGreaterEqual(
|
|
len(sort_args), 2,
|
|
f"order_by should have at least 2 args, got: {sort_args}"
|
|
)
|
|
|
|
|
|
class TestListReceiptConfirmedSortOrder(unittest.TestCase):
|
|
"""list_receipt_confirmed_details should sort by update_time DESC."""
|
|
|
|
@patch("app.crud.collection_protocol.BizProtocolPaymentDetail")
|
|
@patch("app.crud.collection_protocol.BizProtocolPayment")
|
|
@patch("app.crud.collection_protocol.BizCollectionProtocol")
|
|
def test_receipt_confirmed_sorts_by_update_time_desc(self, MockProtocol, MockPayment, MockDetail):
|
|
from app.crud.collection_protocol import list_receipt_confirmed_details
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_query.join.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.order_by.return_value = mock_query
|
|
mock_query.with_entities.return_value = mock_query
|
|
mock_query.scalar.return_value = 0
|
|
mock_query.offset.return_value = mock_query
|
|
mock_query.limit.return_value = mock_query
|
|
mock_query.all.return_value = []
|
|
mock_db.query.return_value = mock_query
|
|
|
|
list_receipt_confirmed_details(mock_db, tenant_no="t1")
|
|
|
|
order_by_calls = mock_query.order_by.call_args_list
|
|
self.assertGreaterEqual(len(order_by_calls), 1)
|
|
|
|
|
|
class TestListRefundConfirmedSortOrder(unittest.TestCase):
|
|
"""list_refund_confirmed_details should sort by update_time DESC."""
|
|
|
|
@patch("app.crud.collection_protocol.BizProtocolPaymentDetail")
|
|
@patch("app.crud.collection_protocol.BizProtocolPayment")
|
|
@patch("app.crud.collection_protocol.BizCollectionProtocol")
|
|
def test_refund_confirmed_sorts_by_update_time_desc(self, MockProtocol, MockPayment, MockDetail):
|
|
from app.crud.collection_protocol import list_refund_confirmed_details
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_query.join.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.order_by.return_value = mock_query
|
|
mock_query.with_entities.return_value = mock_query
|
|
mock_query.scalar.return_value = 0
|
|
mock_query.offset.return_value = mock_query
|
|
mock_query.limit.return_value = mock_query
|
|
mock_query.all.return_value = []
|
|
mock_db.query.return_value = mock_query
|
|
|
|
list_refund_confirmed_details(mock_db, tenant_no="t1")
|
|
|
|
order_by_calls = mock_query.order_by.call_args_list
|
|
self.assertGreaterEqual(len(order_by_calls), 1)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# M1: Sort order — list_refund_confirm_details should include order_num ASC
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestListRefundConfirmSortOrder(unittest.TestCase):
|
|
"""list_refund_confirm_details should sort by order_num ASC, create_time DESC."""
|
|
|
|
@patch("app.crud.collection_protocol.BizProtocolPaymentDetail")
|
|
@patch("app.crud.collection_protocol.BizProtocolPayment")
|
|
@patch("app.crud.collection_protocol.BizCollectionProtocol")
|
|
def test_refund_confirm_includes_order_num_in_sort(self, MockProtocol, MockPayment, MockDetail):
|
|
"""order_by should include order_num ASC before create_time DESC."""
|
|
from app.crud.collection_protocol import list_refund_confirm_details
|
|
|
|
mock_db = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_query.join.return_value = mock_query
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.order_by.return_value = mock_query
|
|
mock_query.with_entities.return_value = mock_query
|
|
mock_query.scalar.return_value = 0
|
|
mock_query.offset.return_value = mock_query
|
|
mock_query.limit.return_value = mock_query
|
|
mock_query.all.return_value = []
|
|
mock_db.query.return_value = mock_query
|
|
|
|
list_refund_confirm_details(mock_db, tenant_no="t1")
|
|
|
|
order_by_calls = mock_query.order_by.call_args_list
|
|
self.assertGreaterEqual(len(order_by_calls), 1)
|
|
|
|
last_sort_call = order_by_calls[-1]
|
|
sort_args = last_sort_call[0]
|
|
self.assertGreaterEqual(
|
|
len(sort_args), 2,
|
|
f"order_by should have at least 2 args, got: {sort_args}"
|
|
)
|
|
|
|
|
|
class TestNormalizeYyyymmddDate(unittest.TestCase):
|
|
def test_compact_and_dashed(self):
|
|
from app.crud.collection_protocol import _normalize_yyyymmdd_date
|
|
|
|
self.assertEqual(_normalize_yyyymmdd_date("20260506"), "20260506")
|
|
self.assertEqual(_normalize_yyyymmdd_date("2026-05-06"), "20260506")
|
|
self.assertIsNone(_normalize_yyyymmdd_date(""))
|
|
self.assertIsNone(_normalize_yyyymmdd_date(None))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|