import inspect import unittest from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import InstrumentedAttribute 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 from module_business.entity.do.master_do import BizUserSite, BizUserWarehouse from module_business.service.access_service import AccessService class AccessServiceTest(unittest.IsolatedAsyncioTestCase): @patch('module_business.service.access_service.AccessDao.user_has_warehouse', new_callable=AsyncMock) @patch('module_business.service.access_service.AccessDao.user_has_site', new_callable=AsyncMock) async def test_admin_bypasses_mapping(self, user_has_site: AsyncMock, user_has_warehouse: AsyncMock) -> None: user = SimpleNamespace(user=SimpleNamespace(user_id=1, admin=True)) db = AsyncMock(spec=AsyncSession) await AccessService.assert_warehouse_access(db, user, 99) await AccessService.assert_site_access(db, user, 7) user_has_warehouse.assert_not_awaited() user_has_site.assert_not_awaited() @patch('module_business.service.access_service.AccessDao.user_has_warehouse', new_callable=AsyncMock) async def test_mapped_warehouse_is_allowed(self, user_has_warehouse: AsyncMock) -> None: user_has_warehouse.return_value = True user = SimpleNamespace(user=SimpleNamespace(user_id=8, admin=False)) db = AsyncMock(spec=AsyncSession) await AccessService.assert_warehouse_access(db, user, 99) user_has_warehouse.assert_awaited_once_with(db, 8, 99) @patch('module_business.service.access_service.AccessDao.user_has_warehouse', new_callable=AsyncMock) async def test_locking_warehouse_access_is_forwarded(self, user_has_warehouse: AsyncMock) -> None: lock_parameter = inspect.signature(AccessService.assert_warehouse_access).parameters.get('for_update') self.assertIsNotNone(lock_parameter) self.assertIs(inspect.Parameter.KEYWORD_ONLY, lock_parameter.kind) self.assertIs(False, lock_parameter.default) user_has_warehouse.return_value = True user = SimpleNamespace(user=SimpleNamespace(user_id=8, admin=False)) db = AsyncMock(spec=AsyncSession) await AccessService.assert_warehouse_access(db, user, 99, for_update=True) user_has_warehouse.assert_awaited_once_with(db, 8, 99, for_update=True) @patch('module_business.service.access_service.AccessDao.user_has_warehouse', new_callable=AsyncMock) async def test_admin_bypasses_locking_warehouse_mapping(self, user_has_warehouse: AsyncMock) -> None: lock_parameter = inspect.signature(AccessService.assert_warehouse_access).parameters.get('for_update') self.assertIsNotNone(lock_parameter) user = SimpleNamespace(user=SimpleNamespace(user_id=1, admin=True)) await AccessService.assert_warehouse_access( AsyncMock(spec=AsyncSession), user, 99, for_update=True, ) user_has_warehouse.assert_not_awaited() @patch('module_business.service.access_service.AccessDao.user_has_warehouse', new_callable=AsyncMock) async def test_unmapped_warehouse_is_forbidden(self, user_has_warehouse: AsyncMock) -> None: user_has_warehouse.return_value = False user = SimpleNamespace(user=SimpleNamespace(user_id=8, admin=False)) with self.assertRaises(PermissionException) as raised: await AccessService.assert_warehouse_access(AsyncMock(spec=AsyncSession), user, 99) self.assertEqual('', raised.exception.data) self.assertEqual('无权访问该仓库', raised.exception.message) @patch('module_business.service.access_service.AccessDao.user_has_site', new_callable=AsyncMock) async def test_mapped_site_is_allowed(self, user_has_site: AsyncMock) -> None: 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) 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_unmapped_site_is_forbidden(self, user_has_site: AsyncMock) -> None: user_has_site.return_value = False user = SimpleNamespace(user=SimpleNamespace(user_id=8, admin=False)) with self.assertRaises(PermissionException) as raised: await AccessService.assert_site_access(AsyncMock(spec=AsyncSession), user, 7) self.assertEqual('', raised.exception.data) self.assertEqual('无权访问该客户现场', raised.exception.message) class AccessDaoTest(unittest.IsolatedAsyncioTestCase): @staticmethod def _db_with_result(result: MagicMock) -> AsyncMock: db = AsyncMock(spec=AsyncSession) db.execute.return_value = result return db def _assert_statement_targets( self, statement: Select[tuple[int]], model: type[BizUserWarehouse] | type[BizUserSite], selected_column: InstrumentedAttribute[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__)) 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)): with self.subTest(scalar=scalar): result = MagicMock() result.scalar_one_or_none.return_value = scalar db = self._db_with_result(result) actual = await AccessDao.user_has_warehouse(db, user_id=8, warehouse_id=99) self.assertIs(expected, actual) statement = db.execute.await_args.args[0] self._assert_statement_targets( statement, BizUserWarehouse, BizUserWarehouse.user_id, ( (BizUserWarehouse.user_id, 8), (BizUserWarehouse.warehouse_id, 99), ), ) async def test_user_has_warehouse_current_read_is_optional_keyword_only_lock(self) -> None: lock_parameter = inspect.signature(AccessDao.user_has_warehouse).parameters.get('for_update') self.assertIsNotNone(lock_parameter) self.assertIs(inspect.Parameter.KEYWORD_ONLY, lock_parameter.kind) self.assertIs(False, lock_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_warehouse(db, 8, 99, for_update=True) else: 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) async def test_user_has_site_returns_mapping_existence(self) -> None: for scalar, expected in ((8, True), (None, False)): with self.subTest(scalar=scalar): result = MagicMock() result.scalar_one_or_none.return_value = scalar db = self._db_with_result(result) actual = await AccessDao.user_has_site(db, user_id=8, site_id=7) self.assertIs(expected, actual) statement = db.execute.await_args.args[0] 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() 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) 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),), ) 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) 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),), ) async def test_replace_warehouse_users_replaces_complete_mapping(self) -> None: db = AsyncMock(spec=AsyncSession) await AccessDao.replace_warehouse_users(db, warehouse_id=99, user_ids=[8, 9]) delete_statement = db.execute.await_args.args[0] self.assertTrue(delete_statement.table.compare(BizUserWarehouse.__table__)) self.assertTrue(delete_statement.whereclause.left.compare(BizUserWarehouse.warehouse_id.property.columns[0])) self.assertEqual({99}, set(delete_statement.compile().params.values())) mappings = db.add_all.call_args.args[0] self.assertEqual([(8, 99), (9, 99)], [(item.user_id, item.warehouse_id) for item in mappings]) self.assertTrue(all(isinstance(item, BizUserWarehouse) for item in mappings)) self._assert_flush_without_transaction_completion(db) async def test_replace_warehouse_users_supports_empty_list(self) -> None: db = AsyncMock(spec=AsyncSession) await AccessDao.replace_warehouse_users(db, warehouse_id=99, user_ids=[]) delete_statement = db.execute.await_args.args[0] self.assertTrue(delete_statement.table.compare(BizUserWarehouse.__table__)) self.assertTrue(delete_statement.whereclause.left.compare(BizUserWarehouse.warehouse_id.property.columns[0])) self.assertEqual({99}, set(delete_statement.compile().params.values())) db.add_all.assert_called_once_with([]) self._assert_flush_without_transaction_completion(db) async def test_replace_site_users_replaces_complete_mapping(self) -> None: db = AsyncMock(spec=AsyncSession) await AccessDao.replace_site_users(db, site_id=7, user_ids=[8, 9]) delete_statement = db.execute.await_args.args[0] self.assertTrue(delete_statement.table.compare(BizUserSite.__table__)) self.assertTrue(delete_statement.whereclause.left.compare(BizUserSite.site_id.property.columns[0])) self.assertEqual({7}, set(delete_statement.compile().params.values())) mappings = db.add_all.call_args.args[0] self.assertEqual([(8, 7), (9, 7)], [(item.user_id, item.site_id) for item in mappings]) self.assertTrue(all(isinstance(item, BizUserSite) for item in mappings)) self._assert_flush_without_transaction_completion(db) async def test_replace_site_users_supports_empty_list(self) -> None: db = AsyncMock(spec=AsyncSession) await AccessDao.replace_site_users(db, site_id=7, user_ids=[]) delete_statement = db.execute.await_args.args[0] self.assertTrue(delete_statement.table.compare(BizUserSite.__table__)) self.assertTrue(delete_statement.whereclause.left.compare(BizUserSite.site_id.property.columns[0])) self.assertEqual({7}, set(delete_statement.compile().params.values())) db.add_all.assert_called_once_with([]) self._assert_flush_without_transaction_completion(db) def _assert_flush_without_transaction_completion(self, db: AsyncMock) -> None: db.flush.assert_awaited_once_with() db.commit.assert_not_awaited() db.rollback.assert_not_awaited()