修复：收紧入库数据范围校验

shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py

@@ -121,7 +121,7 @@ class TestDaoPublicContract(unittest.TestCase):
 
     def test_operation_dao_critical_parameter_names_and_defaults_are_stable(self) -> None:
         expected = {
-            'get_inbound_for_update': ('db', 'inbound_id'),
+            'get_inbound_for_update': ('db', 'inbound_id', 'warehouse_ids'),
             'get_shipment_for_update': ('db', 'shipment_id'),
             'get_return_for_update': ('db', 'return_id'),
             'get_active_filling_for_update': ('db', 'filling_id'),
@@ -131,7 +131,7 @@ class TestDaoPublicContract(unittest.TestCase):
             'get_usage_by_idempotency_key': ('db', 'idempotency_key'),
             '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'),
+            'get_inbound_by_id': ('db', 'inbound_id', 'warehouse_ids'),
             'add_inbound': ('db', 'inbound'),
             'update_inbound': ('db', 'inbound'),
             'delete_inbound': ('db', 'inbound_id'),
@@ -164,6 +164,14 @@ class TestDaoPublicContract(unittest.TestCase):
         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_for_update', 'get_inbound_by_id'):
+            with self.subTest(method=method_name):
+                signature = inspect.signature(getattr(OperationDao, method_name))
+                warehouse_ids = signature.parameters.get('warehouse_ids')
+                self.assertIsNotNone(warehouse_ids)
+                if warehouse_ids is not None:
+                    self.assertIsNone(warehouse_ids.default)
+
         for method_name in (
             'get_filling_list',
             'get_shipment_list',
@@ -446,6 +454,30 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
                 )
                 self.assertIn('FOR UPDATE', item_sql)
 
+    async def test_inbound_for_update_applies_scope_before_locking_items(self) -> None:
+        self.assertIn(
+            'warehouse_ids',
+            inspect.signature(OperationDao.get_inbound_for_update).parameters,
+        )
+        parent_result = MagicMock()
+        parent_result.scalar_one_or_none.return_value = None
+        self.db.execute.return_value = parent_result
+
+        actual = await OperationDao.get_inbound_for_update(
+            self.db,
+            inbound_id=11,
+            warehouse_ids=[7, 3, 7],
+        )
+
+        self.assertIsNone(actual)
+        self.db.execute.assert_awaited_once()
+        parent_statement = self.db.execute.await_args.args[0]
+        for dialect in (mysql.dialect(), postgresql.dialect()):
+            with self.subTest(dialect=dialect.name):
+                sql = compile_sql(parent_statement, dialect)
+                self.assertIn('biz_inbound_order.warehouse_id IN (3, 7)', sql)
+                self.assertIn('FOR UPDATE', sql)
+
     async def test_idempotency_queries_use_the_exact_declared_column(self) -> None:
         cases = (
             (
@@ -753,6 +785,29 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
 
         self.db.execute.assert_awaited_once()
 
+    async def test_inbound_detail_empty_scope_is_false_and_skips_items(self) -> None:
+        self.assertIn(
+            'warehouse_ids',
+            inspect.signature(OperationDao.get_inbound_by_id).parameters,
+        )
+        parent_result = MagicMock()
+        parent_result.scalar_one_or_none.return_value = None
+        self.db.execute.return_value = parent_result
+
+        actual = await OperationDao.get_inbound_by_id(
+            self.db,
+            inbound_id=11,
+            warehouse_ids=[],
+        )
+
+        self.assertIsNone(actual)
+        self.db.execute.assert_awaited_once()
+        parent_statement = self.db.execute.await_args.args[0]
+        mysql_sql = compile_sql(parent_statement, mysql.dialect())
+        postgresql_sql = compile_sql(parent_statement, postgresql.dialect())
+        self.assertIn('false = 1', mysql_sql)
+        self.assertIn('false', postgresql_sql)
+
     async def test_simple_details_target_primary_keys(self) -> None:
         cases = (
             (OperationDao.get_filling_by_id, BizFillingRecord, 'filling_id', 14),