niuniu/ShiErFeng
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

功能：实现客户现场使用记录

Browse Source
This commit is contained in:
焦龙言
2026-07-26 21:14:01 +08:00
parent 3076194328
commit bc774c74e6
7 changed files with 1968 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py
+84 -5
View File
@@ -133,7 +133,7 @@ class TestDaoPublicContract(unittest.TestCase):
'site_ids',
),
'get_filling_by_idempotency_key': ('db', 'idempotency_key', 'warehouse_ids'),
'get_usage_by_idempotency_key': ('db', 'idempotency_key'),
'get_usage_by_idempotency_key': ('db', 'idempotency_key', 'site_ids'),
'get_return_by_request_idempotency_key': ('db', 'idempotency_key'),
'get_return_by_confirm_idempotency_key': ('db', 'idempotency_key'),
'get_inbound_by_id': ('db', 'inbound_id', 'warehouse_ids'),
@@ -149,7 +149,7 @@ class TestDaoPublicContract(unittest.TestCase):
'update_shipment': ('db', 'shipment'),
'delete_shipment': ('db', 'shipment_id'),
'replace_shipment_items': ('db', 'shipment_id', 'items'),
'get_usage_by_id': ('db', 'usage_id'),
'get_usage_by_id': ('db', 'usage_id', 'site_ids'),
'add_usage_record': ('db', 'usage'),
'get_return_by_id': ('db', 'return_id'),
'add_return_order': ('db', 'return_order'),
@@ -209,11 +209,25 @@ class TestDaoPublicContract(unittest.TestCase):
self.assertIsNone(shipment_list_signature.parameters['warehouse_ids'].default)
self.assertIsNone(shipment_list_signature.parameters['site_ids'].default)
for method_name in ('get_usage_list', 'get_return_list'):
usage_list_signature = inspect.signature(OperationDao.get_usage_list)
self.assertEqual(
('db', 'query_object', 'is_page', 'site_ids'),
tuple(usage_list_signature.parameters),
)
self.assertIs(True, usage_list_signature.parameters['is_page'].default)
self.assertIsNone(usage_list_signature.parameters['site_ids'].default)
for method_name in (
'get_usage_by_idempotency_key',
'get_usage_by_id',
):
with self.subTest(method=method_name):
signature = inspect.signature(getattr(OperationDao, method_name))
self.assertEqual(('db', 'query_object', 'is_page'), tuple(signature.parameters))
self.assertIs(True, signature.parameters['is_page'].default)
self.assertIsNone(signature.parameters['site_ids'].default)
return_list_signature = inspect.signature(OperationDao.get_return_list)
self.assertEqual(('db', 'query_object', 'is_page'), tuple(return_list_signature.parameters))
self.assertIs(True, return_list_signature.parameters['is_page'].default)
def test_dao_source_never_controls_transaction_boundary(self) -> None:
for module in (inventory_dao_module, operation_dao_module):
@@ -864,6 +878,71 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
self.assertIn("biz_shipment_order.confirm_idempotency_key = 'confirm-key'", key_sql)
self.assertNotIn('FOR UPDATE', key_sql)
async def test_usage_reads_push_site_scope_and_empty_scope_is_always_false(self) -> None:
query = UsagePageQueryModel(page_num=2, page_size=7)
marker = object()
with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as paginate:
actual = await OperationDao.get_usage_list(
self.db,
query,
is_page=True,
site_ids=[11, 9, 11],
)
self.assertIs(marker, actual)
list_statement = paginate.await_args.args[1]
scalar_result = MagicMock()
scalar_result.scalar_one_or_none.return_value = None
self.db.execute.return_value = scalar_result
await OperationDao.get_usage_by_id(self.db, 15, site_ids=[11, 9, 11])
detail_statement = self.db.execute.await_args.args[0]
self.db.reset_mock()
self.db.execute.return_value = scalar_result
await OperationDao.get_usage_by_idempotency_key(
self.db,
'usage-key',
site_ids=[11, 9, 11],
)
key_statement = self.db.execute.await_args.args[0]
for dialect in (mysql.dialect(), postgresql.dialect()):
with self.subTest(dialect=dialect.name):
for statement in (list_statement, detail_statement, key_statement):
sql = compile_sql(statement, dialect)
self.assertIn('biz_usage_record.site_id IN (9, 11)', sql)
with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as empty_paginate:
await OperationDao.get_usage_list(self.db, query, site_ids=[])
empty_list = empty_paginate.await_args.args[1]
self.db.reset_mock()
self.db.execute.return_value = scalar_result
await OperationDao.get_usage_by_id(self.db, 15, site_ids=[])
empty_detail = self.db.execute.await_args.args[0]
self.db.reset_mock()
self.db.execute.return_value = scalar_result
await OperationDao.get_usage_by_idempotency_key(self.db, 'usage-key', site_ids=[])
empty_key = self.db.execute.await_args.args[0]
for dialect in (mysql.dialect(), postgresql.dialect()):
with self.subTest(dialect=f'{dialect.name}-empty'):
for statement in (empty_list, empty_detail, empty_key):
sql = compile_sql(statement, dialect)
self.assertTrue('false' in sql.lower() or '0 = 1' in sql, sql)
async def test_usage_none_scope_adds_no_authorization_predicate(self) -> None:
query = UsagePageQueryModel()
marker = object()
with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as paginate:
await OperationDao.get_usage_list(self.db, query, site_ids=None)
sql = compile_sql(paginate.await_args.args[1])
self.assertNotIn('site_id IN', sql)
self.assertNotIn('WHERE false', sql)
def test_time_conditions_include_fractional_last_second_with_exclusive_next_day(self) -> None:
lower, upper = OperationDao._time_conditions(
BizUsageRecord.occurred_at,