diff --git a/shierfeng-fastapi-backend/tests/module_business/test_access_service.py b/shierfeng-fastapi-backend/tests/module_business/test_access_service.py index 4f1ecba..57ce566 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_access_service.py +++ b/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)