import hashlib import inspect import json import unittest from datetime import datetime from types import SimpleNamespace from typing import Any, get_type_hints from unittest.mock import AsyncMock, MagicMock, patch from fastapi import FastAPI from fastapi.routing import APIRoute from pydantic import ValidationError from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from common.aspect.interface_auth import CheckUserInterfaceAuth from common.aspect.pre_auth import PreAuth from common.vo import CrudResponseModel, DataResponseModel, PageModel, PageResponseModel, ResponseBaseModel from exceptions.exception import PermissionException, ServiceException from module_admin.entity.vo.user_vo import CurrentUserModel, UserInfoModel from module_admin.service.login_service import LoginService from module_business.controller import filling_controller as filling_controller_module from module_business.dao.access_dao import AccessDao from module_business.dao.inventory_dao import InventoryDao from module_business.dao.master_dao import MasterDao from module_business.dao.operation_dao import OperationDao from module_business.domain.enums import PalletEventType, PalletFlowState from module_business.entity.do.master_do import BizPallet, BizStrain from module_business.entity.do.operation_do import BizFillingRecord, BizPalletEvent, BizPalletInventory from module_business.entity.vo.operation_vo import ( BusinessOptionModel, FillingCreateModel, FillingModel, FillingPageQueryModel, PalletScanResponseModel, ) from module_business.service.access_service import AccessService from module_business.service.filling_service import FillingService def integrity_error() -> IntegrityError: return IntegrityError('statement', {}, RuntimeError('constraint')) def make_user(user_id: int = 7) -> CurrentUserModel: return CurrentUserModel( permissions=[], roles=[], user=UserInfoModel(userId=user_id, userName='admin' if user_id == 1 else f'user-{user_id}'), ) def make_page( *, qr_token: str = 'qr-2', warehouse_id: int = 7, strain_id: int = 11, batch_no: str = 'B-20260726', package_spec: str = '20kg', initial_packages: int = 18, idempotency_key: str = 'fill-key', ) -> FillingCreateModel: return FillingCreateModel( qrToken=qr_token, warehouseId=warehouse_id, strainId=strain_id, batchNo=batch_no, packageSpec=package_spec, initialPackages=initial_packages, idempotencyKey=idempotency_key, ) def make_pallet( pallet_id: int = 2, *, qr_token: str = 'qr-2', capacity_packages: int = 20, asset_status: str = '0', ) -> BizPallet: return BizPallet( pallet_id=pallet_id, pallet_code=f'P-{pallet_id:03d}', qr_token=qr_token, capacity_packages=capacity_packages, asset_status=asset_status, ) def make_inventory( pallet_id: int = 2, *, flow_state: str = PalletFlowState.EMPTY_IN_WAREHOUSE.value, warehouse_id: int | None = 7, site_id: int | None = None, current_line_id: int | None = None, active_filling_id: int | None = None, remaining_packages: int = 0, last_event_id: int | None = 40, version: int = 3, ) -> BizPalletInventory: return BizPalletInventory( pallet_id=pallet_id, flow_state=flow_state, warehouse_id=warehouse_id, site_id=site_id, current_line_id=current_line_id, active_filling_id=active_filling_id, remaining_packages=remaining_packages, last_event_id=last_event_id, version=version, updated_at=datetime(2026, 7, 25, 9, 0), ) def make_strain( strain_id: int = 11, *, status: str = '0', code: str = 'S-011', name: str = '香菇 11', ) -> BizStrain: return BizStrain( strain_id=strain_id, strain_code=code, strain_name=name, status=status, ) def make_filling( filling_id: int = 31, *, warehouse_id: int = 7, idempotency_key: str = 'fill-key', request_hash: str | None = None, ) -> BizFillingRecord: page = make_page(warehouse_id=warehouse_id, idempotency_key=idempotency_key) return BizFillingRecord( filling_id=filling_id, filling_no=f'FL-{filling_id}', pallet_id=2, warehouse_id=warehouse_id, strain_id=11, strain_code_snapshot='S-011', strain_name_snapshot='香菇 11', batch_no=page.batch_no, package_spec=page.package_spec, initial_packages=page.initial_packages, active_slot='ACTIVE', idempotency_key=idempotency_key, request_hash=request_hash or FillingService._build_request_hash(page), operator_id=7, operated_at=datetime(2026, 7, 26, 10, 30), create_by='user-7', create_time=datetime(2026, 7, 26, 10, 30), update_by='user-7', update_time=datetime(2026, 7, 26, 10, 30), ) class TestFillingPublicContract(unittest.TestCase): def test_service_and_controller_are_available(self) -> None: self.assertIsNotNone(FillingService) self.assertIsNotNone(filling_controller_module.filling_controller) def test_fill_signature_and_types_are_exact(self) -> None: signature = inspect.signature(FillingService.fill_pallet_services) self.assertEqual(('query_db', 'page_object', 'current_user'), tuple(signature.parameters)) hints = get_type_hints(FillingService.fill_pallet_services) self.assertIs(AsyncSession, hints['query_db']) self.assertIs(FillingCreateModel, hints['page_object']) self.assertIs(CurrentUserModel, hints['current_user']) self.assertIs(CrudResponseModel, hints['return']) def test_request_hash_is_stable_normalized_sha256_and_excludes_key(self) -> None: page = make_page(batch_no=' B-20260726 ', package_spec=' 20kg ', idempotency_key='key-a') expected_payload = json.dumps( { 'batch_no': 'B-20260726', 'initial_packages': 18, 'package_spec': '20kg', 'qr_token': 'qr-2', 'strain_id': 11, 'warehouse_id': 7, }, sort_keys=True, separators=(',', ':'), ).encode() expected = hashlib.sha256(expected_payload).hexdigest() self.assertEqual(expected, FillingService._build_request_hash(page)) self.assertEqual( expected, FillingService._build_request_hash(page.model_copy(update={'idempotency_key': 'key-b'})) ) self.assertNotEqual( expected, FillingService._build_request_hash(page.model_copy(update={'initial_packages': 17})) ) def test_create_model_requires_strict_positive_integer_packages(self) -> None: for invalid in (True, 0, -1, 1.5): with self.subTest(invalid=invalid), self.assertRaises(ValidationError): make_page(initial_packages=invalid) class TestFillingReadServices(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.db = MagicMock(spec=AsyncSession) self.user = make_user() async def test_list_pushes_user_warehouse_scope_and_converts_rows(self) -> None: query = FillingPageQueryModel(pageNum=1, pageSize=10) row = make_filling() page = PageModel(rows=[row], pageNum=1, pageSize=10, total=1, hasNext=False) get_list = AsyncMock(return_value=page) with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[9, 7, 7])), patch.object(OperationDao, 'get_filling_list', new=get_list), ): result = await FillingService.get_filling_list_services(self.db, query, self.user) get_list.assert_awaited_once_with(self.db, query, True, warehouse_ids=[7, 9]) self.assertIsInstance(result.rows[0], FillingModel) self.assertEqual(1, result.total) async def test_list_with_empty_scope_returns_empty_page_without_global_rows(self) -> None: query = FillingPageQueryModel() empty_page = PageModel(rows=[], pageNum=1, pageSize=10, total=0, hasNext=False) get_list = AsyncMock(return_value=empty_page) with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[])), patch.object(OperationDao, 'get_filling_list', new=get_list), ): result = await FillingService.get_filling_list_services(self.db, query, self.user) get_list.assert_awaited_once_with(self.db, query, True, warehouse_ids=[]) self.assertEqual([], result.rows) self.assertEqual(0, result.total) async def test_list_rejects_explicit_warehouse_outside_scope(self) -> None: query = FillingPageQueryModel(warehouseId=8) get_list = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(OperationDao, 'get_filling_list', new=get_list), self.assertRaises(PermissionException), ): await FillingService.get_filling_list_services(self.db, query, self.user) get_list.assert_not_awaited() async def test_admin_list_is_unscoped(self) -> None: query = FillingPageQueryModel() get_list = AsyncMock(return_value=[]) scope = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=scope), patch.object(OperationDao, 'get_filling_list', new=get_list), ): result = await FillingService.get_filling_list_services(self.db, query, make_user(1), is_page=False) scope.assert_not_awaited() get_list.assert_awaited_once_with(self.db, query, False, warehouse_ids=None) self.assertEqual([], result) async def test_detail_is_scoped_before_lookup_and_hides_missing_for_normal_user(self) -> None: get_detail = AsyncMock(return_value=None) with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(OperationDao, 'get_filling_by_id', new=get_detail), self.assertRaises(PermissionException) as raised, ): await FillingService.filling_detail_services(self.db, 31, self.user) get_detail.assert_awaited_once_with(self.db, 31, warehouse_ids=[7]) self.assertEqual('无权访问该装填记录', raised.exception.message) async def test_admin_detail_keeps_precise_not_found_error(self) -> None: get_detail = AsyncMock(return_value=None) with ( patch.object(OperationDao, 'get_filling_by_id', new=get_detail), self.assertRaises(ServiceException) as raised, ): await FillingService.filling_detail_services(self.db, 31, make_user(1)) get_detail.assert_awaited_once_with(self.db, 31, warehouse_ids=None) self.assertEqual('装填记录不存在', raised.exception.message) async def test_strain_options_return_only_enabled_business_options(self) -> None: strains = [ {'strainId': 3, 'strainCode': 'S-3', 'strainName': '菌种三', 'status': '0'}, {'strainId': 2, 'strainCode': 'S-2', 'strainName': '菌种二', 'status': '1'}, {'strainId': 1, 'strainCode': 'S-1', 'strainName': '菌种一', 'status': '0'}, ] list_strain = AsyncMock(return_value=strains) with patch.object(MasterDao, 'list_strain', new=list_strain): result = await FillingService.get_strain_options_services(self.db, self.user) query = list_strain.await_args.args[1] self.assertEqual('0', query.status) self.assertFalse(list_strain.await_args.args[2]) self.assertEqual([1, 3], [option.id for option in result]) self.assertTrue(all(isinstance(option, BusinessOptionModel) for option in result)) async def test_scan_derives_scope_from_inventory_and_is_read_only(self) -> None: pallet = make_pallet() inventory = make_inventory() access = AsyncMock() pallet_lock = AsyncMock() inventory_lock = AsyncMock() with ( patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=pallet)), patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=inventory)), patch.object(AccessService, 'assert_warehouse_access', new=access), patch.object(InventoryDao, 'get_inventory_for_update', new=inventory_lock), patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_lock), ): result = await FillingService.scan_filling_pallet_services(self.db, 'qr-2', self.user) access.assert_awaited_once_with(self.db, self.user, 7) inventory_lock.assert_not_awaited() pallet_lock.assert_not_awaited() self.db.commit.assert_not_awaited() self.assertEqual(PalletFlowState.EMPTY_IN_WAREHOUSE, result.flow_state) self.assertEqual(['filling'], result.allowed_actions) self.assertEqual(20, result.capacity_packages) self.assertEqual(0, result.remaining_packages) async def test_scan_checks_access_before_exposing_invalid_state(self) -> None: access = AsyncMock(side_effect=PermissionException(data='', message='无权访问该仓库')) with ( patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=make_pallet())), patch.object( MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=make_inventory(flow_state=PalletFlowState.FILLED_IN_WAREHOUSE.value)), ), patch.object(AccessService, 'assert_warehouse_access', new=access), self.assertRaises(PermissionException), ): await FillingService.scan_filling_pallet_services(self.db, 'qr-2', self.user) access.assert_awaited_once_with(self.db, self.user, 7) async def test_scan_hides_nonexistent_pallet_from_normal_user_but_admin_gets_precise_error(self) -> None: for user, exception_type in ((self.user, PermissionException), (make_user(1), ServiceException)): with ( self.subTest(admin=user.user.admin), patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=None)), self.assertRaises(exception_type), ): await FillingService.scan_filling_pallet_services(self.db, 'missing', user) async def test_scan_admin_distinguishes_missing_inventory(self) -> None: with ( patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=make_pallet())), patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=None)), self.assertRaises(ServiceException) as raised, ): await FillingService.scan_filling_pallet_services(self.db, 'qr-2', make_user(1)) self.assertEqual('托盘库存不存在', raised.exception.message) class TestFillPalletService(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.db = MagicMock(spec=AsyncSession) self.user = make_user() self.page = make_page() self.pallet = make_pallet() self.inventory = make_inventory() self.strain = make_strain() self.warehouse = SimpleNamespace(warehouse_id=7, status='0') self.fixed_time = datetime(2026, 7, 26, 10, 30) async def test_fill_locks_inventory_then_pallet_and_writes_one_atomic_event( # noqa: PLR0915 self, ) -> None: actions: list[str] = [] captured: dict[str, Any] = {} async def lock_inventory(*_args: Any) -> BizPalletInventory: actions.append('inventory-lock') return self.inventory async def lock_pallets(*_args: Any) -> list[BizPallet]: actions.append('pallet-lock') return [self.pallet] async def get_strain(*_args: Any) -> BizStrain: actions.append('strain-read') return self.strain async def add_filling(_db: AsyncSession, filling: BizFillingRecord) -> BizFillingRecord: actions.append('filling-flush') filling.filling_id = 31 captured['filling'] = filling return filling async def add_event( _db: AsyncSession, inventory: BizPalletInventory, event: BizPalletEvent, ) -> BizPalletEvent: actions.append('event-flush') event.event_id = 91 event.event_seq = 6 inventory.last_event_id = 91 captured['event'] = event return event async def commit() -> None: actions.append('commit') captured['filling'].__dict__.pop('filling_id', None) self.db.commit.side_effect = commit with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object(OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(return_value=None)), patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(side_effect=lock_inventory)), patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=lock_pallets)), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), patch.object(MasterDao, 'get_strain', new=AsyncMock(side_effect=get_strain)), patch.object(OperationDao, 'add_filling_record', new=AsyncMock(side_effect=add_filling)), patch.object(InventoryDao, 'add_event', new=AsyncMock(side_effect=add_event)), patch('module_business.service.filling_service.datetime') as mocked_datetime, ): mocked_datetime.now.return_value = self.fixed_time result = await FillingService.fill_pallet_services(self.db, self.page, self.user) self.assertEqual( ['inventory-lock', 'pallet-lock', 'strain-read', 'filling-flush', 'event-flush', 'commit'], actions, ) filling = captured['filling'] self.assertTrue(filling.filling_no.startswith('FL-')) self.assertEqual(2, filling.pallet_id) self.assertEqual(7, filling.warehouse_id) self.assertEqual(11, filling.strain_id) self.assertEqual('S-011', filling.strain_code_snapshot) self.assertEqual('香菇 11', filling.strain_name_snapshot) self.assertEqual('B-20260726', filling.batch_no) self.assertEqual('20kg', filling.package_spec) self.assertEqual(18, filling.initial_packages) self.assertEqual('ACTIVE', filling.active_slot) self.assertEqual('fill-key', filling.idempotency_key) self.assertEqual(FillingService._build_request_hash(self.page), filling.request_hash) self.assertEqual(7, filling.operator_id) self.assertEqual(self.fixed_time, filling.operated_at) self.assertEqual('user-7', filling.create_by) self.assertEqual(self.fixed_time, filling.create_time) self.assertEqual('user-7', filling.update_by) self.assertEqual(self.fixed_time, filling.update_time) self.assertEqual(31, self.inventory.active_filling_id) self.assertEqual(18, self.inventory.remaining_packages) self.assertEqual(PalletFlowState.FILLED_IN_WAREHOUSE.value, self.inventory.flow_state) self.assertEqual(4, self.inventory.version) self.assertEqual(self.fixed_time, self.inventory.updated_at) self.assertEqual(91, self.inventory.last_event_id) event = captured['event'] self.assertEqual(PalletEventType.FILLING_CONFIRMED.value, event.event_type) self.assertEqual('FILLING', event.source_type) self.assertEqual(31, event.source_id) self.assertEqual(PalletFlowState.EMPTY_IN_WAREHOUSE.value, event.before_state) self.assertEqual(PalletFlowState.FILLED_IN_WAREHOUSE.value, event.after_state) self.assertEqual(7, event.before_warehouse_id) self.assertEqual(7, event.after_warehouse_id) self.assertIsNone(event.before_site_id) self.assertIsNone(event.after_site_id) self.assertIsNone(event.before_line_id) self.assertIsNone(event.after_line_id) self.assertEqual(0, event.before_remaining_packages) self.assertEqual(18, event.after_remaining_packages) self.assertEqual(7, event.operator_id) self.assertEqual(self.fixed_time, event.occurred_at) self.assertEqual('fill-key', event.idempotency_key) self.assertEqual(filling.request_hash, event.request_hash) self.db.commit.assert_awaited_once_with() self.db.rollback.assert_not_awaited() self.assertTrue(result.is_success) self.assertEqual('装填成功', result.message) self.assertEqual(31, result.result) async def _assert_fill_rejected( self, *, page: FillingCreateModel | None = None, pallet: BizPallet | None = None, inventory: BizPalletInventory | None = None, strain: BizStrain | None = None, ) -> None: add_filling = AsyncMock() add_event = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object(OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(return_value=None)), patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=pallet or self.pallet)), patch.object( InventoryDao, 'get_inventory_for_update', new=AsyncMock(return_value=inventory or self.inventory) ), patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[pallet or self.pallet])), patch.object(OperationDao, 'get_active_filling_for_update', new=AsyncMock(return_value=None)), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), patch.object(MasterDao, 'get_strain', new=AsyncMock(return_value=strain or self.strain)), patch.object(OperationDao, 'add_filling_record', new=add_filling), patch.object(InventoryDao, 'add_event', new=add_event), self.assertRaises(ServiceException), ): await FillingService.fill_pallet_services(self.db, page or self.page, self.user) add_filling.assert_not_awaited() add_event.assert_not_awaited() self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() async def test_fill_rejects_packages_over_persisted_pallet_capacity(self) -> None: await self._assert_fill_rejected(page=make_page(initial_packages=21)) async def test_fill_rejects_stopped_pallet_and_stopped_or_missing_strain(self) -> None: await self._assert_fill_rejected(pallet=make_pallet(asset_status='1')) self.db.reset_mock() await self._assert_fill_rejected(strain=make_strain(status='1')) self.db.reset_mock() add_filling = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object(OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(return_value=None)), patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(return_value=self.inventory)), patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[self.pallet])), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), patch.object(MasterDao, 'get_strain', new=AsyncMock(return_value=None)), patch.object(OperationDao, 'add_filling_record', new=add_filling), self.assertRaises(ServiceException), ): await FillingService.fill_pallet_services(self.db, self.page, self.user) add_filling.assert_not_awaited() self.db.rollback.assert_awaited_once() async def test_fill_rechecks_persisted_qr_and_all_empty_inventory_invariants_after_locks(self) -> None: invalid_inventories = ( make_inventory(flow_state=PalletFlowState.FILLED_IN_WAREHOUSE.value), make_inventory(warehouse_id=8), make_inventory(site_id=3), make_inventory(current_line_id=4), make_inventory(active_filling_id=30), make_inventory(remaining_packages=1), ) for inventory in invalid_inventories: with self.subTest(values=vars(inventory)): self.db.reset_mock() await self._assert_fill_rejected(inventory=inventory) self.db.reset_mock() await self._assert_fill_rejected(pallet=make_pallet(qr_token='persisted-other')) async def test_same_idempotency_key_and_hash_returns_existing_result_before_locks(self) -> None: existing = make_filling() inventory_lock = AsyncMock() access = AsyncMock() async def rollback() -> None: existing.__dict__.pop('filling_id', None) self.db.rollback.side_effect = rollback with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessService, 'assert_warehouse_access', new=access), patch.object( OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(return_value=existing) ) as get_key, patch.object(InventoryDao, 'get_inventory_for_update', new=inventory_lock), ): result = await FillingService.fill_pallet_services(self.db, self.page, self.user) get_key.assert_awaited_once_with(self.db, 'fill-key', warehouse_ids=[7]) access.assert_awaited_once_with(self.db, self.user, 7) inventory_lock.assert_not_awaited() self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() self.assertTrue(result.is_success) self.assertEqual(31, result.result) async def test_same_idempotency_key_with_different_hash_is_rejected(self) -> None: existing = make_filling(request_hash='different-hash') with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object(OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(return_value=existing)), self.assertRaises(ServiceException) as raised, ): await FillingService.fill_pallet_services(self.db, self.page, self.user) self.assertIn('幂等键', raised.exception.message) self.db.rollback.assert_awaited_once() async def test_idempotency_retry_cannot_read_outside_current_scope(self) -> None: get_key = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[])), patch.object(OperationDao, 'get_filling_by_idempotency_key', new=get_key), self.assertRaises(PermissionException), ): await FillingService.fill_pallet_services(self.db, self.page, self.user) get_key.assert_not_awaited() self.db.rollback.assert_awaited_once() async def test_fill_hides_qr_whose_locked_inventory_is_outside_actual_scope(self) -> None: foreign_warehouse_id = 8 inventory = make_inventory( warehouse_id=foreign_warehouse_id, flow_state=PalletFlowState.FILLED_IN_WAREHOUSE.value, ) pallet_lock = AsyncMock(return_value=[make_pallet(asset_status='1')]) async def assert_access(_db: AsyncSession, _user: CurrentUserModel, warehouse_id: int) -> None: if warehouse_id == foreign_warehouse_id: raise PermissionException(data='', message='无权访问该仓库') access = AsyncMock(side_effect=assert_access) with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessService, 'assert_warehouse_access', new=access), patch.object(OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(return_value=None)), patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(return_value=inventory)), patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_lock), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), self.assertRaises(PermissionException), ): await FillingService.fill_pallet_services(self.db, self.page, self.user) self.assertEqual([7, foreign_warehouse_id], [call.args[2] for call in access.await_args_list]) pallet_lock.assert_not_awaited() self.db.rollback.assert_awaited_once() async def test_concurrent_same_key_recovers_after_waiting_for_inventory_lock(self) -> None: recovered = make_filling() inventory = make_inventory( flow_state=PalletFlowState.FILLED_IN_WAREHOUSE.value, active_filling_id=31, remaining_packages=18, version=4, ) actions: list[str] = [] async def lock_inventory(*_args: Any) -> BizPalletInventory: actions.append('inventory-lock') return inventory async def lock_pallet(*_args: Any) -> list[BizPallet]: actions.append('pallet-lock') return [self.pallet] async def lock_active(*_args: Any) -> BizFillingRecord: actions.append('filling-lock') return recovered add_filling = AsyncMock() add_event = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object( OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(side_effect=[None, recovered]), ) as get_key, patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(side_effect=lock_inventory)), patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=lock_pallet)), patch.object(OperationDao, 'get_active_filling_for_update', new=AsyncMock(side_effect=lock_active)), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), patch.object(OperationDao, 'add_filling_record', new=add_filling), patch.object(InventoryDao, 'add_event', new=add_event), ): result = await FillingService.fill_pallet_services(self.db, self.page, self.user) self.assertEqual(['inventory-lock', 'pallet-lock', 'filling-lock'], actions) self.assertEqual(2, get_key.await_count) self.assertEqual(2, self.db.rollback.await_count) self.assertEqual(31, result.result) add_filling.assert_not_awaited() add_event.assert_not_awaited() self.db.commit.assert_not_awaited() async def test_concurrent_same_key_with_different_hash_is_stably_rejected(self) -> None: recovered = make_filling(request_hash='different-hash') inventory = make_inventory( flow_state=PalletFlowState.FILLED_IN_WAREHOUSE.value, active_filling_id=31, remaining_packages=18, version=4, ) add_filling = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object( OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(side_effect=[None, recovered]), ) as get_key, patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(return_value=inventory)), patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[self.pallet])), patch.object(OperationDao, 'get_active_filling_for_update', new=AsyncMock(return_value=recovered)), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), patch.object(OperationDao, 'add_filling_record', new=add_filling), self.assertRaises(ServiceException) as raised, ): await FillingService.fill_pallet_services(self.db, self.page, self.user) self.assertEqual('装填失败，幂等键已被其他请求使用', raised.exception.message) self.assertEqual(2, get_key.await_count) self.assertEqual(2, self.db.rollback.await_count) add_filling.assert_not_awaited() self.db.commit.assert_not_awaited() async def test_any_write_failure_rolls_back_without_commit(self) -> None: with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object(OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(return_value=None)), patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(return_value=self.inventory)), patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[self.pallet])), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), patch.object(MasterDao, 'get_strain', new=AsyncMock(return_value=self.strain)), patch.object(OperationDao, 'add_filling_record', new=AsyncMock(side_effect=RuntimeError('flush failed'))), self.assertRaises(RuntimeError), ): await FillingService.fill_pallet_services(self.db, self.page, self.user) self.db.rollback.assert_awaited_once() self.db.commit.assert_not_awaited() async def test_request_hash_failure_also_rolls_back(self) -> None: with ( patch.object(FillingService, '_build_request_hash', side_effect=RuntimeError('hash failed')), self.assertRaises(RuntimeError), ): await FillingService.fill_pallet_services(self.db, self.page, self.user) self.db.rollback.assert_awaited_once() self.db.commit.assert_not_awaited() async def test_integrity_error_recovers_only_exact_authorized_idempotent_result(self) -> None: recovered = make_filling() key_reads = AsyncMock(side_effect=[None, recovered]) access = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessService, 'assert_warehouse_access', new=access), patch.object(OperationDao, 'get_filling_by_idempotency_key', new=key_reads), patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(return_value=self.inventory)), patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[self.pallet])), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), patch.object(MasterDao, 'get_strain', new=AsyncMock(return_value=self.strain)), patch.object(OperationDao, 'add_filling_record', new=AsyncMock(side_effect=integrity_error())), ): result = await FillingService.fill_pallet_services(self.db, self.page, self.user) self.assertEqual(2, key_reads.await_count) self.assertEqual(2, self.db.rollback.await_count) access.assert_any_await(self.db, self.user, 7) self.assertTrue(result.is_success) async def test_integrity_error_without_scoped_idempotent_result_is_hidden_and_never_succeeds(self) -> None: with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object( OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(side_effect=[None, None]), ), patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(return_value=self.inventory)), patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[self.pallet])), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), patch.object(MasterDao, 'get_strain', new=AsyncMock(return_value=self.strain)), patch.object(OperationDao, 'add_filling_record', new=AsyncMock(side_effect=integrity_error())), self.assertRaises(PermissionException), ): await FillingService.fill_pallet_services(self.db, self.page, self.user) self.assertEqual(2, self.db.rollback.await_count) self.db.commit.assert_not_awaited() class TestFillingOperationDaoScope(unittest.IsolatedAsyncioTestCase): @staticmethod def _compiled(statement: Any) -> str: return str(statement.compile(compile_kwargs={'literal_binds': True})).lower() async def test_filling_list_pushes_warehouse_scope_into_sql(self) -> None: query = FillingPageQueryModel() paginate = AsyncMock(return_value=[]) db = MagicMock(spec=AsyncSession) with patch.object(OperationDao, '_paginate', new=paginate): await OperationDao.get_filling_list(db, query, False, warehouse_ids=[9, 7, 7]) sql = self._compiled(paginate.await_args.args[1]) self.assertIn('biz_filling_record.warehouse_id in (7, 9)', sql) async def test_filling_list_empty_scope_compiles_to_always_false_sql(self) -> None: query = FillingPageQueryModel() paginate = AsyncMock(return_value=[]) db = MagicMock(spec=AsyncSession) with patch.object(OperationDao, '_paginate', new=paginate): await OperationDao.get_filling_list(db, query, False, warehouse_ids=[]) sql = self._compiled(paginate.await_args.args[1]) self.assertTrue('false' in sql or '0 = 1' in sql, sql) async def test_filling_detail_and_idempotency_reads_push_scope_into_sql(self) -> None: for method, key in ( (OperationDao.get_filling_by_id, 31), (OperationDao.get_filling_by_idempotency_key, 'fill-key'), ): with self.subTest(method=method.__name__): result = MagicMock() result.scalar_one_or_none.return_value = None db = MagicMock(spec=AsyncSession) db.execute.return_value = result await method(db, key, warehouse_ids=[7]) sql = self._compiled(db.execute.await_args.args[0]) self.assertIn('biz_filling_record.warehouse_id in (7)', sql) class TestFillingController(unittest.TestCase): @staticmethod def _routes() -> dict[tuple[str, str], APIRoute]: return { (route.path, next(iter(route.methods))): route for route in filling_controller_module.filling_controller.routes if isinstance(route, APIRoute) } def test_openapi_has_exact_five_operations_and_response_models(self) -> None: app = FastAPI() app.include_router(filling_controller_module.filling_controller) schema = app.openapi() expected = { ('/business/filling/list', 'get'), ('/business/filling/strain-options', 'get'), ('/business/filling/scan/{qr_token}', 'get'), ('/business/filling/{filling_id}', 'get'), ('/business/filling', 'post'), } self.assertEqual(expected, {(path, method) for path, methods in schema['paths'].items() for method in methods}) scan_parameters = schema['paths']['/business/filling/scan/{qr_token}']['get']['parameters'] self.assertEqual({'qr_token'}, {parameter['name'] for parameter in scan_parameters}) routes = self._routes() self.assertEqual(PageResponseModel[FillingModel], routes[('/business/filling/list', 'GET')].response_model) self.assertEqual( DataResponseModel[list[BusinessOptionModel]], routes[('/business/filling/strain-options', 'GET')].response_model, ) self.assertEqual( DataResponseModel[PalletScanResponseModel], routes[('/business/filling/scan/{qr_token}', 'GET')].response_model, ) self.assertEqual( DataResponseModel[FillingModel], routes[('/business/filling/{filling_id}', 'GET')].response_model, ) self.assertEqual(ResponseBaseModel, routes[('/business/filling', 'POST')].response_model) def test_static_routes_are_declared_before_dynamic_detail(self) -> None: paths = [route.path for route in filling_controller_module.filling_controller.routes] detail_index = paths.index('/business/filling/{filling_id}') self.assertLess(paths.index('/business/filling/strain-options'), detail_index) self.assertLess(paths.index('/business/filling/scan/{qr_token}'), detail_index) def test_each_route_has_exact_permission_and_three_unique_values(self) -> None: expected = { ('/business/filling/list', 'GET'): 'business:filling:list', ('/business/filling/strain-options', 'GET'): 'business:filling:query', ('/business/filling/scan/{qr_token}', 'GET'): 'business:filling:query', ('/business/filling/{filling_id}', 'GET'): 'business:filling:query', ('/business/filling', 'POST'): 'business:filling:add', } actual = {} for key, route in self._routes().items(): permissions = [ dependency.dependency for dependency in route.dependencies if isinstance(dependency.dependency, CheckUserInterfaceAuth) ] self.assertEqual(1, len(permissions), key) self.assertTrue(any(isinstance(dependency.dependency, PreAuth) for dependency in route.dependencies)) actual[key] = permissions[0].perm self.assertEqual(expected, actual) self.assertEqual( {'business:filling:list', 'business:filling:query', 'business:filling:add'}, set(actual.values()), ) def test_all_routes_inject_current_user_and_only_post_has_log(self) -> None: for key, route in self._routes().items(): self.assertTrue( any( dependency.name == 'current_user' and dependency.call == LoginService.get_current_user for dependency in route.dependant.dependencies ), key, ) self.assertEqual(key == ('/business/filling', 'POST'), hasattr(route.endpoint, '__wrapped__')) def test_controller_does_not_add_scan_export_or_draft_permissions(self) -> None: source = inspect.getsource(filling_controller_module) for forbidden in ( 'business:filling:scan', 'business:filling:export', 'business:filling:edit', 'business:filling:remove', 'business:filling:confirm', ): self.assertNotIn(forbidden, source) if __name__ == '__main__': unittest.main()