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