测试：强化数据权限查询谓词校验

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

@@ -4,7 +4,8 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import InstrumentedAttribute
-from sqlalchemy.sql import Select
+from sqlalchemy.sql import Select, operators
+from sqlalchemy.sql.elements import BinaryExpression, BindParameter, BooleanClauseList
 
 from exceptions.exception import PermissionException
 from module_business.dao.access_dao import AccessDao
@@ -80,11 +81,30 @@ class AccessDaoTest(unittest.IsolatedAsyncioTestCase):
         statement: Select[tuple[int]],
         model: type[BizUserWarehouse] | type[BizUserSite],
         selected_column: InstrumentedAttribute[int],
-        expected_params: set[int],
+        expected_predicates: tuple[tuple[InstrumentedAttribute[int], int], ...],
     ) -> None:
         self.assertTrue(statement.selected_columns[0].compare(selected_column.property.columns[0]))
         self.assertTrue(statement.get_final_froms()[0].compare(model.__table__))
-        self.assertEqual(expected_params, set(statement.compile().params.values()))
+
+        whereclause = statement.whereclause
+        self.assertIsNotNone(whereclause)
+        clauses = list(whereclause.clauses) if isinstance(whereclause, BooleanClauseList) else [whereclause]
+        binary_predicates = [clause for clause in clauses if isinstance(clause, BinaryExpression)]
+        self.assertEqual(len(clauses), len(binary_predicates))
+        self.assertEqual(len(expected_predicates), len(binary_predicates))
+
+        unmatched_predicates = list(binary_predicates)
+        for expected_column, expected_value in expected_predicates:
+            column = expected_column.property.columns[0]
+            matching_predicates = [predicate for predicate in unmatched_predicates if predicate.left.compare(column)]
+            self.assertEqual(1, len(matching_predicates))
+            predicate = matching_predicates[0]
+            self.assertIs(predicate.operator, operators.eq)
+            self.assertIsInstance(predicate.right, BindParameter)
+            self.assertEqual(expected_value, predicate.right.value)
+            unmatched_predicates = [candidate for candidate in unmatched_predicates if candidate is not predicate]
+
+        self.assertEqual([], unmatched_predicates)
 
     async def test_user_has_warehouse_returns_mapping_existence(self) -> None:
         for scalar, expected in ((8, True), (None, False)):
@@ -101,7 +121,10 @@ class AccessDaoTest(unittest.IsolatedAsyncioTestCase):
                     statement,
                     BizUserWarehouse,
                     BizUserWarehouse.user_id,
-                    {8, 99},
+                    (
+                        (BizUserWarehouse.user_id, 8),
+                        (BizUserWarehouse.warehouse_id, 99),
+                    ),
                 )
 
     async def test_user_has_site_returns_mapping_existence(self) -> None:
@@ -115,7 +138,15 @@ class AccessDaoTest(unittest.IsolatedAsyncioTestCase):
 
                 self.assertIs(expected, actual)
                 statement = db.execute.await_args.args[0]
-                self._assert_statement_targets(statement, BizUserSite, BizUserSite.user_id, {7, 8})
+                self._assert_statement_targets(
+                    statement,
+                    BizUserSite,
+                    BizUserSite.user_id,
+                    (
+                        (BizUserSite.user_id, 8),
+                        (BizUserSite.site_id, 7),
+                    ),
+                )
 
     async def test_list_user_warehouse_ids_returns_scalar_ids(self) -> None:
         result = MagicMock()
@@ -126,7 +157,12 @@ class AccessDaoTest(unittest.IsolatedAsyncioTestCase):
 
         self.assertEqual([11, 13], warehouse_ids)
         statement = db.execute.await_args.args[0]
-        self._assert_statement_targets(statement, BizUserWarehouse, BizUserWarehouse.warehouse_id, {8})
+        self._assert_statement_targets(
+            statement,
+            BizUserWarehouse,
+            BizUserWarehouse.warehouse_id,
+            ((BizUserWarehouse.user_id, 8),),
+        )
 
     async def test_list_user_site_ids_returns_scalar_ids(self) -> None:
         result = MagicMock()
@@ -137,7 +173,12 @@ class AccessDaoTest(unittest.IsolatedAsyncioTestCase):
 
         self.assertEqual([3, 5], site_ids)
         statement = db.execute.await_args.args[0]
-        self._assert_statement_targets(statement, BizUserSite, BizUserSite.site_id, {8})
+        self._assert_statement_targets(
+            statement,
+            BizUserSite,
+            BizUserSite.site_id,
+            ((BizUserSite.user_id, 8),),
+        )
 
     async def test_replace_warehouse_users_replaces_complete_mapping(self) -> None:
         db = AsyncMock(spec=AsyncSession)