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 15:19:25 +08:00
parent 317a0fbf68
commit 6f22cf0282
5 changed files with 2106 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py
+96 -8
View File
@@ -156,8 +156,15 @@ class TestDaoPublicContract(unittest.TestCase):
signature = inspect.signature(getattr(OperationDao, method_name))
self.assertEqual(parameter_names, tuple(signature.parameters))
inbound_list_signature = inspect.signature(OperationDao.get_inbound_list)
self.assertEqual(
('db', 'query_object', 'is_page', 'warehouse_ids'),
tuple(inbound_list_signature.parameters),
)
self.assertIs(True, inbound_list_signature.parameters['is_page'].default)
self.assertIsNone(inbound_list_signature.parameters['warehouse_ids'].default)
for method_name in (
'get_inbound_list',
'get_filling_list',
'get_shipment_list',
'get_usage_list',
@@ -360,13 +367,6 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
async def test_for_update_queries_target_primary_keys_and_lock_rows(self) -> None:
cases = (
(
OperationDao.get_inbound_for_update,
BizInboundOrder,
'inbound_id',
11,
(),
),
(
OperationDao.get_shipment_for_update,
BizShipmentOrder,
@@ -410,6 +410,42 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
for fragment in extra_fragments:
self.assertIn(fragment, sql)
async def test_inbound_for_update_locks_parent_then_current_items(self) -> None:
inbound = SimpleNamespace(inbound_id=11)
items = [
SimpleNamespace(inbound_item_id=2, inbound_id=11, pallet_id=2),
SimpleNamespace(inbound_item_id=5, inbound_id=11, pallet_id=5),
]
inbound_result = MagicMock()
inbound_result.scalar_one_or_none.return_value = inbound
item_result = MagicMock()
item_result.scalars.return_value.all.return_value = items
self.db.execute.side_effect = [inbound_result, item_result]
actual = await OperationDao.get_inbound_for_update(self.db, inbound_id=11)
self.assertIs(inbound, actual)
self.assertEqual(items, actual.items)
self.assertEqual(2, self.db.execute.await_count)
parent_statement = self.db.execute.await_args_list[0].args[0]
item_statement = self.db.execute.await_args_list[1].args[0]
self.assertTrue(parent_statement.get_final_froms()[0].compare(BizInboundOrder.__table__))
self.assertTrue(item_statement.get_final_froms()[0].compare(BizInboundItem.__table__))
self.assertIs(True, parent_statement.get_execution_options().get('populate_existing'))
self.assertIs(True, item_statement.get_execution_options().get('populate_existing'))
for dialect in (mysql.dialect(), postgresql.dialect()):
with self.subTest(dialect=dialect.name):
parent_sql = compile_sql(parent_statement, dialect)
item_sql = compile_sql(item_statement, dialect)
self.assertIn('biz_inbound_order.inbound_id = 11', parent_sql)
self.assertIn('FOR UPDATE', parent_sql)
self.assertIn('biz_inbound_item.inbound_id = 11', item_sql)
self.assertIn(
'ORDER BY biz_inbound_item.pallet_id, biz_inbound_item.inbound_item_id',
item_sql,
)
self.assertIn('FOR UPDATE', item_sql)
async def test_idempotency_queries_use_the_exact_declared_column(self) -> None:
cases = (
(
@@ -588,6 +624,58 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
for fragment in fragments:
self.assertIn(fragment, sql)
async def test_inbound_list_applies_warehouse_scope_inside_paginated_statement(self) -> None:
query = InboundPageQueryModel(page_num=2, page_size=7)
marker = object()
with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as paginate:
actual = await OperationDao.get_inbound_list(
self.db,
query,
is_page=True,
warehouse_ids=[5, 2, 5],
)
self.assertIs(marker, actual)
statement = paginate.await_args.args[1]
for dialect in (mysql.dialect(), postgresql.dialect()):
with self.subTest(dialect=dialect.name):
sql = compile_sql(statement, dialect)
self.assertIn('biz_inbound_order.warehouse_id IN (2, 5)', sql)
self.assertIn('ORDER BY biz_inbound_order.inbound_id', sql)
async def test_inbound_list_empty_scope_compiles_to_false_for_both_dialects(self) -> None:
query = InboundPageQueryModel(page_num=4, page_size=6)
marker = object()
with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as paginate:
actual = await OperationDao.get_inbound_list(
self.db,
query,
is_page=True,
warehouse_ids=[],
)
self.assertIs(marker, actual)
statement = paginate.await_args.args[1]
mysql_sql = compile_sql(statement, mysql.dialect())
postgresql_sql = compile_sql(statement, postgresql.dialect())
self.assertIn('WHERE false = 1', mysql_sql)
self.assertIn('WHERE false', postgresql_sql)
self.assertEqual((query.page_num, query.page_size, True), paginate.await_args.args[2:])
async def test_inbound_list_none_scope_has_no_authorization_predicate(self) -> None:
query = InboundPageQueryModel()
with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=object())) as paginate:
await OperationDao.get_inbound_list(
self.db,
query,
is_page=True,
warehouse_ids=None,
)
sql = compile_sql(paginate.await_args.args[1])
self.assertNotIn('warehouse_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,