ShiErFeng/shierfeng-fastapi-backend/tests/module_business/test_shipment_service.py

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.dialects import postgresql
from sqlalchemy.exc import IntegrityError, MultipleResultsFound
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, ServiceWarning
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.domain.test_batch_scope import TestBatchScope
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
from module_business.service.test_batch_access_service import TestBatchAccessService
from tests.module_business.test_batch_fixtures import make_test_scope


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',
    test_batch_id: int | None = None,
) -> 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,
        test_batch_id=test_batch_id,
    )


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,
    test_batch_id: int | None = None,
) -> 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,
        test_batch_id=test_batch_id,
    )


def make_filling(
    pallet_id: int,
    *,
    warehouse_id: int = 7,
    test_batch_id: int | None = None,
) -> 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(),
        test_batch_id=test_batch_id,
    )


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,
    test_batch_id: int | 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,
        test_batch_id=test_batch_id,
    )
    shipment.items = [
        BizShipmentItem(
            shipment_item_id=index,
            shipment_id=shipment_id,
            pallet_id=pallet_id,
            test_batch_id=test_batch_id,
        )
        for index, pallet_id in enumerate(pallet_ids, start=1)
    ]
    return shipment


class ShipmentServiceBootstrapTest(unittest.TestCase):
    def test_public_business_methods_require_explicit_test_scope(self) -> None:
        for method_name in (
            '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',
        ):
            with self.subTest(method=method_name):
                self.assertIn('test_scope', inspect.signature(getattr(ShipmentService, method_name)).parameters)

    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', 'test_scope'),
            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',
                'warehouse_ids',
                'site_ids',
                'scope',
                'is_page',
            ),
            OperationDao.get_shipment_by_id: (
                'db',
                'shipment_id',
                'scope',
                'warehouse_ids',
                'site_ids',
            ),
            OperationDao.get_shipment_for_update: (
                'db',
                'shipment_id',
                'scope',
                'warehouse_ids',
                'site_ids',
            ),
            OperationDao.get_shipment_by_confirm_idempotency_key: (
                'db',
                'idempotency_key',
                'scope',
                '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 ShipmentDetailDaoTest(unittest.IsolatedAsyncioTestCase):
    async def test_draft_detail_loads_only_current_shippable_remaining_quantity_with_one_left_join(self) -> None:
        db = MagicMock(spec=AsyncSession)
        shipment = make_shipment(pallet_ids=(5, 2, 8))
        order_result = MagicMock()
        order_result.scalar_one_or_none.return_value = shipment
        item_result = MagicMock()
        item_result.all.return_value = [
            (shipment.items[0], 13),
            (shipment.items[1], 6),
            (shipment.items[2], None),
        ]
        db.execute.side_effect = [order_result, item_result]

        actual = await OperationDao.get_shipment_by_id(
            db,
            12,
            warehouse_ids=[7],
            site_ids=[9],
            scope=TestBatchScope.formal(),
        )

        self.assertIs(shipment, actual)
        self.assertEqual(2, db.execute.await_count)
        self.assertEqual([13, 6, None], [item.remaining_packages for item in shipment.items])
        self.assertEqual([None, None, None], [item.package_qty for item in shipment.items])
        item_statement = db.execute.await_args_list[1].args[0]
        item_sql = ' '.join(
            str(
                item_statement.compile(
                    dialect=postgresql.dialect(),
                    compile_kwargs={'literal_binds': True},
                )
            ).split()
        )
        self.assertIn(
            'LEFT OUTER JOIN biz_pallet_inventory '
            'ON biz_pallet_inventory.pallet_id = biz_shipment_item.pallet_id '
            'AND biz_pallet_inventory.warehouse_id = 7 '
            "AND biz_pallet_inventory.flow_state = 'FILLED_IN_WAREHOUSE' "
            'AND biz_pallet_inventory.site_id IS NULL '
            'AND biz_pallet_inventory.current_line_id IS NULL '
            'AND biz_pallet_inventory.active_filling_id IS NOT NULL '
            'AND biz_pallet_inventory.remaining_packages > 0',
            item_sql,
        )
        self.assertIn('WHERE biz_shipment_item.shipment_id = 12', item_sql)
        self.assertIn('ORDER BY biz_shipment_item.shipment_item_id', item_sql)

    async def test_confirmed_detail_preserves_snapshot_and_does_not_read_current_inventory(self) -> None:
        db = MagicMock(spec=AsyncSession)
        shipment = make_shipment(status='CONFIRMED', pallet_ids=(5, 2))
        shipment.items[0].active_filling_id = 105
        shipment.items[0].package_qty = 13
        shipment.items[1].active_filling_id = 102
        shipment.items[1].package_qty = 6
        order_result = MagicMock()
        order_result.scalar_one_or_none.return_value = shipment
        item_result = MagicMock()
        item_result.all.return_value = [
            (shipment.items[0], 1),
            (shipment.items[1], 2),
        ]
        item_result.scalars.return_value.all.return_value = shipment.items
        db.execute.side_effect = [order_result, item_result]

        actual = await OperationDao.get_shipment_by_id(
            db,
            12,
            warehouse_ids=[7],
            site_ids=[9],
            scope=TestBatchScope.formal(),
        )

        self.assertIs(shipment, actual)
        self.assertEqual(2, db.execute.await_count)
        self.assertEqual([13, 6], [item.package_qty for item in shipment.items])
        self.assertEqual([None, None], [item.remaining_packages for item in shipment.items])
        item_statement = db.execute.await_args_list[1].args[0]
        item_sql = ' '.join(
            str(
                item_statement.compile(
                    dialect=postgresql.dialect(),
                    compile_kwargs={'literal_binds': True},
                )
            ).split()
        )
        self.assertNotIn('biz_pallet_inventory', item_sql)
        self.assertIn('WHERE biz_shipment_item.shipment_id = 12', item_sql)
        self.assertIn('ORDER BY biz_shipment_item.shipment_item_id', item_sql)


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,
                    TestBatchScope.formal(),
                    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,
                TestBatchScope.formal(),
            )
            try:
                detail = await ShipmentService.shipment_detail_services(self.db, 12, self.user, TestBatchScope.formal())
            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,
            warehouse_ids=[7],
            site_ids=[9],
            scope=TestBatchScope.formal(),
            is_page=True,
        )
        get_detail.assert_awaited_once_with(
            self.db,
            12,
            TestBatchScope.formal(),
            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, TestBatchScope.formal()
                )
                site_options = await ShipmentService.get_site_options_services(
                    self.db, self.user, TestBatchScope.formal()
                )
            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, warehouse_id=3)
        lock_inventory = AsyncMock()
        lock_pallets = AsyncMock()
        lock_filling = AsyncMock()
        get_filling = AsyncMock(return_value=filling)
        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=get_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,
                    TestBatchScope.formal(),
                )
            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)
        get_filling.assert_awaited_once_with(
            self.db,
            inventory.active_filling_id,
            TestBatchScope.formal(),
        )
        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, TestBatchScope.formal())

                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, TestBatchScope.formal())

        self.assertEqual('无权访问该发货单', user_raised.exception.message)
        get_detail.assert_awaited_once_with(
            self.db,
            404,
            TestBatchScope.formal(),
            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), TestBatchScope.formal())
        self.assertEqual('发货单不存在', admin_raised.exception.message)
        get_detail.assert_awaited_once_with(
            self.db,
            404,
            TestBatchScope.formal(),
            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,
                        TestBatchScope.formal(),
                    )
                self.assertEqual('无权访问该托盘', raised.exception.message)

    async def test_test_scope_scan_rejects_cross_batch_pallet(self) -> None:
        scope = make_test_scope(4)
        query = ShipmentScanQueryModel(sourceWarehouseId=7, targetSiteId=9)
        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',
                        test_batch_id=4,
                    )
                ),
            ),
            patch.object(
                MasterDao,
                'get_site',
                new=AsyncMock(
                    return_value=BizCustomerSite(
                        site_id=9,
                        customer_id=1,
                        site_code='S-9',
                        site_name='九号现场',
                        status='0',
                        test_batch_id=4,
                    )
                ),
            ),
            patch.object(
                InventoryDao,
                'get_pallet_by_qr_token',
                new=AsyncMock(return_value=make_pallet(2, test_batch_id=5)),
            ),
            patch.object(
                MasterDao,
                'get_pallet_inventory',
                new=AsyncMock(return_value=make_inventory(2, test_batch_id=4)),
            ),
            patch.object(
                OperationDao,
                'get_filling_by_id',
                new=AsyncMock(return_value=make_filling(2, test_batch_id=4)),
            ),
            self.assertRaisesRegex(ServiceException, '不属于当前数据模式或测试批次'),
        ):
            await ShipmentService.scan_shipment_pallet_services(self.db, 'qr-2', query, self.user, scope)


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_residual_return_can_be_re_shipped_from_current_inventory_warehouse(self) -> None:
        inventory = make_inventory(2, warehouse_id=7, remaining_packages=6)
        pallet = make_pallet(2)
        origin_filling = make_filling(2, warehouse_id=3)

        with (
            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=origin_filling),
            ),
        ):
            inventories, pallets, fillings = await ShipmentService._lock_and_validate_pallets(
                self.db,
                [2],
                7,
                TestBatchScope.formal(),
            )

        self.assertEqual([inventory], inventories)
        self.assertEqual([pallet], pallets)
        self.assertEqual([origin_filling], fillings)

    async def test_test_scope_write_rejects_cross_batch_inventory_and_rolls_back(self) -> None:
        scope = make_test_scope(4)
        page = ShipmentCreateModel(sourceWarehouseId=7, targetSiteId=9, palletIds=[2])
        add_shipment = AsyncMock()
        with (
            patch.object(TestBatchAccessService, 'lock_active_batch', new=AsyncMock()),
            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',
                        test_batch_id=4,
                    )
                ),
            ),
            patch.object(
                MasterDao,
                'get_site',
                new=AsyncMock(
                    return_value=BizCustomerSite(
                        site_id=9,
                        customer_id=1,
                        site_code='S-9',
                        site_name='九号现场',
                        status='0',
                        test_batch_id=4,
                    )
                ),
            ),
            patch.object(
                InventoryDao,
                'get_inventories_for_update',
                new=AsyncMock(return_value=[make_inventory(2, test_batch_id=5)]),
            ),
            patch.object(
                InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[make_pallet(2, test_batch_id=4)])
            ),
            patch.object(OperationDao, 'add_shipment', new=add_shipment),
            self.assertRaisesRegex(ServiceException, '不属于当前数据模式或测试批次'),
        ):
            await ShipmentService.add_shipment_services(self.db, page, self.user, scope)

        add_shipment.assert_not_awaited()
        self.db.commit.assert_not_awaited()
        self.db.rollback.assert_awaited_once()

    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,
            _scope: TestBatchScope,
        ) -> 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, TestBatchScope.formal())
            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], TestBatchScope.formal())
        pallet_lock.assert_awaited_once_with(self.db, [2, 5], TestBatchScope.formal())
        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, TestBatchScope.formal())

        self.assertEqual('无权访问该仓库', raised.exception.message)
        self.assertEqual(2, warehouse_read.await_count)
        warehouse_read.assert_awaited_with(self.db, 7, TestBatchScope.formal(), 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, TestBatchScope.formal())

        self.assertEqual('来源仓库已停用', raised.exception.message)
        warehouse_read.assert_awaited_with(self.db, 7, TestBatchScope.formal(), 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, TestBatchScope.formal())
            except NotImplementedError:
                self.fail('发货草稿编辑尚未实现')

        self.assertTrue(result.is_success)
        lock_order.assert_awaited_once_with(
            self.db,
            12,
            TestBatchScope.formal(),
            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,
                    _scope: TestBatchScope,
                    **_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,
                    _scope: TestBatchScope,
                    _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, TestBatchScope.formal())

                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, TestBatchScope.formal(), 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,
                    _scope: TestBatchScope,
                    _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,
                    _scope: TestBatchScope,
                    _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, TestBatchScope.formal())

                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,
            _scope: TestBatchScope,
            **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,
            _scope: TestBatchScope,
            **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, TestBatchScope.formal())

        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:
        lock = AsyncMock()
        delete = AsyncMock()
        with (
            patch.object(OperationDao, 'get_shipment_for_update', new=lock),
            patch.object(OperationDao, 'delete_shipment', new=delete),
            self.assertRaises(ServiceWarning) as raised,
        ):
            await ShipmentService.delete_shipment_services(self.db, [12, 11], self.user, TestBatchScope.formal())

        self.assertEqual('不支持单条物理删除，请通过测试数据批次清理', raised.exception.message)
        lock.assert_not_awaited()
        delete.assert_not_awaited()
        self.db.commit.assert_not_awaited()

    async def test_batch_delete_rechecks_all_current_routes_before_first_delete(self) -> None:
        lock = AsyncMock()
        delete = AsyncMock()
        with (
            patch.object(OperationDao, 'get_shipment_for_update', new=lock),
            patch.object(OperationDao, 'delete_shipment', new=delete),
            self.assertRaises(ServiceWarning) as raised,
        ):
            await ShipmentService.delete_shipment_services(self.db, [12, 11], self.user, TestBatchScope.formal())

        self.assertEqual('不支持单条物理删除，请通过测试数据批次清理', raised.exception.message)
        lock.assert_not_awaited()
        delete.assert_not_awaited()
        self.db.commit.assert_not_awaited()

    async def test_batch_delete_locks_deduplicated_routes_stably_without_requiring_active(self) -> None:
        db = MagicMock(spec=AsyncSession)
        lock = AsyncMock()
        delete = AsyncMock()
        with (
            patch.object(OperationDao, 'get_shipment_for_update', new=lock),
            patch.object(OperationDao, 'delete_shipment', new=delete),
            self.assertRaises(ServiceWarning) as raised,
        ):
            await ShipmentService.delete_shipment_services(
                db,
                [13, 11, 12],
                self.user,
                TestBatchScope.formal(),
            )

        self.assertEqual('不支持单条物理删除，请通过测试数据批次清理', raised.exception.message)
        lock.assert_not_awaited()
        delete.assert_not_awaited()
        db.commit.assert_not_awaited()


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,
            _scope: TestBatchScope,
        ) -> 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,
            _scope: TestBatchScope,
        ) -> 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,
                    TestBatchScope.formal(),
                )
            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,
            TestBatchScope.formal(),
            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,
                TestBatchScope.formal(),
            )

        self.assertTrue(result.is_success)
        warehouse_lock.assert_awaited_once_with(self.db, 7, TestBatchScope.formal(), 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, TestBatchScope.formal(), 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,
                        TestBatchScope.formal(),
                    )

                self.assertTrue(result.is_success)
                warehouse_lock.assert_awaited_once_with(db, 7, TestBatchScope.formal(), for_update=True)
                warehouse_access.assert_awaited_once_with(db, self.user, 7, for_update=True)
                site_lock.assert_awaited_once_with(db, 9, TestBatchScope.formal(), 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,
                TestBatchScope.formal(),
            )

        self.assertEqual('无权访问该发货单', raised.exception.message)
        warehouse_lock.assert_awaited_once_with(self.db, 7, TestBatchScope.formal(), 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, TestBatchScope.formal(), 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,
                TestBatchScope.formal(),
            )

        self.assertTrue(result.is_success)
        lock_validate.assert_awaited_once_with(self.db, [2], 7, TestBatchScope.formal())
        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, TestBatchScope.formal()
            )

        self.assertEqual('确认发货失败，幂等键已被其他请求使用', raised.exception.message)
        parent_lock.assert_not_awaited()
        self.db.rollback.assert_awaited_once()

    async def test_same_key_retry_uses_exact_route_when_user_can_access_duplicate_keys(self) -> None:
        target = make_shipment(
            status='CONFIRMED',
            idempotency_key='confirm-key',
            request_hash=self.request_hash,
        )
        other_domain = make_shipment(
            13,
            source_warehouse_id=8,
            target_site_id=10,
            status='CONFIRMED',
            idempotency_key='confirm-key',
            request_hash=ShipmentService._build_confirm_request_hash(13),
        )

        async def read_key(
            _db: AsyncSession,
            _key: str,
            _scope: TestBatchScope,
            *,
            warehouse_ids: list[int] | None = None,
            site_ids: list[int] | None = None,
        ) -> BizShipmentOrder | None:
            matches = [
                shipment
                for shipment in (target, other_domain)
                if (warehouse_ids is None or shipment.source_warehouse_id in warehouse_ids)
                and (site_ids is None or shipment.target_site_id in site_ids)
            ]
            if len(matches) > 1:
                raise MultipleResultsFound('广域幂等键查询返回多条发货单')
            return matches[0] if matches else None

        get_detail = AsyncMock(return_value=target)
        key_read = AsyncMock(side_effect=read_key)
        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_by_id', new=get_detail),
            patch.object(OperationDao, 'get_shipment_by_confirm_idempotency_key', new=key_read),
            patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=target)),
            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()),
        ):
            result = await ShipmentService.confirm_shipment_services(
                self.db,
                12,
                self.command,
                self.user,
                TestBatchScope.formal(),
            )

        self.assertTrue(result.is_success)
        get_detail.assert_awaited_once_with(
            self.db,
            12,
            TestBatchScope.formal(),
            warehouse_ids=[7, 8],
            site_ids=[9, 10],
        )
        key_read.assert_awaited_once_with(
            self.db,
            'confirm-key',
            TestBatchScope.formal(),
            warehouse_ids=[7],
            site_ids=[9],
        )
        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, TestBatchScope.formal()
            )

        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, TestBatchScope.formal()
            )

        self.assertEqual('无权访问该发货单', raised.exception.message)
        self.assertEqual(2, parent_lock.await_count)
        parent_lock.assert_awaited_with(
            self.db,
            12,
            TestBatchScope.formal(),
            warehouse_ids=[7],
            site_ids=[9],
        )
        warehouse_lock.assert_awaited_once_with(self.db, 7, TestBatchScope.formal(), 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, TestBatchScope.formal(), 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, TestBatchScope.formal()
            )

        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, TestBatchScope.formal(), 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,
                TestBatchScope.formal(),
            )

        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_integrity_recovery_uses_exact_route_with_duplicate_visible_keys(self) -> None:
        initial = make_shipment(status='DRAFT')
        recovered = make_shipment(
            status='CONFIRMED',
            idempotency_key='confirm-key',
            request_hash=self.request_hash,
        )
        other_domain = make_shipment(
            13,
            source_warehouse_id=8,
            target_site_id=10,
            status='CONFIRMED',
            idempotency_key='confirm-key',
            request_hash=ShipmentService._build_confirm_request_hash(13),
        )
        read_count = 0

        async def read_key(
            _db: AsyncSession,
            _key: str,
            _scope: TestBatchScope,
            *,
            warehouse_ids: list[int] | None = None,
            site_ids: list[int] | None = None,
        ) -> BizShipmentOrder | None:
            nonlocal read_count
            read_count += 1
            if read_count == 1:
                return None
            matches = [
                shipment
                for shipment in (recovered, other_domain)
                if (warehouse_ids is None or shipment.source_warehouse_id in warehouse_ids)
                and (site_ids is None or shipment.target_site_id in site_ids)
            ]
            if len(matches) > 1:
                raise MultipleResultsFound('广域幂等键恢复查询返回多条发货单')
            return matches[0] if matches else None

        warehouse_scope = AsyncMock(side_effect=[[7, 8], [7, 8]])
        site_scope = AsyncMock(side_effect=[[9, 10], [9, 10]])
        key_read = AsyncMock(side_effect=read_key)
        parent_lock = AsyncMock(side_effect=[integrity_error(), 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=parent_lock),
            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()),
        ):
            result = await ShipmentService.confirm_shipment_services(
                self.db,
                12,
                self.command,
                self.user,
                TestBatchScope.formal(),
            )

        self.assertTrue(result.is_success)
        self.assertEqual(2, warehouse_scope.await_count)
        self.assertEqual(2, site_scope.await_count)
        self.assertEqual(2, parent_lock.await_count)
        self.assertEqual(
            [
                unittest.mock.call(
                    self.db,
                    'confirm-key',
                    TestBatchScope.formal(),
                    warehouse_ids=[7],
                    site_ids=[9],
                ),
                unittest.mock.call(
                    self.db,
                    'confirm-key',
                    TestBatchScope.formal(),
                    warehouse_ids=[7],
                    site_ids=[9],
                ),
            ],
            key_read.await_args_list,
        )
        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, TestBatchScope.formal()
            )

        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, TestBatchScope.formal(), 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,
            _scope: TestBatchScope,
        ) -> 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, TestBatchScope.formal()
            )

        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()