安全：隔离报表选项扫码与首页数据

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

@@ -11,6 +11,7 @@ from sqlalchemy.sql.elements import BinaryExpression, BindParameter, BooleanClau
 
 from exceptions.exception import PermissionException
 from module_business.dao.access_dao import AccessDao
+from module_business.domain.test_batch_scope import TestBatchScope
 from module_business.entity.do.master_do import BizUserSite, BizUserWarehouse
 from module_business.service.access_service import AccessService
 
@@ -260,32 +261,28 @@ class AccessDaoTest(unittest.IsolatedAsyncioTestCase):
         result.scalars.return_value.all.return_value = (11, 13)
         db = self._db_with_result(result)
 
-        warehouse_ids = await AccessDao.list_user_warehouse_ids(db, user_id=8)
+        warehouse_ids = await AccessDao.list_user_warehouse_ids(db, user_id=8, scope=TestBatchScope.formal())
 
         self.assertEqual([11, 13], warehouse_ids)
         statement = db.execute.await_args.args[0]
-        self._assert_statement_targets(
-            statement,
-            BizUserWarehouse,
-            BizUserWarehouse.warehouse_id,
-            ((BizUserWarehouse.user_id, 8),),
-        )
+        sql = ' '.join(str(statement.compile(compile_kwargs={'literal_binds': True})).split())
+        self.assertIn('JOIN biz_warehouse', sql)
+        self.assertIn('biz_user_warehouse.user_id = 8', sql)
+        self.assertIn('biz_warehouse.test_batch_id IS NULL', sql)
 
     async def test_list_user_site_ids_returns_scalar_ids(self) -> None:
         result = MagicMock()
         result.scalars.return_value.all.return_value = (3, 5)
         db = self._db_with_result(result)
 
-        site_ids = await AccessDao.list_user_site_ids(db, user_id=8)
+        site_ids = await AccessDao.list_user_site_ids(db, user_id=8, scope=TestBatchScope.formal())
 
         self.assertEqual([3, 5], site_ids)
         statement = db.execute.await_args.args[0]
-        self._assert_statement_targets(
-            statement,
-            BizUserSite,
-            BizUserSite.site_id,
-            ((BizUserSite.user_id, 8),),
-        )
+        sql = ' '.join(str(statement.compile(compile_kwargs={'literal_binds': True})).split())
+        self.assertIn('JOIN biz_customer_site', sql)
+        self.assertIn('biz_user_site.user_id = 8', sql)
+        self.assertIn('biz_customer_site.test_batch_id IS NULL', sql)
 
     async def test_replace_warehouse_users_replaces_complete_mapping(self) -> None:
         db = AsyncMock(spec=AsyncSession)