功能：实现多托盘发货出库

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

@@ -3,6 +3,7 @@ import unittest
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
+from sqlalchemy.dialects import mysql, postgresql
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import InstrumentedAttribute
 from sqlalchemy.sql import Select, operators
@@ -88,6 +89,33 @@ class AccessServiceTest(unittest.IsolatedAsyncioTestCase):
 
         user_has_site.assert_awaited_once_with(db, 8, 7)
 
+    @patch('module_business.service.access_service.AccessDao.user_has_site', new_callable=AsyncMock)
+    async def test_locking_site_access_is_forwarded(self, user_has_site: AsyncMock) -> None:
+        parameter = inspect.signature(AccessService.assert_site_access).parameters.get('for_update')
+        self.assertIsNotNone(parameter)
+        self.assertIs(inspect.Parameter.KEYWORD_ONLY, parameter.kind)
+        self.assertIs(False, parameter.default)
+        user_has_site.return_value = True
+        user = SimpleNamespace(user=SimpleNamespace(user_id=8, admin=False))
+        db = AsyncMock(spec=AsyncSession)
+
+        await AccessService.assert_site_access(db, user, 7, for_update=True)
+
+        user_has_site.assert_awaited_once_with(db, 8, 7, for_update=True)
+
+    @patch('module_business.service.access_service.AccessDao.user_has_site', new_callable=AsyncMock)
+    async def test_admin_bypasses_locking_site_mapping(self, user_has_site: AsyncMock) -> None:
+        user = SimpleNamespace(user=SimpleNamespace(user_id=1, admin=True))
+
+        await AccessService.assert_site_access(
+            AsyncMock(spec=AsyncSession),
+            user,
+            7,
+            for_update=True,
+        )
+
+        user_has_site.assert_not_awaited()
+
     @patch('module_business.service.access_service.AccessDao.user_has_site', new_callable=AsyncMock)
     async def test_unmapped_site_is_forbidden(self, user_has_site: AsyncMock) -> None:
         user_has_site.return_value = False
@@ -176,8 +204,11 @@ class AccessDaoTest(unittest.IsolatedAsyncioTestCase):
                     await AccessDao.user_has_warehouse(db, 8, 99)
 
                 statement = db.execute.await_args.args[0]
-                sql = ' '.join(str(statement.compile(compile_kwargs={'literal_binds': True})).split())
-                self.assertIs(for_update, 'FOR UPDATE' in sql)
+                for dialect in (mysql.dialect(), postgresql.dialect()):
+                    sql = ' '.join(
+                        str(statement.compile(dialect=dialect, compile_kwargs={'literal_binds': True})).split()
+                    )
+                    self.assertIs(for_update, 'FOR UPDATE' in sql)
 
     async def test_user_has_site_returns_mapping_existence(self) -> None:
         for scalar, expected in ((8, True), (None, False)):
@@ -200,6 +231,30 @@ class AccessDaoTest(unittest.IsolatedAsyncioTestCase):
                     ),
                 )
 
+    async def test_user_has_site_current_read_is_optional_keyword_only_lock(self) -> None:
+        parameter = inspect.signature(AccessDao.user_has_site).parameters.get('for_update')
+        self.assertIsNotNone(parameter)
+        self.assertIs(inspect.Parameter.KEYWORD_ONLY, parameter.kind)
+        self.assertIs(False, parameter.default)
+
+        for for_update in (False, True):
+            with self.subTest(for_update=for_update):
+                result = MagicMock()
+                result.scalar_one_or_none.return_value = 8
+                db = self._db_with_result(result)
+
+                if for_update:
+                    await AccessDao.user_has_site(db, 8, 7, for_update=True)
+                else:
+                    await AccessDao.user_has_site(db, 8, 7)
+
+                statement = db.execute.await_args.args[0]
+                for dialect in (mysql.dialect(), postgresql.dialect()):
+                    sql = ' '.join(
+                        str(statement.compile(dialect=dialect, compile_kwargs={'literal_binds': True})).split()
+                    )
+                    self.assertIs(for_update, 'FOR UPDATE' in sql)
+
     async def test_list_user_warehouse_ids_returns_scalar_ids(self) -> None:
         result = MagicMock()
         result.scalars.return_value.all.return_value = (11, 13)