1409 lines
61 KiB
Python
1409 lines
61 KiB
Python
import inspect
|
|
import secrets
|
|
import unittest
|
|
from datetime import datetime
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from unittest.mock import call as mock_call
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.routing import APIRoute
|
|
from pydantic import ValidationError
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from common.aspect.interface_auth import CheckUserInterfaceAuth
|
|
from common.aspect.pre_auth import PreAuth
|
|
from common.vo import CrudResponseModel, DataResponseModel, PageModel, PageResponseModel, ResponseBaseModel
|
|
from exceptions.exception import PermissionException, ServiceException, ServiceWarning
|
|
from module_admin.entity.do.user_do import SysUser
|
|
from module_admin.service.login_service import LoginService
|
|
from module_business.controller import pallet_controller as pallet_controller_module
|
|
from module_business.controller import site_controller as site_controller_module
|
|
from module_business.controller import warehouse_controller as warehouse_controller_module
|
|
from module_business.dao.access_dao import AccessDao
|
|
from module_business.dao.master_dao import MasterDao
|
|
from module_business.dependency.test_batch_dependency import TestBatchScopeResolver
|
|
from module_business.domain.test_batch_scope import TestBatchScope
|
|
from module_business.entity.do.master_do import (
|
|
BizCustomerSite,
|
|
BizPallet,
|
|
BizUserSite,
|
|
BizUserWarehouse,
|
|
BizWarehouse,
|
|
)
|
|
from module_business.entity.do.operation_do import BizPalletInventory
|
|
from module_business.entity.vo.master_vo import (
|
|
BusinessUserOptionModel,
|
|
DeletePalletModel,
|
|
PalletCreateModel,
|
|
PalletEditModel,
|
|
PalletModel,
|
|
PalletPageQueryModel,
|
|
UserAssignModel,
|
|
)
|
|
from module_business.service import pallet_service as pallet_service_module
|
|
from module_business.service.pallet_service import PalletService
|
|
from module_business.service.test_batch_access_service import TestBatchAccessService
|
|
from tests.module_business.test_batch_fixtures import make_admin, make_current_user, make_test_scope
|
|
from utils.page_util import PageUtil
|
|
|
|
|
|
def compile_sql(statement: Any) -> str:
|
|
return ' '.join(str(statement.compile(compile_kwargs={'literal_binds': True})).split())
|
|
|
|
|
|
def integrity_error() -> IntegrityError:
|
|
return IntegrityError('statement', {}, RuntimeError('constraint'))
|
|
|
|
|
|
def make_pallet_record(
|
|
pallet_id: int,
|
|
capacity_packages: int = 20,
|
|
test_batch_id: int | None = None,
|
|
) -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
pallet_id=pallet_id,
|
|
test_batch_id=test_batch_id,
|
|
pallet_code=f'P-{pallet_id:03d}',
|
|
qr_token=f'fixed-token-{pallet_id}',
|
|
capacity_packages=capacity_packages,
|
|
asset_status='0',
|
|
remark=None,
|
|
)
|
|
|
|
|
|
class TestPalletServiceContract(unittest.TestCase):
|
|
def test_service_is_available(self) -> None:
|
|
self.assertIsNotNone(PalletService)
|
|
|
|
|
|
class TestPalletVoContract(unittest.TestCase):
|
|
def test_business_user_option_only_exposes_safe_fields(self) -> None:
|
|
source = SimpleNamespace(
|
|
user_id=7,
|
|
user_name='operator',
|
|
nick_name=None,
|
|
assigned=True,
|
|
password='secret',
|
|
email='operator@example.com',
|
|
phonenumber='13800000000',
|
|
token='token',
|
|
)
|
|
|
|
option = BusinessUserOptionModel.model_validate(source)
|
|
|
|
self.assertEqual(
|
|
{'userId': 7, 'userName': 'operator', 'nickName': None, 'assigned': True},
|
|
option.model_dump(by_alias=True),
|
|
)
|
|
self.assertTrue(
|
|
{'password', 'email', 'phonenumber', 'phone', 'token'}.isdisjoint(BusinessUserOptionModel.model_fields)
|
|
)
|
|
|
|
def test_business_user_option_validates_identity_fields(self) -> None:
|
|
invalid_values = (
|
|
{'userId': 0, 'userName': 'operator', 'assigned': False},
|
|
{'userId': 1, 'userName': '', 'assigned': False},
|
|
{'userId': 1, 'userName': ' ', 'assigned': False},
|
|
)
|
|
for value in invalid_values:
|
|
with self.subTest(value=value), self.assertRaises(ValidationError):
|
|
BusinessUserOptionModel.model_validate(value)
|
|
|
|
def test_pallet_edit_rejects_code_and_qr_token(self) -> None:
|
|
base = {'palletId': 1, 'capacityPackages': 20, 'assetStatus': '0'}
|
|
for identity_field in ({'palletCode': 'P-001'}, {'qrToken': 'fixed-token'}):
|
|
with self.subTest(identity_field=identity_field), self.assertRaises(ValidationError):
|
|
PalletEditModel.model_validate({**base, **identity_field})
|
|
|
|
|
|
class TestPalletMasterDao(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self) -> None:
|
|
self.db = MagicMock(spec=AsyncSession)
|
|
|
|
async def test_list_pallet_filters_and_orders_by_pallet_id(self) -> None:
|
|
query = PalletPageQueryModel(palletCode='P-00', assetStatus='0')
|
|
expected = object()
|
|
|
|
with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=expected)) as paginate:
|
|
actual = await MasterDao.list_pallet(
|
|
self.db,
|
|
query,
|
|
TestBatchScope.formal(),
|
|
is_page=True,
|
|
)
|
|
|
|
self.assertIs(expected, actual)
|
|
statement = paginate.await_args.args[1]
|
|
sql = compile_sql(statement)
|
|
self.assertIn('FROM biz_pallet', sql)
|
|
self.assertIn("biz_pallet.pallet_code LIKE '%P-00%'", sql)
|
|
self.assertIn("biz_pallet.asset_status = '0'", sql)
|
|
self.assertIn('ORDER BY biz_pallet.pallet_id', sql)
|
|
|
|
async def test_add_pallet_and_inventory_only_flush_current_transaction(self) -> None:
|
|
pallet = BizPallet(
|
|
pallet_code='P-001',
|
|
qr_token='fixed-token',
|
|
capacity_packages=20,
|
|
asset_status='0',
|
|
)
|
|
inventory = BizPalletInventory(
|
|
pallet_id=3,
|
|
flow_state='UNSTOCKED',
|
|
remaining_packages=0,
|
|
version=0,
|
|
)
|
|
|
|
actual = await MasterDao.add_pallet(self.db, pallet)
|
|
await MasterDao.add_pallet_inventory(self.db, inventory)
|
|
|
|
self.assertIs(pallet, actual)
|
|
self.assertEqual([pallet, inventory], [call.args[0] for call in self.db.add.call_args_list])
|
|
self.assertEqual(2, self.db.flush.await_count)
|
|
self.db.commit.assert_not_awaited()
|
|
self.db.rollback.assert_not_awaited()
|
|
|
|
async def test_pallet_code_lookup_keeps_global_uniqueness(self) -> None:
|
|
query_result = MagicMock()
|
|
query_result.scalar_one_or_none.return_value = None
|
|
self.db.execute = AsyncMock(return_value=query_result)
|
|
|
|
await MasterDao.find_pallet_by_code(self.db, 'P-001')
|
|
|
|
statement = self.db.execute.await_args.args[0]
|
|
sql = compile_sql(statement)
|
|
self.assertIn("biz_pallet.pallet_code = 'P-001'", sql)
|
|
self.assertNotIn('test_batch_id', compile_sql(statement.whereclause))
|
|
|
|
async def test_update_pallet_sql_never_contains_identity_fields(self) -> None:
|
|
edit = PalletEditModel(palletId=3, capacityPackages=18, assetStatus='1', remark='停用')
|
|
updated_at = datetime(2026, 7, 26, 10, 30)
|
|
|
|
await MasterDao.update_pallet(
|
|
self.db,
|
|
edit,
|
|
TestBatchScope.formal(),
|
|
update_by='admin',
|
|
update_time=updated_at,
|
|
)
|
|
|
|
statement = self.db.execute.await_args.args[0]
|
|
sql = compile_sql(statement)
|
|
self.assertIn('UPDATE biz_pallet', sql)
|
|
self.assertIn('capacity_packages=18', sql)
|
|
self.assertIn("asset_status='1'", sql)
|
|
self.assertIn("remark='停用'", sql)
|
|
self.assertIn("update_by='admin'", sql)
|
|
self.assertIn('WHERE biz_pallet.pallet_id = 3', sql)
|
|
self.assertNotIn('pallet_code=', sql)
|
|
self.assertNotIn('qr_token=', sql)
|
|
self.db.commit.assert_not_awaited()
|
|
self.db.rollback.assert_not_awaited()
|
|
|
|
async def test_inventory_write_read_uses_row_lock(self) -> None:
|
|
result = MagicMock()
|
|
inventory = SimpleNamespace(pallet_id=9)
|
|
result.scalar_one_or_none.return_value = inventory
|
|
self.db.execute.return_value = result
|
|
|
|
actual = await MasterDao.get_pallet_inventory(
|
|
self.db,
|
|
pallet_id=9,
|
|
scope=TestBatchScope.formal(),
|
|
for_update=True,
|
|
)
|
|
|
|
self.assertIs(inventory, actual)
|
|
statement = self.db.execute.await_args.args[0]
|
|
self.assertIn('biz_pallet_inventory.pallet_id = 9', compile_sql(statement))
|
|
self.assertIn('FOR UPDATE', compile_sql(statement))
|
|
|
|
async def test_pallet_write_read_uses_row_lock(self) -> None:
|
|
result = MagicMock()
|
|
pallet = make_pallet_record(9)
|
|
result.scalar_one_or_none.return_value = pallet
|
|
self.db.execute.return_value = result
|
|
|
|
actual = await MasterDao.get_pallet(
|
|
self.db,
|
|
pallet_id=9,
|
|
scope=TestBatchScope.formal(),
|
|
for_update=True,
|
|
)
|
|
|
|
self.assertIs(pallet, actual)
|
|
statement = self.db.execute.await_args.args[0]
|
|
self.assertIn('biz_pallet.pallet_id = 9', compile_sql(statement))
|
|
self.assertIn('FOR UPDATE', compile_sql(statement))
|
|
|
|
async def test_assignment_resource_write_reads_use_row_lock(self) -> None:
|
|
cases = (
|
|
(MasterDao.get_warehouse, BizWarehouse, 'warehouse_id', 7),
|
|
(MasterDao.get_site, BizCustomerSite, 'site_id', 9),
|
|
)
|
|
for method, model, id_field, resource_id in cases:
|
|
with self.subTest(resource=model.__tablename__):
|
|
self.db.reset_mock()
|
|
query_result = MagicMock()
|
|
record = SimpleNamespace()
|
|
query_result.scalar_one_or_none.return_value = record
|
|
self.db.execute.return_value = query_result
|
|
|
|
actual = await method(
|
|
self.db,
|
|
resource_id,
|
|
TestBatchScope.formal(),
|
|
for_update=True,
|
|
)
|
|
|
|
self.assertIs(record, actual)
|
|
statement = self.db.execute.await_args.args[0]
|
|
sql = compile_sql(statement)
|
|
self.assertTrue(statement.get_final_froms()[0].compare(model.__table__))
|
|
self.assertIn(f'{model.__tablename__}.{id_field} = {resource_id}', sql)
|
|
self.assertIn('FOR UPDATE', sql)
|
|
|
|
async def test_assignment_resource_reads_are_unlocked_by_default(self) -> None:
|
|
cases = (
|
|
(MasterDao.get_warehouse, BizWarehouse, 7),
|
|
(MasterDao.get_site, BizCustomerSite, 9),
|
|
)
|
|
for method, model, resource_id in cases:
|
|
with self.subTest(resource=model.__tablename__):
|
|
self.db.reset_mock()
|
|
query_result = MagicMock()
|
|
query_result.scalar_one_or_none.return_value = SimpleNamespace()
|
|
self.db.execute.return_value = query_result
|
|
|
|
await method(self.db, resource_id, TestBatchScope.formal())
|
|
|
|
statement = self.db.execute.await_args.args[0]
|
|
self.assertTrue(statement.get_final_froms()[0].compare(model.__table__))
|
|
self.assertNotIn('FOR UPDATE', compile_sql(statement))
|
|
|
|
async def test_assigned_user_id_queries_target_exact_resource_association(self) -> None:
|
|
cases = (
|
|
(
|
|
MasterDao.list_warehouse_assigned_user_ids,
|
|
BizUserWarehouse,
|
|
BizUserWarehouse.user_id,
|
|
'warehouse_id',
|
|
BizWarehouse,
|
|
7,
|
|
),
|
|
(
|
|
MasterDao.list_site_assigned_user_ids,
|
|
BizUserSite,
|
|
BizUserSite.user_id,
|
|
'site_id',
|
|
BizCustomerSite,
|
|
9,
|
|
),
|
|
)
|
|
for method, model, selected_column, id_field, scope_model, resource_id in cases:
|
|
with self.subTest(resource=model.__tablename__):
|
|
self.db.reset_mock()
|
|
query_result = MagicMock()
|
|
query_result.scalars.return_value.all.return_value = [2, 5]
|
|
self.db.execute.return_value = query_result
|
|
|
|
actual = await method(self.db, resource_id, TestBatchScope.formal())
|
|
|
|
self.assertEqual([2, 5], actual)
|
|
self.db.execute.assert_awaited_once()
|
|
statement = self.db.execute.await_args.args[0]
|
|
self.assertEqual([selected_column], list(statement.selected_columns))
|
|
sql = compile_sql(statement)
|
|
self.assertIn(
|
|
f'FROM {model.__tablename__} JOIN {scope_model.__tablename__}',
|
|
sql,
|
|
)
|
|
self.assertIn(
|
|
f'{model.__tablename__}.{id_field} = {resource_id}',
|
|
sql,
|
|
)
|
|
self.assertIn(f'{scope_model.__tablename__}.test_batch_id IS NULL', sql)
|
|
|
|
async def test_assignable_user_queries_filter_disabled_and_deleted_rows(self) -> None:
|
|
list_result = MagicMock()
|
|
list_result.all.return_value = ()
|
|
selected_result = MagicMock()
|
|
selected_result.all.return_value = ()
|
|
self.db.execute.side_effect = [list_result, selected_result]
|
|
|
|
self.assertEqual([], await MasterDao.list_assignable_users(self.db))
|
|
self.assertEqual([], await MasterDao.list_assignable_users_by_ids(self.db, [2, 5]))
|
|
|
|
for statement in (call.args[0] for call in self.db.execute.await_args_list):
|
|
sql = compile_sql(statement)
|
|
self.assertIn('FROM sys_user', sql)
|
|
self.assertIn("sys_user.status = '0'", sql)
|
|
self.assertIn("sys_user.del_flag = '0'", sql)
|
|
self.assertNotIn('password', sql)
|
|
self.assertEqual(
|
|
[SysUser.user_id, SysUser.user_name, SysUser.nick_name],
|
|
list(statement.selected_columns),
|
|
)
|
|
self.assertIn('ORDER BY sys_user.user_id', compile_sql(self.db.execute.await_args_list[0].args[0]))
|
|
selected_sql = compile_sql(self.db.execute.await_args_list[1].args[0])
|
|
self.assertIn('sys_user.user_id IN (2, 5)', selected_sql)
|
|
|
|
|
|
class TestPalletCreationService(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self) -> None:
|
|
self.db = MagicMock(spec=AsyncSession)
|
|
self.admin = make_admin()
|
|
self.scope = TestBatchScope.formal()
|
|
self.create = PalletCreateModel(palletCode='P-001', capacityPackages=20, remark='测试托盘')
|
|
self.fixed_time = datetime(2026, 7, 26, 11, 30)
|
|
|
|
async def test_add_generates_server_token_and_initial_inventory_in_one_transaction(self) -> None:
|
|
actions: list[str] = []
|
|
captured: dict[str, Any] = {}
|
|
|
|
async def add_pallet(_db: AsyncSession, pallet: BizPallet) -> BizPallet:
|
|
actions.append('pallet')
|
|
captured['pallet'] = pallet
|
|
pallet.pallet_id = 17
|
|
return pallet
|
|
|
|
async def add_inventory(_db: AsyncSession, inventory: BizPalletInventory) -> BizPalletInventory:
|
|
actions.append('inventory')
|
|
captured['inventory'] = inventory
|
|
return inventory
|
|
|
|
async def commit() -> None:
|
|
actions.append('commit')
|
|
|
|
self.db.commit.side_effect = commit
|
|
with (
|
|
patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=None)),
|
|
patch.object(MasterDao, 'add_pallet', new=AsyncMock(side_effect=add_pallet)),
|
|
patch.object(MasterDao, 'add_pallet_inventory', new=AsyncMock(side_effect=add_inventory)),
|
|
patch.object(secrets, 'token_urlsafe', return_value='server-fixed-token') as token_urlsafe,
|
|
patch.object(pallet_service_module, 'datetime', create=True) as service_datetime,
|
|
):
|
|
service_datetime.now.return_value = self.fixed_time
|
|
result = await PalletService.add_pallet(self.db, self.create, self.admin, self.scope)
|
|
|
|
token_urlsafe.assert_called_once_with(32)
|
|
pallet = captured['pallet']
|
|
self.assertEqual('P-001', pallet.pallet_code)
|
|
self.assertEqual('server-fixed-token', pallet.qr_token)
|
|
self.assertEqual(20, pallet.capacity_packages)
|
|
self.assertEqual('0', pallet.asset_status)
|
|
self.assertEqual('admin', pallet.create_by)
|
|
self.assertEqual(self.fixed_time, pallet.create_time)
|
|
self.assertEqual('admin', pallet.update_by)
|
|
self.assertEqual(self.fixed_time, pallet.update_time)
|
|
self.assertIsNone(pallet.test_batch_id)
|
|
inventory = captured['inventory']
|
|
self.assertEqual(17, inventory.pallet_id)
|
|
self.assertEqual('UNSTOCKED', inventory.flow_state)
|
|
self.assertEqual(0, inventory.remaining_packages)
|
|
self.assertEqual(0, inventory.version)
|
|
self.assertIsNone(inventory.test_batch_id)
|
|
self.assertEqual(['pallet', 'inventory', 'commit'], actions)
|
|
self.db.commit.assert_awaited_once()
|
|
self.db.rollback.assert_not_awaited()
|
|
self.assertIsInstance(result, CrudResponseModel)
|
|
self.assertTrue(result.is_success)
|
|
|
|
async def test_add_rolls_back_when_pallet_flush_fails(self) -> None:
|
|
failure = RuntimeError('flush failed')
|
|
add_inventory = AsyncMock()
|
|
with (
|
|
patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=None)),
|
|
patch.object(MasterDao, 'add_pallet', new=AsyncMock(side_effect=failure)),
|
|
patch.object(MasterDao, 'add_pallet_inventory', new=add_inventory),
|
|
self.assertRaisesRegex(RuntimeError, 'flush failed'),
|
|
):
|
|
await PalletService.add_pallet(self.db, self.create, self.admin, self.scope)
|
|
|
|
add_inventory.assert_not_awaited()
|
|
self.db.commit.assert_not_awaited()
|
|
self.db.rollback.assert_awaited_once()
|
|
|
|
async def test_add_rolls_back_when_commit_fails(self) -> None:
|
|
async def assign_id(_db: AsyncSession, pallet: BizPallet) -> BizPallet:
|
|
pallet.pallet_id = 17
|
|
return pallet
|
|
|
|
self.db.commit.side_effect = RuntimeError('commit failed')
|
|
with (
|
|
patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=None)),
|
|
patch.object(MasterDao, 'add_pallet', new=AsyncMock(side_effect=assign_id)),
|
|
patch.object(MasterDao, 'add_pallet_inventory', new=AsyncMock()),
|
|
self.assertRaisesRegex(RuntimeError, 'commit failed'),
|
|
):
|
|
await PalletService.add_pallet(self.db, self.create, self.admin, self.scope)
|
|
|
|
self.db.rollback.assert_awaited_once()
|
|
|
|
async def test_add_converts_integrity_error_to_stable_service_exception(self) -> None:
|
|
with (
|
|
patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=None)),
|
|
patch.object(MasterDao, 'add_pallet', new=AsyncMock(side_effect=integrity_error())),
|
|
self.assertRaises(ServiceException) as raised,
|
|
):
|
|
await PalletService.add_pallet(self.db, self.create, self.admin, self.scope)
|
|
|
|
self.assertEqual('新增托盘失败，托盘编号或二维码标识已存在', raised.exception.message)
|
|
self.db.rollback.assert_awaited_once()
|
|
self.db.commit.assert_not_awaited()
|
|
|
|
async def test_add_rejects_duplicate_code_before_writing(self) -> None:
|
|
add_pallet = AsyncMock()
|
|
with (
|
|
patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=object())),
|
|
patch.object(MasterDao, 'add_pallet', new=add_pallet),
|
|
self.assertRaises(ServiceException) as raised,
|
|
):
|
|
await PalletService.add_pallet(self.db, self.create, self.admin, self.scope)
|
|
|
|
self.assertEqual('托盘编号已存在', raised.exception.message)
|
|
add_pallet.assert_not_awaited()
|
|
self.db.rollback.assert_awaited_once()
|
|
|
|
|
|
class TestPalletQrService(unittest.IsolatedAsyncioTestCase):
|
|
def test_render_qr_png_returns_real_stable_png(self) -> None:
|
|
first = PalletService.render_qr_png('fixed-original-token')
|
|
second = PalletService.render_qr_png('fixed-original-token')
|
|
|
|
self.assertEqual(b'\x89PNG\r\n\x1a\n', first[:8])
|
|
self.assertEqual(first, second)
|
|
|
|
async def test_get_qrcode_uses_original_persisted_token(self) -> None:
|
|
db = MagicMock(spec=AsyncSession)
|
|
scope = TestBatchScope.formal()
|
|
get_pallet = AsyncMock(return_value=SimpleNamespace(qr_token='persisted-token', test_batch_id=None))
|
|
with (
|
|
patch.object(MasterDao, 'get_pallet', new=get_pallet),
|
|
patch.object(PalletService, 'render_qr_png', return_value=b'png-bytes') as render,
|
|
patch.object(secrets, 'token_urlsafe') as token_urlsafe,
|
|
):
|
|
actual = await PalletService.get_pallet_qrcode(db, pallet_id=7, scope=scope)
|
|
|
|
self.assertEqual(b'png-bytes', actual)
|
|
get_pallet.assert_awaited_once_with(db, 7, scope)
|
|
render.assert_called_once_with('persisted-token')
|
|
token_urlsafe.assert_not_called()
|
|
|
|
async def test_get_qrcode_rejects_missing_pallet(self) -> None:
|
|
db = MagicMock(spec=AsyncSession)
|
|
scope = TestBatchScope.formal()
|
|
with (
|
|
patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=None)),
|
|
self.assertRaises(ServiceException) as raised,
|
|
):
|
|
await PalletService.get_pallet_qrcode(db, pallet_id=404, scope=scope)
|
|
|
|
self.assertEqual('托盘不存在', raised.exception.message)
|
|
|
|
|
|
class TestPalletQueryAndEditService(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self) -> None:
|
|
self.db = MagicMock(spec=AsyncSession)
|
|
self.admin = make_admin()
|
|
self.scope = TestBatchScope.formal()
|
|
|
|
async def test_list_returns_models_without_removability_lookup(self) -> None:
|
|
page = PageModel[Any](
|
|
rows=[make_pallet_record(pallet_id) for pallet_id in range(1, 1001)],
|
|
pageNum=1,
|
|
pageSize=1000,
|
|
total=1000,
|
|
hasNext=False,
|
|
)
|
|
query = PalletPageQueryModel()
|
|
list_pallet = AsyncMock(return_value=page)
|
|
with patch.object(MasterDao, 'list_pallet', new=list_pallet):
|
|
actual = await PalletService.list_pallet(self.db, query, self.scope, is_page=True)
|
|
|
|
self.assertIsInstance(actual, PageModel)
|
|
self.assertEqual(1000, len(actual.rows))
|
|
self.assertTrue(all(isinstance(row, PalletModel) for row in actual.rows))
|
|
self.assertNotIn('can_remove', PalletModel.model_fields)
|
|
list_pallet.assert_awaited_once_with(self.db, query, self.scope, True)
|
|
|
|
async def test_detail_returns_persisted_pallet_without_removability_lookup(self) -> None:
|
|
get_pallet = AsyncMock(return_value=make_pallet_record(3))
|
|
with patch.object(MasterDao, 'get_pallet', new=get_pallet):
|
|
actual = await PalletService.detail_pallet(self.db, pallet_id=3, scope=self.scope)
|
|
|
|
self.assertIsInstance(actual, PalletModel)
|
|
self.assertNotIn('can_remove', PalletModel.model_fields)
|
|
get_pallet.assert_awaited_once_with(self.db, 3, self.scope)
|
|
|
|
async def test_detail_rejects_missing_pallet(self) -> None:
|
|
with (
|
|
patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=None)),
|
|
self.assertRaises(ServiceException) as raised,
|
|
):
|
|
await PalletService.detail_pallet(self.db, pallet_id=404, scope=self.scope)
|
|
|
|
self.assertEqual('托盘不存在', raised.exception.message)
|
|
|
|
async def test_edit_updates_only_mutable_fields_with_server_audit(self) -> None:
|
|
edit = PalletEditModel(palletId=3, capacityPackages=12, assetStatus='1', remark='停用')
|
|
updated_at = datetime(2026, 7, 26, 12, 30)
|
|
update_pallet = AsyncMock()
|
|
lock_actions: list[str] = []
|
|
|
|
async def get_pallet(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
|
|
lock_actions.append('pallet')
|
|
return make_pallet_record(3)
|
|
|
|
async def get_inventory(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
|
|
lock_actions.append('inventory')
|
|
return SimpleNamespace(remaining_packages=12, test_batch_id=None)
|
|
|
|
get_pallet_mock = AsyncMock(side_effect=get_pallet)
|
|
get_inventory_mock = AsyncMock(side_effect=get_inventory)
|
|
with (
|
|
patch.object(MasterDao, 'get_pallet', new=get_pallet_mock),
|
|
patch.object(MasterDao, 'get_pallet_inventory', new=get_inventory_mock),
|
|
patch.object(MasterDao, 'update_pallet', new=update_pallet),
|
|
patch.object(pallet_service_module, 'datetime') as service_datetime,
|
|
):
|
|
service_datetime.now.return_value = updated_at
|
|
result = await PalletService.edit_pallet(self.db, edit, self.admin, self.scope)
|
|
|
|
update_pallet.assert_awaited_once_with(
|
|
self.db,
|
|
edit,
|
|
self.scope,
|
|
update_by='admin',
|
|
update_time=updated_at,
|
|
)
|
|
self.assertEqual(['inventory', 'pallet'], lock_actions)
|
|
get_pallet_mock.assert_awaited_once_with(self.db, 3, self.scope, for_update=True)
|
|
get_inventory_mock.assert_awaited_once_with(self.db, 3, self.scope, for_update=True)
|
|
self.db.commit.assert_awaited_once()
|
|
self.db.rollback.assert_not_awaited()
|
|
self.assertTrue(result.is_success)
|
|
|
|
async def test_edit_rejects_capacity_below_remaining_packages(self) -> None:
|
|
edit = PalletEditModel(palletId=3, capacityPackages=11, assetStatus='0')
|
|
update_pallet = AsyncMock()
|
|
with (
|
|
patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=make_pallet_record(3))),
|
|
patch.object(
|
|
MasterDao,
|
|
'get_pallet_inventory',
|
|
new=AsyncMock(return_value=SimpleNamespace(remaining_packages=12, test_batch_id=None)),
|
|
),
|
|
patch.object(MasterDao, 'update_pallet', new=update_pallet),
|
|
self.assertRaises(ServiceException) as raised,
|
|
):
|
|
await PalletService.edit_pallet(self.db, edit, self.admin, self.scope)
|
|
|
|
self.assertEqual('托盘容量不能小于当前剩余包数', raised.exception.message)
|
|
update_pallet.assert_not_awaited()
|
|
self.db.commit.assert_not_awaited()
|
|
self.db.rollback.assert_awaited_once()
|
|
|
|
async def test_edit_rejects_missing_inventory(self) -> None:
|
|
edit = PalletEditModel(palletId=3, capacityPackages=20, assetStatus='0')
|
|
with (
|
|
patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=make_pallet_record(3))),
|
|
patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=None)),
|
|
self.assertRaises(ServiceException) as raised,
|
|
):
|
|
await PalletService.edit_pallet(self.db, edit, self.admin, self.scope)
|
|
|
|
self.assertEqual('托盘库存不存在', raised.exception.message)
|
|
self.db.rollback.assert_awaited_once()
|
|
|
|
|
|
class TestBusinessUserAssignmentService(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self) -> None:
|
|
self.db = MagicMock(spec=AsyncSession)
|
|
self.admin = make_admin()
|
|
self.scope = TestBatchScope.formal()
|
|
self.users = [
|
|
SimpleNamespace(
|
|
user_id=2,
|
|
user_name='alice',
|
|
nick_name='艾丽丝',
|
|
password='secret-a',
|
|
email='alice@example.com',
|
|
phonenumber='13800000000',
|
|
),
|
|
SimpleNamespace(
|
|
user_id=5,
|
|
user_name='bob',
|
|
nick_name=None,
|
|
password='secret-b',
|
|
email='bob@example.com',
|
|
phonenumber='13900000000',
|
|
),
|
|
]
|
|
|
|
async def test_assignment_get_returns_only_safe_fields_with_assigned_marker(self) -> None:
|
|
for resource, resource_id in (('warehouse', 7), ('site', 9)):
|
|
with self.subTest(resource=resource):
|
|
get_resource = AsyncMock(return_value=SimpleNamespace(test_batch_id=None))
|
|
list_users = AsyncMock(return_value=self.users)
|
|
list_assigned_ids = AsyncMock(return_value=[2])
|
|
list_user_resource_ids = AsyncMock()
|
|
with (
|
|
patch.object(
|
|
MasterDao,
|
|
f'get_{resource}',
|
|
new=get_resource,
|
|
),
|
|
patch.object(MasterDao, 'list_assignable_users', new=list_users),
|
|
patch.object(
|
|
MasterDao,
|
|
f'list_{resource}_assigned_user_ids',
|
|
new=list_assigned_ids,
|
|
create=True,
|
|
),
|
|
patch.object(AccessDao, f'list_user_{resource}_ids', new=list_user_resource_ids),
|
|
):
|
|
actual = await getattr(PalletService, f'list_{resource}_users')(
|
|
self.db,
|
|
resource_id,
|
|
self.scope,
|
|
)
|
|
|
|
self.assertEqual(
|
|
[
|
|
{'userId': 2, 'userName': 'alice', 'nickName': '艾丽丝', 'assigned': True},
|
|
{'userId': 5, 'userName': 'bob', 'nickName': None, 'assigned': False},
|
|
],
|
|
[option.model_dump(by_alias=True) for option in actual],
|
|
)
|
|
self.assertTrue(all(isinstance(option, BusinessUserOptionModel) for option in actual))
|
|
get_resource.assert_awaited_once_with(self.db, resource_id, self.scope)
|
|
list_users.assert_awaited_once_with(self.db)
|
|
list_assigned_ids.assert_awaited_once_with(self.db, resource_id, self.scope)
|
|
list_user_resource_ids.assert_not_awaited()
|
|
|
|
async def test_assignment_get_uses_three_fixed_queries_for_1000_users(self) -> None:
|
|
users = [
|
|
SimpleNamespace(user_id=user_id, user_name=f'user-{user_id}', nick_name=None) for user_id in range(1, 1001)
|
|
]
|
|
assigned_ids = [2, 500, 1000]
|
|
for resource, resource_id in (('warehouse', 7), ('site', 9)):
|
|
with self.subTest(resource=resource):
|
|
get_resource = AsyncMock(return_value=SimpleNamespace(test_batch_id=None))
|
|
list_users = AsyncMock(return_value=users)
|
|
list_assigned = AsyncMock(return_value=assigned_ids)
|
|
list_user_resource_ids = AsyncMock()
|
|
with (
|
|
patch.object(MasterDao, f'get_{resource}', new=get_resource),
|
|
patch.object(MasterDao, 'list_assignable_users', new=list_users),
|
|
patch.object(
|
|
MasterDao,
|
|
f'list_{resource}_assigned_user_ids',
|
|
new=list_assigned,
|
|
create=True,
|
|
),
|
|
patch.object(AccessDao, f'list_user_{resource}_ids', new=list_user_resource_ids),
|
|
):
|
|
actual = await getattr(PalletService, f'list_{resource}_users')(
|
|
self.db,
|
|
resource_id,
|
|
self.scope,
|
|
)
|
|
|
|
self.assertEqual(1000, len(actual))
|
|
self.assertEqual(assigned_ids, [option.user_id for option in actual if option.assigned])
|
|
get_resource.assert_awaited_once_with(self.db, resource_id, self.scope)
|
|
list_users.assert_awaited_once_with(self.db)
|
|
list_assigned.assert_awaited_once_with(self.db, resource_id, self.scope)
|
|
list_user_resource_ids.assert_not_awaited()
|
|
|
|
async def test_assignment_get_rejects_missing_resource_before_user_query(self) -> None:
|
|
cases = (('warehouse', 7, '仓库不存在'), ('site', 9, '客户现场不存在'))
|
|
for resource, resource_id, message in cases:
|
|
with self.subTest(resource=resource):
|
|
get_resource = AsyncMock(return_value=None)
|
|
list_users = AsyncMock()
|
|
with (
|
|
patch.object(MasterDao, f'get_{resource}', new=get_resource),
|
|
patch.object(MasterDao, 'list_assignable_users', new=list_users),
|
|
self.assertRaises(ServiceException) as raised,
|
|
):
|
|
await getattr(PalletService, f'list_{resource}_users')(
|
|
self.db,
|
|
resource_id,
|
|
self.scope,
|
|
)
|
|
|
|
self.assertEqual(message, raised.exception.message)
|
|
get_resource.assert_awaited_once_with(self.db, resource_id, self.scope)
|
|
list_users.assert_not_awaited()
|
|
|
|
async def test_assignment_put_validates_users_replaces_and_commits(self) -> None:
|
|
assignment = UserAssignModel(userIds=[2, 5])
|
|
for resource, resource_id in (('warehouse', 7), ('site', 9)):
|
|
with self.subTest(resource=resource):
|
|
self.db.reset_mock()
|
|
calls = MagicMock()
|
|
get_resource_mock = AsyncMock(return_value=SimpleNamespace(test_batch_id=None))
|
|
list_users_mock = AsyncMock(return_value=self.users)
|
|
replace_mock = AsyncMock()
|
|
calls.attach_mock(get_resource_mock, 'get_resource')
|
|
calls.attach_mock(list_users_mock, 'list_users')
|
|
calls.attach_mock(replace_mock, 'replace')
|
|
with (
|
|
patch.object(
|
|
MasterDao,
|
|
f'get_{resource}',
|
|
new=get_resource_mock,
|
|
),
|
|
patch.object(
|
|
MasterDao,
|
|
'list_assignable_users_by_ids',
|
|
new=list_users_mock,
|
|
),
|
|
patch.object(AccessDao, f'replace_{resource}_users', new=replace_mock),
|
|
):
|
|
result = await getattr(PalletService, f'assign_{resource}_users')(
|
|
self.db,
|
|
resource_id,
|
|
assignment,
|
|
self.admin,
|
|
self.scope,
|
|
)
|
|
|
|
self.assertEqual(
|
|
[
|
|
mock_call.get_resource(
|
|
self.db,
|
|
resource_id,
|
|
self.scope,
|
|
for_update=True,
|
|
),
|
|
mock_call.list_users(self.db, [2, 5]),
|
|
mock_call.replace(self.db, resource_id, [2, 5]),
|
|
],
|
|
calls.mock_calls,
|
|
)
|
|
get_resource_mock.assert_awaited_once_with(
|
|
self.db,
|
|
resource_id,
|
|
self.scope,
|
|
for_update=True,
|
|
)
|
|
list_users_mock.assert_awaited_once_with(self.db, [2, 5])
|
|
replace_mock.assert_awaited_once_with(self.db, resource_id, [2, 5])
|
|
self.db.commit.assert_awaited_once()
|
|
self.db.rollback.assert_not_awaited()
|
|
self.assertTrue(result.is_success)
|
|
|
|
async def test_assignment_put_allows_empty_complete_replacement(self) -> None:
|
|
try:
|
|
assignment = UserAssignModel(userIds=[])
|
|
except ValidationError as exc:
|
|
self.fail(f'完整替换用户关联必须允许空数组：{exc}')
|
|
|
|
for resource, resource_id in (('warehouse', 7), ('site', 9)):
|
|
with self.subTest(resource=resource):
|
|
self.db.reset_mock()
|
|
list_users_mock = AsyncMock(return_value=[])
|
|
replace_mock = AsyncMock()
|
|
with (
|
|
patch.object(
|
|
MasterDao,
|
|
f'get_{resource}',
|
|
new=AsyncMock(return_value=SimpleNamespace(test_batch_id=None)),
|
|
),
|
|
patch.object(MasterDao, 'list_assignable_users_by_ids', new=list_users_mock),
|
|
patch.object(AccessDao, f'replace_{resource}_users', new=replace_mock),
|
|
):
|
|
result = await getattr(PalletService, f'assign_{resource}_users')(
|
|
self.db,
|
|
resource_id,
|
|
assignment,
|
|
self.admin,
|
|
self.scope,
|
|
)
|
|
|
|
list_users_mock.assert_awaited_once_with(self.db, [])
|
|
replace_mock.assert_awaited_once_with(self.db, resource_id, [])
|
|
self.db.commit.assert_awaited_once()
|
|
self.db.rollback.assert_not_awaited()
|
|
self.assertTrue(result.is_success)
|
|
|
|
async def test_empty_replacement_deletes_links_without_adding_new_rows(self) -> None:
|
|
cases = (
|
|
('warehouse', 7, 'biz_user_warehouse', 'warehouse_id'),
|
|
('site', 9, 'biz_user_site', 'site_id'),
|
|
)
|
|
for resource, resource_id, table, id_field in cases:
|
|
with self.subTest(resource=resource):
|
|
self.db.reset_mock()
|
|
|
|
await getattr(AccessDao, f'replace_{resource}_users')(self.db, resource_id, [])
|
|
|
|
delete_sql = compile_sql(self.db.execute.await_args.args[0])
|
|
self.assertIn(f'DELETE FROM {table}', delete_sql)
|
|
self.assertIn(f'{table}.{id_field} = {resource_id}', delete_sql)
|
|
self.db.add_all.assert_called_once_with([])
|
|
self.db.flush.assert_awaited_once()
|
|
|
|
async def test_assignment_put_rejects_missing_disabled_or_deleted_user(self) -> None:
|
|
replace = AsyncMock()
|
|
assignment = UserAssignModel(userIds=[2, 5])
|
|
with (
|
|
patch.object(
|
|
MasterDao,
|
|
'get_warehouse',
|
|
new=AsyncMock(return_value=SimpleNamespace(test_batch_id=None)),
|
|
),
|
|
patch.object(
|
|
MasterDao,
|
|
'list_assignable_users_by_ids',
|
|
new=AsyncMock(return_value=[self.users[0]]),
|
|
),
|
|
patch.object(AccessDao, 'replace_warehouse_users', new=replace),
|
|
self.assertRaises(ServiceException) as raised,
|
|
):
|
|
await PalletService.assign_warehouse_users(
|
|
self.db,
|
|
7,
|
|
assignment,
|
|
self.admin,
|
|
self.scope,
|
|
)
|
|
|
|
self.assertEqual('存在不存在、停用或已删除的用户', raised.exception.message)
|
|
replace.assert_not_awaited()
|
|
self.db.commit.assert_not_awaited()
|
|
self.db.rollback.assert_awaited_once()
|
|
|
|
async def test_replace_failure_rolls_back_and_preserves_original_exception(self) -> None:
|
|
assignment = UserAssignModel(userIds=[2, 5])
|
|
with (
|
|
patch.object(
|
|
MasterDao,
|
|
'get_site',
|
|
new=AsyncMock(return_value=SimpleNamespace(test_batch_id=None)),
|
|
),
|
|
patch.object(
|
|
MasterDao,
|
|
'list_assignable_users_by_ids',
|
|
new=AsyncMock(return_value=self.users),
|
|
),
|
|
patch.object(
|
|
AccessDao,
|
|
'replace_site_users',
|
|
new=AsyncMock(side_effect=RuntimeError('replace failed')),
|
|
),
|
|
self.assertRaisesRegex(RuntimeError, 'replace failed'),
|
|
):
|
|
await PalletService.assign_site_users(
|
|
self.db,
|
|
9,
|
|
assignment,
|
|
self.admin,
|
|
self.scope,
|
|
)
|
|
|
|
self.db.commit.assert_not_awaited()
|
|
self.db.rollback.assert_awaited_once()
|
|
|
|
async def test_assignment_commit_failure_rolls_back(self) -> None:
|
|
assignment = UserAssignModel(userIds=[2, 5])
|
|
self.db.commit.side_effect = RuntimeError('commit failed')
|
|
with (
|
|
patch.object(
|
|
MasterDao,
|
|
'get_warehouse',
|
|
new=AsyncMock(return_value=SimpleNamespace(test_batch_id=None)),
|
|
),
|
|
patch.object(
|
|
MasterDao,
|
|
'list_assignable_users_by_ids',
|
|
new=AsyncMock(return_value=self.users),
|
|
),
|
|
patch.object(AccessDao, 'replace_warehouse_users', new=AsyncMock()),
|
|
self.assertRaisesRegex(RuntimeError, 'commit failed'),
|
|
):
|
|
await PalletService.assign_warehouse_users(
|
|
self.db,
|
|
7,
|
|
assignment,
|
|
self.admin,
|
|
self.scope,
|
|
)
|
|
|
|
self.db.rollback.assert_awaited_once()
|
|
|
|
async def test_assignment_integrity_error_has_stable_message_and_rolls_back(self) -> None:
|
|
assignment = UserAssignModel(userIds=[2, 5])
|
|
with (
|
|
patch.object(
|
|
MasterDao,
|
|
'get_warehouse',
|
|
new=AsyncMock(return_value=SimpleNamespace(test_batch_id=None)),
|
|
),
|
|
patch.object(
|
|
MasterDao,
|
|
'list_assignable_users_by_ids',
|
|
new=AsyncMock(return_value=self.users),
|
|
),
|
|
patch.object(
|
|
AccessDao,
|
|
'replace_warehouse_users',
|
|
new=AsyncMock(side_effect=integrity_error()),
|
|
),
|
|
self.assertRaises(ServiceException) as raised,
|
|
):
|
|
await PalletService.assign_warehouse_users(
|
|
self.db,
|
|
7,
|
|
assignment,
|
|
self.admin,
|
|
self.scope,
|
|
)
|
|
|
|
self.assertEqual('分配仓库用户失败，关联数据无效', raised.exception.message)
|
|
self.db.rollback.assert_awaited_once()
|
|
|
|
|
|
class TestPalletBatchScopeService(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self) -> None:
|
|
self.db = MagicMock(spec=AsyncSession)
|
|
self.admin = make_admin()
|
|
self.scope = make_test_scope(9)
|
|
self.create = PalletCreateModel(palletCode='P-T09', capacityPackages=20)
|
|
|
|
async def test_list_detail_and_qrcode_forward_scope_to_scoped_dao_reads(self) -> None:
|
|
query = PalletPageQueryModel()
|
|
page = PageModel(rows=[], pageNum=1, pageSize=10, total=0, hasNext=False)
|
|
list_pallet = AsyncMock(return_value=page)
|
|
get_pallet = AsyncMock(return_value=make_pallet_record(7, test_batch_id=9))
|
|
with (
|
|
patch.object(MasterDao, 'list_pallet', new=list_pallet),
|
|
patch.object(MasterDao, 'get_pallet', new=get_pallet),
|
|
patch.object(PalletService, 'render_qr_png', return_value=b'png'),
|
|
):
|
|
listed = await PalletService.list_pallet(self.db, query, self.scope, is_page=True)
|
|
detailed = await PalletService.detail_pallet(self.db, 7, self.scope)
|
|
qr_png = await PalletService.get_pallet_qrcode(self.db, 7, self.scope)
|
|
|
|
self.assertIsInstance(listed, PageModel)
|
|
self.assertIsInstance(detailed, PalletModel)
|
|
self.assertEqual(b'png', qr_png)
|
|
list_pallet.assert_awaited_once_with(self.db, query, self.scope, True)
|
|
self.assertEqual(
|
|
[mock_call(self.db, 7, self.scope), mock_call(self.db, 7, self.scope)],
|
|
get_pallet.await_args_list,
|
|
)
|
|
|
|
async def test_test_scope_add_locks_first_and_marks_pallet_and_inventory(self) -> None:
|
|
events: list[str] = []
|
|
captured: dict[str, Any] = {}
|
|
|
|
async def lock_batch(*_args: Any) -> None:
|
|
events.append('lock-batch')
|
|
|
|
async def find_duplicate(*_args: Any) -> None:
|
|
events.append('find-duplicate')
|
|
|
|
async def add_pallet(_db: AsyncSession, pallet: BizPallet) -> BizPallet:
|
|
events.append('add-pallet')
|
|
captured['pallet'] = pallet
|
|
pallet.pallet_id = 17
|
|
return pallet
|
|
|
|
async def add_inventory(_db: AsyncSession, inventory: BizPalletInventory) -> BizPalletInventory:
|
|
events.append('add-inventory')
|
|
captured['inventory'] = inventory
|
|
return inventory
|
|
|
|
with (
|
|
patch.object(TestBatchAccessService, 'lock_active_batch', new=AsyncMock(side_effect=lock_batch)),
|
|
patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(side_effect=find_duplicate)),
|
|
patch.object(MasterDao, 'add_pallet', new=AsyncMock(side_effect=add_pallet)),
|
|
patch.object(MasterDao, 'add_pallet_inventory', new=AsyncMock(side_effect=add_inventory)),
|
|
):
|
|
result = await PalletService.add_pallet(self.db, self.create, self.admin, self.scope)
|
|
|
|
self.assertTrue(result.is_success)
|
|
self.assertEqual('lock-batch', events[0])
|
|
self.assertEqual(9, captured['pallet'].test_batch_id)
|
|
self.assertEqual(9, captured['inventory'].test_batch_id)
|
|
self.assertEqual(17, captured['inventory'].pallet_id)
|
|
|
|
async def test_test_scope_edit_locks_batch_before_inventory_or_pallet(self) -> None:
|
|
events: list[str] = []
|
|
|
|
async def lock_batch(*_args: Any) -> None:
|
|
events.append('lock-batch')
|
|
|
|
async def get_inventory(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
|
|
events.append('get-inventory')
|
|
return SimpleNamespace(remaining_packages=10, test_batch_id=9)
|
|
|
|
async def get_pallet(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
|
|
events.append('get-pallet')
|
|
return make_pallet_record(7, test_batch_id=9)
|
|
|
|
edit = PalletEditModel(palletId=7, capacityPackages=20, assetStatus='0')
|
|
update_pallet = AsyncMock()
|
|
with (
|
|
patch.object(TestBatchAccessService, 'lock_active_batch', new=AsyncMock(side_effect=lock_batch)),
|
|
patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(side_effect=get_inventory)),
|
|
patch.object(MasterDao, 'get_pallet', new=AsyncMock(side_effect=get_pallet)),
|
|
patch.object(MasterDao, 'update_pallet', new=update_pallet),
|
|
):
|
|
result = await PalletService.edit_pallet(self.db, edit, self.admin, self.scope)
|
|
|
|
self.assertTrue(result.is_success)
|
|
self.assertEqual(['lock-batch', 'get-inventory', 'get-pallet'], events)
|
|
self.assertIs(self.scope, update_pallet.await_args.args[2])
|
|
|
|
async def test_regular_user_and_closed_batch_reject_pallet_writes_before_business_queries(self) -> None:
|
|
business_query = AsyncMock()
|
|
lock_batch = AsyncMock()
|
|
with (
|
|
patch.object(TestBatchAccessService, 'lock_active_batch', new=lock_batch),
|
|
patch.object(MasterDao, 'find_pallet_by_code', new=business_query),
|
|
self.assertRaisesRegex(PermissionException, '仅超级管理员'),
|
|
):
|
|
await PalletService.add_pallet(self.db, self.create, make_current_user(8), self.scope)
|
|
lock_batch.assert_not_awaited()
|
|
business_query.assert_not_awaited()
|
|
|
|
closed = PermissionException('', '当前测试批次已关闭')
|
|
with (
|
|
patch.object(TestBatchAccessService, 'lock_active_batch', new=AsyncMock(side_effect=closed)),
|
|
patch.object(MasterDao, 'find_pallet_by_code', new=business_query),
|
|
self.assertRaisesRegex(PermissionException, '已关闭'),
|
|
):
|
|
await PalletService.add_pallet(self.db, self.create, self.admin, self.scope)
|
|
business_query.assert_not_awaited()
|
|
|
|
async def test_assignment_reads_are_scoped_and_test_writes_lock_before_resource_and_users(self) -> None:
|
|
for resource, resource_id in (('warehouse', 7), ('site', 9)):
|
|
get_resource = AsyncMock(return_value=SimpleNamespace(test_batch_id=9))
|
|
list_users = AsyncMock(return_value=[])
|
|
list_assigned = AsyncMock(return_value=[])
|
|
with (
|
|
self.subTest(resource=resource, operation='read'),
|
|
patch.object(MasterDao, f'get_{resource}', new=get_resource),
|
|
patch.object(MasterDao, 'list_assignable_users', new=list_users),
|
|
patch.object(MasterDao, f'list_{resource}_assigned_user_ids', new=list_assigned),
|
|
):
|
|
await getattr(PalletService, f'list_{resource}_users')(
|
|
self.db,
|
|
resource_id,
|
|
self.scope,
|
|
)
|
|
get_resource.assert_awaited_once_with(self.db, resource_id, self.scope)
|
|
|
|
events: list[str] = []
|
|
lock_batch = AsyncMock(side_effect=lambda *_args, _events=events: _events.append('lock-batch'))
|
|
get_resource = AsyncMock(
|
|
side_effect=lambda *_args, _events=events, **_kwargs: (
|
|
_events.append('get-resource') or SimpleNamespace(test_batch_id=9)
|
|
)
|
|
)
|
|
list_users = AsyncMock(side_effect=lambda *_args, _events=events: _events.append('list-users') or [])
|
|
replace_users = AsyncMock(side_effect=lambda *_args, _events=events: _events.append('replace-users'))
|
|
with (
|
|
self.subTest(resource=resource, operation='write'),
|
|
patch.object(TestBatchAccessService, 'lock_active_batch', new=lock_batch),
|
|
patch.object(MasterDao, f'get_{resource}', new=get_resource),
|
|
patch.object(MasterDao, 'list_assignable_users_by_ids', new=list_users),
|
|
patch.object(AccessDao, f'replace_{resource}_users', new=replace_users),
|
|
):
|
|
await getattr(PalletService, f'assign_{resource}_users')(
|
|
self.db,
|
|
resource_id,
|
|
UserAssignModel(userIds=[]),
|
|
self.admin,
|
|
self.scope,
|
|
)
|
|
self.assertEqual('lock-batch', events[0])
|
|
get_resource.assert_awaited_once_with(
|
|
self.db,
|
|
resource_id,
|
|
self.scope,
|
|
for_update=True,
|
|
)
|
|
|
|
async def test_regular_user_cannot_assign_test_resource_users(self) -> None:
|
|
lock_batch = AsyncMock()
|
|
get_resource = AsyncMock()
|
|
with (
|
|
patch.object(TestBatchAccessService, 'lock_active_batch', new=lock_batch),
|
|
patch.object(MasterDao, 'get_warehouse', new=get_resource),
|
|
self.assertRaisesRegex(PermissionException, '仅超级管理员'),
|
|
):
|
|
await PalletService.assign_warehouse_users(
|
|
self.db,
|
|
7,
|
|
UserAssignModel(userIds=[]),
|
|
make_current_user(8),
|
|
self.scope,
|
|
)
|
|
lock_batch.assert_not_awaited()
|
|
get_resource.assert_not_awaited()
|
|
|
|
async def test_legacy_pallet_delete_is_rejected_before_any_delete_dao(self) -> None:
|
|
lock_inventory = AsyncMock()
|
|
delete_inventory = AsyncMock()
|
|
delete_pallet = AsyncMock()
|
|
with (
|
|
patch.object(
|
|
PalletService,
|
|
'_lock_pallet_inventory',
|
|
new=lock_inventory,
|
|
),
|
|
patch.object(MasterDao, 'delete_pallet_inventory', new=delete_inventory, create=True),
|
|
patch.object(MasterDao, 'delete_pallet', new=delete_pallet, create=True),
|
|
self.assertRaisesRegex(ServiceWarning, '不支持单条物理删除，请通过测试数据批次清理'),
|
|
):
|
|
await PalletService.remove_pallet(self.db, DeletePalletModel(palletIds=[4]))
|
|
|
|
lock_inventory.assert_not_awaited()
|
|
delete_inventory.assert_not_awaited()
|
|
delete_pallet.assert_not_awaited()
|
|
self.db.commit.assert_not_awaited()
|
|
|
|
|
|
class TestPalletDaoBatchScope(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self) -> None:
|
|
self.db = MagicMock(spec=AsyncSession)
|
|
|
|
async def test_pallet_list_detail_and_inventory_queries_apply_scope(self) -> None:
|
|
query = PalletPageQueryModel()
|
|
for scope, expected in (
|
|
(TestBatchScope.formal(), 'test_batch_id IS NULL'),
|
|
(make_test_scope(9), 'test_batch_id = 9'),
|
|
):
|
|
with (
|
|
self.subTest(scope=scope.mode, operation='list'),
|
|
patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=[])) as paginate,
|
|
):
|
|
await MasterDao.list_pallet(self.db, query, scope, is_page=False)
|
|
sql = compile_sql(paginate.await_args.args[1])
|
|
self.assertIn(f'biz_pallet.{expected}', sql)
|
|
|
|
for method_name, table in (
|
|
('get_pallet', 'biz_pallet'),
|
|
('get_pallet_inventory', 'biz_pallet_inventory'),
|
|
):
|
|
result = MagicMock()
|
|
result.scalar_one_or_none.return_value = None
|
|
self.db.execute = AsyncMock(return_value=result)
|
|
with self.subTest(scope=scope.mode, operation=method_name):
|
|
await getattr(MasterDao, method_name)(self.db, 7, scope, for_update=True)
|
|
sql = compile_sql(self.db.execute.await_args.args[0])
|
|
self.assertIn(f'{table}.{expected}', sql)
|
|
self.assertIn('FOR UPDATE', sql)
|
|
|
|
async def test_pallet_update_is_scoped_and_never_updates_batch_id(self) -> None:
|
|
edit = PalletEditModel(palletId=7, capacityPackages=20, assetStatus='0')
|
|
await MasterDao.update_pallet(
|
|
self.db,
|
|
edit,
|
|
make_test_scope(9),
|
|
update_by='admin',
|
|
update_time=datetime(2026, 9, 22),
|
|
)
|
|
sql = compile_sql(self.db.execute.await_args.args[0])
|
|
self.assertIn('biz_pallet.test_batch_id = 9', sql)
|
|
self.assertNotIn('SET test_batch_id=', sql)
|
|
|
|
|
|
class TestPalletAndAssignmentControllers(unittest.IsolatedAsyncioTestCase):
|
|
@staticmethod
|
|
def _routers() -> tuple[Any, ...]:
|
|
return (
|
|
pallet_controller_module.pallet_controller,
|
|
warehouse_controller_module.warehouse_user_controller,
|
|
site_controller_module.site_user_controller,
|
|
)
|
|
|
|
@classmethod
|
|
def _routes(cls) -> dict[tuple[str, str], APIRoute]:
|
|
routes = {}
|
|
for router in cls._routers():
|
|
for route in router.routes:
|
|
if isinstance(route, APIRoute):
|
|
routes[(route.path, next(iter(route.methods)))] = route
|
|
return routes
|
|
|
|
def test_openapi_has_all_pallet_and_assignment_operations(self) -> None:
|
|
app = FastAPI()
|
|
for router in self._routers():
|
|
app.include_router(router)
|
|
schema = app.openapi()
|
|
|
|
expected_operations = {
|
|
('/master/pallet/list', 'get'),
|
|
('/master/pallet/{pallet_id}', 'get'),
|
|
('/master/pallet', 'post'),
|
|
('/master/pallet', 'put'),
|
|
('/master/pallet/{pallet_ids}', 'delete'),
|
|
('/master/pallet/{pallet_id}/qrcode', 'get'),
|
|
('/master/warehouse/{warehouse_id}/users', 'get'),
|
|
('/master/warehouse/{warehouse_id}/users', 'put'),
|
|
('/master/site/{site_id}/users', 'get'),
|
|
('/master/site/{site_id}/users', 'put'),
|
|
}
|
|
for path, method in expected_operations:
|
|
with self.subTest(path=path, method=method):
|
|
self.assertIn(path, schema['paths'])
|
|
self.assertIn(method, schema['paths'][path])
|
|
self.assertIn(
|
|
'image/png',
|
|
schema['paths']['/master/pallet/{pallet_id}/qrcode']['get']['responses']['200']['content'],
|
|
)
|
|
|
|
def test_legacy_delete_openapi_declares_fixed_rejection(self) -> None:
|
|
app = FastAPI()
|
|
for router in self._routers():
|
|
app.include_router(router)
|
|
operation = app.openapi()['paths']['/master/pallet/{pallet_ids}']['delete']
|
|
|
|
self.assertEqual('兼容旧接口，固定拒绝，请使用测试数据批次清理', operation['summary'])
|
|
self.assertEqual(
|
|
'兼容旧接口，固定拒绝单条物理删除，请使用测试数据批次清理',
|
|
operation['description'],
|
|
)
|
|
self.assertNotIn('用于删除', operation['description'])
|
|
self.assertNotIn('尚未入库', operation['description'])
|
|
self.assertNotIn('没有流转事件', operation['description'])
|
|
|
|
def test_route_response_models_are_exact(self) -> None:
|
|
routes = self._routes()
|
|
self.assertEqual(
|
|
PageResponseModel[PalletModel],
|
|
routes[('/master/pallet/list', 'GET')].response_model,
|
|
)
|
|
self.assertEqual(
|
|
DataResponseModel[PalletModel],
|
|
routes[('/master/pallet/{pallet_id}', 'GET')].response_model,
|
|
)
|
|
for path, method in (
|
|
('/master/pallet', 'POST'),
|
|
('/master/pallet', 'PUT'),
|
|
('/master/pallet/{pallet_ids}', 'DELETE'),
|
|
('/master/warehouse/{warehouse_id}/users', 'PUT'),
|
|
('/master/site/{site_id}/users', 'PUT'),
|
|
):
|
|
self.assertEqual(ResponseBaseModel, routes[(path, method)].response_model)
|
|
user_response = DataResponseModel[list[BusinessUserOptionModel]]
|
|
self.assertEqual(
|
|
user_response,
|
|
routes[('/master/warehouse/{warehouse_id}/users', 'GET')].response_model,
|
|
)
|
|
self.assertEqual(
|
|
user_response,
|
|
routes[('/master/site/{site_id}/users', 'GET')].response_model,
|
|
)
|
|
self.assertIsNone(routes[('/master/pallet/{pallet_id}/qrcode', 'GET')].response_model)
|
|
|
|
def test_permissions_are_exact_and_all_routes_require_pre_auth(self) -> None:
|
|
expected_permissions = {
|
|
('/master/pallet/list', 'GET'): 'master:pallet:list',
|
|
('/master/pallet/{pallet_id}', 'GET'): 'master:pallet:query',
|
|
('/master/pallet', 'POST'): 'master:pallet:add',
|
|
('/master/pallet', 'PUT'): 'master:pallet:edit',
|
|
('/master/pallet/{pallet_ids}', 'DELETE'): 'master:pallet:remove',
|
|
('/master/pallet/{pallet_id}/qrcode', 'GET'): 'master:pallet:print',
|
|
('/master/warehouse/{warehouse_id}/users', 'GET'): 'master:warehouse:assign',
|
|
('/master/warehouse/{warehouse_id}/users', 'PUT'): 'master:warehouse:assign',
|
|
('/master/site/{site_id}/users', 'GET'): 'master:site:assign',
|
|
('/master/site/{site_id}/users', 'PUT'): 'master:site:assign',
|
|
}
|
|
|
|
actual_permissions = {}
|
|
for key, route in self._routes().items():
|
|
dependencies = [
|
|
dependency.dependency
|
|
for dependency in route.dependencies
|
|
if isinstance(dependency.dependency, CheckUserInterfaceAuth)
|
|
]
|
|
self.assertEqual(1, len(dependencies))
|
|
self.assertTrue(any(isinstance(dependency.dependency, PreAuth) for dependency in route.dependencies))
|
|
actual_permissions[key] = dependencies[0].perm
|
|
|
|
self.assertEqual(expected_permissions, actual_permissions)
|
|
self.assertEqual(8, len(set(actual_permissions.values())))
|
|
|
|
def test_write_routes_have_log_and_current_user(self) -> None:
|
|
write_routes = (
|
|
('/master/pallet', 'POST'),
|
|
('/master/pallet', 'PUT'),
|
|
('/master/pallet/{pallet_ids}', 'DELETE'),
|
|
('/master/warehouse/{warehouse_id}/users', 'PUT'),
|
|
('/master/site/{site_id}/users', 'PUT'),
|
|
)
|
|
routes = self._routes()
|
|
for key in write_routes:
|
|
route = routes[key]
|
|
self.assertTrue(hasattr(route.endpoint, '__wrapped__'))
|
|
self.assertTrue(
|
|
any(
|
|
dependency.name == 'current_user' and dependency.call == LoginService.get_current_user
|
|
for dependency in route.dependant.dependencies
|
|
)
|
|
)
|
|
|
|
def test_all_non_delete_routes_have_matching_batch_scope_dependency(self) -> None:
|
|
for (path, method), route in self._routes().items():
|
|
resolvers = [
|
|
dependency.call
|
|
for dependency in route.dependant.dependencies
|
|
if isinstance(dependency.call, TestBatchScopeResolver)
|
|
]
|
|
with self.subTest(path=path, method=method):
|
|
if method == 'DELETE':
|
|
self.assertEqual([], resolvers)
|
|
else:
|
|
self.assertEqual(1, len(resolvers))
|
|
self.assertIs(method in {'POST', 'PUT'}, resolvers[0].write)
|
|
|
|
def test_qrcode_route_is_declared_before_dynamic_detail(self) -> None:
|
|
route_paths = [route.path for route in pallet_controller_module.pallet_controller.routes]
|
|
self.assertLess(
|
|
route_paths.index('/master/pallet/{pallet_id}/qrcode'),
|
|
route_paths.index('/master/pallet/{pallet_id}'),
|
|
)
|
|
|
|
async def test_qrcode_endpoint_returns_real_image_png_body(self) -> None:
|
|
png = PalletService.render_qr_png('persisted-token')
|
|
db = MagicMock(spec=AsyncSession)
|
|
scope = TestBatchScope.formal()
|
|
with patch.object(PalletService, 'get_pallet_qrcode', new=AsyncMock(return_value=png)) as get_qrcode:
|
|
response = await pallet_controller_module.query_master_pallet_qrcode(
|
|
request=MagicMock(),
|
|
pallet_id=7,
|
|
query_db=db,
|
|
scope=scope,
|
|
)
|
|
|
|
get_qrcode.assert_awaited_once_with(db, 7, scope)
|
|
self.assertEqual('image/png', response.media_type)
|
|
self.assertEqual(b'\x89PNG\r\n\x1a\n', response.body[:8])
|
|
|
|
def test_controller_sources_do_not_require_system_user_list_or_cache(self) -> None:
|
|
sources = '\n'.join(
|
|
inspect.getsource(module)
|
|
for module in (
|
|
pallet_controller_module,
|
|
warehouse_controller_module,
|
|
site_controller_module,
|
|
)
|
|
)
|
|
self.assertNotIn('system:user:list', sources)
|
|
self.assertNotIn('ApiCache', sources)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|