198 lines
9.4 KiB
Python
198 lines
9.4 KiB
Python
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()
|