1398 lines
60 KiB
Python
1398 lines
60 KiB
Python
import inspect
|
|
import unittest
|
|
from datetime import datetime
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from sqlalchemy.dialects import mysql
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.sql import operators
|
|
|
|
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.domain.test_batch_scope import TestBatchScope
|
|
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 mysql.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', 'scope'),
|
|
'get_inventories_for_update': ('db', 'pallet_ids', 'scope'),
|
|
'get_pallet_by_qr_token': ('db', 'qr_token', 'scope'),
|
|
'get_pallets_by_ids': ('db', 'pallet_ids', 'scope'),
|
|
'get_event_by_idempotency_key': ('db', 'pallet_id', 'idempotency_key', 'scope'),
|
|
'add_event': ('db', 'inventory', 'event', 'scope'),
|
|
}
|
|
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', 'scope', 'warehouse_ids'),
|
|
'get_shipment_for_update': ('db', 'shipment_id', 'scope', 'warehouse_ids', 'site_ids'),
|
|
'get_return_for_update': ('db', 'return_id', 'scope', 'site_ids', 'warehouse_ids'),
|
|
'get_active_filling_for_update': ('db', 'filling_id', 'scope'),
|
|
'get_inbound_by_confirm_idempotency_key': (
|
|
'db',
|
|
'idempotency_key',
|
|
'scope',
|
|
'warehouse_ids',
|
|
),
|
|
'get_shipment_by_confirm_idempotency_key': (
|
|
'db',
|
|
'idempotency_key',
|
|
'scope',
|
|
'warehouse_ids',
|
|
'site_ids',
|
|
),
|
|
'get_filling_by_idempotency_key': ('db', 'idempotency_key', 'scope', 'warehouse_ids'),
|
|
'get_usage_by_idempotency_key': ('db', 'idempotency_key', 'scope', 'site_ids'),
|
|
'get_return_by_request_idempotency_key': (
|
|
'db',
|
|
'idempotency_key',
|
|
'source_site_id',
|
|
'target_warehouse_id',
|
|
'scope',
|
|
),
|
|
'get_return_by_confirm_idempotency_key': (
|
|
'db',
|
|
'idempotency_key',
|
|
'source_site_id',
|
|
'target_warehouse_id',
|
|
'scope',
|
|
),
|
|
'get_inbound_by_id': ('db', 'inbound_id', 'scope', 'warehouse_ids'),
|
|
'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', 'scope', 'warehouse_ids'),
|
|
'add_filling_record': ('db', 'filling'),
|
|
'update_filling_record': ('db', 'filling'),
|
|
'get_shipment_by_id': ('db', 'shipment_id', 'scope', 'warehouse_ids', 'site_ids'),
|
|
'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', 'scope', 'site_ids'),
|
|
'add_usage_record': ('db', 'usage'),
|
|
'get_return_by_id': ('db', 'return_id', 'scope', 'site_ids', 'warehouse_ids'),
|
|
'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))
|
|
|
|
inbound_list_signature = inspect.signature(OperationDao.get_inbound_list)
|
|
self.assertEqual(
|
|
('db', 'query_object', 'warehouse_ids', 'scope', 'is_page'),
|
|
tuple(inbound_list_signature.parameters),
|
|
)
|
|
self.assertIs(True, inbound_list_signature.parameters['is_page'].default)
|
|
|
|
for method_name in (
|
|
'get_inbound_for_update',
|
|
'get_inbound_by_id',
|
|
'get_filling_by_idempotency_key',
|
|
'get_filling_by_id',
|
|
):
|
|
with self.subTest(method=method_name):
|
|
signature = inspect.signature(getattr(OperationDao, method_name))
|
|
warehouse_ids = signature.parameters.get('warehouse_ids')
|
|
self.assertIsNotNone(warehouse_ids)
|
|
if warehouse_ids is not None:
|
|
self.assertIsNone(warehouse_ids.default)
|
|
|
|
for method_name in (
|
|
'get_shipment_for_update',
|
|
'get_shipment_by_confirm_idempotency_key',
|
|
'get_shipment_by_id',
|
|
):
|
|
with self.subTest(method=method_name):
|
|
signature = inspect.signature(getattr(OperationDao, method_name))
|
|
self.assertIsNone(signature.parameters['warehouse_ids'].default)
|
|
self.assertIsNone(signature.parameters['site_ids'].default)
|
|
|
|
filling_list_signature = inspect.signature(OperationDao.get_filling_list)
|
|
self.assertEqual(
|
|
('db', 'query_object', 'warehouse_ids', 'scope', 'is_page'),
|
|
tuple(filling_list_signature.parameters),
|
|
)
|
|
self.assertIs(True, filling_list_signature.parameters['is_page'].default)
|
|
|
|
shipment_list_signature = inspect.signature(OperationDao.get_shipment_list)
|
|
self.assertEqual(
|
|
('db', 'query_object', 'warehouse_ids', 'site_ids', 'scope', 'is_page'),
|
|
tuple(shipment_list_signature.parameters),
|
|
)
|
|
self.assertIs(True, shipment_list_signature.parameters['is_page'].default)
|
|
|
|
usage_list_signature = inspect.signature(OperationDao.get_usage_list)
|
|
self.assertEqual(
|
|
('db', 'query_object', 'site_ids', 'scope', 'is_page'),
|
|
tuple(usage_list_signature.parameters),
|
|
)
|
|
self.assertIs(True, usage_list_signature.parameters['is_page'].default)
|
|
|
|
for method_name in (
|
|
'get_usage_by_idempotency_key',
|
|
'get_usage_by_id',
|
|
):
|
|
with self.subTest(method=method_name):
|
|
signature = inspect.signature(getattr(OperationDao, method_name))
|
|
self.assertIsNone(signature.parameters['site_ids'].default)
|
|
|
|
self.assertNotIn('for_update', inspect.signature(OperationDao._get_by_key).parameters)
|
|
|
|
return_list_signature = inspect.signature(OperationDao.get_return_list)
|
|
self.assertEqual(
|
|
('db', 'query_object', 'site_ids', 'warehouse_ids', 'scope', 'is_page'),
|
|
tuple(return_list_signature.parameters),
|
|
)
|
|
self.assertIs(True, return_list_signature.parameters['is_page'].default)
|
|
for method_name in (
|
|
'get_return_for_update',
|
|
'get_return_by_id',
|
|
):
|
|
with self.subTest(method=method_name):
|
|
signature = inspect.signature(getattr(OperationDao, method_name))
|
|
self.assertIsNone(signature.parameters['site_ids'].default)
|
|
self.assertIsNone(signature.parameters['warehouse_ids'].default)
|
|
|
|
for method_name in (
|
|
'get_return_by_request_idempotency_key',
|
|
'get_return_by_confirm_idempotency_key',
|
|
):
|
|
with self.subTest(method=method_name):
|
|
signature = inspect.signature(getattr(OperationDao, method_name))
|
|
self.assertIs(inspect.Parameter.empty, signature.parameters['source_site_id'].default)
|
|
self.assertIs(inspect.Parameter.empty, signature.parameters['target_warehouse_id'].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, scope=TestBatchScope.formal())
|
|
|
|
self.assertIs(inventory, actual)
|
|
statement = self.db.execute.await_args.args[0]
|
|
self.assertTrue(statement.get_final_froms()[0].compare(BizPalletInventory.__table__))
|
|
self.assertIs(True, statement.get_execution_options().get('populate_existing'))
|
|
for dialect in (mysql.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], scope=TestBatchScope.formal())
|
|
|
|
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__))
|
|
self.assertIs(True, statement.get_execution_options().get('populate_existing'))
|
|
for dialect in (mysql.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, [], scope=TestBatchScope.formal()))
|
|
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', scope=TestBatchScope.formal()
|
|
)
|
|
|
|
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], scope=TestBatchScope.formal())
|
|
|
|
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__))
|
|
self.assertIs(True, statement.get_execution_options().get('populate_existing'))
|
|
for dialect in (mysql.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, [], scope=TestBatchScope.formal()))
|
|
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',
|
|
scope=TestBatchScope.formal(),
|
|
)
|
|
|
|
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_reads_exact_previous_pointer_and_updates_inventory(self) -> None:
|
|
sequence_result = MagicMock()
|
|
sequence_result.scalar_one.return_value = 4
|
|
sequence_result.scalar_one_or_none.return_value = 4
|
|
self.db.execute.return_value = sequence_result
|
|
inventory = BizPalletInventory(pallet_id=7, last_event_id=40)
|
|
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, TestBatchScope.formal())
|
|
|
|
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]
|
|
self.assertTrue(statement.selected_columns[0].compare(BizPalletEvent.event_seq.property.columns[0]))
|
|
self.assertTrue(statement.get_final_froms()[0].compare(BizPalletEvent.__table__))
|
|
for dialect in (mysql.dialect(),):
|
|
sql = compile_sql(statement, dialect)
|
|
self.assertNotIn('max(', sql.lower())
|
|
self.assertIn('biz_pallet_event.event_id = 40', sql)
|
|
self.assertIn('biz_pallet_event.pallet_id = 7', sql)
|
|
self.assertIn('biz_pallet_event.test_batch_id IS NULL', sql)
|
|
self.assertIn('FOR UPDATE', sql)
|
|
self.assertNotIn('ORDER BY', sql)
|
|
self.assertNotIn('LIMIT', sql)
|
|
|
|
async def test_add_first_event_starts_at_one_without_query(self) -> None:
|
|
sequence_result = MagicMock()
|
|
sequence_result.scalar_one.return_value = None
|
|
sequence_result.scalar_one_or_none.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, TestBatchScope.formal())
|
|
|
|
self.assertEqual(1, event.event_seq)
|
|
self.assertEqual(51, inventory.last_event_id)
|
|
self.db.execute.assert_not_awaited()
|
|
self.db.add.assert_called_once_with(event)
|
|
self.assertEqual(2, self.db.flush.await_count)
|
|
|
|
async def test_add_event_rejects_missing_previous_pointer_without_write(self) -> None:
|
|
sequence_result = MagicMock()
|
|
sequence_result.scalar_one_or_none.return_value = None
|
|
self.db.execute.return_value = sequence_result
|
|
inventory = BizPalletInventory(pallet_id=8, last_event_id=50)
|
|
event = BizPalletEvent(pallet_id=8, idempotency_key='broken-pointer')
|
|
|
|
with self.assertRaisesRegex(RuntimeError, '托盘库存事件指针不一致'):
|
|
await InventoryDao.add_event(self.db, inventory, event, TestBatchScope.formal())
|
|
|
|
self.db.execute.assert_awaited_once()
|
|
self.db.add.assert_not_called()
|
|
self.db.flush.assert_not_awaited()
|
|
self.assertEqual(50, inventory.last_event_id)
|
|
self.assertIsNone(event.event_seq)
|
|
|
|
|
|
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_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, scope=TestBatchScope.formal())
|
|
|
|
self.assertIs(record, actual)
|
|
statement = self.db.execute.await_args.args[0]
|
|
self.assertTrue(statement.get_final_froms()[0].compare(model.__table__))
|
|
self.assertIs(True, statement.get_execution_options().get('populate_existing'))
|
|
for dialect in (mysql.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_shipment_for_update_locks_scoped_parent_then_current_items(self) -> None:
|
|
shipment = SimpleNamespace(shipment_id=12)
|
|
items = [
|
|
SimpleNamespace(shipment_item_id=1, shipment_id=12, pallet_id=5),
|
|
SimpleNamespace(shipment_item_id=3, shipment_id=12, pallet_id=2),
|
|
]
|
|
parent_result = MagicMock()
|
|
parent_result.scalar_one_or_none.return_value = shipment
|
|
item_result = MagicMock()
|
|
item_result.scalars.return_value.all.return_value = items
|
|
self.db.execute.side_effect = [parent_result, item_result]
|
|
|
|
actual = await OperationDao.get_shipment_for_update(
|
|
self.db,
|
|
12,
|
|
warehouse_ids=[7, 3, 7],
|
|
site_ids=[11, 9, 11],
|
|
scope=TestBatchScope.formal(),
|
|
)
|
|
|
|
self.assertIs(shipment, actual)
|
|
self.assertEqual([2, 5], [item.pallet_id for item in actual.items])
|
|
self.assertEqual(2, self.db.execute.await_count)
|
|
parent_statement = self.db.execute.await_args_list[0].args[0]
|
|
item_statement = self.db.execute.await_args_list[1].args[0]
|
|
self.assertTrue(parent_statement.get_final_froms()[0].compare(BizShipmentOrder.__table__))
|
|
self.assertTrue(item_statement.get_final_froms()[0].compare(BizShipmentItem.__table__))
|
|
self.assertIs(True, parent_statement.get_execution_options().get('populate_existing'))
|
|
self.assertIs(True, item_statement.get_execution_options().get('populate_existing'))
|
|
for dialect in (mysql.dialect(),):
|
|
with self.subTest(dialect=dialect.name):
|
|
parent_sql = compile_sql(parent_statement, dialect)
|
|
item_sql = compile_sql(item_statement, dialect)
|
|
self.assertIn('biz_shipment_order.shipment_id = 12', parent_sql)
|
|
self.assertIn('biz_shipment_order.source_warehouse_id IN (3, 7)', parent_sql)
|
|
self.assertIn('biz_shipment_order.target_site_id IN (9, 11)', parent_sql)
|
|
self.assertIn('FOR UPDATE', parent_sql)
|
|
self.assertIn('biz_shipment_item.shipment_id = 12', item_sql)
|
|
self.assertIn(
|
|
'ORDER BY biz_shipment_item.pallet_id, biz_shipment_item.shipment_item_id',
|
|
item_sql,
|
|
)
|
|
self.assertIn('FOR UPDATE', item_sql)
|
|
|
|
async def test_shipment_for_update_missing_scoped_parent_skips_item_lock(self) -> None:
|
|
parent_result = MagicMock()
|
|
parent_result.scalar_one_or_none.return_value = None
|
|
self.db.execute.return_value = parent_result
|
|
|
|
actual = await OperationDao.get_shipment_for_update(
|
|
self.db,
|
|
12,
|
|
warehouse_ids=[],
|
|
site_ids=[9],
|
|
scope=TestBatchScope.formal(),
|
|
)
|
|
|
|
self.assertIsNone(actual)
|
|
self.db.execute.assert_awaited_once()
|
|
parent_statement = self.db.execute.await_args.args[0]
|
|
mysql_sql = compile_sql(parent_statement, mysql.dialect())
|
|
self.assertIn('false = 1', mysql_sql)
|
|
|
|
async def test_inbound_for_update_locks_parent_then_current_items(self) -> None:
|
|
inbound = SimpleNamespace(inbound_id=11)
|
|
items = [
|
|
SimpleNamespace(inbound_item_id=2, inbound_id=11, pallet_id=2),
|
|
SimpleNamespace(inbound_item_id=5, inbound_id=11, pallet_id=5),
|
|
]
|
|
inbound_result = MagicMock()
|
|
inbound_result.scalar_one_or_none.return_value = inbound
|
|
item_result = MagicMock()
|
|
item_result.scalars.return_value.all.return_value = items
|
|
self.db.execute.side_effect = [inbound_result, item_result]
|
|
|
|
actual = await OperationDao.get_inbound_for_update(self.db, inbound_id=11, scope=TestBatchScope.formal())
|
|
|
|
self.assertIs(inbound, actual)
|
|
self.assertEqual(items, actual.items)
|
|
self.assertEqual(2, self.db.execute.await_count)
|
|
parent_statement = self.db.execute.await_args_list[0].args[0]
|
|
item_statement = self.db.execute.await_args_list[1].args[0]
|
|
self.assertTrue(parent_statement.get_final_froms()[0].compare(BizInboundOrder.__table__))
|
|
self.assertTrue(item_statement.get_final_froms()[0].compare(BizInboundItem.__table__))
|
|
self.assertIs(True, parent_statement.get_execution_options().get('populate_existing'))
|
|
self.assertIs(True, item_statement.get_execution_options().get('populate_existing'))
|
|
for dialect in (mysql.dialect(),):
|
|
with self.subTest(dialect=dialect.name):
|
|
parent_sql = compile_sql(parent_statement, dialect)
|
|
item_sql = compile_sql(item_statement, dialect)
|
|
self.assertIn('biz_inbound_order.inbound_id = 11', parent_sql)
|
|
self.assertIn('FOR UPDATE', parent_sql)
|
|
self.assertIn('biz_inbound_item.inbound_id = 11', item_sql)
|
|
self.assertIn(
|
|
'ORDER BY biz_inbound_item.pallet_id, biz_inbound_item.inbound_item_id',
|
|
item_sql,
|
|
)
|
|
self.assertIn('FOR UPDATE', item_sql)
|
|
|
|
async def test_inbound_for_update_applies_scope_before_locking_items(self) -> None:
|
|
self.assertIn(
|
|
'warehouse_ids',
|
|
inspect.signature(OperationDao.get_inbound_for_update).parameters,
|
|
)
|
|
parent_result = MagicMock()
|
|
parent_result.scalar_one_or_none.return_value = None
|
|
self.db.execute.return_value = parent_result
|
|
|
|
actual = await OperationDao.get_inbound_for_update(
|
|
self.db,
|
|
inbound_id=11,
|
|
warehouse_ids=[7, 3, 7],
|
|
scope=TestBatchScope.formal(),
|
|
)
|
|
|
|
self.assertIsNone(actual)
|
|
self.db.execute.assert_awaited_once()
|
|
parent_statement = self.db.execute.await_args.args[0]
|
|
for dialect in (mysql.dialect(),):
|
|
with self.subTest(dialect=dialect.name):
|
|
sql = compile_sql(parent_statement, dialect)
|
|
self.assertIn('biz_inbound_order.warehouse_id IN (3, 7)', sql)
|
|
self.assertIn('FOR UPDATE', 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',
|
|
{'source_site_id': 9, 'target_warehouse_id': 7},
|
|
),
|
|
(
|
|
OperationDao.get_return_by_confirm_idempotency_key,
|
|
BizReturnOrder,
|
|
'confirm_idempotency_key',
|
|
{'source_site_id': 9, 'target_warehouse_id': 7},
|
|
),
|
|
)
|
|
for method, model, key_column, kwargs 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', scope=TestBatchScope.formal(), **kwargs)
|
|
|
|
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)
|
|
if model is BizReturnOrder:
|
|
self.assertIn('biz_return_order.source_site_id = 9', sql)
|
|
self.assertIn('biz_return_order.target_warehouse_id = 7', sql)
|
|
self.assertNotIn(' OR ', sql)
|
|
self.assertNotIn('FOR UPDATE', sql)
|
|
|
|
async def test_inbound_confirm_key_pushes_warehouse_scope_into_sql(self) -> None:
|
|
self.assertIn(
|
|
'warehouse_ids',
|
|
inspect.signature(OperationDao.get_inbound_by_confirm_idempotency_key).parameters,
|
|
)
|
|
result = MagicMock()
|
|
result.scalar_one_or_none.return_value = None
|
|
self.db.execute.return_value = result
|
|
|
|
actual = await OperationDao.get_inbound_by_confirm_idempotency_key(
|
|
self.db,
|
|
'confirm-key',
|
|
warehouse_ids=[7, 3, 7],
|
|
scope=TestBatchScope.formal(),
|
|
)
|
|
|
|
self.assertIsNone(actual)
|
|
statement = self.db.execute.await_args.args[0]
|
|
for dialect in (mysql.dialect(),):
|
|
with self.subTest(dialect=dialect.name):
|
|
sql = compile_sql(statement, dialect)
|
|
self.assertIn("biz_inbound_order.confirm_idempotency_key = 'confirm-key'", sql)
|
|
self.assertIn('biz_inbound_order.warehouse_id IN (3, 7)', 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-03 00:00:00'",
|
|
),
|
|
),
|
|
(
|
|
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-03 00:00:00'",
|
|
),
|
|
),
|
|
(
|
|
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-03 00:00:00'",
|
|
),
|
|
),
|
|
(
|
|
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-03 00:00:00'",
|
|
),
|
|
),
|
|
(
|
|
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-03 00:00:00'",
|
|
),
|
|
),
|
|
)
|
|
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:
|
|
kwargs = {
|
|
'scope': TestBatchScope.formal(),
|
|
'is_page': is_page,
|
|
}
|
|
parameters = inspect.signature(method).parameters
|
|
if 'warehouse_ids' in parameters:
|
|
kwargs['warehouse_ids'] = None
|
|
if 'site_ids' in parameters:
|
|
kwargs['site_ids'] = None
|
|
actual = await method(self.db, query, **kwargs)
|
|
|
|
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_inbound_list_applies_warehouse_scope_inside_paginated_statement(self) -> None:
|
|
query = InboundPageQueryModel(page_num=2, page_size=7)
|
|
marker = object()
|
|
with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as paginate:
|
|
actual = await OperationDao.get_inbound_list(
|
|
self.db,
|
|
query,
|
|
is_page=True,
|
|
warehouse_ids=[5, 2, 5],
|
|
scope=TestBatchScope.formal(),
|
|
)
|
|
|
|
self.assertIs(marker, actual)
|
|
statement = paginate.await_args.args[1]
|
|
for dialect in (mysql.dialect(),):
|
|
with self.subTest(dialect=dialect.name):
|
|
sql = compile_sql(statement, dialect)
|
|
self.assertIn('biz_inbound_order.warehouse_id IN (2, 5)', sql)
|
|
self.assertIn('ORDER BY biz_inbound_order.inbound_id', sql)
|
|
|
|
async def test_inbound_list_empty_scope_compiles_to_false_for_both_dialects(self) -> None:
|
|
query = InboundPageQueryModel(page_num=4, page_size=6)
|
|
marker = object()
|
|
with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as paginate:
|
|
actual = await OperationDao.get_inbound_list(
|
|
self.db,
|
|
query,
|
|
is_page=True,
|
|
warehouse_ids=[],
|
|
scope=TestBatchScope.formal(),
|
|
)
|
|
|
|
self.assertIs(marker, actual)
|
|
statement = paginate.await_args.args[1]
|
|
mysql_sql = compile_sql(statement, mysql.dialect())
|
|
self.assertIn('WHERE false = 1', mysql_sql)
|
|
self.assertEqual((query.page_num, query.page_size, True), paginate.await_args.args[2:])
|
|
|
|
async def test_inbound_list_none_scope_has_no_authorization_predicate(self) -> None:
|
|
query = InboundPageQueryModel()
|
|
with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=object())) as paginate:
|
|
await OperationDao.get_inbound_list(
|
|
self.db,
|
|
query,
|
|
is_page=True,
|
|
warehouse_ids=None,
|
|
scope=TestBatchScope.formal(),
|
|
)
|
|
|
|
sql = compile_sql(paginate.await_args.args[1])
|
|
self.assertNotIn('warehouse_id IN', sql)
|
|
self.assertNotIn('WHERE false', sql)
|
|
|
|
async def test_shipment_list_applies_both_scopes_and_empty_scope_is_false(self) -> None:
|
|
query = ShipmentPageQueryModel(page_num=2, page_size=7)
|
|
marker = object()
|
|
with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as paginate:
|
|
actual = await OperationDao.get_shipment_list(
|
|
self.db,
|
|
query,
|
|
is_page=True,
|
|
warehouse_ids=[7, 3, 7],
|
|
site_ids=[11, 9, 11],
|
|
scope=TestBatchScope.formal(),
|
|
)
|
|
|
|
self.assertIs(marker, actual)
|
|
statement = paginate.await_args.args[1]
|
|
for dialect in (mysql.dialect(),):
|
|
with self.subTest(dialect=dialect.name):
|
|
sql = compile_sql(statement, dialect)
|
|
self.assertIn('biz_shipment_order.source_warehouse_id IN (3, 7)', sql)
|
|
self.assertIn('biz_shipment_order.target_site_id IN (9, 11)', sql)
|
|
|
|
with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as empty_paginate:
|
|
await OperationDao.get_shipment_list(
|
|
self.db,
|
|
query,
|
|
warehouse_ids=[7],
|
|
site_ids=[],
|
|
scope=TestBatchScope.formal(),
|
|
)
|
|
empty_statement = empty_paginate.await_args.args[1]
|
|
self.assertIn('false = 1', compile_sql(empty_statement, mysql.dialect()))
|
|
|
|
async def test_shipment_detail_and_confirm_key_push_both_scopes_into_parent_sql(self) -> None:
|
|
parent_result = MagicMock()
|
|
parent_result.scalar_one_or_none.return_value = None
|
|
self.db.execute.return_value = parent_result
|
|
|
|
self.assertIsNone(
|
|
await OperationDao.get_shipment_by_id(
|
|
self.db,
|
|
12,
|
|
warehouse_ids=[7, 3],
|
|
site_ids=[11, 9],
|
|
scope=TestBatchScope.formal(),
|
|
)
|
|
)
|
|
self.db.execute.assert_awaited_once()
|
|
detail_statement = self.db.execute.await_args.args[0]
|
|
|
|
self.db.reset_mock()
|
|
self.db.execute.return_value = parent_result
|
|
self.assertIsNone(
|
|
await OperationDao.get_shipment_by_confirm_idempotency_key(
|
|
self.db,
|
|
'confirm-key',
|
|
warehouse_ids=[7, 3],
|
|
site_ids=[11, 9],
|
|
scope=TestBatchScope.formal(),
|
|
)
|
|
)
|
|
key_statement = self.db.execute.await_args.args[0]
|
|
for dialect in (mysql.dialect(),):
|
|
with self.subTest(dialect=dialect.name):
|
|
detail_sql = compile_sql(detail_statement, dialect)
|
|
key_sql = compile_sql(key_statement, dialect)
|
|
for sql in (detail_sql, key_sql):
|
|
self.assertIn('biz_shipment_order.source_warehouse_id IN (3, 7)', sql)
|
|
self.assertIn('biz_shipment_order.target_site_id IN (9, 11)', sql)
|
|
self.assertIn("biz_shipment_order.confirm_idempotency_key = 'confirm-key'", key_sql)
|
|
self.assertNotIn('FOR UPDATE', key_sql)
|
|
|
|
async def test_usage_reads_push_site_scope_and_empty_scope_is_always_false(self) -> None:
|
|
query = UsagePageQueryModel(page_num=2, page_size=7)
|
|
marker = object()
|
|
with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as paginate:
|
|
actual = await OperationDao.get_usage_list(
|
|
self.db,
|
|
query,
|
|
is_page=True,
|
|
site_ids=[11, 9, 11],
|
|
scope=TestBatchScope.formal(),
|
|
)
|
|
|
|
self.assertIs(marker, actual)
|
|
list_statement = paginate.await_args.args[1]
|
|
|
|
scalar_result = MagicMock()
|
|
scalar_result.scalar_one_or_none.return_value = None
|
|
self.db.execute.return_value = scalar_result
|
|
await OperationDao.get_usage_by_id(self.db, 15, site_ids=[11, 9, 11], scope=TestBatchScope.formal())
|
|
detail_statement = self.db.execute.await_args.args[0]
|
|
|
|
self.db.reset_mock()
|
|
self.db.execute.return_value = scalar_result
|
|
await OperationDao.get_usage_by_idempotency_key(
|
|
self.db,
|
|
'usage-key',
|
|
site_ids=[11, 9, 11],
|
|
scope=TestBatchScope.formal(),
|
|
)
|
|
key_statement = self.db.execute.await_args.args[0]
|
|
|
|
for dialect in (mysql.dialect(),):
|
|
with self.subTest(dialect=dialect.name):
|
|
for statement in (list_statement, detail_statement, key_statement):
|
|
sql = compile_sql(statement, dialect)
|
|
self.assertIn('biz_usage_record.site_id IN (9, 11)', sql)
|
|
|
|
with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as empty_paginate:
|
|
await OperationDao.get_usage_list(self.db, query, site_ids=[], scope=TestBatchScope.formal())
|
|
empty_list = empty_paginate.await_args.args[1]
|
|
|
|
self.db.reset_mock()
|
|
self.db.execute.return_value = scalar_result
|
|
await OperationDao.get_usage_by_id(self.db, 15, site_ids=[], scope=TestBatchScope.formal())
|
|
empty_detail = self.db.execute.await_args.args[0]
|
|
|
|
self.db.reset_mock()
|
|
self.db.execute.return_value = scalar_result
|
|
await OperationDao.get_usage_by_idempotency_key(
|
|
self.db, 'usage-key', site_ids=[], scope=TestBatchScope.formal()
|
|
)
|
|
empty_key = self.db.execute.await_args.args[0]
|
|
|
|
for dialect in (mysql.dialect(),):
|
|
with self.subTest(dialect=f'{dialect.name}-empty'):
|
|
for statement in (empty_list, empty_detail, empty_key):
|
|
sql = compile_sql(statement, dialect)
|
|
self.assertTrue('false' in sql.lower() or '0 = 1' in sql, sql)
|
|
|
|
async def test_usage_none_scope_adds_no_authorization_predicate(self) -> None:
|
|
query = UsagePageQueryModel()
|
|
marker = object()
|
|
with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as paginate:
|
|
await OperationDao.get_usage_list(self.db, query, site_ids=None, scope=TestBatchScope.formal())
|
|
|
|
sql = compile_sql(paginate.await_args.args[1])
|
|
self.assertNotIn('site_id IN', sql)
|
|
self.assertNotIn('WHERE false', sql)
|
|
|
|
async def test_usage_idempotency_reads_never_lock_and_preserve_scope(self) -> None:
|
|
scalar_result = MagicMock()
|
|
scalar_result.scalar_one_or_none.return_value = None
|
|
self.db.execute.return_value = scalar_result
|
|
|
|
cases = (
|
|
([11, 9, 11], 'biz_usage_record.site_id IN (9, 11)'),
|
|
([], None),
|
|
(None, None),
|
|
)
|
|
for site_ids, expected_scope in cases:
|
|
with self.subTest(site_ids=site_ids):
|
|
self.db.reset_mock()
|
|
self.db.execute.return_value = scalar_result
|
|
await OperationDao.get_usage_by_idempotency_key(
|
|
self.db,
|
|
'usage-key',
|
|
site_ids=site_ids,
|
|
scope=TestBatchScope.formal(),
|
|
)
|
|
statement = self.db.execute.await_args.args[0]
|
|
|
|
self.assertFalse(statement.get_execution_options().get('populate_existing', False))
|
|
for dialect in (mysql.dialect(),):
|
|
sql = compile_sql(statement, dialect)
|
|
self.assertNotIn('FOR UPDATE', sql)
|
|
if expected_scope is not None:
|
|
self.assertIn(expected_scope, sql)
|
|
elif site_ids == []:
|
|
self.assertTrue('false' in sql.lower() or '0 = 1' in sql, sql)
|
|
else:
|
|
self.assertNotIn('site_id IN', sql)
|
|
|
|
def test_time_conditions_include_fractional_last_second_with_exclusive_next_day(self) -> None:
|
|
lower, upper = OperationDao._time_conditions(
|
|
BizUsageRecord.occurred_at,
|
|
'2026-07-02',
|
|
'2026-07-02',
|
|
)
|
|
|
|
self.assertIs(operators.ge, lower.operator)
|
|
self.assertIs(operators.lt, upper.operator)
|
|
self.assertEqual(datetime(2026, 7, 2), lower.right.value)
|
|
self.assertEqual(datetime(2026, 7, 3), upper.right.value)
|
|
fractional_last_second = datetime(2026, 7, 2, 23, 59, 59, 500000)
|
|
self.assertTrue(lower.right.value <= fractional_last_second < upper.right.value)
|
|
|
|
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_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, scope=TestBatchScope.formal())
|
|
|
|
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, scope=TestBatchScope.formal()))
|
|
|
|
self.db.execute.assert_awaited_once()
|
|
|
|
async def test_inbound_detail_empty_scope_is_false_and_skips_items(self) -> None:
|
|
self.assertIn(
|
|
'warehouse_ids',
|
|
inspect.signature(OperationDao.get_inbound_by_id).parameters,
|
|
)
|
|
parent_result = MagicMock()
|
|
parent_result.scalar_one_or_none.return_value = None
|
|
self.db.execute.return_value = parent_result
|
|
|
|
actual = await OperationDao.get_inbound_by_id(
|
|
self.db,
|
|
inbound_id=11,
|
|
warehouse_ids=[],
|
|
scope=TestBatchScope.formal(),
|
|
)
|
|
|
|
self.assertIsNone(actual)
|
|
self.db.execute.assert_awaited_once()
|
|
parent_statement = self.db.execute.await_args.args[0]
|
|
mysql_sql = compile_sql(parent_statement, mysql.dialect())
|
|
self.assertIn('false = 1', mysql_sql)
|
|
|
|
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, scope=TestBatchScope.formal())
|
|
|
|
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_exclude_creation_audit(self) -> None:
|
|
malicious_create_time = datetime(2020, 1, 2, 3, 4, 5)
|
|
cases = (
|
|
(
|
|
OperationDao.update_inbound,
|
|
BizInboundOrder(
|
|
inbound_id=11,
|
|
warehouse_id=3,
|
|
status='CONFIRMED',
|
|
create_by='malicious',
|
|
create_time=malicious_create_time,
|
|
),
|
|
BizInboundOrder,
|
|
'inbound_id',
|
|
11,
|
|
'warehouse_id=3',
|
|
),
|
|
(
|
|
OperationDao.update_filling_record,
|
|
BizFillingRecord(
|
|
filling_id=12,
|
|
active_slot=None,
|
|
create_by='malicious',
|
|
create_time=malicious_create_time,
|
|
),
|
|
BizFillingRecord,
|
|
'filling_id',
|
|
12,
|
|
'active_slot=NULL',
|
|
),
|
|
(
|
|
OperationDao.update_shipment,
|
|
BizShipmentOrder(
|
|
shipment_id=13,
|
|
status='CONFIRMED',
|
|
create_by='malicious',
|
|
create_time=malicious_create_time,
|
|
),
|
|
BizShipmentOrder,
|
|
'shipment_id',
|
|
13,
|
|
"status='CONFIRMED'",
|
|
),
|
|
(
|
|
OperationDao.update_return_order,
|
|
BizReturnOrder(
|
|
return_id=14,
|
|
status='CONFIRMED',
|
|
create_by='malicious',
|
|
create_time=malicious_create_time,
|
|
),
|
|
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.assertNotIn('create_by', sql)
|
|
self.assertNotIn('create_time', sql)
|
|
self.db.flush.assert_awaited_once()
|
|
self.db.commit.assert_not_awaited()
|
|
self.db.rollback.assert_not_awaited()
|
|
|
|
async def test_delete_methods_remove_items_before_parent_without_commit(self) -> None:
|
|
cases = (
|
|
(
|
|
OperationDao.delete_inbound,
|
|
BizInboundItem,
|
|
BizInboundOrder,
|
|
'inbound_id',
|
|
11,
|
|
),
|
|
(
|
|
OperationDao.delete_shipment,
|
|
BizShipmentItem,
|
|
BizShipmentOrder,
|
|
'shipment_id',
|
|
12,
|
|
),
|
|
)
|
|
for method, item_model, order_model, id_field, record_id in cases:
|
|
with self.subTest(method=method.__name__):
|
|
self.db.reset_mock()
|
|
|
|
await method(self.db, record_id)
|
|
|
|
self.assertEqual(2, len(self.db.execute.await_args_list))
|
|
item_statement = self.db.execute.await_args_list[0].args[0]
|
|
order_statement = self.db.execute.await_args_list[1].args[0]
|
|
self.assertTrue(item_statement.table.compare(item_model.__table__))
|
|
self.assertTrue(order_statement.table.compare(order_model.__table__))
|
|
for dialect in (mysql.dialect(),):
|
|
self.assertIn(
|
|
f'{item_model.__tablename__}.{id_field} = {record_id}',
|
|
compile_sql(item_statement, dialect),
|
|
)
|
|
self.assertIn(
|
|
f'{order_model.__tablename__}.{id_field} = {record_id}',
|
|
compile_sql(order_statement, dialect),
|
|
)
|
|
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()
|