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_summary_vo import PalletDetailModel, PalletListItemModel 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 = 100, 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': 100, '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=100, 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=100, 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=100', 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=100, 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(100, 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=100, 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, capacity_packages=11)), ), 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=100, 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=100) 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=100, 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=100, 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[PalletListItemModel], routes[('/master/pallet/list', 'GET')].response_model, ) self.assertEqual( DataResponseModel[PalletDetailModel], 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:list', ('/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(7, 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()