From aba8a96cb8379197cd917f6f0b7a7750fa0ea149 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=84=A6=E9=BE=99=E8=A8=80?= Date: Sun, 26 Jul 2026 12:36:00 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E6=B6=88=E9=99=A4?= =?UTF-8?q?=E6=89=98=E7=9B=98=E4=B8=8E=E7=94=A8=E6=88=B7=E5=88=86=E9=85=8D?= =?UTF-8?q?=E6=9F=A5=E8=AF=A2=E7=AB=9E=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../module_business/dao/master_dao.py | 95 ++++- .../module_business/service/pallet_service.py | 69 ++- .../module_business/test_pallet_service.py | 393 ++++++++++++++---- 3 files changed, 446 insertions(+), 111 deletions(-) diff --git a/shierfeng-fastapi-backend/module_business/dao/master_dao.py b/shierfeng-fastapi-backend/module_business/dao/master_dao.py index e267e93..82d1687 100644 --- a/shierfeng-fastapi-backend/module_business/dao/master_dao.py +++ b/shierfeng-fastapi-backend/module_business/dao/master_dao.py @@ -2,12 +2,13 @@ from datetime import datetime from typing import Any, Literal from pydantic import BaseModel -from sqlalchemy import ColumnElement, Select, delete, func, or_, select, update +from sqlalchemy import ColumnElement, Select, and_, delete, func, or_, select, update from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm.attributes import InstrumentedAttribute from common.vo import PageModel from module_admin.entity.do.user_do import SysUser +from module_business.domain.enums import PalletFlowState from module_business.entity.do.master_do import ( BizCustomer, BizCustomerSite, @@ -20,10 +21,13 @@ from module_business.entity.do.master_do import ( ) from module_business.entity.do.operation_do import ( BizFillingRecord, + BizInboundItem, BizInboundOrder, BizPalletEvent, BizPalletInventory, + BizReturnItem, BizReturnOrder, + BizShipmentItem, BizShipmentOrder, BizUsageRecord, ) @@ -74,6 +78,15 @@ _REFERENCE_COLUMNS: dict[MasterResource, tuple[tuple[InstrumentedAttribute, ...] ), } +_PALLET_REMOVAL_BLOCKING_COLUMNS: tuple[InstrumentedAttribute, ...] = ( + BizInboundItem.pallet_id, + BizFillingRecord.pallet_id, + BizShipmentItem.pallet_id, + BizUsageRecord.pallet_id, + BizReturnItem.pallet_id, + BizPalletEvent.pallet_id, +) + class MasterDao: """菌种、客户、现场、产线和仓库主数据数据库操作层。""" @@ -99,8 +112,13 @@ class MasterDao: model: type[Any], id_column: InstrumentedAttribute, resource_id: int, + *, + for_update: bool = False, ) -> Any | None: - result = await db.execute(select(model).where(id_column == resource_id)) + statement = select(model).where(id_column == resource_id) + if for_update: + statement = statement.with_for_update() + result = await db.execute(statement) return result.scalar_one_or_none() @staticmethod @@ -245,8 +263,20 @@ class MasterDao: return await cls._paginate(db, statement, query_object, is_page) @classmethod - async def get_site(cls, db: AsyncSession, site_id: int) -> BizCustomerSite | None: - return await cls._get(db, BizCustomerSite, BizCustomerSite.site_id, site_id) + async def get_site( + cls, + db: AsyncSession, + site_id: int, + *, + for_update: bool = False, + ) -> BizCustomerSite | None: + return await cls._get( + db, + BizCustomerSite, + BizCustomerSite.site_id, + site_id, + for_update=for_update, + ) @classmethod async def find_site_by_code( @@ -343,8 +373,20 @@ class MasterDao: return await cls._paginate(db, statement, query_object, is_page) @classmethod - async def get_warehouse(cls, db: AsyncSession, warehouse_id: int) -> BizWarehouse | None: - return await cls._get(db, BizWarehouse, BizWarehouse.warehouse_id, warehouse_id) + async def get_warehouse( + cls, + db: AsyncSession, + warehouse_id: int, + *, + for_update: bool = False, + ) -> BizWarehouse | None: + return await cls._get( + db, + BizWarehouse, + BizWarehouse.warehouse_id, + warehouse_id, + for_update=for_update, + ) @classmethod async def find_warehouse_by_code(cls, db: AsyncSession, warehouse_code: str) -> BizWarehouse | None: @@ -391,6 +433,37 @@ class MasterDao: result = await db.execute(statement) return result.scalar_one_or_none() + @classmethod + async def get_pallet_removability( + cls, + db: AsyncSession, + pallet_ids: list[int], + ) -> dict[int, bool]: + """批量判断托盘是否仍为未入库且没有任何业务引用。""" + removability = dict.fromkeys(pallet_ids, False) + if not pallet_ids: + return removability + + has_unstocked_inventory = ( + select(BizPalletInventory.pallet_id) + .where( + BizPalletInventory.pallet_id == BizPallet.pallet_id, + BizPalletInventory.flow_state == PalletFlowState.UNSTOCKED.value, + ) + .exists() + ) + has_no_business_references = tuple( + ~select(column).where(column == BizPallet.pallet_id).exists() for column in _PALLET_REMOVAL_BLOCKING_COLUMNS + ) + statement = select( + BizPallet.pallet_id, + and_(has_unstocked_inventory, *has_no_business_references).label('can_remove'), + ).where(BizPallet.pallet_id.in_(pallet_ids)) + result = await db.execute(statement) + for pallet_id, can_remove in result.all(): + removability[pallet_id] = bool(can_remove) + return removability + @classmethod async def find_pallet_by_code(cls, db: AsyncSession, pallet_code: str) -> BizPallet | None: return await cls._find(db, BizPallet, BizPallet.pallet_code, pallet_code) @@ -484,6 +557,16 @@ class MasterDao: ) return list(result.all()) + @classmethod + async def list_warehouse_assigned_user_ids(cls, db: AsyncSession, warehouse_id: int) -> list[int]: + result = await db.execute(select(BizUserWarehouse.user_id).where(BizUserWarehouse.warehouse_id == warehouse_id)) + return list(result.scalars().all()) + + @classmethod + async def list_site_assigned_user_ids(cls, db: AsyncSession, site_id: int) -> list[int]: + result = await db.execute(select(BizUserSite.user_id).where(BizUserSite.site_id == site_id)) + return list(result.scalars().all()) + @classmethod async def count_references( cls, diff --git a/shierfeng-fastapi-backend/module_business/service/pallet_service.py b/shierfeng-fastapi-backend/module_business/service/pallet_service.py index d4cd515..a227a14 100644 --- a/shierfeng-fastapi-backend/module_business/service/pallet_service.py +++ b/shierfeng-fastapi-backend/module_business/service/pallet_service.py @@ -1,7 +1,7 @@ import secrets from datetime import datetime from io import BytesIO -from typing import Any, Literal +from typing import Literal import qrcode from sqlalchemy.exc import IntegrityError @@ -11,7 +11,6 @@ from common.vo import CrudResponseModel, PageModel from exceptions.exception import ServiceException from module_business.dao.access_dao import AccessDao from module_business.dao.master_dao import MasterDao -from module_business.domain.enums import PalletFlowState from module_business.entity.do.master_do import BizPallet from module_business.entity.do.operation_do import BizPalletInventory from module_business.entity.vo.master_vo import ( @@ -58,18 +57,6 @@ class PalletService: raise ServiceException(message='托盘ID不能重复') return sorted(pallet_ids) - @classmethod - async def _can_remove(cls, db: AsyncSession, pallet_id: int) -> bool: - inventory = await MasterDao.get_pallet_inventory(db, pallet_id) - if inventory is None or inventory.flow_state != PalletFlowState.UNSTOCKED.value: - return False - return await MasterDao.count_pallet_events(db, pallet_id) == 0 - - @classmethod - async def _to_pallet_model(cls, db: AsyncSession, record: Any) -> PalletModel: - model = PalletModel.model_validate(record) - return model.model_copy(update={'can_remove': await cls._can_remove(db, model.pallet_id)}) - @classmethod async def _lock_pallet_inventory(cls, db: AsyncSession, pallet_id: int) -> tuple[BizPallet, BizPalletInventory]: """按库存、托盘顺序加锁，避免与流转事务形成反序死锁。""" @@ -91,17 +78,25 @@ class PalletService: is_page: bool = True, ) -> PageModel | list[PalletModel]: result = await MasterDao.list_pallet(db, query_object, is_page) + records = result.rows if isinstance(result, PageModel) else result + models = [PalletModel.model_validate(record) for record in records] + removability = await MasterDao.get_pallet_removability( + db, + [model.pallet_id for model in models], + ) + rows = [model.model_copy(update={'can_remove': removability.get(model.pallet_id, False)}) for model in models] if isinstance(result, PageModel): - rows = [await cls._to_pallet_model(db, row) for row in result.rows] return result.model_copy(update={'rows': rows}) - return [await cls._to_pallet_model(db, row) for row in result] + return rows @classmethod async def detail_pallet(cls, db: AsyncSession, pallet_id: int) -> PalletModel: pallet = await MasterDao.get_pallet(db, pallet_id) if pallet is None: raise ServiceException(message='托盘不存在') - return await cls._to_pallet_model(db, pallet) + model = PalletModel.model_validate(pallet) + removability = await MasterDao.get_pallet_removability(db, [pallet_id]) + return model.model_copy(update={'can_remove': removability.get(pallet_id, False)}) @classmethod async def add_pallet( @@ -179,11 +174,11 @@ class PalletService: try: pallet_ids = cls._extract_pallet_ids(value) for pallet_id in pallet_ids: - _, inventory = await cls._lock_pallet_inventory(db, pallet_id) - if inventory.flow_state != PalletFlowState.UNSTOCKED.value: - raise ServiceException(message='托盘未处于未入库状态，不能删除') - if await MasterDao.count_pallet_events(db, pallet_id) > 0: - raise ServiceException(message='托盘已有流转事件，不能删除') + await cls._lock_pallet_inventory(db, pallet_id) + + removability = await MasterDao.get_pallet_removability(db, pallet_ids) + if not all(removability.get(pallet_id, False) for pallet_id in pallet_ids): + raise ServiceException(message='托盘状态或关联数据不允许删除') for pallet_id in pallet_ids: await MasterDao.delete_pallet_inventory(db, pallet_id) @@ -208,19 +203,16 @@ class PalletService: raise ServiceException(message=f'{_ASSIGNMENT_LABELS[resource]}不存在') users = await MasterDao.list_assignable_users(db) - result = [] - association_method = getattr(AccessDao, f'list_user_{resource}_ids') - for user in users: - assigned_ids = await association_method(db, user.user_id) - result.append( - BusinessUserOptionModel( - user_id=user.user_id, - user_name=user.user_name, - nick_name=user.nick_name, - assigned=resource_id in assigned_ids, - ) + assigned_user_ids = set(await getattr(MasterDao, f'list_{resource}_assigned_user_ids')(db, resource_id)) + return [ + BusinessUserOptionModel( + user_id=user.user_id, + user_name=user.user_name, + nick_name=user.nick_name, + assigned=user.user_id in assigned_user_ids, ) - return result + for user in users + ] @classmethod async def list_warehouse_users( @@ -248,7 +240,14 @@ class PalletService: ) -> CrudResponseModel: label = _ASSIGNMENT_LABELS[resource] try: - if await getattr(MasterDao, f'get_{resource}')(db, resource_id) is None: + if ( + await getattr(MasterDao, f'get_{resource}')( + db, + resource_id, + for_update=True, + ) + is None + ): raise ServiceException(message=f'{label}不存在') users = await MasterDao.list_assignable_users_by_ids(db, assignment.user_ids) if {user.user_id for user in users} != set(assignment.user_ids): diff --git a/shierfeng-fastapi-backend/tests/module_business/test_pallet_service.py b/shierfeng-fastapi-backend/tests/module_business/test_pallet_service.py index 29a984e..ef5bcc8 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_pallet_service.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_pallet_service.py @@ -5,10 +5,12 @@ from datetime import datetime from types import SimpleNamespace from typing import Any from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import call as mock_call from fastapi import FastAPI from fastapi.routing import APIRoute from pydantic import ValidationError +from sqlalchemy.dialects import mysql, postgresql from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession @@ -23,8 +25,22 @@ from module_business.controller import site_controller as site_controller_module from module_business.controller import warehouse_controller as warehouse_controller_module from module_business.dao.access_dao import AccessDao from module_business.dao.master_dao import MasterDao -from module_business.entity.do.master_do import BizPallet -from module_business.entity.do.operation_do import BizPalletInventory +from module_business.entity.do.master_do import ( + BizCustomerSite, + BizPallet, + BizUserSite, + BizUserWarehouse, + BizWarehouse, +) +from module_business.entity.do.operation_do import ( + BizFillingRecord, + BizInboundItem, + BizPalletEvent, + BizPalletInventory, + BizReturnItem, + BizShipmentItem, + BizUsageRecord, +) from module_business.entity.vo.master_vo import ( BusinessUserOptionModel, DeletePalletModel, @@ -122,6 +138,57 @@ class TestPalletMasterDao(unittest.IsolatedAsyncioTestCase): self.assertIn("biz_pallet.asset_status = '0'", sql) self.assertIn('ORDER BY biz_pallet.pallet_id', sql) + async def test_pallet_removability_uses_one_complete_business_reference_query(self) -> None: + query_result = MagicMock() + query_result.all.return_value = [(2, True), (3, False)] + self.db.execute.return_value = query_result + + actual = await MasterDao.get_pallet_removability(self.db, [1, 2, 3]) + + business_reference_columns = ( + BizInboundItem.pallet_id, + BizFillingRecord.pallet_id, + BizShipmentItem.pallet_id, + BizUsageRecord.pallet_id, + BizReturnItem.pallet_id, + BizPalletEvent.pallet_id, + ) + expected_references = {(column.class_.__tablename__, column.key) for column in business_reference_columns} + metadata_references = { + (table.name, column.name) + for table in BizPallet.metadata.tables.values() + if table.name != BizPalletInventory.__tablename__ + for column in table.columns + if any(foreign_key.target_fullname == 'biz_pallet.pallet_id' for foreign_key in column.foreign_keys) + } + self.assertEqual(expected_references, metadata_references) + self.assertEqual({1: False, 2: True, 3: False}, actual) + self.db.execute.assert_awaited_once() + + statement = self.db.execute.await_args.args[0] + for dialect in (mysql.dialect(), postgresql.dialect()): + with self.subTest(dialect=dialect.name): + sql = ' '.join(str(statement.compile(dialect=dialect, compile_kwargs={'literal_binds': True})).split()) + self.assertIn('FROM biz_pallet', sql) + self.assertIn( + 'EXISTS (SELECT biz_pallet_inventory.pallet_id FROM biz_pallet_inventory ' + 'WHERE biz_pallet_inventory.pallet_id = biz_pallet.pallet_id ' + "AND biz_pallet_inventory.flow_state = 'UNSTOCKED')", + sql, + ) + self.assertEqual(6, sql.count('NOT (EXISTS')) + for table_name, _ in expected_references: + self.assertIn( + f'NOT (EXISTS (SELECT {table_name}.pallet_id FROM {table_name} ' + f'WHERE {table_name}.pallet_id = biz_pallet.pallet_id))', + sql, + ) + + async def test_empty_pallet_removability_skips_database_query(self) -> None: + self.assertEqual({}, await MasterDao.get_pallet_removability(self.db, [])) + + self.db.execute.assert_not_awaited() + async def test_add_pallet_and_inventory_only_flush_current_transaction(self) -> None: pallet = BizPallet( pallet_code='P-001', @@ -208,6 +275,82 @@ class TestPalletMasterDao(unittest.IsolatedAsyncioTestCase): self.assertIn('biz_pallet.pallet_id = 9', compile_sql(statement)) self.assertIn('FOR UPDATE', compile_sql(statement)) + async def test_assignment_resource_write_reads_use_row_lock(self) -> None: + cases = ( + (MasterDao.get_warehouse, BizWarehouse, 'warehouse_id', 7), + (MasterDao.get_site, BizCustomerSite, 'site_id', 9), + ) + for method, model, id_field, resource_id in cases: + with self.subTest(resource=model.__tablename__): + self.db.reset_mock() + query_result = MagicMock() + record = SimpleNamespace() + query_result.scalar_one_or_none.return_value = record + self.db.execute.return_value = query_result + + actual = await method(self.db, resource_id, for_update=True) + + self.assertIs(record, actual) + statement = self.db.execute.await_args.args[0] + sql = compile_sql(statement) + self.assertTrue(statement.get_final_froms()[0].compare(model.__table__)) + self.assertIn(f'{model.__tablename__}.{id_field} = {resource_id}', sql) + self.assertIn('FOR UPDATE', sql) + + async def test_assignment_resource_reads_are_unlocked_by_default(self) -> None: + cases = ( + (MasterDao.get_warehouse, BizWarehouse, 7), + (MasterDao.get_site, BizCustomerSite, 9), + ) + for method, model, resource_id in cases: + with self.subTest(resource=model.__tablename__): + self.db.reset_mock() + query_result = MagicMock() + query_result.scalar_one_or_none.return_value = SimpleNamespace() + self.db.execute.return_value = query_result + + await method(self.db, resource_id) + + statement = self.db.execute.await_args.args[0] + self.assertTrue(statement.get_final_froms()[0].compare(model.__table__)) + self.assertNotIn('FOR UPDATE', compile_sql(statement)) + + async def test_assigned_user_id_queries_target_exact_resource_association(self) -> None: + cases = ( + ( + MasterDao.list_warehouse_assigned_user_ids, + BizUserWarehouse, + BizUserWarehouse.user_id, + 'warehouse_id', + 7, + ), + ( + MasterDao.list_site_assigned_user_ids, + BizUserSite, + BizUserSite.user_id, + 'site_id', + 9, + ), + ) + for method, model, selected_column, id_field, resource_id in cases: + with self.subTest(resource=model.__tablename__): + self.db.reset_mock() + query_result = MagicMock() + query_result.scalars.return_value.all.return_value = [2, 5] + self.db.execute.return_value = query_result + + actual = await method(self.db, resource_id) + + self.assertEqual([2, 5], actual) + self.db.execute.assert_awaited_once() + statement = self.db.execute.await_args.args[0] + self.assertEqual([selected_column], list(statement.selected_columns)) + self.assertTrue(statement.get_final_froms()[0].compare(model.__table__)) + self.assertIn( + f'{model.__tablename__}.{id_field} = {resource_id}', + compile_sql(statement), + ) + async def test_assignable_user_queries_filter_disabled_and_deleted_rows(self) -> None: list_result = MagicMock() list_result.all.return_value = () @@ -386,49 +529,56 @@ class TestPalletQueryAndEditService(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.db = MagicMock(spec=AsyncSession) - async def test_list_computes_can_remove_from_inventory_and_events(self) -> None: + async def test_list_computes_can_remove_with_one_bulk_query(self) -> None: page = PageModel[Any]( - rows=[make_pallet_record(1), make_pallet_record(2)], + rows=[make_pallet_record(pallet_id) for pallet_id in range(1, 1001)], pageNum=1, - pageSize=10, - total=2, + pageSize=1000, + total=1000, hasNext=False, ) query = PalletPageQueryModel() + list_pallet = AsyncMock(return_value=page) + removability = AsyncMock(return_value={pallet_id: pallet_id % 2 == 1 for pallet_id in range(1, 1001)}) + get_inventory = AsyncMock(return_value=SimpleNamespace(flow_state='UNSTOCKED')) + count_events = AsyncMock(return_value=0) with ( - patch.object(MasterDao, 'list_pallet', new=AsyncMock(return_value=page)), - patch.object( - MasterDao, - 'get_pallet_inventory', - new=AsyncMock( - side_effect=[ - SimpleNamespace(flow_state='UNSTOCKED'), - SimpleNamespace(flow_state='UNSTOCKED'), - ] - ), - ), - patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(side_effect=[0, 1])), + patch.object(MasterDao, 'list_pallet', new=list_pallet), + patch.object(MasterDao, 'get_pallet_removability', new=removability, create=True), + patch.object(MasterDao, 'get_pallet_inventory', new=get_inventory), + patch.object(MasterDao, 'count_pallet_events', new=count_events), ): actual = await PalletService.list_pallet(self.db, query, is_page=True) self.assertIsInstance(actual, PageModel) + self.assertEqual(1000, len(actual.rows)) self.assertTrue(all(isinstance(row, PalletModel) for row in actual.rows)) - self.assertEqual([True, False], [row.can_remove for row in actual.rows]) + self.assertEqual( + [pallet_id % 2 == 1 for pallet_id in range(1, 1001)], + [row.can_remove for row in actual.rows], + ) + list_pallet.assert_awaited_once_with(self.db, query, True) + removability.assert_awaited_once_with(self.db, list(range(1, 1001))) + get_inventory.assert_not_awaited() + count_events.assert_not_awaited() async def test_detail_computes_false_when_inventory_is_not_unstocked(self) -> None: + removability = AsyncMock(return_value={3: False}) + get_inventory = AsyncMock(return_value=SimpleNamespace(flow_state='EMPTY_IN_WAREHOUSE')) + count_events = AsyncMock(return_value=0) with ( patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=make_pallet_record(3))), - patch.object( - MasterDao, - 'get_pallet_inventory', - new=AsyncMock(return_value=SimpleNamespace(flow_state='EMPTY_IN_WAREHOUSE')), - ), - patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(return_value=0)), + patch.object(MasterDao, 'get_pallet_removability', new=removability, create=True), + patch.object(MasterDao, 'get_pallet_inventory', new=get_inventory), + patch.object(MasterDao, 'count_pallet_events', new=count_events), ): actual = await PalletService.detail_pallet(self.db, pallet_id=3) self.assertIsInstance(actual, PalletModel) self.assertFalse(actual.can_remove) + removability.assert_awaited_once_with(self.db, [3]) + get_inventory.assert_not_awaited() + count_events.assert_not_awaited() async def test_detail_rejects_missing_pallet(self) -> None: with ( @@ -539,6 +689,10 @@ class TestPalletRemovalService(unittest.IsolatedAsyncioTestCase): actions.append(f'count-events-{pallet_id}') return 0 + async def get_removability(_db: AsyncSession, pallet_ids: list[int]) -> dict[int, bool]: + actions.append(f'get-removability-{pallet_ids}') + return dict.fromkeys(pallet_ids, True) + async def delete_inventory(_db: AsyncSession, pallet_id: int) -> None: actions.append(f'delete-inventory-{pallet_id}') @@ -553,6 +707,12 @@ class TestPalletRemovalService(unittest.IsolatedAsyncioTestCase): patch.object(MasterDao, 'get_pallet', new=AsyncMock(side_effect=get_pallet)), patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(side_effect=get_inventory)), patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(side_effect=count_events)), + patch.object( + MasterDao, + 'get_pallet_removability', + new=AsyncMock(side_effect=get_removability), + create=True, + ), patch.object(MasterDao, 'delete_pallet_inventory', new=AsyncMock(side_effect=delete_inventory)), patch.object(MasterDao, 'delete_pallet', new=AsyncMock(side_effect=delete_pallet)), ): @@ -562,10 +722,9 @@ class TestPalletRemovalService(unittest.IsolatedAsyncioTestCase): [ 'get-inventory-4-True', 'get-pallet-4-True', - 'count-events-4', 'get-inventory-5-True', 'get-pallet-5-True', - 'count-events-5', + 'get-removability-[4, 5]', 'delete-inventory-4', 'delete-pallet-4', 'delete-inventory-5', @@ -578,32 +737,26 @@ class TestPalletRemovalService(unittest.IsolatedAsyncioTestCase): self.db.rollback.assert_not_awaited() self.assertTrue(result.is_success) - async def test_remove_rejects_invalid_state_before_any_delete(self) -> None: + async def test_remove_rejects_missing_pallet_or_inventory(self) -> None: cases = ( - ('托盘不存在', None, None, 0), - ('托盘库存不存在', make_pallet_record(4), None, 0), - ( - '托盘未处于未入库状态，不能删除', - make_pallet_record(4), - SimpleNamespace(flow_state='EMPTY_IN_WAREHOUSE'), - 0, - ), - ( - '托盘已有流转事件，不能删除', - make_pallet_record(4), - SimpleNamespace(flow_state='UNSTOCKED'), - 1, - ), + ('托盘不存在', None), + ('托盘库存不存在', make_pallet_record(4)), ) - for message, pallet, inventory, event_count in cases: + for message, pallet in cases: with self.subTest(message=message): self.db.reset_mock() delete_inventory = AsyncMock() delete_pallet = AsyncMock() + removability = AsyncMock() with ( patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=pallet)), - patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=inventory)), - patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(return_value=event_count)), + patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=None)), + patch.object( + MasterDao, + 'get_pallet_removability', + new=removability, + create=True, + ), patch.object(MasterDao, 'delete_pallet_inventory', new=delete_inventory), patch.object(MasterDao, 'delete_pallet', new=delete_pallet), self.assertRaises(ServiceException) as raised, @@ -611,12 +764,55 @@ class TestPalletRemovalService(unittest.IsolatedAsyncioTestCase): await PalletService.remove_pallet(self.db, 4) self.assertEqual(message, raised.exception.message) + removability.assert_not_awaited() delete_inventory.assert_not_awaited() delete_pallet.assert_not_awaited() self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() + async def test_remove_rejects_any_ineligible_pallet_before_any_delete(self) -> None: + delete_inventory = AsyncMock() + delete_pallet = AsyncMock() + removability = AsyncMock(return_value={4: True, 5: False}) + with ( + patch.object( + MasterDao, + 'get_pallet', + new=AsyncMock(side_effect=[make_pallet_record(4), make_pallet_record(5)]), + ), + patch.object( + MasterDao, + 'get_pallet_inventory', + new=AsyncMock( + side_effect=[ + SimpleNamespace(flow_state='UNSTOCKED'), + SimpleNamespace(flow_state='UNSTOCKED'), + ] + ), + ), + patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(return_value=0)), + patch.object( + MasterDao, + 'get_pallet_removability', + new=removability, + create=True, + ), + patch.object(MasterDao, 'delete_pallet_inventory', new=delete_inventory), + patch.object(MasterDao, 'delete_pallet', new=delete_pallet), + self.assertRaises(ServiceException) as raised, + ): + await PalletService.remove_pallet(self.db, DeletePalletModel(palletIds=[5, 4])) + + self.assertEqual('托盘状态或关联数据不允许删除', raised.exception.message) + removability.assert_awaited_once_with(self.db, [4, 5]) + delete_inventory.assert_not_awaited() + delete_pallet.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + async def test_remove_integrity_error_rolls_back_with_stable_failure(self) -> None: + removability = AsyncMock(return_value={4: True}) + count_events = AsyncMock(return_value=0) with ( patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=make_pallet_record(4))), patch.object( @@ -624,7 +820,13 @@ class TestPalletRemovalService(unittest.IsolatedAsyncioTestCase): 'get_pallet_inventory', new=AsyncMock(return_value=SimpleNamespace(flow_state='UNSTOCKED')), ), - patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(return_value=0)), + patch.object(MasterDao, 'count_pallet_events', new=count_events), + patch.object( + MasterDao, + 'get_pallet_removability', + new=removability, + create=True, + ), patch.object(MasterDao, 'delete_pallet_inventory', new=AsyncMock()), patch.object(MasterDao, 'delete_pallet', new=AsyncMock(side_effect=integrity_error())), self.assertRaises(ServiceException) as raised, @@ -632,6 +834,8 @@ class TestPalletRemovalService(unittest.IsolatedAsyncioTestCase): await PalletService.remove_pallet(self.db, 4) self.assertEqual('删除托盘失败，托盘状态或关联数据已变化', raised.exception.message) + removability.assert_awaited_once_with(self.db, [4]) + count_events.assert_not_awaited() self.db.rollback.assert_awaited_once() self.db.commit.assert_not_awaited() @@ -659,25 +863,26 @@ class TestBusinessUserAssignmentService(unittest.IsolatedAsyncioTestCase): ] async def test_assignment_get_returns_only_safe_fields_with_assigned_marker(self) -> None: - cases = ( - ('warehouse', 7, [7], []), - ('site', 9, [9], []), - ) - for resource, resource_id, first_ids, second_ids in cases: + for resource, resource_id in (('warehouse', 7), ('site', 9)): with self.subTest(resource=resource): - association_method = f'list_user_{resource}_ids' + get_resource = AsyncMock(return_value=SimpleNamespace()) + list_users = AsyncMock(return_value=self.users) + list_assigned_ids = AsyncMock(return_value=[2]) + list_user_resource_ids = AsyncMock() with ( patch.object( MasterDao, f'get_{resource}', - new=AsyncMock(return_value=SimpleNamespace()), + new=get_resource, ), - patch.object(MasterDao, 'list_assignable_users', new=AsyncMock(return_value=self.users)), + patch.object(MasterDao, 'list_assignable_users', new=list_users), patch.object( - AccessDao, - association_method, - new=AsyncMock(side_effect=[first_ids, second_ids]), - ) as list_ids, + MasterDao, + f'list_{resource}_assigned_user_ids', + new=list_assigned_ids, + create=True, + ), + patch.object(AccessDao, f'list_user_{resource}_ids', new=list_user_resource_ids), ): actual = await getattr(PalletService, f'list_{resource}_users')(self.db, resource_id) @@ -689,24 +894,57 @@ class TestBusinessUserAssignmentService(unittest.IsolatedAsyncioTestCase): [option.model_dump(by_alias=True) for option in actual], ) self.assertTrue(all(isinstance(option, BusinessUserOptionModel) for option in actual)) - self.assertEqual( - [(self.db, 2), (self.db, 5)], - [call.args for call in list_ids.await_args_list], - ) + get_resource.assert_awaited_once_with(self.db, resource_id) + list_users.assert_awaited_once_with(self.db) + list_assigned_ids.assert_awaited_once_with(self.db, resource_id) + list_user_resource_ids.assert_not_awaited() + + async def test_assignment_get_uses_three_fixed_queries_for_1000_users(self) -> None: + users = [ + SimpleNamespace(user_id=user_id, user_name=f'user-{user_id}', nick_name=None) for user_id in range(1, 1001) + ] + assigned_ids = [2, 500, 1000] + for resource, resource_id in (('warehouse', 7), ('site', 9)): + with self.subTest(resource=resource): + get_resource = AsyncMock(return_value=SimpleNamespace()) + list_users = AsyncMock(return_value=users) + list_assigned = AsyncMock(return_value=assigned_ids) + list_user_resource_ids = AsyncMock() + with ( + patch.object(MasterDao, f'get_{resource}', new=get_resource), + patch.object(MasterDao, 'list_assignable_users', new=list_users), + patch.object( + MasterDao, + f'list_{resource}_assigned_user_ids', + new=list_assigned, + create=True, + ), + patch.object(AccessDao, f'list_user_{resource}_ids', new=list_user_resource_ids), + ): + actual = await getattr(PalletService, f'list_{resource}_users')(self.db, resource_id) + + self.assertEqual(1000, len(actual)) + self.assertEqual(assigned_ids, [option.user_id for option in actual if option.assigned]) + get_resource.assert_awaited_once_with(self.db, resource_id) + list_users.assert_awaited_once_with(self.db) + list_assigned.assert_awaited_once_with(self.db, resource_id) + list_user_resource_ids.assert_not_awaited() async def test_assignment_get_rejects_missing_resource_before_user_query(self) -> None: cases = (('warehouse', 7, '仓库不存在'), ('site', 9, '客户现场不存在')) for resource, resource_id, message in cases: with self.subTest(resource=resource): + get_resource = AsyncMock(return_value=None) list_users = AsyncMock() with ( - patch.object(MasterDao, f'get_{resource}', new=AsyncMock(return_value=None)), + patch.object(MasterDao, f'get_{resource}', new=get_resource), patch.object(MasterDao, 'list_assignable_users', new=list_users), self.assertRaises(ServiceException) as raised, ): await getattr(PalletService, f'list_{resource}_users')(self.db, resource_id) self.assertEqual(message, raised.exception.message) + get_resource.assert_awaited_once_with(self.db, resource_id) list_users.assert_not_awaited() async def test_assignment_put_validates_users_replaces_and_commits(self) -> None: @@ -714,19 +952,25 @@ class TestBusinessUserAssignmentService(unittest.IsolatedAsyncioTestCase): for resource, resource_id in (('warehouse', 7), ('site', 9)): with self.subTest(resource=resource): self.db.reset_mock() - replace = AsyncMock() + calls = MagicMock() + get_resource_mock = AsyncMock(return_value=SimpleNamespace()) + list_users_mock = AsyncMock(return_value=self.users) + replace_mock = AsyncMock() + calls.attach_mock(get_resource_mock, 'get_resource') + calls.attach_mock(list_users_mock, 'list_users') + calls.attach_mock(replace_mock, 'replace') with ( patch.object( MasterDao, f'get_{resource}', - new=AsyncMock(return_value=SimpleNamespace()), + new=get_resource_mock, ), patch.object( MasterDao, 'list_assignable_users_by_ids', - new=AsyncMock(return_value=self.users), - ) as list_users, - patch.object(AccessDao, f'replace_{resource}_users', new=replace), + new=list_users_mock, + ), + patch.object(AccessDao, f'replace_{resource}_users', new=replace_mock), ): result = await getattr(PalletService, f'assign_{resource}_users')( self.db, @@ -734,8 +978,17 @@ class TestBusinessUserAssignmentService(unittest.IsolatedAsyncioTestCase): assignment, ) - list_users.assert_awaited_once_with(self.db, [2, 5]) - replace.assert_awaited_once_with(self.db, resource_id, [2, 5]) + self.assertEqual( + [ + mock_call.get_resource(self.db, resource_id, for_update=True), + mock_call.list_users(self.db, [2, 5]), + mock_call.replace(self.db, resource_id, [2, 5]), + ], + calls.mock_calls, + ) + get_resource_mock.assert_awaited_once_with(self.db, resource_id, for_update=True) + list_users_mock.assert_awaited_once_with(self.db, [2, 5]) + replace_mock.assert_awaited_once_with(self.db, resource_id, [2, 5]) self.db.commit.assert_awaited_once() self.db.rollback.assert_not_awaited() self.assertTrue(result.is_success)