From 763ea1efccf55dc25724dc2bd8ed084dc8be1af9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=84=A6=E9=BE=99=E8=A8=80?= Date: Tue, 22 Sep 2026 15:11:53 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=89=E5=85=A8=EF=BC=9A=E5=A2=9E=E5=8A=A0?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=E6=89=B9=E6=AC=A1=E4=B8=8A=E4=B8=8B=E6=96=87?= =?UTF-8?q?=E7=A1=AC=E6=A0=A1=E9=AA=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- shierfeng-fastapi-backend/common/context.py | 24 ++ .../middlewares/context_middleware.py | 8 +- .../module_business/dao/test_batch_dao.py | 47 ++ .../module_business/dependency/__init__.py | 0 .../dependency/test_batch_dependency.py | 47 ++ .../domain/test_batch_scope.py | 33 ++ .../service/test_batch_access_service.py | 81 ++++ .../module_business/test_batch_fixtures.py | 28 ++ .../module_business/test_test_batch_access.py | 402 ++++++++++++++++++ 9 files changed, 666 insertions(+), 4 deletions(-) create mode 100644 shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py create mode 100644 shierfeng-fastapi-backend/module_business/dependency/__init__.py create mode 100644 shierfeng-fastapi-backend/module_business/dependency/test_batch_dependency.py create mode 100644 shierfeng-fastapi-backend/module_business/domain/test_batch_scope.py create mode 100644 shierfeng-fastapi-backend/module_business/service/test_batch_access_service.py create mode 100644 shierfeng-fastapi-backend/tests/module_business/test_batch_fixtures.py create mode 100644 shierfeng-fastapi-backend/tests/module_business/test_test_batch_access.py diff --git a/shierfeng-fastapi-backend/common/context.py b/shierfeng-fastapi-backend/common/context.py index 67e933c..1b36672 100644 --- a/shierfeng-fastapi-backend/common/context.py +++ b/shierfeng-fastapi-backend/common/context.py @@ -4,6 +4,7 @@ from typing import Literal from exceptions.exception import LoginException from module_admin.entity.vo.user_vo import CurrentUserModel +from module_business.domain.test_batch_scope import TestBatchScope # 定义上下文变量 # 存储当前请求的编译后的排除路由模式列表 @@ -12,6 +13,8 @@ current_exclude_patterns: ContextVar[ ] = ContextVar('current_exclude_patterns', default=None) # 存储当前用户信息 current_user: ContextVar[CurrentUserModel | None] = ContextVar('current_user', default=None) +# 存储当前请求已校验的测试批次作用域 +current_test_batch_scope: ContextVar[TestBatchScope | None] = ContextVar('current_test_batch_scope', default=None) class RequestContext: @@ -69,6 +72,21 @@ class RequestContext: raise LoginException(data='', message='当前用户信息为空，请检查是否已登录') return _current_user + @staticmethod + def set_current_test_batch_scope(scope: TestBatchScope) -> Token: + """ + 设置当前请求已校验的测试批次作用域 + + :param scope: 测试批次作用域 + :return: 上下文变量令牌，用于重置 + """ + return current_test_batch_scope.set(scope) + + @staticmethod + def get_current_test_batch_scope() -> TestBatchScope | None: + """获取当前请求已校验的测试批次作用域。""" + return current_test_batch_scope.get() + @staticmethod def reset_current_exclude_patterns(token: Token) -> None: """ @@ -87,6 +105,11 @@ class RequestContext: """ current_user.reset(token) + @staticmethod + def reset_current_test_batch_scope(token: Token) -> None: + """重置当前请求的测试批次作用域。""" + current_test_batch_scope.reset(token) + @staticmethod def clear_all() -> None: """ @@ -94,3 +117,4 @@ class RequestContext: """ current_exclude_patterns.set(None) current_user.set(None) + current_test_batch_scope.set(None) diff --git a/shierfeng-fastapi-backend/middlewares/context_middleware.py b/shierfeng-fastapi-backend/middlewares/context_middleware.py index 89c9054..457e97d 100644 --- a/shierfeng-fastapi-backend/middlewares/context_middleware.py +++ b/shierfeng-fastapi-backend/middlewares/context_middleware.py @@ -14,10 +14,10 @@ class ContextCleanupMiddleware(BaseHTTPMiddleware): """ 在每个请求处理完成后清理上下文信息 """ - response = await call_next(request) - # 请求处理完成后清理所有上下文变量 - RequestContext.clear_all() - return response + try: + return await call_next(request) + finally: + RequestContext.clear_all() def add_context_cleanup_middleware(app: FastAPI) -> None: diff --git a/shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py b/shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py new file mode 100644 index 0000000..3f0e0d2 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/dao/test_batch_dao.py @@ -0,0 +1,47 @@ +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from module_business.domain.enums import TestBatchStatus +from module_business.entity.do.test_batch_do import BizTestBatch, BizTestBatchUser + + +class TestBatchDao: + """测试批次访问数据库操作层。""" + + @classmethod + async def get_batch(cls, db: AsyncSession, batch_id: int) -> BizTestBatch | None: + result = await db.execute(select(BizTestBatch).where(BizTestBatch.batch_id == batch_id)) + return result.scalar_one_or_none() + + @classmethod + async def get_batch_for_update(cls, db: AsyncSession, batch_id: int) -> BizTestBatch | None: + result = await db.execute( + select(BizTestBatch) + .where(BizTestBatch.batch_id == batch_id) + .with_for_update() + .execution_options(populate_existing=True) + ) + return result.scalar_one_or_none() + + @classmethod + async def user_is_authorized(cls, db: AsyncSession, batch_id: int, user_id: int) -> bool: + result = await db.execute( + select(BizTestBatchUser.batch_id).where( + BizTestBatchUser.batch_id == batch_id, + BizTestBatchUser.user_id == user_id, + ) + ) + return result.scalar_one_or_none() is not None + + @classmethod + async def list_available_batches( + cls, + db: AsyncSession, + user_id: int, + is_admin: bool, + ) -> list[BizTestBatch]: + statement = select(BizTestBatch).where(BizTestBatch.status == TestBatchStatus.ACTIVE.value) + if not is_admin: + statement = statement.join(BizTestBatchUser).where(BizTestBatchUser.user_id == user_id) + result = await db.execute(statement.order_by(BizTestBatch.batch_id.desc())) + return list(result.scalars().all()) diff --git a/shierfeng-fastapi-backend/module_business/dependency/__init__.py b/shierfeng-fastapi-backend/module_business/dependency/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/shierfeng-fastapi-backend/module_business/dependency/test_batch_dependency.py b/shierfeng-fastapi-backend/module_business/dependency/test_batch_dependency.py new file mode 100644 index 0000000..9fe68ff --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/dependency/test_batch_dependency.py @@ -0,0 +1,47 @@ +from fastapi import Depends, Request, params +from sqlalchemy.ext.asyncio import AsyncSession + +from common.aspect.pre_auth import CurrentUserDependency +from common.context import RequestContext +from config.get_db import get_db +from exceptions.exception import ServiceWarning +from module_admin.entity.vo.user_vo import CurrentUserModel +from module_business.domain.test_batch_scope import TestBatchScope +from module_business.service.test_batch_access_service import TestBatchAccessService + + +def parse_optional_positive_batch_id(raw: str | None) -> int | None: + """解析可选的测试批次请求头。""" + if raw is None: + return None + try: + batch_id = int(raw) + except (TypeError, ValueError): + raise ServiceWarning('', '测试批次标识必须为正整数') from None + if batch_id <= 0: + raise ServiceWarning('', '测试批次标识必须为正整数') + return batch_id + + +class TestBatchScopeResolver: + """将请求头解析为已验证的测试批次作用域。""" + + def __init__(self, *, write: bool = False) -> None: + self.write = write + + async def __call__( + self, + request: Request, + db: AsyncSession = Depends(get_db), + current_user: CurrentUserModel = CurrentUserDependency(), + ) -> TestBatchScope: + raw = request.headers.get('X-Test-Batch-Id') + batch_id = parse_optional_positive_batch_id(raw) + scope = await TestBatchAccessService.resolve_scope(db, current_user, batch_id, write=self.write) + RequestContext.set_current_test_batch_scope(scope) + return scope + + +def TestBatchScopeDependency(*, write: bool = False) -> params.Depends: # noqa: N802 + """构造可配置读写语义的批次作用域依赖。""" + return Depends(TestBatchScopeResolver(write=write)) diff --git a/shierfeng-fastapi-backend/module_business/domain/test_batch_scope.py b/shierfeng-fastapi-backend/module_business/domain/test_batch_scope.py new file mode 100644 index 0000000..e83f2fc --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/domain/test_batch_scope.py @@ -0,0 +1,33 @@ +from dataclasses import dataclass +from typing import Any, Literal + +from sqlalchemy.orm import InstrumentedAttribute +from sqlalchemy.sql.elements import ColumnElement + +from exceptions.exception import ServiceException +from module_business.domain.enums import TestBatchStatus + + +@dataclass(frozen=True, slots=True) +class TestBatchScope: + """当前请求可信任的数据批次作用域。""" + + batch_id: int | None + mode: Literal['FORMAL', 'TEST'] + status: TestBatchStatus | None + writable: bool + + @classmethod + def formal(cls) -> 'TestBatchScope': + return cls(batch_id=None, mode='FORMAL', status=None, writable=True) + + +def batch_scope_condition(column: InstrumentedAttribute, scope: TestBatchScope) -> ColumnElement[bool]: + """生成正式数据或当前测试批次的查询条件。""" + return column.is_(None) if scope.batch_id is None else column == scope.batch_id + + +def assert_record_in_scope(record: Any, scope: TestBatchScope, label: str) -> None: + """防止已加载记录跨数据模式或测试批次被使用。""" + if record is None or record.test_batch_id != scope.batch_id: + raise ServiceException('', f'{label}不属于当前数据模式或测试批次') diff --git a/shierfeng-fastapi-backend/module_business/service/test_batch_access_service.py b/shierfeng-fastapi-backend/module_business/service/test_batch_access_service.py new file mode 100644 index 0000000..440cead --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/service/test_batch_access_service.py @@ -0,0 +1,81 @@ +from sqlalchemy.ext.asyncio import AsyncSession + +from exceptions.exception import PermissionException +from module_admin.entity.vo.user_vo import CurrentUserModel +from module_business.dao.test_batch_dao import TestBatchDao +from module_business.domain.enums import TestBatchStatus +from module_business.domain.test_batch_scope import TestBatchScope +from module_business.entity.do.test_batch_do import BizTestBatch + + +class TestBatchAccessService: + """测试批次作用域与管理权限校验。""" + + @staticmethod + def _is_super_admin(current_user: CurrentUserModel) -> bool: + return current_user.user is not None and current_user.user.admin is True + + @staticmethod + def _raise_unavailable() -> None: + raise PermissionException({'reason': 'TEST_BATCH_UNAVAILABLE'}, '当前测试批次不可用') + + @staticmethod + def _raise_closed() -> None: + raise PermissionException({'reason': 'TEST_BATCH_UNAVAILABLE'}, '当前测试批次已关闭') + + @classmethod + def assert_super_admin(cls, current_user: CurrentUserModel) -> None: + """确保当前用户由超级管理员标记明确授信。""" + if not cls._is_super_admin(current_user): + raise PermissionException('', '仅超级管理员可执行测试批次管理操作') + + @classmethod + async def resolve_scope( + cls, + db: AsyncSession, + current_user: CurrentUserModel, + batch_id: int | None, + *, + write: bool = False, + ) -> TestBatchScope: + """解析并验证业务请求可使用的数据作用域。""" + if batch_id is None: + return TestBatchScope.formal() + + batch = await TestBatchDao.get_batch(db, batch_id) + if batch is None or batch.status == TestBatchStatus.PURGED.value: + cls._raise_unavailable() + + if not cls._is_super_admin(current_user): + user_id = current_user.user.user_id if current_user.user is not None else None + if user_id is None or not await TestBatchDao.user_is_authorized(db, batch_id, user_id): + cls._raise_unavailable() + + if batch.status == TestBatchStatus.CLOSED.value: + if write: + cls._raise_closed() + cls._raise_unavailable() + if batch.status != TestBatchStatus.ACTIVE.value: + cls._raise_unavailable() + + return TestBatchScope( + batch_id=batch_id, + mode='TEST', + status=TestBatchStatus.ACTIVE, + writable=True, + ) + + @classmethod + async def lock_active_batch(cls, db: AsyncSession, scope: TestBatchScope) -> BizTestBatch | None: + """在写入前锁定并重新确认测试批次仍为 ACTIVE。""" + if scope.mode == 'FORMAL': + return None + + batch = await TestBatchDao.get_batch_for_update(db, scope.batch_id) + if batch is None or batch.status == TestBatchStatus.PURGED.value: + cls._raise_unavailable() + if batch.status == TestBatchStatus.CLOSED.value: + cls._raise_closed() + if batch.status != TestBatchStatus.ACTIVE.value: + cls._raise_unavailable() + return batch diff --git a/shierfeng-fastapi-backend/tests/module_business/test_batch_fixtures.py b/shierfeng-fastapi-backend/tests/module_business/test_batch_fixtures.py new file mode 100644 index 0000000..64df883 --- /dev/null +++ b/shierfeng-fastapi-backend/tests/module_business/test_batch_fixtures.py @@ -0,0 +1,28 @@ +from module_admin.entity.vo.user_vo import CurrentUserModel, UserInfoModel +from module_business.domain.enums import TestBatchStatus +from module_business.domain.test_batch_scope import TestBatchScope + + +def make_current_user( + user_id: int = 7, + permissions: list[str] | None = None, + *, + admin: bool = False, +) -> CurrentUserModel: + return CurrentUserModel( + permissions=permissions or [], + roles=[], + user=UserInfoModel( + userId=user_id, + userName='admin' if user_id == 1 else f'user-{user_id}', + admin=admin, + ), + ) + + +def make_admin() -> CurrentUserModel: + return make_current_user(1, ['*:*:*'], admin=True) + + +def make_test_scope(batch_id: int, status: str = 'ACTIVE', writable: bool = True) -> TestBatchScope: + return TestBatchScope(batch_id=batch_id, mode='TEST', status=TestBatchStatus(status), writable=writable) diff --git a/shierfeng-fastapi-backend/tests/module_business/test_test_batch_access.py b/shierfeng-fastapi-backend/tests/module_business/test_test_batch_access.py new file mode 100644 index 0000000..46e182c --- /dev/null +++ b/shierfeng-fastapi-backend/tests/module_business/test_test_batch_access.py @@ -0,0 +1,402 @@ +import unittest +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +from fastapi import params +from sqlalchemy.dialects import postgresql +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=postgresql.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) + + +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()