diff --git a/shierfeng-fastapi-backend/module_business/dao/inventory_dao.py b/shierfeng-fastapi-backend/module_business/dao/inventory_dao.py new file mode 100644 index 0000000..d3bbf6b --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/dao/inventory_dao.py @@ -0,0 +1,101 @@ +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from module_business.entity.do.master_do import BizPallet +from module_business.entity.do.operation_do import BizPalletEvent, BizPalletInventory + + +class InventoryDao: + """托盘库存锁、资产锁和只追加事件数据库操作层。""" + + @classmethod + async def get_inventory_for_update( + cls, + db: AsyncSession, + pallet_id: int, + ) -> BizPalletInventory | None: + result = await db.execute( + select(BizPalletInventory).where(BizPalletInventory.pallet_id == pallet_id).with_for_update() + ) + return result.scalar_one_or_none() + + @classmethod + async def get_inventories_for_update( + cls, + db: AsyncSession, + pallet_ids: list[int], + ) -> list[BizPalletInventory]: + ordered_ids = sorted(set(pallet_ids)) + if not ordered_ids: + return [] + + result = await db.execute( + select(BizPalletInventory) + .where(BizPalletInventory.pallet_id.in_(ordered_ids)) + .order_by(BizPalletInventory.pallet_id) + .with_for_update() + ) + return sorted(result.scalars().all(), key=lambda inventory: inventory.pallet_id) + + @classmethod + async def get_pallet_by_qr_token( + cls, + db: AsyncSession, + qr_token: str, + ) -> BizPallet | None: + result = await db.execute(select(BizPallet).where(BizPallet.qr_token == qr_token)) + return result.scalar_one_or_none() + + @classmethod + async def get_pallets_by_ids( + cls, + db: AsyncSession, + pallet_ids: list[int], + ) -> list[BizPallet]: + """在调用方已锁库存后，按托盘 ID 升序锁定资产行。""" + ordered_ids = sorted(set(pallet_ids)) + if not ordered_ids: + return [] + + result = await db.execute( + select(BizPallet) + .where(BizPallet.pallet_id.in_(ordered_ids)) + .order_by(BizPallet.pallet_id) + .with_for_update() + ) + return sorted(result.scalars().all(), key=lambda pallet: pallet.pallet_id) + + @classmethod + async def get_event_by_idempotency_key( + cls, + db: AsyncSession, + pallet_id: int, + idempotency_key: str, + ) -> BizPalletEvent | None: + result = await db.execute( + select(BizPalletEvent).where( + BizPalletEvent.pallet_id == pallet_id, + BizPalletEvent.idempotency_key == idempotency_key, + ) + ) + return result.scalar_one_or_none() + + @classmethod + async def add_event( + cls, + db: AsyncSession, + inventory: BizPalletInventory, + event: BizPalletEvent, + ) -> BizPalletEvent: + """在调用方持有库存行锁时，为托盘追加下一序号事件。""" + sequence_result = await db.execute( + select(func.max(BizPalletEvent.event_seq)).where(BizPalletEvent.pallet_id == inventory.pallet_id) + ) + last_sequence = sequence_result.scalar_one() + event.pallet_id = inventory.pallet_id + event.event_seq = (last_sequence or 0) + 1 + db.add(event) + await db.flush() + inventory.last_event_id = event.event_id + await db.flush() + return event diff --git a/shierfeng-fastapi-backend/module_business/dao/operation_dao.py b/shierfeng-fastapi-backend/module_business/dao/operation_dao.py new file mode 100644 index 0000000..048874e --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/dao/operation_dao.py @@ -0,0 +1,667 @@ +from datetime import datetime, time +from typing import Any + +from sqlalchemy import ColumnElement, Select, delete, select, update +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm.attributes import InstrumentedAttribute + +from common.vo import PageModel +from module_business.entity.do.master_do import BizPallet +from module_business.entity.do.operation_do import ( + BizFillingRecord, + BizInboundItem, + BizInboundOrder, + BizReturnItem, + BizReturnOrder, + BizShipmentItem, + BizShipmentOrder, + BizUsageRecord, +) +from module_business.entity.vo.operation_vo import ( + FillingPageQueryModel, + InboundPageQueryModel, + ReturnPageQueryModel, + ShipmentPageQueryModel, + UsagePageQueryModel, +) +from utils.page_util import PageUtil + + +class OperationDao: + """托盘入库、装填、发货、使用和退回单据数据库操作层。""" + + @staticmethod + def _time_conditions( + column: InstrumentedAttribute, + begin_time: str | None, + end_time: str | None, + ) -> list[ColumnElement[bool]]: + conditions = [] + if begin_time: + begin_at = datetime.combine( + datetime.strptime(begin_time, '%Y-%m-%d').date(), + time.min, + ) + conditions.append(column >= begin_at) + if end_time: + end_at = datetime.combine( + datetime.strptime(end_time, '%Y-%m-%d').date(), + time(23, 59, 59), + ) + conditions.append(column <= end_at) + return conditions + + @staticmethod + async def _paginate( + db: AsyncSession, + statement: Select, + query_object: Any, + is_page: bool, + ) -> PageModel | list[dict[str, Any]]: + return await PageUtil.paginate( + db, + statement, + query_object.page_num, + query_object.page_size, + is_page, + ) + + @staticmethod + async def _get_for_update( + db: AsyncSession, + model: type[Any], + id_column: InstrumentedAttribute, + record_id: int, + *extra_conditions: ColumnElement[bool], + ) -> Any | None: + result = await db.execute(select(model).where(id_column == record_id, *extra_conditions).with_for_update()) + return result.scalar_one_or_none() + + @staticmethod + async def _get_by_key( + db: AsyncSession, + model: type[Any], + key_column: InstrumentedAttribute, + idempotency_key: str, + ) -> Any | None: + result = await db.execute(select(model).where(key_column == idempotency_key)) + return result.scalar_one_or_none() + + @staticmethod + async def _get_by_id( + db: AsyncSession, + model: type[Any], + id_column: InstrumentedAttribute, + record_id: int, + ) -> Any | None: + result = await db.execute(select(model).where(id_column == record_id)) + return result.scalar_one_or_none() + + @classmethod + async def _get_order_detail( + cls, + db: AsyncSession, + order_model: type[Any], + order_id_column: InstrumentedAttribute, + order_id: int, + item_model: type[Any], + item_order_id_column: InstrumentedAttribute, + item_id_column: InstrumentedAttribute, + ) -> Any | None: + order = await cls._get_by_id(db, order_model, order_id_column, order_id) + if order is None: + return None + + item_result = await db.execute( + select(item_model).where(item_order_id_column == order_id).order_by(item_id_column) + ) + order.items = list(item_result.scalars().all()) + return order + + @staticmethod + async def _add(db: AsyncSession, record: Any) -> Any: + db.add(record) + await db.flush() + return record + + @staticmethod + async def _update( + db: AsyncSession, + model: type[Any], + id_column: InstrumentedAttribute, + record: Any, + ) -> Any: + primary_key = id_column.key + payload = { + column.key: vars(record)[column.key] + for column in model.__table__.columns + if column.key != primary_key and column.key in vars(record) + } + if payload: + await db.execute(update(model).where(id_column == getattr(record, primary_key)).values(**payload)) + await db.flush() + return record + + @staticmethod + async def _delete( + db: AsyncSession, + model: type[Any], + id_column: InstrumentedAttribute, + record_id: int, + ) -> None: + await db.execute(delete(model).where(id_column == record_id)) + await db.flush() + + @staticmethod + async def _replace_items( + db: AsyncSession, + item_model: type[Any], + order_id_column: InstrumentedAttribute, + order_id: int, + items: list[Any], + ) -> list[Any]: + await db.execute(delete(item_model).where(order_id_column == order_id)) + for item in items: + setattr(item, order_id_column.key, order_id) + if items: + db.add_all(items) + await db.flush() + return items + + @staticmethod + async def _add_items(db: AsyncSession, items: list[Any]) -> list[Any]: + if items: + db.add_all(items) + await db.flush() + return items + + @classmethod + async def get_inbound_for_update( + cls, + db: AsyncSession, + inbound_id: int, + ) -> BizInboundOrder | None: + return await cls._get_for_update( + db, + BizInboundOrder, + BizInboundOrder.inbound_id, + inbound_id, + ) + + @classmethod + async def get_shipment_for_update( + cls, + db: AsyncSession, + shipment_id: int, + ) -> BizShipmentOrder | None: + return await cls._get_for_update( + db, + BizShipmentOrder, + BizShipmentOrder.shipment_id, + shipment_id, + ) + + @classmethod + async def get_return_for_update( + cls, + db: AsyncSession, + return_id: int, + ) -> BizReturnOrder | None: + return await cls._get_for_update( + db, + BizReturnOrder, + BizReturnOrder.return_id, + return_id, + ) + + @classmethod + async def get_active_filling_for_update( + cls, + db: AsyncSession, + filling_id: int, + ) -> BizFillingRecord | None: + """按已锁库存的 active_filling_id 锁定仍处于活动状态的装填周期。""" + return await cls._get_for_update( + db, + BizFillingRecord, + BizFillingRecord.filling_id, + filling_id, + BizFillingRecord.active_slot == 'ACTIVE', + ) + + @classmethod + async def get_inbound_by_confirm_idempotency_key( + cls, + db: AsyncSession, + idempotency_key: str, + ) -> BizInboundOrder | None: + return await cls._get_by_key( + db, + BizInboundOrder, + BizInboundOrder.confirm_idempotency_key, + idempotency_key, + ) + + @classmethod + async def get_shipment_by_confirm_idempotency_key( + cls, + db: AsyncSession, + idempotency_key: str, + ) -> BizShipmentOrder | None: + return await cls._get_by_key( + db, + BizShipmentOrder, + BizShipmentOrder.confirm_idempotency_key, + idempotency_key, + ) + + @classmethod + async def get_filling_by_idempotency_key( + cls, + db: AsyncSession, + idempotency_key: str, + ) -> BizFillingRecord | None: + return await cls._get_by_key( + db, + BizFillingRecord, + BizFillingRecord.idempotency_key, + idempotency_key, + ) + + @classmethod + async def get_usage_by_idempotency_key( + cls, + db: AsyncSession, + idempotency_key: str, + ) -> BizUsageRecord | None: + return await cls._get_by_key( + db, + BizUsageRecord, + BizUsageRecord.idempotency_key, + idempotency_key, + ) + + @classmethod + async def get_return_by_request_idempotency_key( + cls, + db: AsyncSession, + idempotency_key: str, + ) -> BizReturnOrder | None: + return await cls._get_by_key( + db, + BizReturnOrder, + BizReturnOrder.request_idempotency_key, + idempotency_key, + ) + + @classmethod + async def get_return_by_confirm_idempotency_key( + cls, + db: AsyncSession, + idempotency_key: str, + ) -> BizReturnOrder | None: + return await cls._get_by_key( + db, + BizReturnOrder, + BizReturnOrder.confirm_idempotency_key, + idempotency_key, + ) + + @classmethod + async def get_inbound_list( + cls, + db: AsyncSession, + query_object: InboundPageQueryModel, + is_page: bool = True, + ) -> PageModel | list[dict[str, Any]]: + conditions = [] + if query_object.inbound_no: + conditions.append(BizInboundOrder.inbound_no.like(f'%{query_object.inbound_no}%')) + if query_object.warehouse_id: + conditions.append(BizInboundOrder.warehouse_id == query_object.warehouse_id) + if query_object.status: + conditions.append(BizInboundOrder.status == query_object.status.value) + conditions.extend( + cls._time_conditions( + BizInboundOrder.create_time, + query_object.begin_time, + query_object.end_time, + ) + ) + statement = select(BizInboundOrder).where(*conditions).order_by(BizInboundOrder.inbound_id) + return await cls._paginate(db, statement, query_object, is_page) + + @classmethod + async def get_inbound_by_id( + cls, + db: AsyncSession, + inbound_id: int, + ) -> BizInboundOrder | None: + return await cls._get_order_detail( + db, + BizInboundOrder, + BizInboundOrder.inbound_id, + inbound_id, + BizInboundItem, + BizInboundItem.inbound_id, + BizInboundItem.inbound_item_id, + ) + + @classmethod + async def add_inbound( + cls, + db: AsyncSession, + inbound: BizInboundOrder, + ) -> BizInboundOrder: + return await cls._add(db, inbound) + + @classmethod + async def update_inbound( + cls, + db: AsyncSession, + inbound: BizInboundOrder, + ) -> BizInboundOrder: + return await cls._update( + db, + BizInboundOrder, + BizInboundOrder.inbound_id, + inbound, + ) + + @classmethod + async def delete_inbound(cls, db: AsyncSession, inbound_id: int) -> None: + await cls._delete(db, BizInboundOrder, BizInboundOrder.inbound_id, inbound_id) + + @classmethod + async def replace_inbound_items( + cls, + db: AsyncSession, + inbound_id: int, + items: list[BizInboundItem], + ) -> list[BizInboundItem]: + return await cls._replace_items( + db, + BizInboundItem, + BizInboundItem.inbound_id, + inbound_id, + items, + ) + + @classmethod + async def get_filling_list( + cls, + db: AsyncSession, + query_object: FillingPageQueryModel, + is_page: bool = True, + ) -> PageModel | list[dict[str, Any]]: + conditions = [] + if query_object.filling_no: + conditions.append(BizFillingRecord.filling_no.like(f'%{query_object.filling_no}%')) + if query_object.warehouse_id: + conditions.append(BizFillingRecord.warehouse_id == query_object.warehouse_id) + if query_object.strain_id: + conditions.append(BizFillingRecord.strain_id == query_object.strain_id) + if query_object.batch_no: + conditions.append(BizFillingRecord.batch_no.like(f'%{query_object.batch_no}%')) + if query_object.pallet_code: + conditions.append(BizPallet.pallet_code.like(f'%{query_object.pallet_code}%')) + conditions.extend( + cls._time_conditions( + BizFillingRecord.operated_at, + query_object.begin_time, + query_object.end_time, + ) + ) + statement = select(BizFillingRecord) + if query_object.pallet_code: + statement = statement.join( + BizPallet, + BizPallet.pallet_id == BizFillingRecord.pallet_id, + ) + statement = statement.where(*conditions).order_by(BizFillingRecord.filling_id) + return await cls._paginate(db, statement, query_object, is_page) + + @classmethod + async def get_filling_by_id( + cls, + db: AsyncSession, + filling_id: int, + ) -> BizFillingRecord | None: + return await cls._get_by_id( + db, + BizFillingRecord, + BizFillingRecord.filling_id, + filling_id, + ) + + @classmethod + async def add_filling_record( + cls, + db: AsyncSession, + filling: BizFillingRecord, + ) -> BizFillingRecord: + return await cls._add(db, filling) + + @classmethod + async def update_filling_record( + cls, + db: AsyncSession, + filling: BizFillingRecord, + ) -> BizFillingRecord: + return await cls._update( + db, + BizFillingRecord, + BizFillingRecord.filling_id, + filling, + ) + + @classmethod + async def get_shipment_list( + cls, + db: AsyncSession, + query_object: ShipmentPageQueryModel, + is_page: bool = True, + ) -> PageModel | list[dict[str, Any]]: + conditions = [] + if query_object.shipment_no: + conditions.append(BizShipmentOrder.shipment_no.like(f'%{query_object.shipment_no}%')) + if query_object.source_warehouse_id: + conditions.append(BizShipmentOrder.source_warehouse_id == query_object.source_warehouse_id) + if query_object.target_site_id: + conditions.append(BizShipmentOrder.target_site_id == query_object.target_site_id) + if query_object.status: + conditions.append(BizShipmentOrder.status == query_object.status.value) + conditions.extend( + cls._time_conditions( + BizShipmentOrder.create_time, + query_object.begin_time, + query_object.end_time, + ) + ) + statement = select(BizShipmentOrder).where(*conditions).order_by(BizShipmentOrder.shipment_id) + return await cls._paginate(db, statement, query_object, is_page) + + @classmethod + async def get_shipment_by_id( + cls, + db: AsyncSession, + shipment_id: int, + ) -> BizShipmentOrder | None: + return await cls._get_order_detail( + db, + BizShipmentOrder, + BizShipmentOrder.shipment_id, + shipment_id, + BizShipmentItem, + BizShipmentItem.shipment_id, + BizShipmentItem.shipment_item_id, + ) + + @classmethod + async def add_shipment( + cls, + db: AsyncSession, + shipment: BizShipmentOrder, + ) -> BizShipmentOrder: + return await cls._add(db, shipment) + + @classmethod + async def update_shipment( + cls, + db: AsyncSession, + shipment: BizShipmentOrder, + ) -> BizShipmentOrder: + return await cls._update( + db, + BizShipmentOrder, + BizShipmentOrder.shipment_id, + shipment, + ) + + @classmethod + async def delete_shipment(cls, db: AsyncSession, shipment_id: int) -> None: + await cls._delete( + db, + BizShipmentOrder, + BizShipmentOrder.shipment_id, + shipment_id, + ) + + @classmethod + async def replace_shipment_items( + cls, + db: AsyncSession, + shipment_id: int, + items: list[BizShipmentItem], + ) -> list[BizShipmentItem]: + return await cls._replace_items( + db, + BizShipmentItem, + BizShipmentItem.shipment_id, + shipment_id, + items, + ) + + @classmethod + async def get_usage_list( + cls, + db: AsyncSession, + query_object: UsagePageQueryModel, + is_page: bool = True, + ) -> PageModel | list[dict[str, Any]]: + conditions = [] + if query_object.usage_no: + conditions.append(BizUsageRecord.usage_no.like(f'%{query_object.usage_no}%')) + if query_object.site_id: + conditions.append(BizUsageRecord.site_id == query_object.site_id) + if query_object.line_id: + conditions.append(BizUsageRecord.line_id == query_object.line_id) + if query_object.pallet_code: + conditions.append(BizPallet.pallet_code.like(f'%{query_object.pallet_code}%')) + conditions.extend( + cls._time_conditions( + BizUsageRecord.occurred_at, + query_object.begin_time, + query_object.end_time, + ) + ) + statement = select(BizUsageRecord) + if query_object.pallet_code: + statement = statement.join( + BizPallet, + BizPallet.pallet_id == BizUsageRecord.pallet_id, + ) + statement = statement.where(*conditions).order_by(BizUsageRecord.usage_id) + return await cls._paginate(db, statement, query_object, is_page) + + @classmethod + async def get_usage_by_id( + cls, + db: AsyncSession, + usage_id: int, + ) -> BizUsageRecord | None: + return await cls._get_by_id( + db, + BizUsageRecord, + BizUsageRecord.usage_id, + usage_id, + ) + + @classmethod + async def add_usage_record( + cls, + db: AsyncSession, + usage: BizUsageRecord, + ) -> BizUsageRecord: + return await cls._add(db, usage) + + @classmethod + async def get_return_list( + cls, + db: AsyncSession, + query_object: ReturnPageQueryModel, + is_page: bool = True, + ) -> PageModel | list[dict[str, Any]]: + conditions = [] + if query_object.return_no: + conditions.append(BizReturnOrder.return_no.like(f'%{query_object.return_no}%')) + if query_object.source_site_id: + conditions.append(BizReturnOrder.source_site_id == query_object.source_site_id) + if query_object.target_warehouse_id: + conditions.append(BizReturnOrder.target_warehouse_id == query_object.target_warehouse_id) + if query_object.status: + conditions.append(BizReturnOrder.status == query_object.status.value) + conditions.extend( + cls._time_conditions( + BizReturnOrder.requested_at, + query_object.begin_time, + query_object.end_time, + ) + ) + statement = select(BizReturnOrder).where(*conditions).order_by(BizReturnOrder.return_id) + return await cls._paginate(db, statement, query_object, is_page) + + @classmethod + async def get_return_by_id( + cls, + db: AsyncSession, + return_id: int, + ) -> BizReturnOrder | None: + return await cls._get_order_detail( + db, + BizReturnOrder, + BizReturnOrder.return_id, + return_id, + BizReturnItem, + BizReturnItem.return_id, + BizReturnItem.return_item_id, + ) + + @classmethod + async def add_return_order( + cls, + db: AsyncSession, + return_order: BizReturnOrder, + ) -> BizReturnOrder: + return await cls._add(db, return_order) + + @classmethod + async def update_return_order( + cls, + db: AsyncSession, + return_order: BizReturnOrder, + ) -> BizReturnOrder: + return await cls._update( + db, + BizReturnOrder, + BizReturnOrder.return_id, + return_order, + ) + + @classmethod + async def add_return_items( + cls, + db: AsyncSession, + items: list[BizReturnItem], + ) -> list[BizReturnItem]: + return await cls._add_items(db, items) diff --git a/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py b/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py new file mode 100644 index 0000000..d51cc62 --- /dev/null +++ b/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py @@ -0,0 +1,823 @@ +import inspect +import unittest +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +from sqlalchemy.dialects import mysql, postgresql +from sqlalchemy.ext.asyncio import AsyncSession + +from module_business.dao import inventory_dao as inventory_dao_module +from module_business.dao import operation_dao as operation_dao_module +from module_business.dao.inventory_dao import InventoryDao +from module_business.dao.operation_dao import OperationDao +from module_business.entity.do.master_do import BizPallet +from module_business.entity.do.operation_do import ( + BizFillingRecord, + BizInboundItem, + BizInboundOrder, + BizPalletEvent, + BizPalletInventory, + BizReturnItem, + BizReturnOrder, + BizShipmentItem, + BizShipmentOrder, + BizUsageRecord, +) +from module_business.entity.vo.operation_vo import ( + FillingPageQueryModel, + InboundPageQueryModel, + ReturnPageQueryModel, + ShipmentPageQueryModel, + UsagePageQueryModel, +) +from utils.page_util import PageUtil + + +def compile_sql(statement: Any, dialect: Any | None = None) -> str: + compiled = statement.compile( + dialect=dialect or postgresql.dialect(), + compile_kwargs={'literal_binds': True}, + ) + return ' '.join(str(compiled).split()).replace('%%', '%') + + +def public_methods(dao: type[Any]) -> set[str]: + return {name for name in vars(dao) if not name.startswith('_') and callable(getattr(dao, name))} + + +class TestDaoPublicContract(unittest.TestCase): + def test_inventory_dao_exposes_only_fixed_append_only_api(self) -> None: + self.assertEqual( + { + 'get_inventory_for_update', + 'get_inventories_for_update', + 'get_pallet_by_qr_token', + 'get_pallets_by_ids', + 'get_event_by_idempotency_key', + 'add_event', + }, + public_methods(InventoryDao), + ) + self.assertFalse(hasattr(InventoryDao, 'update_event')) + self.assertFalse(hasattr(InventoryDao, 'delete_event')) + + def test_inventory_dao_critical_parameter_names_are_stable(self) -> None: + expected = { + 'get_inventory_for_update': ('db', 'pallet_id'), + 'get_inventories_for_update': ('db', 'pallet_ids'), + 'get_pallet_by_qr_token': ('db', 'qr_token'), + 'get_pallets_by_ids': ('db', 'pallet_ids'), + 'get_event_by_idempotency_key': ('db', 'pallet_id', 'idempotency_key'), + 'add_event': ('db', 'inventory', 'event'), + } + for method_name, parameter_names in expected.items(): + with self.subTest(method=method_name): + signature = inspect.signature(getattr(InventoryDao, method_name)) + self.assertEqual(parameter_names, tuple(signature.parameters)) + + def test_operation_dao_exposes_fixed_api(self) -> None: + self.assertEqual( + { + 'get_inbound_for_update', + 'get_shipment_for_update', + 'get_return_for_update', + 'get_active_filling_for_update', + 'get_inbound_by_confirm_idempotency_key', + 'get_shipment_by_confirm_idempotency_key', + 'get_filling_by_idempotency_key', + 'get_usage_by_idempotency_key', + 'get_return_by_request_idempotency_key', + 'get_return_by_confirm_idempotency_key', + 'get_inbound_list', + 'get_inbound_by_id', + 'add_inbound', + 'update_inbound', + 'delete_inbound', + 'replace_inbound_items', + 'get_filling_list', + 'get_filling_by_id', + 'add_filling_record', + 'update_filling_record', + 'get_shipment_list', + 'get_shipment_by_id', + 'add_shipment', + 'update_shipment', + 'delete_shipment', + 'replace_shipment_items', + 'get_usage_list', + 'get_usage_by_id', + 'add_usage_record', + 'get_return_list', + 'get_return_by_id', + 'add_return_order', + 'update_return_order', + 'add_return_items', + }, + public_methods(OperationDao), + ) + + def test_operation_dao_critical_parameter_names_and_defaults_are_stable(self) -> None: + expected = { + 'get_inbound_for_update': ('db', 'inbound_id'), + 'get_shipment_for_update': ('db', 'shipment_id'), + 'get_return_for_update': ('db', 'return_id'), + 'get_active_filling_for_update': ('db', 'filling_id'), + 'get_inbound_by_confirm_idempotency_key': ('db', 'idempotency_key'), + 'get_shipment_by_confirm_idempotency_key': ('db', 'idempotency_key'), + 'get_filling_by_idempotency_key': ('db', 'idempotency_key'), + 'get_usage_by_idempotency_key': ('db', 'idempotency_key'), + 'get_return_by_request_idempotency_key': ('db', 'idempotency_key'), + 'get_return_by_confirm_idempotency_key': ('db', 'idempotency_key'), + 'get_inbound_by_id': ('db', 'inbound_id'), + 'add_inbound': ('db', 'inbound'), + 'update_inbound': ('db', 'inbound'), + 'delete_inbound': ('db', 'inbound_id'), + 'replace_inbound_items': ('db', 'inbound_id', 'items'), + 'get_filling_by_id': ('db', 'filling_id'), + 'add_filling_record': ('db', 'filling'), + 'update_filling_record': ('db', 'filling'), + 'get_shipment_by_id': ('db', 'shipment_id'), + 'add_shipment': ('db', 'shipment'), + 'update_shipment': ('db', 'shipment'), + 'delete_shipment': ('db', 'shipment_id'), + 'replace_shipment_items': ('db', 'shipment_id', 'items'), + 'get_usage_by_id': ('db', 'usage_id'), + 'add_usage_record': ('db', 'usage'), + 'get_return_by_id': ('db', 'return_id'), + 'add_return_order': ('db', 'return_order'), + 'update_return_order': ('db', 'return_order'), + 'add_return_items': ('db', 'items'), + } + for method_name, parameter_names in expected.items(): + with self.subTest(method=method_name): + signature = inspect.signature(getattr(OperationDao, method_name)) + self.assertEqual(parameter_names, tuple(signature.parameters)) + + for method_name in ( + 'get_inbound_list', + 'get_filling_list', + 'get_shipment_list', + 'get_usage_list', + 'get_return_list', + ): + with self.subTest(method=method_name): + signature = inspect.signature(getattr(OperationDao, method_name)) + self.assertEqual(('db', 'query_object', 'is_page'), tuple(signature.parameters)) + self.assertIs(True, signature.parameters['is_page'].default) + + def test_dao_source_never_controls_transaction_boundary(self) -> None: + for module in (inventory_dao_module, operation_dao_module): + with self.subTest(module=module.__name__): + source = inspect.getsource(module) + self.assertNotIn('.commit(', source) + self.assertNotIn('.rollback(', source) + + +class TestInventoryDao(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + + async def test_get_inventory_for_update_targets_primary_key_with_lock(self) -> None: + inventory = SimpleNamespace(pallet_id=7) + result = MagicMock() + result.scalar_one_or_none.return_value = inventory + self.db.execute.return_value = result + + actual = await InventoryDao.get_inventory_for_update(self.db, pallet_id=7) + + self.assertIs(inventory, actual) + statement = self.db.execute.await_args.args[0] + self.assertTrue(statement.get_final_froms()[0].compare(BizPalletInventory.__table__)) + for dialect in (mysql.dialect(), postgresql.dialect()): + sql = compile_sql(statement, dialect) + self.assertIn('biz_pallet_inventory.pallet_id = 7', sql) + self.assertIn('FOR UPDATE', sql) + + async def test_get_inventories_for_update_deduplicates_sorts_and_locks(self) -> None: + inventories = [ + SimpleNamespace(pallet_id=9), + SimpleNamespace(pallet_id=2), + SimpleNamespace(pallet_id=5), + ] + result = MagicMock() + result.scalars.return_value.all.return_value = inventories + self.db.execute.return_value = result + + actual = await InventoryDao.get_inventories_for_update(self.db, [9, 2, 5, 2]) + + self.assertEqual([2, 5, 9], [inventory.pallet_id for inventory in actual]) + statement = self.db.execute.await_args.args[0] + self.assertTrue(statement.get_final_froms()[0].compare(BizPalletInventory.__table__)) + for dialect in (mysql.dialect(), postgresql.dialect()): + sql = compile_sql(statement, dialect) + self.assertIn('biz_pallet_inventory.pallet_id IN (2, 5, 9)', sql) + self.assertIn('ORDER BY biz_pallet_inventory.pallet_id', sql) + self.assertIn('FOR UPDATE', sql) + + async def test_get_inventories_for_update_empty_input_skips_sql(self) -> None: + self.assertEqual([], await InventoryDao.get_inventories_for_update(self.db, [])) + self.db.execute.assert_not_awaited() + + async def test_get_pallet_by_qr_token_uses_exact_token_without_lock(self) -> None: + pallet = SimpleNamespace(pallet_id=4) + result = MagicMock() + result.scalar_one_or_none.return_value = pallet + self.db.execute.return_value = result + + actual = await InventoryDao.get_pallet_by_qr_token(self.db, qr_token='fixed-token') + + self.assertIs(pallet, actual) + statement = self.db.execute.await_args.args[0] + sql = compile_sql(statement) + self.assertTrue(statement.get_final_froms()[0].compare(BizPallet.__table__)) + self.assertIn("biz_pallet.qr_token = 'fixed-token'", sql) + self.assertNotIn('FOR UPDATE', sql) + + async def test_get_pallets_by_ids_deduplicates_sorts_and_locks_after_inventory(self) -> None: + pallets = [ + SimpleNamespace(pallet_id=9), + SimpleNamespace(pallet_id=2), + SimpleNamespace(pallet_id=5), + ] + result = MagicMock() + result.scalars.return_value.all.return_value = pallets + self.db.execute.return_value = result + + actual = await InventoryDao.get_pallets_by_ids(self.db, [9, 2, 5, 2]) + + self.assertEqual([2, 5, 9], [pallet.pallet_id for pallet in actual]) + statement = self.db.execute.await_args.args[0] + self.assertTrue(statement.get_final_froms()[0].compare(BizPallet.__table__)) + for dialect in (mysql.dialect(), postgresql.dialect()): + sql = compile_sql(statement, dialect) + self.assertIn('biz_pallet.pallet_id IN (2, 5, 9)', sql) + self.assertIn('ORDER BY biz_pallet.pallet_id', sql) + self.assertIn('FOR UPDATE', sql) + + async def test_get_pallets_by_ids_empty_input_skips_sql(self) -> None: + self.assertEqual([], await InventoryDao.get_pallets_by_ids(self.db, [])) + self.db.execute.assert_not_awaited() + + async def test_get_event_by_idempotency_key_uses_pallet_scoped_key(self) -> None: + event = SimpleNamespace(event_id=19) + result = MagicMock() + result.scalar_one_or_none.return_value = event + self.db.execute.return_value = result + + actual = await InventoryDao.get_event_by_idempotency_key( + self.db, + pallet_id=7, + idempotency_key='event-key', + ) + + self.assertIs(event, actual) + statement = self.db.execute.await_args.args[0] + sql = compile_sql(statement) + self.assertTrue(statement.get_final_froms()[0].compare(BizPalletEvent.__table__)) + self.assertIn('biz_pallet_event.pallet_id = 7', sql) + self.assertIn("biz_pallet_event.idempotency_key = 'event-key'", sql) + self.assertNotIn('FOR UPDATE', sql) + + async def test_add_event_allocates_next_sequence_and_updates_inventory_pointer(self) -> None: + sequence_result = MagicMock() + sequence_result.scalar_one.return_value = 4 + sequence_result.scalar_one_or_none.return_value = 4 + sequence_result.scalar.return_value = 4 + self.db.execute.return_value = sequence_result + inventory = BizPalletInventory(pallet_id=7, last_event_id=None) + event = BizPalletEvent(pallet_id=99, event_seq=99, idempotency_key='event-key') + flush_count = 0 + + async def assign_event_id() -> None: + nonlocal flush_count + flush_count += 1 + if flush_count == 1: + event.event_id = 41 + + self.db.flush.side_effect = assign_event_id + + actual = await InventoryDao.add_event(self.db, inventory, event) + + self.assertIs(event, actual) + self.assertEqual(7, event.pallet_id) + self.assertEqual(5, event.event_seq) + self.assertEqual(41, inventory.last_event_id) + self.db.add.assert_called_once_with(event) + self.assertEqual(2, self.db.flush.await_count) + statement = self.db.execute.await_args.args[0] + sql = compile_sql(statement) + self.assertIn('max(biz_pallet_event.event_seq)', sql.lower()) + self.assertIn('biz_pallet_event.pallet_id = 7', sql) + + async def test_add_first_event_starts_sequence_at_one(self) -> None: + sequence_result = MagicMock() + sequence_result.scalar_one.return_value = None + sequence_result.scalar_one_or_none.return_value = None + sequence_result.scalar.return_value = None + self.db.execute.return_value = sequence_result + inventory = BizPalletInventory(pallet_id=8, last_event_id=None) + event = BizPalletEvent(pallet_id=8, idempotency_key='first-key', event_id=51) + + await InventoryDao.add_event(self.db, inventory, event) + + self.assertEqual(1, event.event_seq) + self.assertEqual(51, inventory.last_event_id) + + +class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + + async def test_for_update_queries_target_primary_keys_and_lock_rows(self) -> None: + cases = ( + ( + OperationDao.get_inbound_for_update, + BizInboundOrder, + 'inbound_id', + 11, + (), + ), + ( + OperationDao.get_shipment_for_update, + BizShipmentOrder, + 'shipment_id', + 12, + (), + ), + ( + OperationDao.get_return_for_update, + BizReturnOrder, + 'return_id', + 13, + (), + ), + ( + OperationDao.get_active_filling_for_update, + BizFillingRecord, + 'filling_id', + 14, + ("biz_filling_record.active_slot = 'ACTIVE'",), + ), + ) + for method, model, id_field, record_id, extra_fragments in cases: + with self.subTest(table=model.__tablename__): + self.db.reset_mock() + record = SimpleNamespace() + result = MagicMock() + result.scalar_one_or_none.return_value = record + self.db.execute.return_value = result + + actual = await method(self.db, record_id) + + self.assertIs(record, actual) + statement = self.db.execute.await_args.args[0] + self.assertTrue(statement.get_final_froms()[0].compare(model.__table__)) + for dialect in (mysql.dialect(), postgresql.dialect()): + sql = compile_sql(statement, dialect) + self.assertIn(f'{model.__tablename__}.{id_field} = {record_id}', sql) + self.assertIn('FOR UPDATE', sql) + for fragment in extra_fragments: + self.assertIn(fragment, sql) + + async def test_idempotency_queries_use_the_exact_declared_column(self) -> None: + cases = ( + ( + OperationDao.get_inbound_by_confirm_idempotency_key, + BizInboundOrder, + 'confirm_idempotency_key', + ), + ( + OperationDao.get_shipment_by_confirm_idempotency_key, + BizShipmentOrder, + 'confirm_idempotency_key', + ), + (OperationDao.get_filling_by_idempotency_key, BizFillingRecord, 'idempotency_key'), + (OperationDao.get_usage_by_idempotency_key, BizUsageRecord, 'idempotency_key'), + ( + OperationDao.get_return_by_request_idempotency_key, + BizReturnOrder, + 'request_idempotency_key', + ), + ( + OperationDao.get_return_by_confirm_idempotency_key, + BizReturnOrder, + 'confirm_idempotency_key', + ), + ) + for method, model, key_column in cases: + with self.subTest(method=method.__name__): + self.db.reset_mock() + record = SimpleNamespace() + result = MagicMock() + result.scalar_one_or_none.return_value = record + self.db.execute.return_value = result + + actual = await method(self.db, 'same-key') + + 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__}.{key_column} = 'same-key'", sql) + self.assertNotIn('FOR UPDATE', sql) + + async def test_list_queries_apply_every_declared_filter_sort_and_pagination(self) -> None: + cases = ( + ( + OperationDao.get_inbound_list, + InboundPageQueryModel( + page_num=2, + page_size=7, + inbound_no='IN-26', + warehouse_id=3, + status='DRAFT', + begin_time='2026-07-01', + end_time='2026-07-02', + ), + BizInboundOrder, + 'inbound_id', + ( + "biz_inbound_order.inbound_no LIKE '%IN-26%'", + 'biz_inbound_order.warehouse_id = 3', + "biz_inbound_order.status = 'DRAFT'", + "biz_inbound_order.create_time >= '2026-07-01 00:00:00'", + "biz_inbound_order.create_time <= '2026-07-02 23:59:59'", + ), + ), + ( + OperationDao.get_filling_list, + FillingPageQueryModel( + page_num=2, + page_size=7, + filling_no='FI-26', + warehouse_id=3, + strain_id=4, + batch_no='BATCH-A', + pallet_code='P-00', + begin_time='2026-07-01', + end_time='2026-07-02', + ), + BizFillingRecord, + 'filling_id', + ( + "biz_filling_record.filling_no LIKE '%FI-26%'", + 'biz_filling_record.warehouse_id = 3', + 'biz_filling_record.strain_id = 4', + "biz_filling_record.batch_no LIKE '%BATCH-A%'", + "biz_pallet.pallet_code LIKE '%P-00%'", + "biz_filling_record.operated_at >= '2026-07-01 00:00:00'", + "biz_filling_record.operated_at <= '2026-07-02 23:59:59'", + ), + ), + ( + OperationDao.get_shipment_list, + ShipmentPageQueryModel( + page_num=2, + page_size=7, + shipment_no='SH-26', + source_warehouse_id=3, + target_site_id=4, + status='CONFIRMED', + begin_time='2026-07-01', + end_time='2026-07-02', + ), + BizShipmentOrder, + 'shipment_id', + ( + "biz_shipment_order.shipment_no LIKE '%SH-26%'", + 'biz_shipment_order.source_warehouse_id = 3', + 'biz_shipment_order.target_site_id = 4', + "biz_shipment_order.status = 'CONFIRMED'", + "biz_shipment_order.create_time >= '2026-07-01 00:00:00'", + "biz_shipment_order.create_time <= '2026-07-02 23:59:59'", + ), + ), + ( + OperationDao.get_usage_list, + UsagePageQueryModel( + page_num=2, + page_size=7, + usage_no='US-26', + site_id=3, + line_id=4, + pallet_code='P-00', + begin_time='2026-07-01', + end_time='2026-07-02', + ), + BizUsageRecord, + 'usage_id', + ( + "biz_usage_record.usage_no LIKE '%US-26%'", + 'biz_usage_record.site_id = 3', + 'biz_usage_record.line_id = 4', + "biz_pallet.pallet_code LIKE '%P-00%'", + "biz_usage_record.occurred_at >= '2026-07-01 00:00:00'", + "biz_usage_record.occurred_at <= '2026-07-02 23:59:59'", + ), + ), + ( + OperationDao.get_return_list, + ReturnPageQueryModel( + page_num=2, + page_size=7, + return_no='RE-26', + source_site_id=3, + target_warehouse_id=4, + status='REQUESTED', + begin_time='2026-07-01', + end_time='2026-07-02', + ), + BizReturnOrder, + 'return_id', + ( + "biz_return_order.return_no LIKE '%RE-26%'", + 'biz_return_order.source_site_id = 3', + 'biz_return_order.target_warehouse_id = 4', + "biz_return_order.status = 'REQUESTED'", + "biz_return_order.requested_at >= '2026-07-01 00:00:00'", + "biz_return_order.requested_at <= '2026-07-02 23:59:59'", + ), + ), + ) + for index, (method, query, model, id_field, fragments) in enumerate(cases): + with self.subTest(method=method.__name__): + marker = object() + is_page = index % 2 == 0 + with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as paginate: + actual = await method(self.db, query, is_page=is_page) + + self.assertIs(marker, actual) + args = paginate.await_args.args + self.assertIs(self.db, args[0]) + self.assertEqual((query.page_num, query.page_size, is_page), args[2:]) + statement = args[1] + sql = compile_sql(statement) + self.assertIn(f'FROM {model.__tablename__}', sql) + self.assertIn(f'ORDER BY {model.__tablename__}.{id_field}', sql) + for fragment in fragments: + self.assertIn(fragment, sql) + + async def test_order_details_load_items_explicitly_in_stable_order(self) -> None: + cases = ( + ( + OperationDao.get_inbound_by_id, + BizInboundOrder, + 'inbound_id', + BizInboundItem, + 'inbound_item_id', + 11, + ), + ( + OperationDao.get_shipment_by_id, + BizShipmentOrder, + 'shipment_id', + BizShipmentItem, + 'shipment_item_id', + 12, + ), + ( + OperationDao.get_return_by_id, + BizReturnOrder, + 'return_id', + BizReturnItem, + 'return_item_id', + 13, + ), + ) + for method, order_model, order_id_field, item_model, item_id_field, order_id in cases: + with self.subTest(method=method.__name__): + self.db.reset_mock() + order = SimpleNamespace() + items = [SimpleNamespace(item_id=2), SimpleNamespace(item_id=5)] + order_result = MagicMock() + order_result.scalar_one_or_none.return_value = order + item_result = MagicMock() + item_result.scalars.return_value.all.return_value = items + self.db.execute.side_effect = [order_result, item_result] + + actual = await method(self.db, order_id) + + self.assertIs(order, actual) + self.assertEqual(items, order.items) + order_statement = self.db.execute.await_args_list[0].args[0] + item_statement = self.db.execute.await_args_list[1].args[0] + self.assertTrue(order_statement.get_final_froms()[0].compare(order_model.__table__)) + self.assertIn( + f'{order_model.__tablename__}.{order_id_field} = {order_id}', + compile_sql(order_statement), + ) + self.assertTrue(item_statement.get_final_froms()[0].compare(item_model.__table__)) + item_sql = compile_sql(item_statement) + self.assertIn(f'{item_model.__tablename__}.{order_id_field} = {order_id}', item_sql) + self.assertIn(f'ORDER BY {item_model.__tablename__}.{item_id_field}', item_sql) + + async def test_missing_order_detail_skips_item_query(self) -> None: + result = MagicMock() + result.scalar_one_or_none.return_value = None + self.db.execute.return_value = result + + self.assertIsNone(await OperationDao.get_inbound_by_id(self.db, inbound_id=404)) + + self.db.execute.assert_awaited_once() + + async def test_simple_details_target_primary_keys(self) -> None: + cases = ( + (OperationDao.get_filling_by_id, BizFillingRecord, 'filling_id', 14), + (OperationDao.get_usage_by_id, BizUsageRecord, 'usage_id', 15), + ) + for method, model, id_field, record_id in cases: + with self.subTest(method=method.__name__): + self.db.reset_mock() + record = SimpleNamespace() + result = MagicMock() + result.scalar_one_or_none.return_value = record + self.db.execute.return_value = result + + actual = await method(self.db, record_id) + + self.assertIs(record, actual) + statement = self.db.execute.await_args.args[0] + self.assertTrue(statement.get_final_froms()[0].compare(model.__table__)) + self.assertIn(f'{model.__tablename__}.{id_field} = {record_id}', compile_sql(statement)) + + +class TestOperationDaoWrites(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + + async def test_add_methods_add_exact_orm_and_flush_without_commit(self) -> None: + cases = ( + (OperationDao.add_inbound, BizInboundOrder(inbound_no='IN-1', warehouse_id=3)), + ( + OperationDao.add_filling_record, + BizFillingRecord(filling_no='FI-1', pallet_id=4, warehouse_id=3), + ), + ( + OperationDao.add_shipment, + BizShipmentOrder(shipment_no='SH-1', source_warehouse_id=3, target_site_id=4), + ), + (OperationDao.add_usage_record, BizUsageRecord(usage_no='US-1', pallet_id=4)), + ( + OperationDao.add_return_order, + BizReturnOrder(return_no='RE-1', source_site_id=4, target_warehouse_id=3), + ), + ) + for method, record in cases: + with self.subTest(method=method.__name__): + self.db.reset_mock() + + actual = await method(self.db, record) + + self.assertIs(record, actual) + self.db.add.assert_called_once_with(record) + self.db.flush.assert_awaited_once() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_not_awaited() + + async def test_update_methods_target_primary_key_and_flush_without_commit(self) -> None: + cases = ( + ( + OperationDao.update_inbound, + BizInboundOrder(inbound_id=11, warehouse_id=3, status='CONFIRMED'), + BizInboundOrder, + 'inbound_id', + 11, + 'warehouse_id=3', + ), + ( + OperationDao.update_filling_record, + BizFillingRecord(filling_id=12, active_slot=None), + BizFillingRecord, + 'filling_id', + 12, + 'active_slot=NULL', + ), + ( + OperationDao.update_shipment, + BizShipmentOrder(shipment_id=13, status='CONFIRMED'), + BizShipmentOrder, + 'shipment_id', + 13, + "status='CONFIRMED'", + ), + ( + OperationDao.update_return_order, + BizReturnOrder(return_id=14, status='CONFIRMED'), + BizReturnOrder, + 'return_id', + 14, + "status='CONFIRMED'", + ), + ) + for method, record, model, id_field, record_id, value_fragment in cases: + with self.subTest(method=method.__name__): + self.db.reset_mock() + + actual = await method(self.db, record) + + self.assertIs(record, actual) + statement = self.db.execute.await_args.args[0] + self.assertTrue(statement.table.compare(model.__table__)) + sql = compile_sql(statement) + self.assertIn(f'WHERE {model.__tablename__}.{id_field} = {record_id}', sql) + self.assertIn(value_fragment, sql) + self.db.flush.assert_awaited_once() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_not_awaited() + + async def test_delete_methods_target_one_order_without_commit(self) -> None: + cases = ( + (OperationDao.delete_inbound, BizInboundOrder, 'inbound_id', 11), + (OperationDao.delete_shipment, BizShipmentOrder, 'shipment_id', 12), + ) + for method, model, id_field, record_id in cases: + with self.subTest(method=method.__name__): + self.db.reset_mock() + + await method(self.db, record_id) + + statement = self.db.execute.await_args.args[0] + self.assertTrue(statement.table.compare(model.__table__)) + self.assertIn( + f'{model.__tablename__}.{id_field} = {record_id}', + compile_sql(statement), + ) + self.db.flush.assert_awaited_once() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_not_awaited() + + async def test_replace_items_deletes_whole_order_then_adds_normalized_items(self) -> None: + cases = ( + ( + OperationDao.replace_inbound_items, + BizInboundItem, + 'inbound_id', + 11, + [BizInboundItem(inbound_id=99, pallet_id=2), BizInboundItem(inbound_id=99, pallet_id=5)], + ), + ( + OperationDao.replace_shipment_items, + BizShipmentItem, + 'shipment_id', + 12, + [BizShipmentItem(shipment_id=99, pallet_id=2), BizShipmentItem(shipment_id=99, pallet_id=5)], + ), + ) + for method, item_model, order_id_field, order_id, items in cases: + with self.subTest(method=method.__name__): + self.db.reset_mock() + + actual = await method(self.db, order_id, items) + + self.assertEqual(items, actual) + statement = self.db.execute.await_args.args[0] + self.assertTrue(statement.table.compare(item_model.__table__)) + self.assertIn( + f'{item_model.__tablename__}.{order_id_field} = {order_id}', + compile_sql(statement), + ) + self.assertEqual([order_id, order_id], [getattr(item, order_id_field) for item in items]) + self.db.add_all.assert_called_once_with(items) + self.db.flush.assert_awaited_once() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_not_awaited() + + async def test_replace_items_empty_list_means_clear_all_existing_items(self) -> None: + cases = ( + (OperationDao.replace_inbound_items, BizInboundItem, 'inbound_id', 11), + (OperationDao.replace_shipment_items, BizShipmentItem, 'shipment_id', 12), + ) + for method, item_model, order_id_field, order_id in cases: + with self.subTest(method=method.__name__): + self.db.reset_mock() + + actual = await method(self.db, order_id, []) + + self.assertEqual([], actual) + statement = self.db.execute.await_args.args[0] + self.assertTrue(statement.table.compare(item_model.__table__)) + self.assertIn( + f'{item_model.__tablename__}.{order_id_field} = {order_id}', + compile_sql(statement), + ) + self.db.add_all.assert_not_called() + self.db.flush.assert_awaited_once() + + async def test_add_return_items_adds_all_and_flushes_without_commit(self) -> None: + items = [ + BizReturnItem(return_id=13, pallet_id=2, remaining_packages_snapshot=0), + BizReturnItem(return_id=13, pallet_id=5, remaining_packages_snapshot=4), + ] + + actual = await OperationDao.add_return_items(self.db, items) + + self.assertEqual(items, actual) + self.db.add_all.assert_called_once_with(items) + self.db.flush.assert_awaited_once() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_not_awaited() + + +if __name__ == '__main__': + unittest.main()