import hashlib import importlib.util 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 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 shipment_controller as shipment_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.domain.pallet_state_machine import PalletStateMachine from module_business.entity.do.master_do import BizCustomerSite, BizPallet, BizWarehouse from module_business.entity.do.operation_do import ( BizFillingRecord, BizPalletEvent, BizPalletInventory, BizShipmentItem, BizShipmentOrder, ) from module_business.entity.vo.operation_vo import ( BusinessOptionModel, ConfirmCommandModel, PalletScanResponseModel, ShipmentCreateModel, ShipmentModel, ShipmentPageQueryModel, ShipmentScanQueryModel, ShipmentUpdateModel, ) from module_business.service import shipment_service as shipment_service_module from module_business.service.access_service import AccessService from module_business.service.shipment_service import ShipmentService 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 integrity_error() -> IntegrityError: return IntegrityError('statement', {}, RuntimeError('constraint')) def make_pallet(pallet_id: int, *, capacity: int = 20, asset_status: str = '0') -> BizPallet: return BizPallet( pallet_id=pallet_id, pallet_code=f'P-{pallet_id:03d}', qr_token=f'qr-{pallet_id}', capacity_packages=capacity, asset_status=asset_status, ) def make_inventory( pallet_id: int, *, warehouse_id: int = 7, site_id: int | None = None, current_line_id: int | None = None, active_filling_id: int | None = None, remaining_packages: int = 8, flow_state: str = 'FILLED_IN_WAREHOUSE', version: int = 3, ) -> BizPalletInventory: return BizPalletInventory( pallet_id=pallet_id, warehouse_id=warehouse_id, site_id=site_id, current_line_id=current_line_id, active_filling_id=active_filling_id if active_filling_id is not None else 100 + pallet_id, remaining_packages=remaining_packages, flow_state=flow_state, version=version, ) def make_filling(pallet_id: int, *, warehouse_id: int = 7) -> BizFillingRecord: return BizFillingRecord( filling_id=100 + pallet_id, filling_no=f'FI-{pallet_id}', pallet_id=pallet_id, warehouse_id=warehouse_id, strain_id=3, strain_code_snapshot='S-03', strain_name_snapshot='菌种三', batch_no='B-202607', package_spec='袋装', initial_packages=20, active_slot='ACTIVE', idempotency_key=f'fill-{pallet_id}', request_hash='f' * 64, operator_id=7, operated_at=SimpleNamespace(), ) def make_shipment( shipment_id: int = 12, *, source_warehouse_id: int = 7, target_site_id: int = 9, pallet_ids: tuple[int, ...] = (5, 2), status: str = 'DRAFT', idempotency_key: str | None = None, request_hash: str | None = None, ) -> BizShipmentOrder: shipment = BizShipmentOrder( shipment_id=shipment_id, shipment_no=f'SH-{shipment_id}', source_warehouse_id=source_warehouse_id, target_site_id=target_site_id, status=status, confirm_idempotency_key=idempotency_key, confirm_request_hash=request_hash, ) shipment.items = [ BizShipmentItem( shipment_item_id=index, shipment_id=shipment_id, pallet_id=pallet_id, ) for index, pallet_id in enumerate(pallet_ids, start=1) ] return shipment class ShipmentServiceBootstrapTest(unittest.TestCase): def test_shipment_service_module_exists(self) -> None: spec = importlib.util.find_spec('module_business.service.shipment_service') self.assertIsNotNone(spec, 'ShipmentService 尚未实现') def test_confirm_signature_and_types_are_exact(self) -> None: self.assertTrue( hasattr(ShipmentService, 'confirm_shipment_services'), 'confirm_shipment_services 尚未实现', ) signature = inspect.signature(ShipmentService.confirm_shipment_services) self.assertEqual( ('query_db', 'shipment_id', 'page_object', 'current_user'), tuple(signature.parameters), ) hints = get_type_hints(ShipmentService.confirm_shipment_services) self.assertIs(AsyncSession, hints['query_db']) self.assertIs(int, hints['shipment_id']) self.assertIs(ConfirmCommandModel, hints['page_object']) self.assertIs(CurrentUserModel, hints['current_user']) self.assertIs(CrudResponseModel, hints['return']) def test_confirm_hash_is_canonical_and_only_contains_shipment_id(self) -> None: self.assertTrue( hasattr(ShipmentService, '_build_confirm_request_hash'), '_build_confirm_request_hash 尚未实现', ) expected = hashlib.sha256( json.dumps( {'shipment_id': 12}, sort_keys=True, separators=(',', ':'), ).encode() ).hexdigest() self.assertEqual(expected, ShipmentService._build_confirm_request_hash(12)) self.assertNotEqual(expected, ShipmentService._build_confirm_request_hash(13)) def test_service_exposes_complete_shipment_workflow(self) -> None: expected = { 'get_shipment_list_services', 'shipment_detail_services', 'get_warehouse_options_services', 'get_site_options_services', 'scan_shipment_pallet_services', 'add_shipment_services', 'edit_shipment_services', 'delete_shipment_services', 'confirm_shipment_services', } self.assertEqual( expected, {name for name in expected if hasattr(ShipmentService, name)}, ) class ShipmentInfrastructureContractTest(unittest.TestCase): def test_shipment_dao_reads_accept_both_data_scopes(self) -> None: expected = { OperationDao.get_shipment_list: ( 'db', 'query_object', 'is_page', 'warehouse_ids', 'site_ids', ), OperationDao.get_shipment_by_id: ( 'db', 'shipment_id', 'warehouse_ids', 'site_ids', ), OperationDao.get_shipment_for_update: ( 'db', 'shipment_id', 'warehouse_ids', 'site_ids', ), OperationDao.get_shipment_by_confirm_idempotency_key: ( 'db', 'idempotency_key', 'warehouse_ids', 'site_ids', ), } for method, parameter_names in expected.items(): with self.subTest(method=method.__name__): self.assertEqual(parameter_names, tuple(inspect.signature(method).parameters)) def test_site_access_supports_optional_current_read(self) -> None: for method in (AccessDao.user_has_site, AccessService.assert_site_access): with self.subTest(method=method.__qualname__): parameter = inspect.signature(method).parameters.get('for_update') self.assertIsNotNone(parameter) self.assertIs(inspect.Parameter.KEYWORD_ONLY, parameter.kind) self.assertIs(False, parameter.default) class ShipmentControllerBootstrapTest(unittest.TestCase): def test_shipment_controller_module_exists(self) -> None: spec = importlib.util.find_spec('module_business.controller.shipment_controller') self.assertIsNotNone(spec, 'shipment_controller 尚未实现') class ShipmentControllerContractTest(unittest.TestCase): @staticmethod def _routes() -> dict[tuple[str, str], APIRoute]: return { (route.path, next(iter(route.methods))): route for route in shipment_controller_module.shipment_controller.routes if isinstance(route, APIRoute) } def test_openapi_has_nine_operations_aliases_and_response_models(self) -> None: app = FastAPI() app.include_router(shipment_controller_module.shipment_controller) schema = app.openapi() expected = { ('/business/shipment/list', 'get'), ('/business/shipment/warehouse-options', 'get'), ('/business/shipment/site-options', 'get'), ('/business/shipment/scan/{qr_token}', 'get'), ('/business/shipment/{shipment_id}', 'get'), ('/business/shipment', 'post'), ('/business/shipment', 'put'), ('/business/shipment/{shipment_ids}', 'delete'), ('/business/shipment/{shipment_id}/confirm', 'put'), } self.assertEqual(expected, {(path, method) for path, methods in schema['paths'].items() for method in methods}) scan_parameters = schema['paths']['/business/shipment/scan/{qr_token}']['get']['parameters'] self.assertEqual( {'qr_token', 'sourceWarehouseId', 'targetSiteId'}, {parameter['name'] for parameter in scan_parameters}, ) routes = self._routes() self.assertEqual(PageResponseModel[ShipmentModel], routes[('/business/shipment/list', 'GET')].response_model) for path in ('warehouse-options', 'site-options'): self.assertEqual( DataResponseModel[list[BusinessOptionModel]], routes[(f'/business/shipment/{path}', 'GET')].response_model, ) self.assertEqual( DataResponseModel[PalletScanResponseModel], routes[('/business/shipment/scan/{qr_token}', 'GET')].response_model, ) self.assertEqual( DataResponseModel[ShipmentModel], routes[('/business/shipment/{shipment_id}', 'GET')].response_model, ) for key in ( ('/business/shipment', 'POST'), ('/business/shipment', 'PUT'), ('/business/shipment/{shipment_ids}', 'DELETE'), ('/business/shipment/{shipment_id}/confirm', 'PUT'), ): self.assertEqual(ResponseBaseModel, routes[key].response_model) def test_static_options_and_scan_precede_dynamic_detail(self) -> None: paths = [route.path for route in shipment_controller_module.shipment_controller.routes] detail_index = paths.index('/business/shipment/{shipment_id}') for static_path in ( '/business/shipment/warehouse-options', '/business/shipment/site-options', '/business/shipment/scan/{qr_token}', ): self.assertLess(paths.index(static_path), detail_index) def test_routes_have_exact_permissions_current_user_and_write_logs(self) -> None: expected = { ('/business/shipment/list', 'GET'): 'business:shipment:list', ('/business/shipment/warehouse-options', 'GET'): 'business:shipment:query', ('/business/shipment/site-options', 'GET'): 'business:shipment:query', ('/business/shipment/scan/{qr_token}', 'GET'): 'business:shipment:query', ('/business/shipment/{shipment_id}', 'GET'): 'business:shipment:query', ('/business/shipment', 'POST'): 'business:shipment:add', ('/business/shipment', 'PUT'): 'business:shipment:edit', ('/business/shipment/{shipment_ids}', 'DELETE'): 'business:shipment:remove', ('/business/shipment/{shipment_id}/confirm', 'PUT'): 'business:shipment:confirm', } 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)) self.assertTrue( any( dependency.name == 'current_user' and dependency.call == LoginService.get_current_user for dependency in route.dependant.dependencies ), key, ) if key[1] in {'POST', 'PUT', 'DELETE'}: self.assertTrue(hasattr(route.endpoint, '__wrapped__'), key) actual[key] = permissions[0].perm self.assertEqual(expected, actual) self.assertEqual(6, len(set(actual.values()))) def test_controller_defines_no_scan_or_export_permission(self) -> None: source = inspect.getsource(shipment_controller_module) self.assertNotIn('business:shipment:scan', source) self.assertNotIn('business:shipment:export', source) class ShipmentReadServiceTest(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.db = MagicMock(spec=AsyncSession) self.user = make_user() async def test_empty_either_scope_returns_empty_page_without_global_query(self) -> None: query = ShipmentPageQueryModel(pageNum=2, pageSize=7) get_list = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[])), patch.object(OperationDao, 'get_shipment_list', new=get_list), ): try: result = await ShipmentService.get_shipment_list_services( self.db, query, self.user, is_page=True, ) except NotImplementedError: self.fail('双范围空集处理尚未实现') self.assertIsInstance(result, PageModel) self.assertEqual([], result.rows) self.assertEqual(0, result.total) self.assertEqual(2, result.page_num) self.assertEqual(7, result.page_size) self.assertFalse(result.has_next) get_list.assert_not_awaited() async def test_list_and_detail_push_both_scopes_before_reading(self) -> None: shipment = make_shipment() page = PageModel(rows=[shipment], pageNum=1, pageSize=10, total=1, hasNext=False) get_list = AsyncMock(return_value=page) get_detail = AsyncMock(return_value=shipment) with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 7])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 9])), patch.object(OperationDao, 'get_shipment_list', new=get_list), patch.object(OperationDao, 'get_shipment_by_id', new=get_detail), ): listed = await ShipmentService.get_shipment_list_services( self.db, ShipmentPageQueryModel(), self.user, ) try: detail = await ShipmentService.shipment_detail_services(self.db, 12, self.user) except NotImplementedError: self.fail('双范围单据详情尚未实现') self.assertEqual(1, listed.total) self.assertEqual(12, detail.shipment_id) get_list.assert_awaited_once_with( self.db, unittest.mock.ANY, True, warehouse_ids=[7], site_ids=[9], ) get_detail.assert_awaited_once_with( self.db, 12, warehouse_ids=[7], site_ids=[9], ) async def test_options_only_return_enabled_assigned_warehouse_and_site(self) -> None: warehouses = [ BizWarehouse(warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status='0'), BizWarehouse(warehouse_id=8, warehouse_code='W-8', warehouse_name='八号仓', status='0'), BizWarehouse(warehouse_id=10, warehouse_code='W-10', warehouse_name='停用仓', status='1'), ] sites = [ BizCustomerSite(site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status='0'), BizCustomerSite(site_id=11, customer_id=1, site_code='S-11', site_name='十一号现场', status='0'), BizCustomerSite(site_id=12, customer_id=1, site_code='S-12', site_name='停用现场', status='1'), ] with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), patch.object(MasterDao, 'list_warehouse', new=AsyncMock(return_value=warehouses)), patch.object(MasterDao, 'list_site', new=AsyncMock(return_value=sites)), ): try: warehouse_options = await ShipmentService.get_warehouse_options_services(self.db, self.user) site_options = await ShipmentService.get_site_options_services(self.db, self.user) except NotImplementedError: self.fail('发货仓库/现场选项尚未实现') self.assertEqual([(7, 'W-7', '七号仓')], [(item.id, item.code, item.name) for item in warehouse_options]) self.assertEqual([(9, 'S-9', '九号现场')], [(item.id, item.code, item.name) for item in site_options]) async def test_scan_is_read_only_and_returns_backend_remaining_packages(self) -> None: query = ShipmentScanQueryModel(sourceWarehouseId=7, targetSiteId=9) pallet = make_pallet(2) inventory = make_inventory(2, remaining_packages=8) filling = make_filling(2) lock_inventory = AsyncMock() lock_pallets = AsyncMock() lock_filling = AsyncMock() with ( patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object(AccessService, 'assert_site_access', new=AsyncMock()), patch.object( MasterDao, 'get_warehouse', new=AsyncMock( return_value=BizWarehouse( warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status='0', ) ), ), patch.object( MasterDao, 'get_site', new=AsyncMock( return_value=BizCustomerSite( site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status='0', ) ), ), 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(OperationDao, 'get_filling_by_id', new=AsyncMock(return_value=filling)), patch.object(InventoryDao, 'get_inventories_for_update', new=lock_inventory), patch.object(InventoryDao, 'get_pallets_by_ids', new=lock_pallets), patch.object(OperationDao, 'get_active_filling_for_update', new=lock_filling), ): try: result = await ShipmentService.scan_shipment_pallet_services( self.db, 'qr-2', query, self.user, ) except NotImplementedError: self.fail('整托发货扫码尚未实现') self.assertEqual(2, result.pallet_id) self.assertEqual(PalletFlowState.FILLED_IN_WAREHOUSE, result.flow_state) self.assertEqual(8, result.remaining_packages) self.assertEqual(['shipment'], result.allowed_actions) lock_inventory.assert_not_awaited() lock_pallets.assert_not_awaited() lock_filling.assert_not_awaited() self.db.commit.assert_not_awaited() self.db.rollback.assert_not_awaited() async def test_explicit_filter_outside_either_scope_is_rejected_before_dao(self) -> None: cases = ( (ShipmentPageQueryModel(sourceWarehouseId=8), '无权访问该仓库'), (ShipmentPageQueryModel(targetSiteId=10), '无权访问该客户现场'), ) for query, expected_message in cases: with self.subTest(expected_message=expected_message): get_list = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), patch.object(OperationDao, 'get_shipment_list', new=get_list), self.assertRaises(PermissionException) as raised, ): await ShipmentService.get_shipment_list_services(self.db, query, self.user) self.assertEqual(expected_message, raised.exception.message) get_list.assert_not_awaited() async def test_missing_detail_is_hidden_for_user_but_precise_for_admin(self) -> None: get_detail = AsyncMock(return_value=None) with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), patch.object(OperationDao, 'get_shipment_by_id', new=get_detail), self.assertRaises(PermissionException) as user_raised, ): await ShipmentService.shipment_detail_services(self.db, 404, self.user) self.assertEqual('无权访问该发货单', user_raised.exception.message) get_detail.assert_awaited_once_with( self.db, 404, warehouse_ids=[7], site_ids=[9], ) get_detail.reset_mock() with ( patch.object(OperationDao, 'get_shipment_by_id', new=get_detail), self.assertRaises(ServiceException) as admin_raised, ): await ShipmentService.shipment_detail_services(self.db, 404, make_user(1)) self.assertEqual('发货单不存在', admin_raised.exception.message) get_detail.assert_awaited_once_with( self.db, 404, warehouse_ids=None, site_ids=None, ) async def test_scan_hides_missing_qr_external_warehouse_and_site_location(self) -> None: route_mocks = ( patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object(AccessService, 'assert_site_access', new=AsyncMock()), patch.object( MasterDao, 'get_warehouse', new=AsyncMock( return_value=BizWarehouse( warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status='0', ) ), ), patch.object( MasterDao, 'get_site', new=AsyncMock( return_value=BizCustomerSite( site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status='0', ) ), ), ) scenarios = ( (None, None), (make_pallet(2), make_inventory(2, warehouse_id=8)), (make_pallet(2), make_inventory(2, site_id=11)), ) for pallet, inventory in scenarios: with self.subTest(pallet=pallet, inventory=inventory): with ( route_mocks[0], route_mocks[1], route_mocks[2], route_mocks[3], patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=pallet)), patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=inventory)), self.assertRaises(PermissionException) as raised, ): await ShipmentService.scan_shipment_pallet_services( self.db, 'hidden-token', ShipmentScanQueryModel(sourceWarehouseId=7, targetSiteId=9), self.user, ) self.assertEqual('无权访问该托盘', raised.exception.message) class ShipmentDraftServiceTest(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.db = MagicMock(spec=AsyncSession) self.user = make_user() self.fixed_time = datetime(2026, 7, 26, 16, 30) async def test_add_only_persists_pallet_ids_with_null_snapshots(self) -> None: # noqa: PLR0915 page = ShipmentCreateModel(sourceWarehouseId=7, targetSiteId=9, palletIds=[5, 2]) inventories = [make_inventory(2), make_inventory(5)] pallets = [make_pallet(2), make_pallet(5)] fillings = {102: make_filling(2), 105: make_filling(5)} captured: dict[str, Any] = {} actions: list[str] = [] async def warehouse_access(*_args: Any, **_kwargs: Any) -> None: actions.append('warehouse-access') async def site_access(*_args: Any, **_kwargs: Any) -> None: actions.append('site-access') async def get_warehouse(*_args: Any, **_kwargs: Any) -> BizWarehouse: actions.append('warehouse-read') return BizWarehouse( warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status='0', ) async def get_site(*_args: Any, **_kwargs: Any) -> BizCustomerSite: actions.append('site-read') return BizCustomerSite( site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status='0', ) async def lock_inventories(*_args: Any) -> list[BizPalletInventory]: actions.append('inventory-lock') return inventories async def lock_pallets(*_args: Any) -> list[BizPallet]: actions.append('pallet-lock') return pallets async def lock_filling(_db: AsyncSession, filling_id: int) -> BizFillingRecord: actions.append(f'filling-lock-{filling_id}') return fillings[filling_id] async def add_order(_db: AsyncSession, shipment: BizShipmentOrder) -> BizShipmentOrder: actions.append('add-order') shipment.shipment_id = 12 captured['shipment'] = shipment return shipment async def replace_items( _db: AsyncSession, shipment_id: int, items: list[BizShipmentItem], ) -> list[BizShipmentItem]: actions.append('replace-items') captured['items'] = items self.assertEqual(12, shipment_id) return items async def commit() -> None: actions.append('commit') self.db.commit.side_effect = commit add_event = AsyncMock() with ( patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock(side_effect=warehouse_access)), patch.object(AccessService, 'assert_site_access', new=AsyncMock(side_effect=site_access)), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=get_warehouse)), patch.object(MasterDao, 'get_site', new=AsyncMock(side_effect=get_site)), patch.object( InventoryDao, 'get_inventories_for_update', new=AsyncMock(side_effect=lock_inventories), ) as inventory_lock, patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=lock_pallets)) as pallet_lock, patch.object( OperationDao, 'get_active_filling_for_update', new=AsyncMock(side_effect=lock_filling), ), patch.object(OperationDao, 'add_shipment', new=AsyncMock(side_effect=add_order)), patch.object(OperationDao, 'replace_shipment_items', new=AsyncMock(side_effect=replace_items)), patch.object(InventoryDao, 'add_event', new=add_event), patch.object(shipment_service_module, 'datetime', create=True) as service_datetime, patch.object(shipment_service_module, 'secrets', create=True) as service_secrets, ): service_datetime.now.return_value = self.fixed_time service_secrets.token_hex.return_value = 'draft-token' try: result = await ShipmentService.add_shipment_services(self.db, page, self.user) except NotImplementedError: self.fail('发货草稿新增尚未实现') self.assertTrue(result.is_success) self.assertEqual('DRAFT', captured['shipment'].status) self.assertEqual('SH-draft-token', captured['shipment'].shipment_no) self.assertEqual([2, 5], [item.pallet_id for item in captured['items']]) self.assertTrue(all(item.active_filling_id is None for item in captured['items'])) self.assertTrue(all(item.package_qty is None for item in captured['items'])) inventory_lock.assert_awaited_once_with(self.db, [2, 5]) pallet_lock.assert_awaited_once_with(self.db, [2, 5]) self.assertEqual( [ 'warehouse-access', 'warehouse-read', 'site-access', 'site-read', 'inventory-lock', 'pallet-lock', 'filling-lock-102', 'filling-lock-105', 'warehouse-read', 'warehouse-access', 'site-read', 'site-access', 'add-order', 'replace-items', 'commit', ], actions, ) self.assertEqual(['FILLED_IN_WAREHOUSE', 'FILLED_IN_WAREHOUSE'], [item.flow_state for item in inventories]) add_event.assert_not_awaited() self.db.commit.assert_awaited_once() self.db.rollback.assert_not_awaited() async def test_add_rechecks_revoked_mapping_after_pallet_locks_before_writing(self) -> None: page = ShipmentCreateModel(sourceWarehouseId=7, targetSiteId=9, palletIds=[2]) async def warehouse_access(*_args: Any, **kwargs: Any) -> None: if kwargs.get('for_update'): raise PermissionException(data='', message='无权访问该仓库') add_order = AsyncMock(return_value=make_shipment(pallet_ids=(2,))) replace_items = AsyncMock() warehouse_mapping = AsyncMock(side_effect=warehouse_access) with ( patch.object(AccessService, 'assert_warehouse_access', new=warehouse_mapping), patch.object(AccessService, 'assert_site_access', new=AsyncMock()), patch.object( MasterDao, 'get_warehouse', new=AsyncMock( return_value=BizWarehouse( warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status='0', ) ), ) as warehouse_read, patch.object( MasterDao, 'get_site', new=AsyncMock( return_value=BizCustomerSite( site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status='0', ) ), ), patch.object( ShipmentService, '_lock_and_validate_pallets', new=AsyncMock(return_value=([], [], [])), ), patch.object(OperationDao, 'add_shipment', new=add_order), patch.object(OperationDao, 'replace_shipment_items', new=replace_items), self.assertRaises(PermissionException) as raised, ): await ShipmentService.add_shipment_services(self.db, page, self.user) self.assertEqual('无权访问该仓库', raised.exception.message) self.assertEqual(2, warehouse_read.await_count) warehouse_read.assert_awaited_with(self.db, 7, for_update=True) self.assertEqual(2, warehouse_mapping.await_count) warehouse_mapping.assert_awaited_with(self.db, self.user, 7, for_update=True) add_order.assert_not_awaited() replace_items.assert_not_awaited() self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() async def test_add_rechecks_disabled_master_after_pallet_locks_before_writing(self) -> None: page = ShipmentCreateModel(sourceWarehouseId=7, targetSiteId=9, palletIds=[2]) async def get_warehouse(*_args: Any, **kwargs: Any) -> BizWarehouse: return BizWarehouse( warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status='1' if kwargs.get('for_update') else '0', ) add_order = AsyncMock(return_value=make_shipment(pallet_ids=(2,))) replace_items = AsyncMock() warehouse_read = AsyncMock(side_effect=get_warehouse) with ( patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object(AccessService, 'assert_site_access', new=AsyncMock()), patch.object(MasterDao, 'get_warehouse', new=warehouse_read), patch.object( MasterDao, 'get_site', new=AsyncMock( return_value=BizCustomerSite( site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status='0', ) ), ), patch.object( ShipmentService, '_lock_and_validate_pallets', new=AsyncMock(return_value=([], [], [])), ), patch.object(OperationDao, 'add_shipment', new=add_order), patch.object(OperationDao, 'replace_shipment_items', new=replace_items), self.assertRaises(ServiceException) as raised, ): await ShipmentService.add_shipment_services(self.db, page, self.user) self.assertEqual('来源仓库已停用', raised.exception.message) warehouse_read.assert_awaited_with(self.db, 7, for_update=True) add_order.assert_not_awaited() replace_items.assert_not_awaited() self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() async def test_edit_revalidates_current_draft_and_keeps_snapshots_null(self) -> None: page = ShipmentUpdateModel( shipmentId=12, sourceWarehouseId=8, targetSiteId=10, palletIds=[3], ) shipment = make_shipment() inventory = make_inventory(3, warehouse_id=8) pallet = make_pallet(3) filling = make_filling(3, warehouse_id=8) captured: dict[str, Any] = {} async def update_order(_db: AsyncSession, updated: BizShipmentOrder) -> BizShipmentOrder: captured['shipment'] = updated return updated async def replace_items( _db: AsyncSession, _shipment_id: int, items: list[BizShipmentItem], ) -> list[BizShipmentItem]: captured['items'] = items return items with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 8])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 10])), patch.object( OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=shipment), ) as lock_order, patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object(AccessService, 'assert_site_access', new=AsyncMock()), patch.object( MasterDao, 'get_warehouse', new=AsyncMock( return_value=BizWarehouse( warehouse_id=8, warehouse_code='W-8', warehouse_name='八号仓', status='0', ) ), ), patch.object( MasterDao, 'get_site', new=AsyncMock( return_value=BizCustomerSite( site_id=10, customer_id=1, site_code='S-10', site_name='十号现场', status='0', ) ), ), patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=[inventory])), patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[pallet])), patch.object( OperationDao, 'get_active_filling_for_update', new=AsyncMock(return_value=filling), ), patch.object(OperationDao, 'update_shipment', new=AsyncMock(side_effect=update_order)), patch.object(OperationDao, 'replace_shipment_items', new=AsyncMock(side_effect=replace_items)), patch.object(shipment_service_module, 'datetime') as service_datetime, ): service_datetime.now.return_value = self.fixed_time try: result = await ShipmentService.edit_shipment_services(self.db, page, self.user) except NotImplementedError: self.fail('发货草稿编辑尚未实现') self.assertTrue(result.is_success) lock_order.assert_awaited_once_with( self.db, 12, warehouse_ids=[7, 8], site_ids=[9, 10], ) self.assertEqual(8, captured['shipment'].source_warehouse_id) self.assertEqual(10, captured['shipment'].target_site_id) self.assertEqual([3], [item.pallet_id for item in captured['items']]) self.assertIsNone(captured['items'][0].active_filling_id) self.assertIsNone(captured['items'][0].package_qty) self.assertEqual('FILLED_IN_WAREHOUSE', inventory.flow_state) self.db.commit.assert_awaited_once() async def test_edit_rechecks_old_and_new_routes_before_writing(self) -> None: scenarios = ( ('current-warehouse-revoked', PermissionException, '无权访问该发货单'), ('new-site-revoked', PermissionException, '无权访问该客户现场'), ('new-site-disabled', ServiceException, '目标现场已停用'), ) for scenario, exception_type, expected_message in scenarios: with self.subTest(scenario=scenario): db = MagicMock(spec=AsyncSession) shipment = make_shipment(pallet_ids=(2,)) page = ShipmentUpdateModel( shipmentId=12, sourceWarehouseId=8, targetSiteId=10, palletIds=[2], ) current_warehouse_id = shipment.source_warehouse_id new_site_id = page.target_site_id async def warehouse_access( _db: AsyncSession, _user: CurrentUserModel, warehouse_id: int, _scenario: str = scenario, _current_warehouse_id: int = current_warehouse_id, **kwargs: Any, ) -> None: if ( _scenario == 'current-warehouse-revoked' and warehouse_id == _current_warehouse_id and kwargs.get('for_update') ): raise PermissionException(data='', message='无权访问该仓库') async def site_access( _db: AsyncSession, _user: CurrentUserModel, site_id: int, _scenario: str = scenario, _new_site_id: int = new_site_id, **kwargs: Any, ) -> None: if _scenario == 'new-site-revoked' and site_id == _new_site_id and kwargs.get('for_update'): raise PermissionException(data='', message='无权访问该客户现场') async def get_warehouse( _db: AsyncSession, warehouse_id: int, **_kwargs: Any, ) -> BizWarehouse: return BizWarehouse( warehouse_id=warehouse_id, warehouse_code=f'W-{warehouse_id}', warehouse_name=f'{warehouse_id}号仓', status='0', ) async def get_site( _db: AsyncSession, site_id: int, _scenario: str = scenario, _new_site_id: int = new_site_id, **kwargs: Any, ) -> BizCustomerSite: return BizCustomerSite( site_id=site_id, customer_id=1, site_code=f'S-{site_id}', site_name=f'{site_id}号现场', status=( '1' if _scenario == 'new-site-disabled' and site_id == _new_site_id and kwargs.get('for_update') else '0' ), ) update = AsyncMock() replace_items = AsyncMock() warehouse_mapping = AsyncMock(side_effect=warehouse_access) site_mapping = AsyncMock(side_effect=site_access) site_read = AsyncMock(side_effect=get_site) with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 8])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 10])), patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=shipment)), patch.object(AccessService, 'assert_warehouse_access', new=warehouse_mapping), patch.object(AccessService, 'assert_site_access', new=site_mapping), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=get_warehouse)), patch.object(MasterDao, 'get_site', new=site_read), patch.object( ShipmentService, '_lock_and_validate_pallets', new=AsyncMock(return_value=([], [], [])), ), patch.object(OperationDao, 'update_shipment', new=update), patch.object(OperationDao, 'replace_shipment_items', new=replace_items), self.assertRaises(exception_type) as raised, ): await ShipmentService.edit_shipment_services(db, page, self.user) self.assertEqual(expected_message, raised.exception.message) if scenario == 'current-warehouse-revoked': warehouse_mapping.assert_any_await(db, self.user, 7, for_update=True) else: site_read.assert_any_await(db, 10, for_update=True) if scenario == 'new-site-revoked': site_mapping.assert_any_await(db, self.user, 10, for_update=True) update.assert_not_awaited() replace_items.assert_not_awaited() db.commit.assert_not_awaited() db.rollback.assert_awaited_once() async def test_edit_hidden_current_route_denial_wins_after_all_stable_route_locks(self) -> None: scenarios = ( ('both-warehouses-revoked', {7, 8}, set(), set(), set()), ('new-warehouse-and-current-site-revoked', {7}, {10}, set(), set()), ('current-warehouse-disabled-and-revoked', {8}, set(), {8}, set()), ('current-site-disabled-and-revoked', set(), {10}, set(), {10}), ) expected_locks = [ 'warehouse-master-7', 'warehouse-mapping-7', 'warehouse-master-8', 'warehouse-mapping-8', 'site-master-9', 'site-mapping-9', 'site-master-10', 'site-mapping-10', ] for ( scenario, denied_warehouse_ids, denied_site_ids, disabled_warehouse_ids, disabled_site_ids, ) in scenarios: with self.subTest(scenario=scenario): db = MagicMock(spec=AsyncSession) shipment = make_shipment( source_warehouse_id=8, target_site_id=10, pallet_ids=(2,), ) page = ShipmentUpdateModel( shipmentId=12, sourceWarehouseId=7, targetSiteId=9, palletIds=[2], ) locks: list[str] = [] async def warehouse_access( _db: AsyncSession, _user: CurrentUserModel, warehouse_id: int, _denied_ids: set[int] = denied_warehouse_ids, _locks: list[str] = locks, **kwargs: Any, ) -> None: if not kwargs.get('for_update'): return _locks.append(f'warehouse-mapping-{warehouse_id}') if warehouse_id in _denied_ids: raise PermissionException(data='', message='无权访问该仓库') async def site_access( _db: AsyncSession, _user: CurrentUserModel, site_id: int, _denied_ids: set[int] = denied_site_ids, _locks: list[str] = locks, **kwargs: Any, ) -> None: if not kwargs.get('for_update'): return _locks.append(f'site-mapping-{site_id}') if site_id in _denied_ids: raise PermissionException(data='', message='无权访问该客户现场') async def get_warehouse( _db: AsyncSession, warehouse_id: int, _disabled_ids: set[int] = disabled_warehouse_ids, _locks: list[str] = locks, **kwargs: Any, ) -> BizWarehouse: if kwargs.get('for_update'): _locks.append(f'warehouse-master-{warehouse_id}') return BizWarehouse( warehouse_id=warehouse_id, warehouse_code=f'W-{warehouse_id}', warehouse_name=f'{warehouse_id}号仓', status='1' if kwargs.get('for_update') and warehouse_id in _disabled_ids else '0', ) async def get_site( _db: AsyncSession, site_id: int, _disabled_ids: set[int] = disabled_site_ids, _locks: list[str] = locks, **kwargs: Any, ) -> BizCustomerSite: if kwargs.get('for_update'): _locks.append(f'site-master-{site_id}') return BizCustomerSite( site_id=site_id, customer_id=1, site_code=f'S-{site_id}', site_name=f'{site_id}号现场', status='1' if kwargs.get('for_update') and site_id in _disabled_ids else '0', ) update = AsyncMock() replace_items = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 8])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 10])), patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=shipment)), patch.object( AccessService, 'assert_warehouse_access', new=AsyncMock(side_effect=warehouse_access), ), patch.object(AccessService, 'assert_site_access', new=AsyncMock(side_effect=site_access)), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=get_warehouse)), patch.object(MasterDao, 'get_site', new=AsyncMock(side_effect=get_site)), patch.object( ShipmentService, '_lock_and_validate_pallets', new=AsyncMock(return_value=([], [], [])), ), patch.object(OperationDao, 'update_shipment', new=update), patch.object(OperationDao, 'replace_shipment_items', new=replace_items), self.assertRaises(PermissionException) as raised, ): await ShipmentService.edit_shipment_services(db, page, self.user) self.assertEqual('无权访问该发货单', raised.exception.message) self.assertEqual(expected_locks, locks) update.assert_not_awaited() replace_items.assert_not_awaited() db.commit.assert_not_awaited() db.rollback.assert_awaited_once() async def test_edit_locks_old_and_new_routes_in_stable_grouped_order(self) -> None: db = MagicMock(spec=AsyncSession) shipment = make_shipment( source_warehouse_id=8, target_site_id=10, pallet_ids=(2,), ) page = ShipmentUpdateModel( shipmentId=12, sourceWarehouseId=7, targetSiteId=9, palletIds=[2], ) actions: list[str] = [] async def warehouse_access( _db: AsyncSession, _user: CurrentUserModel, warehouse_id: int, **kwargs: Any, ) -> None: if kwargs.get('for_update'): actions.append(f'warehouse-mapping-{warehouse_id}') async def site_access( _db: AsyncSession, _user: CurrentUserModel, site_id: int, **kwargs: Any, ) -> None: if kwargs.get('for_update'): actions.append(f'site-mapping-{site_id}') async def get_warehouse( _db: AsyncSession, warehouse_id: int, **kwargs: Any, ) -> BizWarehouse: if kwargs.get('for_update'): actions.append(f'warehouse-master-{warehouse_id}') return BizWarehouse( warehouse_id=warehouse_id, warehouse_code=f'W-{warehouse_id}', warehouse_name=f'{warehouse_id}号仓', status='0', ) async def get_site( _db: AsyncSession, site_id: int, **kwargs: Any, ) -> BizCustomerSite: if kwargs.get('for_update'): actions.append(f'site-master-{site_id}') return BizCustomerSite( site_id=site_id, customer_id=1, site_code=f'S-{site_id}', site_name=f'{site_id}号现场', status='0', ) async def lock_pallets(*_args: Any, **_kwargs: Any) -> tuple[list[Any], list[Any], list[Any]]: actions.append('pallet-locks') return [], [], [] async def update(*_args: Any, **_kwargs: Any) -> BizShipmentOrder: actions.append('update') return shipment async def replace(*_args: Any, **_kwargs: Any) -> list[BizShipmentItem]: actions.append('replace') return [] async def commit() -> None: actions.append('commit') db.commit.side_effect = commit with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 8])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 10])), patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=shipment)), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock(side_effect=warehouse_access)), patch.object(AccessService, 'assert_site_access', new=AsyncMock(side_effect=site_access)), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=get_warehouse)), patch.object(MasterDao, 'get_site', new=AsyncMock(side_effect=get_site)), patch.object(ShipmentService, '_lock_and_validate_pallets', new=AsyncMock(side_effect=lock_pallets)), patch.object(OperationDao, 'update_shipment', new=AsyncMock(side_effect=update)), patch.object(OperationDao, 'replace_shipment_items', new=AsyncMock(side_effect=replace)), ): result = await ShipmentService.edit_shipment_services(db, page, self.user) self.assertTrue(result.is_success) self.assertEqual( [ 'pallet-locks', 'warehouse-master-7', 'warehouse-mapping-7', 'warehouse-master-8', 'warehouse-mapping-8', 'site-master-9', 'site-mapping-9', 'site-master-10', 'site-mapping-10', 'update', 'replace', 'commit', ], actions, ) async def test_batch_delete_preflights_all_drafts_before_any_delete(self) -> None: documents = { 11: make_shipment(11, status='DRAFT'), 12: make_shipment(12, status='CONFIRMED'), } async def lock_order( _db: AsyncSession, shipment_id: int, **_kwargs: Any, ) -> BizShipmentOrder: return documents[shipment_id] delete = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(side_effect=lock_order)) as lock, patch.object(OperationDao, 'delete_shipment', new=delete), ): try: with self.assertRaises(ServiceException): await ShipmentService.delete_shipment_services(self.db, [12, 11], self.user) except NotImplementedError: self.fail('发货草稿原子批量删除尚未实现') self.assertEqual([11, 12], [call.args[1] for call in lock.await_args_list]) delete.assert_not_awaited() self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() async def test_batch_delete_rechecks_all_current_routes_before_first_delete(self) -> None: documents = { 11: make_shipment(11, source_warehouse_id=8, target_site_id=10, status='DRAFT'), 12: make_shipment(12, source_warehouse_id=7, target_site_id=9, status='DRAFT'), } revoked_warehouse_id = documents[11].source_warehouse_id parent_lock_order: list[int] = [] async def lock_order( _db: AsyncSession, shipment_id: int, **_kwargs: Any, ) -> BizShipmentOrder: parent_lock_order.append(shipment_id) return documents[shipment_id] async def get_warehouse( _db: AsyncSession, warehouse_id: int, **_kwargs: Any, ) -> BizWarehouse: return BizWarehouse( warehouse_id=warehouse_id, warehouse_code=f'W-{warehouse_id}', warehouse_name=f'{warehouse_id}号仓', status='0', ) async def get_site( _db: AsyncSession, site_id: int, **_kwargs: Any, ) -> BizCustomerSite: return BizCustomerSite( site_id=site_id, customer_id=1, site_code=f'S-{site_id}', site_name=f'{site_id}号现场', status='0', ) async def warehouse_access( _db: AsyncSession, _user: CurrentUserModel, warehouse_id: int, **kwargs: Any, ) -> None: if warehouse_id == revoked_warehouse_id and kwargs.get('for_update'): raise PermissionException(data='', message='无权访问该仓库') delete = AsyncMock() warehouse_mapping = AsyncMock(side_effect=warehouse_access) site_lock = AsyncMock(side_effect=get_site) site_mapping = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 8])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 10])), patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(side_effect=lock_order)), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=get_warehouse)), patch.object(AccessService, 'assert_warehouse_access', new=warehouse_mapping), patch.object(MasterDao, 'get_site', new=site_lock), patch.object(AccessService, 'assert_site_access', new=site_mapping), patch.object(OperationDao, 'delete_shipment', new=delete), self.assertRaises(PermissionException) as raised, ): await ShipmentService.delete_shipment_services(self.db, [12, 11], self.user) self.assertEqual('无权访问该发货单', raised.exception.message) self.assertEqual([11, 12], parent_lock_order) self.assertEqual([7, 8], [call.args[2] for call in warehouse_mapping.await_args_list]) self.assertTrue(all(call.kwargs.get('for_update') for call in warehouse_mapping.await_args_list)) self.assertEqual([9, 10], [call.args[1] for call in site_lock.await_args_list]) self.assertEqual([9, 10], [call.args[2] for call in site_mapping.await_args_list]) self.assertTrue(all(call.kwargs.get('for_update') for call in site_mapping.await_args_list)) delete.assert_not_awaited() self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() async def test_batch_delete_locks_deduplicated_routes_stably_without_requiring_active(self) -> None: db = MagicMock(spec=AsyncSession) documents = { 11: make_shipment(11, source_warehouse_id=8, target_site_id=10, status='DRAFT'), 12: make_shipment(12, source_warehouse_id=7, target_site_id=10, status='DRAFT'), 13: make_shipment(13, source_warehouse_id=8, target_site_id=9, status='DRAFT'), } actions: list[str] = [] async def lock_order( _db: AsyncSession, shipment_id: int, **_kwargs: Any, ) -> BizShipmentOrder: actions.append(f'parent-{shipment_id}') return documents[shipment_id] async def get_warehouse( _db: AsyncSession, warehouse_id: int, **kwargs: Any, ) -> BizWarehouse: self.assertTrue(kwargs.get('for_update')) actions.append(f'warehouse-master-{warehouse_id}') return BizWarehouse( warehouse_id=warehouse_id, warehouse_code=f'W-{warehouse_id}', warehouse_name=f'{warehouse_id}号仓', status='1', ) async def warehouse_access( _db: AsyncSession, _user: CurrentUserModel, warehouse_id: int, **kwargs: Any, ) -> None: self.assertTrue(kwargs.get('for_update')) actions.append(f'warehouse-mapping-{warehouse_id}') async def get_site( _db: AsyncSession, site_id: int, **kwargs: Any, ) -> BizCustomerSite: self.assertTrue(kwargs.get('for_update')) actions.append(f'site-master-{site_id}') return BizCustomerSite( site_id=site_id, customer_id=1, site_code=f'S-{site_id}', site_name=f'{site_id}号现场', status='1', ) async def site_access( _db: AsyncSession, _user: CurrentUserModel, site_id: int, **kwargs: Any, ) -> None: self.assertTrue(kwargs.get('for_update')) actions.append(f'site-mapping-{site_id}') async def delete(_db: AsyncSession, shipment_id: int) -> None: actions.append(f'delete-{shipment_id}') async def commit() -> None: actions.append('commit') db.commit.side_effect = commit with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 8])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 10])), patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(side_effect=lock_order)), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=get_warehouse)), patch.object( AccessService, 'assert_warehouse_access', new=AsyncMock(side_effect=warehouse_access), ), patch.object(MasterDao, 'get_site', new=AsyncMock(side_effect=get_site)), patch.object(AccessService, 'assert_site_access', new=AsyncMock(side_effect=site_access)), patch.object(OperationDao, 'delete_shipment', new=AsyncMock(side_effect=delete)), ): result = await ShipmentService.delete_shipment_services(db, [13, 11, 12], self.user) self.assertTrue(result.is_success) self.assertEqual( [ 'parent-11', 'parent-12', 'parent-13', 'warehouse-master-7', 'warehouse-mapping-7', 'warehouse-master-8', 'warehouse-mapping-8', 'site-master-9', 'site-mapping-9', 'site-master-10', 'site-mapping-10', 'delete-11', 'delete-12', 'delete-13', 'commit', ], actions, ) class ShipmentConfirmServiceTest(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.db = MagicMock(spec=AsyncSession) self.user = make_user() self.command = ConfirmCommandModel(idempotencyKey='confirm-key') self.fixed_time = datetime(2026, 7, 26, 17, 20) self.request_hash = ShipmentService._build_confirm_request_hash(12) async def test_confirm_snapshots_full_remaining_and_moves_every_pallet_atomically( # noqa: PLR0915 self, ) -> None: shipment = make_shipment() inventories = [ make_inventory(2, remaining_packages=8, version=4), make_inventory(5, remaining_packages=13, version=8), ] pallets = [make_pallet(2), make_pallet(5)] fillings = {102: make_filling(2), 105: make_filling(5)} actions: list[str] = [] events: list[BizPalletEvent] = [] async def detail(*_args: Any, **_kwargs: Any) -> BizShipmentOrder: actions.append('detail') return shipment async def idempotency(*_args: Any, **_kwargs: Any) -> None: actions.append('idempotency-read') async def order_lock(*_args: Any, **_kwargs: Any) -> BizShipmentOrder: actions.append('order-lock') return shipment async def inventory_lock(*_args: Any) -> list[BizPalletInventory]: actions.append('inventory-lock') return inventories async def pallet_lock(*_args: Any) -> list[BizPallet]: actions.append('pallet-lock') return pallets async def filling_lock(_db: AsyncSession, filling_id: int) -> BizFillingRecord: actions.append(f'filling-lock-{filling_id}') return fillings[filling_id] async def warehouse_lock(*_args: Any, **kwargs: Any) -> BizWarehouse: self.assertTrue(kwargs.get('for_update')) actions.append('warehouse-lock') return BizWarehouse( warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status='0', ) async def warehouse_mapping(*_args: Any, **kwargs: Any) -> None: self.assertTrue(kwargs.get('for_update')) actions.append('warehouse-mapping-lock') async def site_lock(*_args: Any, **kwargs: Any) -> BizCustomerSite: self.assertTrue(kwargs.get('for_update')) actions.append('site-lock') return BizCustomerSite( site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status='0', ) async def site_mapping(*_args: Any, **kwargs: Any) -> None: self.assertTrue(kwargs.get('for_update')) actions.append('site-mapping-lock') def transition(current: PalletFlowState, target: PalletFlowState) -> None: actions.append(f'transition-{current.value}-{target.value}') async def update(*_args: Any) -> BizShipmentOrder: actions.append('update-order') return shipment async def add_event( _db: AsyncSession, inventory: BizPalletInventory, event: BizPalletEvent, ) -> BizPalletEvent: actions.append(f'event-{inventory.pallet_id}') event.event_id = 200 + inventory.pallet_id inventory.last_event_id = event.event_id events.append(event) return event async def commit() -> None: actions.append('commit') self.db.commit.side_effect = commit with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(side_effect=detail)) as detail_read, patch.object( OperationDao, 'get_shipment_by_confirm_idempotency_key', new=AsyncMock(side_effect=idempotency), ), patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(side_effect=order_lock)), patch.object( InventoryDao, 'get_inventories_for_update', new=AsyncMock(side_effect=inventory_lock), ), patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=pallet_lock)), patch.object( OperationDao, 'get_active_filling_for_update', new=AsyncMock(side_effect=filling_lock), ), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=warehouse_lock)), patch.object( AccessService, 'assert_warehouse_access', new=AsyncMock(side_effect=warehouse_mapping), ), patch.object(MasterDao, 'get_site', new=AsyncMock(side_effect=site_lock)), patch.object(AccessService, 'assert_site_access', new=AsyncMock(side_effect=site_mapping)), patch.object(PalletStateMachine, 'ensure_transition', side_effect=transition), patch.object(OperationDao, 'update_shipment', new=AsyncMock(side_effect=update)), patch.object(InventoryDao, 'add_event', new=AsyncMock(side_effect=add_event)), patch.object(shipment_service_module, 'datetime') as service_datetime, ): service_datetime.now.return_value = self.fixed_time try: result = await ShipmentService.confirm_shipment_services( self.db, 12, self.command, self.user, ) except NotImplementedError: self.fail('多托盘整托确认尚未实现') self.assertTrue(result.is_success) self.assertEqual( [ 'detail', 'idempotency-read', 'order-lock', 'inventory-lock', 'pallet-lock', 'filling-lock-102', 'filling-lock-105', 'warehouse-lock', 'warehouse-mapping-lock', 'site-lock', 'site-mapping-lock', 'transition-FILLED_IN_WAREHOUSE-AT_SITE_UNUSED', 'transition-FILLED_IN_WAREHOUSE-AT_SITE_UNUSED', 'update-order', 'event-2', 'event-5', 'commit', ], actions, ) detail_read.assert_awaited_once_with( self.db, 12, warehouse_ids=[7], site_ids=[9], ) self.assertEqual('CONFIRMED', shipment.status) self.assertEqual('confirm-key', shipment.confirm_idempotency_key) self.assertEqual(self.request_hash, shipment.confirm_request_hash) self.assertEqual(7, shipment.confirmed_by) self.assertEqual(self.fixed_time, shipment.confirmed_at) snapshots = {item.pallet_id: (item.active_filling_id, item.package_qty) for item in shipment.items} self.assertEqual({2: (102, 8), 5: (105, 13)}, snapshots) for inventory, original_version, remaining in zip( inventories, (4, 8), (8, 13), strict=True, ): self.assertEqual('AT_SITE_UNUSED', inventory.flow_state) self.assertIsNone(inventory.warehouse_id) self.assertEqual(9, inventory.site_id) self.assertIsNone(inventory.current_line_id) self.assertEqual(100 + inventory.pallet_id, inventory.active_filling_id) self.assertEqual(remaining, inventory.remaining_packages) self.assertEqual(original_version + 1, inventory.version) self.assertEqual(self.fixed_time, inventory.updated_at) self.assertEqual([2, 5], [event.pallet_id for event in events]) for event in events: self.assertEqual(PalletEventType.SHIPMENT_CONFIRMED.value, event.event_type) self.assertEqual('SHIPMENT', event.source_type) self.assertEqual(12, event.source_id) self.assertEqual('FILLED_IN_WAREHOUSE', event.before_state) self.assertEqual('AT_SITE_UNUSED', event.after_state) self.assertEqual(7, event.before_warehouse_id) self.assertIsNone(event.before_site_id) self.assertIsNone(event.after_warehouse_id) self.assertEqual(9, event.after_site_id) self.assertEqual(event.before_remaining_packages, event.after_remaining_packages) self.assertEqual(7, event.operator_id) self.assertEqual(self.fixed_time, event.occurred_at) self.assertEqual('confirm-key', event.idempotency_key) self.assertEqual(self.request_hash, event.request_hash) self.db.commit.assert_awaited_once() self.db.rollback.assert_not_awaited() async def test_rr_current_parent_recovers_exact_confirm_without_repeating_inventory_writes(self) -> None: initial = make_shipment(status='DRAFT') confirmed = make_shipment( status='CONFIRMED', idempotency_key='confirm-key', request_hash=self.request_hash, ) inventory_lock = AsyncMock() update = AsyncMock() add_event = AsyncMock() warehouse_access = AsyncMock() site_access = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=initial)), patch.object( OperationDao, 'get_shipment_by_confirm_idempotency_key', new=AsyncMock(return_value=None), ), patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=confirmed)), patch.object(InventoryDao, 'get_inventories_for_update', new=inventory_lock), patch.object( MasterDao, 'get_warehouse', new=AsyncMock( return_value=BizWarehouse( warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status='0', ) ), ) as warehouse_lock, patch.object(AccessService, 'assert_warehouse_access', new=warehouse_access), patch.object( MasterDao, 'get_site', new=AsyncMock( return_value=BizCustomerSite( site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status='0', ) ), ) as site_lock, patch.object(AccessService, 'assert_site_access', new=site_access), patch.object(OperationDao, 'update_shipment', new=update), patch.object(InventoryDao, 'add_event', new=add_event), ): result = await ShipmentService.confirm_shipment_services( self.db, 12, self.command, self.user, ) self.assertTrue(result.is_success) warehouse_lock.assert_awaited_once_with(self.db, 7, for_update=True) warehouse_access.assert_awaited_once_with(self.db, self.user, 7, for_update=True) site_lock.assert_awaited_once_with(self.db, 9, for_update=True) site_access.assert_awaited_once_with(self.db, self.user, 9, for_update=True) inventory_lock.assert_not_awaited() update.assert_not_awaited() add_event.assert_not_awaited() self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() async def test_exact_confirm_retry_ignores_later_route_disable_but_rechecks_current_access( self, ) -> None: for warehouse_status, site_status in (('1', '0'), ('0', '1')): with self.subTest(warehouse_status=warehouse_status, site_status=site_status): db = MagicMock(spec=AsyncSession) confirmed = make_shipment( status='CONFIRMED', idempotency_key='confirm-key', request_hash=self.request_hash, ) for item in confirmed.items: item.active_filling_id = 100 + item.pallet_id item.package_qty = 8 original_snapshots = [ (item.pallet_id, item.active_filling_id, item.package_qty) for item in confirmed.items ] inventory_lock = AsyncMock() update = AsyncMock() add_event = AsyncMock() warehouse_access = AsyncMock() site_access = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=confirmed)), patch.object( OperationDao, 'get_shipment_by_confirm_idempotency_key', new=AsyncMock(return_value=confirmed), ), patch.object( OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=confirmed), ), patch.object(InventoryDao, 'get_inventories_for_update', new=inventory_lock), patch.object( MasterDao, 'get_warehouse', new=AsyncMock( return_value=BizWarehouse( warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status=warehouse_status, ) ), ) as warehouse_lock, patch.object(AccessService, 'assert_warehouse_access', new=warehouse_access), patch.object( MasterDao, 'get_site', new=AsyncMock( return_value=BizCustomerSite( site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status=site_status, ) ), ) as site_lock, patch.object(AccessService, 'assert_site_access', new=site_access), patch.object(OperationDao, 'update_shipment', new=update), patch.object(InventoryDao, 'add_event', new=add_event), ): result = await ShipmentService.confirm_shipment_services( db, 12, self.command, self.user, ) self.assertTrue(result.is_success) warehouse_lock.assert_awaited_once_with(db, 7, for_update=True) warehouse_access.assert_awaited_once_with(db, self.user, 7, for_update=True) site_lock.assert_awaited_once_with(db, 9, for_update=True) site_access.assert_awaited_once_with(db, self.user, 9, for_update=True) self.assertEqual( original_snapshots, [(item.pallet_id, item.active_filling_id, item.package_qty) for item in confirmed.items], ) inventory_lock.assert_not_awaited() update.assert_not_awaited() add_event.assert_not_awaited() db.commit.assert_not_awaited() db.rollback.assert_awaited_once() async def test_exact_confirm_retry_hides_document_when_current_mapping_is_revoked(self) -> None: confirmed = make_shipment( status='CONFIRMED', idempotency_key='confirm-key', request_hash=self.request_hash, ) inventory_lock = AsyncMock() update = AsyncMock() add_event = AsyncMock() site_lock = AsyncMock( return_value=BizCustomerSite( site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status='0', ) ) site_mapping = AsyncMock() warehouse_mapping = AsyncMock(side_effect=PermissionException(data='', message='无权访问该仓库')) with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=confirmed)), patch.object( OperationDao, 'get_shipment_by_confirm_idempotency_key', new=AsyncMock(return_value=confirmed), ), patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=confirmed)), patch.object(InventoryDao, 'get_inventories_for_update', new=inventory_lock), patch.object( MasterDao, 'get_warehouse', new=AsyncMock( return_value=BizWarehouse( warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status='1', ) ), ) as warehouse_lock, patch.object(AccessService, 'assert_warehouse_access', new=warehouse_mapping), patch.object(MasterDao, 'get_site', new=site_lock), patch.object(AccessService, 'assert_site_access', new=site_mapping), patch.object(OperationDao, 'update_shipment', new=update), patch.object(InventoryDao, 'add_event', new=add_event), self.assertRaises(PermissionException) as raised, ): await ShipmentService.confirm_shipment_services( self.db, 12, self.command, self.user, ) self.assertEqual('无权访问该发货单', raised.exception.message) warehouse_lock.assert_awaited_once_with(self.db, 7, for_update=True) warehouse_mapping.assert_awaited_once_with(self.db, self.user, 7, for_update=True) site_lock.assert_awaited_once_with(self.db, 9, for_update=True) site_mapping.assert_awaited_once_with(self.db, self.user, 9, for_update=True) inventory_lock.assert_not_awaited() update.assert_not_awaited() add_event.assert_not_awaited() self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() async def test_confirm_uses_locked_current_items_not_initial_rr_snapshot(self) -> None: initial = make_shipment(pallet_ids=(99,)) locked = make_shipment(pallet_ids=(2,)) inventory = make_inventory(2, remaining_packages=6) lock_validate = AsyncMock(return_value=([inventory], [make_pallet(2)], [make_filling(2)])) with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=initial)), patch.object( OperationDao, 'get_shipment_by_confirm_idempotency_key', new=AsyncMock(return_value=None), ), patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=locked)), patch.object(ShipmentService, '_lock_and_validate_pallets', new=lock_validate), patch.object( MasterDao, 'get_warehouse', new=AsyncMock( return_value=BizWarehouse( warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status='0', ) ), ), patch.object( MasterDao, 'get_site', new=AsyncMock( return_value=BizCustomerSite( site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status='0', ) ), ), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object(AccessService, 'assert_site_access', new=AsyncMock()), patch.object(PalletStateMachine, 'ensure_transition'), patch.object(OperationDao, 'update_shipment', new=AsyncMock()), patch.object(InventoryDao, 'add_event', new=AsyncMock()), ): result = await ShipmentService.confirm_shipment_services( self.db, 12, self.command, self.user, ) self.assertTrue(result.is_success) lock_validate.assert_awaited_once_with(self.db, [2], 7) self.assertEqual((102, 6), (locked.items[0].active_filling_id, locked.items[0].package_qty)) self.assertIsNone(initial.items[0].active_filling_id) self.assertIsNone(initial.items[0].package_qty) async def test_same_key_on_other_shipment_is_rejected_before_parent_lock(self) -> None: target = make_shipment() other = make_shipment( 13, status='CONFIRMED', idempotency_key='confirm-key', request_hash=ShipmentService._build_confirm_request_hash(13), ) parent_lock = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=target)), patch.object( OperationDao, 'get_shipment_by_confirm_idempotency_key', new=AsyncMock(return_value=other), ), patch.object(OperationDao, 'get_shipment_for_update', new=parent_lock), self.assertRaises(ServiceException) as raised, ): await ShipmentService.confirm_shipment_services(self.db, 12, self.command, self.user) self.assertEqual('确认发货失败，幂等键已被其他请求使用', raised.exception.message) parent_lock.assert_not_awaited() self.db.rollback.assert_awaited_once() async def test_integrity_recovery_nonexact_result_is_stable_conflict_not_success(self) -> None: initial = make_shipment() nonexact = make_shipment( 13, status='CONFIRMED', idempotency_key='confirm-key', request_hash=ShipmentService._build_confirm_request_hash(13), ) with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(side_effect=[[7], [7]])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(side_effect=[[9], [9]])), patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=initial)), patch.object( OperationDao, 'get_shipment_by_confirm_idempotency_key', new=AsyncMock(side_effect=[None, nonexact]), ), patch.object( OperationDao, 'get_shipment_for_update', new=AsyncMock(side_effect=integrity_error()), ), self.assertRaises(ServiceException) as raised, ): await ShipmentService.confirm_shipment_services(self.db, 12, self.command, self.user) self.assertEqual('确认发货失败，幂等键已被其他请求使用', raised.exception.message) self.assertEqual(2, self.db.rollback.await_count) self.db.commit.assert_not_awaited() async def test_integrity_recovery_locks_parent_and_rechecks_revoked_current_mapping(self) -> None: initial = make_shipment(status='DRAFT') recovered = make_shipment( status='CONFIRMED', idempotency_key='confirm-key', request_hash=self.request_hash, ) parent_lock_count = 0 async def lock_parent(*_args: Any, **_kwargs: Any) -> BizShipmentOrder: nonlocal parent_lock_count parent_lock_count += 1 if parent_lock_count == 1: raise integrity_error() return recovered async def warehouse_access(*_args: Any, **kwargs: Any) -> None: if kwargs.get('for_update'): raise PermissionException(data='', message='无权访问该仓库') parent_lock = AsyncMock(side_effect=lock_parent) warehouse_mapping = AsyncMock(side_effect=warehouse_access) site_lock = AsyncMock( return_value=BizCustomerSite( site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status='0', ) ) with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(side_effect=[[7], [7]])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(side_effect=[[9], [9]])), patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=initial)), patch.object( OperationDao, 'get_shipment_by_confirm_idempotency_key', new=AsyncMock(side_effect=[None, recovered]), ), patch.object(OperationDao, 'get_shipment_for_update', new=parent_lock), patch.object( MasterDao, 'get_warehouse', new=AsyncMock( return_value=BizWarehouse( warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status='0', ) ), ) as warehouse_lock, patch.object(AccessService, 'assert_warehouse_access', new=warehouse_mapping), patch.object(MasterDao, 'get_site', new=site_lock), patch.object(AccessService, 'assert_site_access', new=AsyncMock()), self.assertRaises(PermissionException) as raised, ): await ShipmentService.confirm_shipment_services(self.db, 12, self.command, self.user) self.assertEqual('无权访问该发货单', raised.exception.message) self.assertEqual(2, parent_lock.await_count) parent_lock.assert_awaited_with( self.db, 12, warehouse_ids=[7], site_ids=[9], ) warehouse_lock.assert_awaited_once_with(self.db, 7, for_update=True) warehouse_mapping.assert_awaited_once_with(self.db, self.user, 7, for_update=True) site_lock.assert_awaited_once_with(self.db, 9, for_update=True) self.db.commit.assert_not_awaited() self.assertEqual(2, self.db.rollback.await_count) async def test_disabled_source_after_pallet_locks_rolls_back_before_mutation(self) -> None: shipment = make_shipment(pallet_ids=(2,)) inventory = make_inventory(2, version=4) update = AsyncMock() add_event = AsyncMock() mapping = AsyncMock() site_lock = AsyncMock( return_value=BizCustomerSite( site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status='0', ) ) site_mapping = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=shipment)), patch.object( OperationDao, 'get_shipment_by_confirm_idempotency_key', new=AsyncMock(return_value=None), ), patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=shipment)), patch.object( ShipmentService, '_lock_and_validate_pallets', new=AsyncMock(return_value=([inventory], [make_pallet(2)], [make_filling(2)])), ), patch.object( MasterDao, 'get_warehouse', new=AsyncMock( return_value=BizWarehouse( warehouse_id=7, warehouse_code='W-7', warehouse_name='停用仓', status='1', ) ), ), patch.object(AccessService, 'assert_warehouse_access', new=mapping), patch.object(MasterDao, 'get_site', new=site_lock), patch.object(AccessService, 'assert_site_access', new=site_mapping), patch.object(OperationDao, 'update_shipment', new=update), patch.object(InventoryDao, 'add_event', new=add_event), self.assertRaises(ServiceException) as raised, ): await ShipmentService.confirm_shipment_services(self.db, 12, self.command, self.user) self.assertEqual('来源仓库已停用', raised.exception.message) self.assertEqual('DRAFT', shipment.status) self.assertEqual('FILLED_IN_WAREHOUSE', inventory.flow_state) self.assertEqual(4, inventory.version) mapping.assert_awaited_once_with(self.db, self.user, 7, for_update=True) site_lock.assert_awaited_once_with(self.db, 9, for_update=True) site_mapping.assert_awaited_once_with(self.db, self.user, 9, for_update=True) update.assert_not_awaited() add_event.assert_not_awaited() self.db.rollback.assert_awaited_once() async def test_integrity_error_recovers_only_fresh_scoped_exact_result(self) -> None: initial = make_shipment(status='DRAFT') recovered = make_shipment( status='CONFIRMED', idempotency_key='confirm-key', request_hash=self.request_hash, ) warehouse_scope = AsyncMock(side_effect=[[7], [7]]) site_scope = AsyncMock(side_effect=[[9], [9]]) key_read = AsyncMock(side_effect=[None, recovered]) with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=warehouse_scope), patch.object(AccessDao, 'list_user_site_ids', new=site_scope), patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=initial)), patch.object(OperationDao, 'get_shipment_by_confirm_idempotency_key', new=key_read), patch.object( OperationDao, 'get_shipment_for_update', new=AsyncMock(side_effect=[integrity_error(), recovered]), ), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object(AccessService, 'assert_site_access', new=AsyncMock()), patch.object( MasterDao, 'get_warehouse', new=AsyncMock( return_value=BizWarehouse( warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status='0', ) ), ), patch.object( MasterDao, 'get_site', new=AsyncMock( return_value=BizCustomerSite( site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status='0', ) ), ), ): result = await ShipmentService.confirm_shipment_services( self.db, 12, self.command, self.user, ) self.assertTrue(result.is_success) self.assertEqual(2, warehouse_scope.await_count) self.assertEqual(2, site_scope.await_count) self.assertEqual(2, key_read.await_count) for call in key_read.await_args_list: self.assertEqual([7], call.kwargs['warehouse_ids']) self.assertEqual([9], call.kwargs['site_ids']) self.assertEqual(2, self.db.rollback.await_count) self.db.commit.assert_not_awaited() async def test_revoked_mapping_after_locks_hides_document_and_mutates_nothing(self) -> None: shipment = make_shipment(pallet_ids=(2,)) inventory = make_inventory(2, version=4) update = AsyncMock() add_event = AsyncMock() transition = MagicMock() site_lock = AsyncMock( return_value=BizCustomerSite( site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status='0', ) ) site_mapping = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=shipment)), patch.object( OperationDao, 'get_shipment_by_confirm_idempotency_key', new=AsyncMock(return_value=None), ), patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=shipment)), patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=[inventory])), patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[make_pallet(2)])), patch.object( OperationDao, 'get_active_filling_for_update', new=AsyncMock(return_value=make_filling(2)), ), patch.object( MasterDao, 'get_warehouse', new=AsyncMock( return_value=BizWarehouse( warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status='0', ) ), ), patch.object( AccessService, 'assert_warehouse_access', new=AsyncMock(side_effect=PermissionException(data='', message='无权访问该仓库')), ), patch.object(MasterDao, 'get_site', new=site_lock), patch.object(AccessService, 'assert_site_access', new=site_mapping), patch.object(PalletStateMachine, 'ensure_transition', new=transition), patch.object(OperationDao, 'update_shipment', new=update), patch.object(InventoryDao, 'add_event', new=add_event), self.assertRaises(PermissionException) as raised, ): await ShipmentService.confirm_shipment_services(self.db, 12, self.command, self.user) self.assertEqual('无权访问该发货单', raised.exception.message) self.assertEqual('DRAFT', shipment.status) self.assertEqual('FILLED_IN_WAREHOUSE', inventory.flow_state) self.assertEqual(7, inventory.warehouse_id) self.assertIsNone(inventory.site_id) self.assertEqual(4, inventory.version) site_lock.assert_awaited_once_with(self.db, 9, for_update=True) site_mapping.assert_awaited_once_with(self.db, self.user, 9, for_update=True) transition.assert_not_called() update.assert_not_awaited() add_event.assert_not_awaited() self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() async def test_all_transitions_validate_before_first_mutation(self) -> None: shipment = make_shipment() inventories = [make_inventory(2, version=4), make_inventory(5, version=8)] fillings = {102: make_filling(2), 105: make_filling(5)} update = AsyncMock() add_event = AsyncMock() async def get_filling(_db: AsyncSession, filling_id: int) -> BizFillingRecord: return fillings[filling_id] transition = MagicMock(side_effect=[None, ValueError('second pallet invalid')]) with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=shipment)), patch.object( OperationDao, 'get_shipment_by_confirm_idempotency_key', new=AsyncMock(return_value=None), ), patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=shipment)), patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=inventories)), patch.object( InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[make_pallet(2), make_pallet(5)]), ), patch.object( OperationDao, 'get_active_filling_for_update', new=AsyncMock(side_effect=get_filling), ), patch.object( MasterDao, 'get_warehouse', new=AsyncMock( return_value=BizWarehouse( warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status='0', ) ), ), patch.object( MasterDao, 'get_site', new=AsyncMock( return_value=BizCustomerSite( site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status='0', ) ), ), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object(AccessService, 'assert_site_access', new=AsyncMock()), patch.object(PalletStateMachine, 'ensure_transition', new=transition), patch.object(OperationDao, 'update_shipment', new=update), patch.object(InventoryDao, 'add_event', new=add_event), self.assertRaisesRegex(ValueError, 'second pallet invalid'), ): await ShipmentService.confirm_shipment_services(self.db, 12, self.command, self.user) self.assertEqual('DRAFT', shipment.status) self.assertTrue(all(item.active_filling_id is None and item.package_qty is None for item in shipment.items)) self.assertEqual(['FILLED_IN_WAREHOUSE', 'FILLED_IN_WAREHOUSE'], [item.flow_state for item in inventories]) self.assertEqual([4, 8], [item.version for item in inventories]) update.assert_not_awaited() add_event.assert_not_awaited() self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() if __name__ == '__main__': unittest.main()