412 lines
18 KiB
Python
412 lines
18 KiB
Python
import unittest
|
||
from types import SimpleNamespace
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
from fastapi import params
|
||
from sqlalchemy.dialects import mysql
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy.sql import select
|
||
from starlette.requests import Request
|
||
from starlette.responses import Response
|
||
|
||
from common.context import RequestContext
|
||
from exceptions.exception import LoginException, PermissionException, ServiceException, ServiceWarning
|
||
from middlewares.context_middleware import ContextCleanupMiddleware
|
||
from module_business.dao.test_batch_dao import TestBatchDao
|
||
from module_business.dependency.test_batch_dependency import (
|
||
TestBatchScopeDependency,
|
||
TestBatchScopeResolver,
|
||
parse_optional_positive_batch_id,
|
||
)
|
||
from module_business.domain.enums import TestBatchStatus
|
||
from module_business.domain.test_batch_scope import TestBatchScope, assert_record_in_scope, batch_scope_condition
|
||
from module_business.entity.do.master_do import BizWarehouse
|
||
from module_business.service.test_batch_access_service import TestBatchAccessService
|
||
from tests.module_business.test_batch_fixtures import make_admin, make_current_user, make_test_scope
|
||
|
||
|
||
def make_request(batch_id: str | None = None) -> Request:
|
||
headers = [] if batch_id is None else [(b'x-test-batch-id', batch_id.encode())]
|
||
return Request({'type': 'http', 'method': 'GET', 'path': '/', 'headers': headers})
|
||
|
||
|
||
def compile_sql(statement: object) -> str:
|
||
return ' '.join(
|
||
str(statement.compile(dialect=mysql.dialect(), compile_kwargs={'literal_binds': True})).split()
|
||
)
|
||
|
||
|
||
class TestBatchFixtureTest(unittest.TestCase):
|
||
def test_fixtures_set_admin_flag_explicitly(self) -> None:
|
||
self.assertIs(False, make_current_user(user_id=2, permissions=['*:*:*']).user.admin)
|
||
self.assertIs(True, make_admin().user.admin)
|
||
|
||
|
||
class TestBatchScopeTest(unittest.TestCase):
|
||
def test_formal_scope_is_writable_without_batch_status(self) -> None:
|
||
self.assertEqual(
|
||
TestBatchScope(batch_id=None, mode='FORMAL', status=None, writable=True),
|
||
TestBatchScope.formal(),
|
||
)
|
||
|
||
def test_batch_scope_condition_separates_formal_and_test_rows(self) -> None:
|
||
formal_statement = select(BizWarehouse.warehouse_id).where(
|
||
batch_scope_condition(BizWarehouse.test_batch_id, TestBatchScope.formal())
|
||
)
|
||
test_statement = select(BizWarehouse.warehouse_id).where(
|
||
batch_scope_condition(BizWarehouse.test_batch_id, make_test_scope(7))
|
||
)
|
||
|
||
self.assertIn('biz_warehouse.test_batch_id IS NULL', compile_sql(formal_statement))
|
||
self.assertIn('biz_warehouse.test_batch_id = 7', compile_sql(test_statement))
|
||
|
||
def test_record_scope_assertion_accepts_only_matching_scope(self) -> None:
|
||
assert_record_in_scope(SimpleNamespace(test_batch_id=None), TestBatchScope.formal(), '仓库')
|
||
assert_record_in_scope(SimpleNamespace(test_batch_id=7), make_test_scope(7), '仓库')
|
||
|
||
for record in (None, SimpleNamespace(test_batch_id=None), SimpleNamespace(test_batch_id=8)):
|
||
with self.subTest(record=record), self.assertRaises(ServiceException) as raised:
|
||
assert_record_in_scope(record, make_test_scope(7), '仓库')
|
||
self.assertEqual('仓库不属于当前数据模式或测试批次', raised.exception.message)
|
||
|
||
|
||
class TestBatchHeaderTest(unittest.TestCase):
|
||
def test_missing_header_selects_formal_mode(self) -> None:
|
||
self.assertIsNone(parse_optional_positive_batch_id(None))
|
||
|
||
def test_positive_integer_header_is_parsed(self) -> None:
|
||
self.assertEqual(7, parse_optional_positive_batch_id('7'))
|
||
|
||
def test_invalid_header_is_rejected(self) -> None:
|
||
for raw in ('', ' ', '0', '-1', '1.5', 'batch-7'):
|
||
with self.subTest(raw=raw), self.assertRaisesRegex(ServiceWarning, '测试批次标识必须为正整数'):
|
||
parse_optional_positive_batch_id(raw)
|
||
|
||
def test_python_coercible_values_are_not_valid_ascii_decimal_literals(self) -> None:
|
||
for raw in (' 7', '7 ', '+7', '1_0', '７', True, 7, b'7'):
|
||
with self.subTest(raw=raw), self.assertRaisesRegex(ServiceWarning, '测试批次标识必须为正整数'):
|
||
parse_optional_positive_batch_id(raw)
|
||
|
||
def test_overlong_ascii_decimal_is_mapped_to_service_warning(self) -> None:
|
||
with self.assertRaisesRegex(ServiceWarning, '测试批次标识必须为正整数'):
|
||
parse_optional_positive_batch_id('9' * 5000)
|
||
|
||
|
||
class TestBatchAccessTest(unittest.IsolatedAsyncioTestCase):
|
||
async def asyncSetUp(self) -> None:
|
||
self.db = AsyncMock(spec=AsyncSession)
|
||
|
||
async def test_permission_character_cannot_replace_super_admin_flag(self) -> None:
|
||
current_user = make_current_user(user_id=2, permissions=['*:*:*', 'system:testbatch:purge'])
|
||
|
||
with self.assertRaisesRegex(PermissionException, '仅超级管理员'):
|
||
TestBatchAccessService.assert_super_admin(current_user)
|
||
|
||
async def test_super_admin_flag_passes_hard_gate(self) -> None:
|
||
TestBatchAccessService.assert_super_admin(make_admin())
|
||
|
||
async def test_formal_mode_skips_batch_queries(self) -> None:
|
||
get_batch = AsyncMock()
|
||
user_is_authorized = AsyncMock()
|
||
with (
|
||
patch.object(TestBatchDao, 'get_batch', get_batch),
|
||
patch.object(TestBatchDao, 'user_is_authorized', user_is_authorized),
|
||
):
|
||
scope = await TestBatchAccessService.resolve_scope(self.db, make_current_user(), None, write=True)
|
||
|
||
self.assertEqual(TestBatchScope.formal(), scope)
|
||
get_batch.assert_not_awaited()
|
||
user_is_authorized.assert_not_awaited()
|
||
|
||
async def test_authorized_active_batch_resolves_writable_scope(self) -> None:
|
||
get_batch = AsyncMock(return_value=SimpleNamespace(batch_id=7, status='ACTIVE'))
|
||
user_is_authorized = AsyncMock(return_value=True)
|
||
with (
|
||
patch.object(TestBatchDao, 'get_batch', get_batch),
|
||
patch.object(TestBatchDao, 'user_is_authorized', user_is_authorized),
|
||
):
|
||
scope = await TestBatchAccessService.resolve_scope(self.db, make_current_user(8), 7, write=True)
|
||
|
||
self.assertEqual(
|
||
TestBatchScope(batch_id=7, mode='TEST', status=TestBatchStatus.ACTIVE, writable=True),
|
||
scope,
|
||
)
|
||
get_batch.assert_awaited_once_with(self.db, 7)
|
||
user_is_authorized.assert_awaited_once_with(self.db, 7, 8)
|
||
|
||
async def test_admin_active_batch_bypasses_user_mapping(self) -> None:
|
||
get_batch = AsyncMock(return_value=SimpleNamespace(batch_id=7, status='ACTIVE'))
|
||
user_is_authorized = AsyncMock()
|
||
with (
|
||
patch.object(TestBatchDao, 'get_batch', get_batch),
|
||
patch.object(TestBatchDao, 'user_is_authorized', user_is_authorized),
|
||
):
|
||
scope = await TestBatchAccessService.resolve_scope(self.db, make_admin(), 7, write=False)
|
||
|
||
self.assertEqual(make_test_scope(7), scope)
|
||
user_is_authorized.assert_not_awaited()
|
||
|
||
async def test_missing_batch_is_unavailable(self) -> None:
|
||
get_batch = AsyncMock(return_value=None)
|
||
user_is_authorized = AsyncMock()
|
||
with (
|
||
patch.object(TestBatchDao, 'get_batch', get_batch),
|
||
patch.object(TestBatchDao, 'user_is_authorized', user_is_authorized),
|
||
self.assertRaisesRegex(PermissionException, '当前测试批次不可用') as raised,
|
||
):
|
||
await TestBatchAccessService.resolve_scope(self.db, make_admin(), 7, write=False)
|
||
|
||
self.assertEqual({'reason': 'TEST_BATCH_UNAVAILABLE'}, raised.exception.data)
|
||
user_is_authorized.assert_not_awaited()
|
||
|
||
async def test_unauthorized_batch_is_unavailable(self) -> None:
|
||
get_batch = AsyncMock(return_value=SimpleNamespace(batch_id=7, status='ACTIVE'))
|
||
user_is_authorized = AsyncMock(return_value=False)
|
||
with (
|
||
patch.object(TestBatchDao, 'get_batch', get_batch),
|
||
patch.object(TestBatchDao, 'user_is_authorized', user_is_authorized),
|
||
self.assertRaisesRegex(PermissionException, '当前测试批次不可用') as raised,
|
||
):
|
||
await TestBatchAccessService.resolve_scope(self.db, make_current_user(8), 7, write=False)
|
||
|
||
self.assertEqual({'reason': 'TEST_BATCH_UNAVAILABLE'}, raised.exception.data)
|
||
|
||
async def test_purged_batch_is_unavailable(self) -> None:
|
||
with (
|
||
patch.object(
|
||
TestBatchDao,
|
||
'get_batch',
|
||
AsyncMock(return_value=SimpleNamespace(batch_id=7, status='PURGED')),
|
||
),
|
||
self.assertRaisesRegex(PermissionException, '当前测试批次不可用') as raised,
|
||
):
|
||
await TestBatchAccessService.resolve_scope(self.db, make_admin(), 7, write=False)
|
||
|
||
self.assertEqual({'reason': 'TEST_BATCH_UNAVAILABLE'}, raised.exception.data)
|
||
|
||
async def test_closed_batch_is_unavailable_for_business_read(self) -> None:
|
||
with (
|
||
patch.object(
|
||
TestBatchDao,
|
||
'get_batch',
|
||
AsyncMock(return_value=SimpleNamespace(batch_id=7, status='CLOSED')),
|
||
),
|
||
self.assertRaisesRegex(PermissionException, '当前测试批次不可用') as raised,
|
||
):
|
||
await TestBatchAccessService.resolve_scope(self.db, make_admin(), 7, write=False)
|
||
|
||
self.assertEqual({'reason': 'TEST_BATCH_UNAVAILABLE'}, raised.exception.data)
|
||
|
||
async def test_closed_batch_rejects_business_write(self) -> None:
|
||
with (
|
||
patch.object(
|
||
TestBatchDao,
|
||
'get_batch',
|
||
AsyncMock(return_value=SimpleNamespace(batch_id=7, status='CLOSED')),
|
||
),
|
||
self.assertRaisesRegex(PermissionException, '当前测试批次已关闭') as raised,
|
||
):
|
||
await TestBatchAccessService.resolve_scope(self.db, make_admin(), 7, write=True)
|
||
|
||
self.assertEqual({'reason': 'TEST_BATCH_UNAVAILABLE'}, raised.exception.data)
|
||
|
||
async def test_unauthorized_closed_batch_does_not_disclose_status(self) -> None:
|
||
with (
|
||
patch.object(
|
||
TestBatchDao,
|
||
'get_batch',
|
||
AsyncMock(return_value=SimpleNamespace(batch_id=7, status='CLOSED')),
|
||
),
|
||
patch.object(TestBatchDao, 'user_is_authorized', AsyncMock(return_value=False)),
|
||
self.assertRaises(PermissionException) as raised,
|
||
):
|
||
await TestBatchAccessService.resolve_scope(self.db, make_current_user(8), 7, write=True)
|
||
|
||
self.assertEqual('当前测试批次不可用', raised.exception.message)
|
||
|
||
async def test_formal_scope_does_not_lock_batch(self) -> None:
|
||
get_batch_for_update = AsyncMock()
|
||
with patch.object(TestBatchDao, 'get_batch_for_update', get_batch_for_update):
|
||
batch = await TestBatchAccessService.lock_active_batch(self.db, TestBatchScope.formal())
|
||
|
||
self.assertIsNone(batch)
|
||
get_batch_for_update.assert_not_awaited()
|
||
|
||
async def test_active_batch_lock_returns_current_batch(self) -> None:
|
||
expected = SimpleNamespace(batch_id=7, status='ACTIVE')
|
||
get_batch_for_update = AsyncMock(return_value=expected)
|
||
with patch.object(TestBatchDao, 'get_batch_for_update', get_batch_for_update):
|
||
batch = await TestBatchAccessService.lock_active_batch(self.db, make_test_scope(7))
|
||
|
||
self.assertIs(expected, batch)
|
||
get_batch_for_update.assert_awaited_once_with(self.db, 7)
|
||
|
||
async def test_batch_closed_after_scope_resolution_rejects_lock(self) -> None:
|
||
with (
|
||
patch.object(
|
||
TestBatchDao,
|
||
'get_batch_for_update',
|
||
AsyncMock(return_value=SimpleNamespace(batch_id=7, status='CLOSED')),
|
||
),
|
||
self.assertRaisesRegex(PermissionException, '当前测试批次已关闭') as raised,
|
||
):
|
||
await TestBatchAccessService.lock_active_batch(self.db, make_test_scope(7))
|
||
|
||
self.assertEqual({'reason': 'TEST_BATCH_UNAVAILABLE'}, raised.exception.data)
|
||
|
||
|
||
class TestBatchDaoTest(unittest.IsolatedAsyncioTestCase):
|
||
@staticmethod
|
||
def db_with_result(result: MagicMock) -> AsyncMock:
|
||
db = AsyncMock(spec=AsyncSession)
|
||
db.execute.return_value = result
|
||
return db
|
||
|
||
async def test_get_batch_filters_by_batch_id_without_locking(self) -> None:
|
||
expected = SimpleNamespace(batch_id=7)
|
||
result = MagicMock()
|
||
result.scalar_one_or_none.return_value = expected
|
||
db = self.db_with_result(result)
|
||
|
||
batch = await TestBatchDao.get_batch(db, 7)
|
||
|
||
self.assertIs(expected, batch)
|
||
sql = compile_sql(db.execute.await_args.args[0])
|
||
self.assertIn('biz_test_batch.batch_id = 7', sql)
|
||
self.assertNotIn('FOR UPDATE', sql)
|
||
db.commit.assert_not_awaited()
|
||
|
||
async def test_get_batch_for_update_locks_and_refreshes_identity_map(self) -> None:
|
||
expected = SimpleNamespace(batch_id=7)
|
||
result = MagicMock()
|
||
result.scalar_one_or_none.return_value = expected
|
||
db = self.db_with_result(result)
|
||
|
||
batch = await TestBatchDao.get_batch_for_update(db, 7)
|
||
|
||
self.assertIs(expected, batch)
|
||
statement = db.execute.await_args.args[0]
|
||
self.assertIn('FOR UPDATE', compile_sql(statement))
|
||
self.assertIs(True, statement.get_execution_options().get('populate_existing'))
|
||
db.commit.assert_not_awaited()
|
||
|
||
async def test_user_authorization_checks_batch_and_user_mapping(self) -> None:
|
||
for scalar, expected in ((7, True), (None, False)):
|
||
with self.subTest(scalar=scalar):
|
||
result = MagicMock()
|
||
result.scalar_one_or_none.return_value = scalar
|
||
db = self.db_with_result(result)
|
||
|
||
authorized = await TestBatchDao.user_is_authorized(db, batch_id=7, user_id=8)
|
||
|
||
self.assertIs(expected, authorized)
|
||
sql = compile_sql(db.execute.await_args.args[0])
|
||
self.assertIn('biz_test_batch_user.batch_id = 7', sql)
|
||
self.assertIn('biz_test_batch_user.user_id = 8', sql)
|
||
db.commit.assert_not_awaited()
|
||
|
||
async def test_list_available_batches_filters_active_and_orders_newest_first(self) -> None:
|
||
for is_admin in (False, True):
|
||
with self.subTest(is_admin=is_admin):
|
||
first = SimpleNamespace(batch_id=9)
|
||
second = SimpleNamespace(batch_id=7)
|
||
result = MagicMock()
|
||
result.scalars.return_value.all.return_value = (first, second)
|
||
db = self.db_with_result(result)
|
||
|
||
batches = await TestBatchDao.list_available_batches(db, user_id=8, is_admin=is_admin)
|
||
|
||
self.assertEqual([first, second], batches)
|
||
sql = compile_sql(db.execute.await_args.args[0])
|
||
self.assertIn("biz_test_batch.status = 'ACTIVE'", sql)
|
||
self.assertIn('ORDER BY biz_test_batch.batch_id DESC', sql)
|
||
if is_admin:
|
||
self.assertNotIn('JOIN biz_test_batch_user', sql)
|
||
else:
|
||
self.assertIn('JOIN biz_test_batch_user', sql)
|
||
self.assertIn('biz_test_batch_user.user_id = 8', sql)
|
||
db.commit.assert_not_awaited()
|
||
|
||
|
||
class TestBatchDependencyTest(unittest.IsolatedAsyncioTestCase):
|
||
async def asyncSetUp(self) -> None:
|
||
RequestContext.clear_all()
|
||
self.db = AsyncMock(spec=AsyncSession)
|
||
|
||
async def asyncTearDown(self) -> None:
|
||
RequestContext.clear_all()
|
||
|
||
async def test_resolver_forwards_write_mode_and_sets_trusted_context(self) -> None:
|
||
expected = make_test_scope(7)
|
||
resolve_scope = AsyncMock(return_value=expected)
|
||
current_user = make_current_user(8)
|
||
with patch.object(TestBatchAccessService, 'resolve_scope', resolve_scope):
|
||
scope = await TestBatchScopeResolver(write=True)(make_request('7'), self.db, current_user)
|
||
|
||
self.assertIs(expected, scope)
|
||
self.assertIs(expected, RequestContext.get_current_test_batch_scope())
|
||
resolve_scope.assert_awaited_once_with(self.db, current_user, 7, write=True)
|
||
|
||
async def test_invalid_header_does_not_call_access_service_or_set_context(self) -> None:
|
||
resolve_scope = AsyncMock()
|
||
with (
|
||
patch.object(TestBatchAccessService, 'resolve_scope', resolve_scope),
|
||
self.assertRaisesRegex(ServiceWarning, '测试批次标识必须为正整数'),
|
||
):
|
||
await TestBatchScopeResolver()(make_request('0'), self.db, make_current_user())
|
||
|
||
resolve_scope.assert_not_awaited()
|
||
self.assertIsNone(RequestContext.get_current_test_batch_scope())
|
||
|
||
async def test_dependency_factory_wraps_configured_resolver(self) -> None:
|
||
dependency = TestBatchScopeDependency(write=True)
|
||
|
||
self.assertIsInstance(dependency, params.Depends)
|
||
self.assertIsInstance(dependency.dependency, TestBatchScopeResolver)
|
||
self.assertIs(True, dependency.dependency.write)
|
||
|
||
|
||
class ContextCleanupMiddlewareTest(unittest.IsolatedAsyncioTestCase):
|
||
async def asyncSetUp(self) -> None:
|
||
RequestContext.clear_all()
|
||
self.middleware = ContextCleanupMiddleware(lambda scope, receive, send: None)
|
||
self.request = make_request()
|
||
|
||
async def asyncTearDown(self) -> None:
|
||
RequestContext.clear_all()
|
||
|
||
def seed_context(self) -> None:
|
||
RequestContext.set_current_exclude_patterns([{'path': '/private'}])
|
||
RequestContext.set_current_user(make_admin())
|
||
RequestContext.set_current_test_batch_scope(make_test_scope(7))
|
||
|
||
def assert_context_is_clear(self) -> None:
|
||
self.assertEqual([], RequestContext.get_current_exclude_patterns())
|
||
with self.assertRaises(LoginException):
|
||
RequestContext.get_current_user()
|
||
self.assertIsNone(RequestContext.get_current_test_batch_scope())
|
||
|
||
async def test_dispatch_clears_all_context_after_response(self) -> None:
|
||
self.seed_context()
|
||
expected = Response('ok')
|
||
|
||
response = await self.middleware.dispatch(self.request, AsyncMock(return_value=expected))
|
||
|
||
self.assertIs(expected, response)
|
||
self.assert_context_is_clear()
|
||
|
||
async def test_dispatch_clears_all_context_after_exception(self) -> None:
|
||
self.seed_context()
|
||
|
||
with self.assertRaisesRegex(RuntimeError, 'endpoint failed'):
|
||
await self.middleware.dispatch(
|
||
self.request,
|
||
AsyncMock(side_effect=RuntimeError('endpoint failed')),
|
||
)
|
||
|
||
self.assert_context_is_clear()
|
||
|
||
|
||
if __name__ == '__main__':
|
||
unittest.main()
|