From 83a834f87ef8dbb0d279efb6ba46328ab1efaae6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=84=A6=E9=BE=99=E8=A8=80?= Date: Sun, 26 Jul 2026 09:42:12 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8A=9F=E8=83=BD=EF=BC=9A=E6=96=B0=E5=A2=9E?= =?UTF-8?q?=E4=BB=93=E5=BA=93=E4=B8=8E=E5=AE=A2=E6=88=B7=E7=8E=B0=E5=9C=BA?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E6=9D=83=E9=99=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../module_business/dao/access_dao.py | 64 ++++++ .../module_business/service/access_service.py | 35 ++++ .../module_business/test_access_service.py | 197 ++++++++++++++++++ 3 files changed, 296 insertions(+) create mode 100644 shierfeng-fastapi-backend/module_business/dao/access_dao.py create mode 100644 shierfeng-fastapi-backend/module_business/service/access_service.py create mode 100644 shierfeng-fastapi-backend/tests/module_business/test_access_service.py diff --git a/shierfeng-fastapi-backend/module_business/dao/access_dao.py b/shierfeng-fastapi-backend/module_business/dao/access_dao.py new file mode 100644 index 0000000..5935034 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/dao/access_dao.py @@ -0,0 +1,64 @@ +from sqlalchemy import delete, select +from sqlalchemy.ext.asyncio import AsyncSession + +from module_business.entity.do.master_do import BizUserSite, BizUserWarehouse + + +class AccessDao: + """仓库与客户现场数据权限数据库操作层。""" + + @classmethod + async def user_has_warehouse(cls, db: AsyncSession, user_id: int, warehouse_id: int) -> bool: + """判断用户是否关联指定仓库。""" + mapped_user_id = ( + await db.execute( + select(BizUserWarehouse.user_id).where( + BizUserWarehouse.user_id == user_id, + BizUserWarehouse.warehouse_id == warehouse_id, + ) + ) + ).scalar_one_or_none() + + return mapped_user_id is not None + + @classmethod + async def user_has_site(cls, db: AsyncSession, user_id: int, site_id: int) -> bool: + """判断用户是否关联指定客户现场。""" + mapped_user_id = ( + await db.execute( + select(BizUserSite.user_id).where( + BizUserSite.user_id == user_id, + BizUserSite.site_id == site_id, + ) + ) + ).scalar_one_or_none() + + return mapped_user_id is not None + + @classmethod + async def list_user_warehouse_ids(cls, db: AsyncSession, user_id: int) -> list[int]: + """获取用户关联的仓库 ID 列表。""" + result = await db.execute(select(BizUserWarehouse.warehouse_id).where(BizUserWarehouse.user_id == user_id)) + + return list(result.scalars().all()) + + @classmethod + async def list_user_site_ids(cls, db: AsyncSession, user_id: int) -> list[int]: + """获取用户关联的客户现场 ID 列表。""" + result = await db.execute(select(BizUserSite.site_id).where(BizUserSite.user_id == user_id)) + + return list(result.scalars().all()) + + @classmethod + async def replace_warehouse_users(cls, db: AsyncSession, warehouse_id: int, user_ids: list[int]) -> None: + """完整替换指定仓库的用户关联，仅刷新当前事务。""" + await db.execute(delete(BizUserWarehouse).where(BizUserWarehouse.warehouse_id == warehouse_id)) + db.add_all([BizUserWarehouse(user_id=user_id, warehouse_id=warehouse_id) for user_id in user_ids]) + await db.flush() + + @classmethod + async def replace_site_users(cls, db: AsyncSession, site_id: int, user_ids: list[int]) -> None: + """完整替换指定客户现场的用户关联，仅刷新当前事务。""" + await db.execute(delete(BizUserSite).where(BizUserSite.site_id == site_id)) + db.add_all([BizUserSite(user_id=user_id, site_id=site_id) for user_id in user_ids]) + await db.flush() diff --git a/shierfeng-fastapi-backend/module_business/service/access_service.py b/shierfeng-fastapi-backend/module_business/service/access_service.py new file mode 100644 index 0000000..5f73764 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/service/access_service.py @@ -0,0 +1,35 @@ +from sqlalchemy.ext.asyncio import AsyncSession + +from exceptions.exception import PermissionException +from module_admin.entity.vo.user_vo import CurrentUserModel +from module_business.dao.access_dao import AccessDao + + +class AccessService: + """仓库与客户现场数据权限服务层。""" + + @classmethod + async def assert_warehouse_access( + cls, + db: AsyncSession, + current_user: CurrentUserModel, + warehouse_id: int, + ) -> None: + """校验当前用户是否有权访问指定仓库。""" + if current_user.user.admin: + return + if not await AccessDao.user_has_warehouse(db, current_user.user.user_id, warehouse_id): + raise PermissionException(data='', message='无权访问该仓库') + + @classmethod + async def assert_site_access( + cls, + db: AsyncSession, + current_user: CurrentUserModel, + site_id: int, + ) -> None: + """校验当前用户是否有权访问指定客户现场。""" + if current_user.user.admin: + return + if not await AccessDao.user_has_site(db, current_user.user.user_id, site_id): + raise PermissionException(data='', message='无权访问该客户现场') diff --git a/shierfeng-fastapi-backend/tests/module_business/test_access_service.py b/shierfeng-fastapi-backend/tests/module_business/test_access_service.py new file mode 100644 index 0000000..4f1ecba --- /dev/null +++ b/shierfeng-fastapi-backend/tests/module_business/test_access_service.py @@ -0,0 +1,197 @@ +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 + +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_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_params: set[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())) + + 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, + {8, 99}, + ) + + 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, {7, 8}) + + 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, {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, {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()