修复：收紧托盘业务权限与数据契约

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

@@ -125,7 +125,11 @@ class TestDaoPublicContract(unittest.TestCase):
             'get_shipment_for_update': ('db', 'shipment_id', 'warehouse_ids', 'site_ids'),
             'get_return_for_update': ('db', 'return_id', 'site_ids', 'warehouse_ids'),
             'get_active_filling_for_update': ('db', 'filling_id'),
-            'get_inbound_by_confirm_idempotency_key': ('db', 'idempotency_key'),
+            'get_inbound_by_confirm_idempotency_key': (
+                'db',
+                'idempotency_key',
+                'warehouse_ids',
+            ),
             'get_shipment_by_confirm_idempotency_key': (
                 'db',
                 'idempotency_key',
@@ -641,6 +645,30 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
                 self.assertIn(f"{model.__tablename__}.{key_column} = 'same-key'", sql)
                 self.assertNotIn('FOR UPDATE', sql)
 
+    async def test_inbound_confirm_key_pushes_warehouse_scope_into_sql(self) -> None:
+        self.assertIn(
+            'warehouse_ids',
+            inspect.signature(OperationDao.get_inbound_by_confirm_idempotency_key).parameters,
+        )
+        result = MagicMock()
+        result.scalar_one_or_none.return_value = None
+        self.db.execute.return_value = result
+
+        actual = await OperationDao.get_inbound_by_confirm_idempotency_key(
+            self.db,
+            'confirm-key',
+            warehouse_ids=[7, 3, 7],
+        )
+
+        self.assertIsNone(actual)
+        statement = self.db.execute.await_args.args[0]
+        for dialect in (mysql.dialect(), postgresql.dialect()):
+            with self.subTest(dialect=dialect.name):
+                sql = compile_sql(statement, dialect)
+                self.assertIn("biz_inbound_order.confirm_idempotency_key = 'confirm-key'", sql)
+                self.assertIn('biz_inbound_order.warehouse_id IN (3, 7)', sql)
+                self.assertNotIn('FOR UPDATE', sql)
+
     async def test_list_queries_apply_every_declared_filter_sort_and_pagination(self) -> None:
         cases = (
             (