From cda87a4d656a0db740a6713594c8b5ecf9ccbf83 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=84=A6=E9=BE=99=E8=A8=80?= Date: Sun, 26 Jul 2026 22:34:46 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E6=B6=88=E9=99=A4?= =?UTF-8?q?=E4=BD=BF=E7=94=A8=E5=B9=82=E7=AD=89=E7=BC=BA=E5=A4=B1=E9=94=AE?= =?UTF-8?q?=E9=97=B4=E9=9A=99=E9=94=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../module_business/dao/operation_dao.py | 9 +- .../module_business/service/usage_service.py | 16 +- .../module_business/test_inventory_dao.py | 36 +- .../module_business/test_usage_service.py | 330 +++++++++++++++--- 4 files changed, 291 insertions(+), 100 deletions(-) diff --git a/shierfeng-fastapi-backend/module_business/dao/operation_dao.py b/shierfeng-fastapi-backend/module_business/dao/operation_dao.py index 59c5a14..ef329b2 100644 --- a/shierfeng-fastapi-backend/module_business/dao/operation_dao.py +++ b/shierfeng-fastapi-backend/module_business/dao/operation_dao.py @@ -132,12 +132,8 @@ class OperationDao: key_column: InstrumentedAttribute, idempotency_key: str, *extra_conditions: ColumnElement[bool], - for_update: bool = False, ) -> Any | None: - statement = select(model).where(key_column == idempotency_key, *extra_conditions) - if for_update: - statement = statement.with_for_update().execution_options(populate_existing=True) - result = await db.execute(statement) + result = await db.execute(select(model).where(key_column == idempotency_key, *extra_conditions)) return result.scalar_one_or_none() @staticmethod @@ -377,8 +373,6 @@ class OperationDao: db: AsyncSession, idempotency_key: str, site_ids: list[int] | None = None, - *, - for_update: bool = False, ) -> BizUsageRecord | None: return await cls._get_by_key( db, @@ -386,7 +380,6 @@ class OperationDao: BizUsageRecord.idempotency_key, idempotency_key, *cls._usage_scope_conditions(site_ids), - for_update=for_update, ) @classmethod diff --git a/shierfeng-fastapi-backend/module_business/service/usage_service.py b/shierfeng-fastapi-backend/module_business/service/usage_service.py index 7ba1861..2862375 100644 --- a/shierfeng-fastapi-backend/module_business/service/usage_service.py +++ b/shierfeng-fastapi-backend/module_business/service/usage_service.py @@ -169,7 +169,6 @@ class UsageService: query_db, page_object.idempotency_key, site_ids=site_ids, - for_update=False, ) if recovered is None: raise ServiceException(message='使用失败，数据冲突') @@ -340,17 +339,16 @@ class UsageService: try: request_hash = cls._build_request_hash(page_object) user_id, user_name = cls._current_identity(current_user) - site_ids = await cls._site_scope(query_db, current_user) + preflight_site_ids = await cls._site_scope(query_db, current_user) existing = await OperationDao.get_usage_by_idempotency_key( query_db, page_object.idempotency_key, - site_ids=site_ids, - for_update=False, + site_ids=preflight_site_ids, ) if existing is not None: if not cls._is_exact_idempotent_result(existing, page_object, request_hash): raise ServiceException(message='使用失败，幂等键已被其他请求使用') - if site_ids is not None and existing.site_id not in site_ids: + if preflight_site_ids is not None and existing.site_id not in preflight_site_ids: cls._raise_usage_not_accessible(current_user) await cls._lock_and_reauthorize_usage_site( query_db, @@ -367,10 +365,12 @@ class UsageService: ) if scanned_pallet is None: cls._raise_pallet_not_accessible(current_user) + scanned_pallet_id = int(scanned_pallet.pallet_id) + await query_db.rollback() inventory = await InventoryDao.get_inventory_for_update( query_db, - scanned_pallet.pallet_id, + scanned_pallet_id, ) if inventory is None: cls._raise_pallet_not_accessible(current_user, '托盘库存不存在') @@ -378,6 +378,7 @@ class UsageService: cls._raise_pallet_not_accessible(current_user, '托盘不在客户现场') expected_site_id = inventory.site_id await cls._assert_pallet_site_access(query_db, current_user, inventory.site_id) + locked_site_ids = await cls._site_scope(query_db, current_user) locked_pallets = await InventoryDao.get_pallets_by_ids( query_db, @@ -398,8 +399,7 @@ class UsageService: locked_existing = await OperationDao.get_usage_by_idempotency_key( query_db, page_object.idempotency_key, - site_ids=site_ids, - for_update=True, + site_ids=locked_site_ids, ) if locked_existing is not None: if locked_existing.site_id != inventory.site_id or not cls._is_exact_idempotent_result( diff --git a/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py b/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py index c1d9f49..2202139 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py @@ -133,12 +133,7 @@ class TestDaoPublicContract(unittest.TestCase): 'site_ids', ), 'get_filling_by_idempotency_key': ('db', 'idempotency_key', 'warehouse_ids'), - 'get_usage_by_idempotency_key': ( - 'db', - 'idempotency_key', - 'site_ids', - 'for_update', - ), + 'get_usage_by_idempotency_key': ('db', 'idempotency_key', 'site_ids'), 'get_return_by_request_idempotency_key': ('db', 'idempotency_key'), 'get_return_by_confirm_idempotency_key': ('db', 'idempotency_key'), 'get_inbound_by_id': ('db', 'inbound_id', 'warehouse_ids'), @@ -230,12 +225,7 @@ class TestDaoPublicContract(unittest.TestCase): signature = inspect.signature(getattr(OperationDao, method_name)) self.assertIsNone(signature.parameters['site_ids'].default) - usage_key_signature = inspect.signature(OperationDao.get_usage_by_idempotency_key) - usage_key_lock = usage_key_signature.parameters.get('for_update') - if usage_key_lock is None: - self.fail('get_usage_by_idempotency_key 缺少 for_update 参数') - self.assertIs(inspect.Parameter.KEYWORD_ONLY, usage_key_lock.kind) - self.assertFalse(usage_key_lock.default) + self.assertNotIn('for_update', inspect.signature(OperationDao._get_by_key).parameters) return_list_signature = inspect.signature(OperationDao.get_return_list) self.assertEqual(('db', 'query_object', 'is_page'), tuple(return_list_signature.parameters)) @@ -955,11 +945,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase): self.assertNotIn('site_id IN', sql) self.assertNotIn('WHERE false', sql) - async def test_usage_idempotency_current_read_locks_refreshes_and_preserves_scope(self) -> None: - self.assertIn( - 'for_update', - inspect.signature(OperationDao.get_usage_by_idempotency_key).parameters, - ) + async def test_usage_idempotency_reads_never_lock_and_preserve_scope(self) -> None: scalar_result = MagicMock() scalar_result.scalar_one_or_none.return_value = None self.db.execute.return_value = scalar_result @@ -977,14 +963,13 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase): self.db, 'usage-key', site_ids=site_ids, - for_update=True, ) statement = self.db.execute.await_args.args[0] - self.assertTrue(statement.get_execution_options().get('populate_existing')) + self.assertFalse(statement.get_execution_options().get('populate_existing', False)) for dialect in (mysql.dialect(), postgresql.dialect()): sql = compile_sql(statement, dialect) - self.assertIn('FOR UPDATE', sql) + self.assertNotIn('FOR UPDATE', sql) if expected_scope is not None: self.assertIn(expected_scope, sql) elif site_ids == []: @@ -992,17 +977,6 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase): else: self.assertNotIn('site_id IN', sql) - self.db.reset_mock() - self.db.execute.return_value = scalar_result - await OperationDao.get_usage_by_idempotency_key( - self.db, - 'usage-key', - site_ids=[9], - ) - unlocked_statement = self.db.execute.await_args.args[0] - self.assertNotIn('FOR UPDATE', compile_sql(unlocked_statement, mysql.dialect())) - self.assertFalse(unlocked_statement.get_execution_options().get('populate_existing', False)) - def test_time_conditions_include_fractional_last_second_with_exclusive_next_day(self) -> None: lower, upper = OperationDao._time_conditions( BizUsageRecord.occurred_at, diff --git a/shierfeng-fastapi-backend/tests/module_business/test_usage_service.py b/shierfeng-fastapi-backend/tests/module_business/test_usage_service.py index 3ee75e0..885c5de 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_usage_service.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_usage_service.py @@ -6,7 +6,7 @@ import json import unittest from contextlib import ExitStack from datetime import datetime -from typing import Any, get_type_hints +from typing import Any, NoReturn, get_type_hints from unittest.mock import AsyncMock, MagicMock, patch from fastapi import FastAPI @@ -610,6 +610,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): *, site_scope: list[int] | None | AsyncMock = None, key_read: BizUsageRecord | None | AsyncMock = None, + scanned_pallet: Any | None = None, pallet: BizPallet | None = None, inventory: BizPalletInventory | None = None, filling: BizFillingRecord | None = None, @@ -627,6 +628,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): scope_read = AsyncMock(return_value=[9] if site_scope is None else site_scope) usage_key_read = key_read if isinstance(key_read, AsyncMock) else AsyncMock(return_value=key_read) selected_pallet = pallet or self.pallet + selected_scanned_pallet = selected_pallet if scanned_pallet is None else scanned_pallet selected_inventory = inventory or self.inventory selected_filling = filling or self.filling selected_site = self.site if site is None else site @@ -640,6 +642,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): usage_write = add_usage or AsyncMock(side_effect=assign_usage_id) filling_write = update_filling or AsyncMock() event_write = add_event or AsyncMock() + scan_pallet = AsyncMock(return_value=selected_scanned_pallet) inventory_lock = AsyncMock(return_value=selected_inventory) pallet_lock = AsyncMock(return_value=[selected_pallet]) filling_lock = AsyncMock(return_value=selected_filling) @@ -649,9 +652,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): stack.enter_context(patch.object(AccessDao, 'list_user_site_ids', new=scope_read)) stack.enter_context(patch.object(AccessService, 'assert_site_access', new=site_access)) stack.enter_context(patch.object(OperationDao, 'get_usage_by_idempotency_key', new=usage_key_read)) - stack.enter_context( - patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=selected_pallet)) - ) + stack.enter_context(patch.object(InventoryDao, 'get_pallet_by_qr_token', new=scan_pallet)) stack.enter_context(patch.object(InventoryDao, 'get_inventory_for_update', new=inventory_lock)) stack.enter_context(patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_lock)) stack.enter_context(patch.object(OperationDao, 'get_active_filling_for_update', new=filling_lock)) @@ -665,6 +666,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): return stack, { 'scope_read': scope_read, 'key_read': usage_key_read, + 'scan_pallet': scan_pallet, 'access': site_access, 'inventory_lock': inventory_lock, 'pallet_lock': pallet_lock, @@ -681,6 +683,21 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): ) -> None: actions: list[str] = [] captured: dict[str, Any] = {} + scope_call_count = 0 + key_call_count = 0 + + async def read_site_scope(*_args: Any) -> list[int]: + nonlocal scope_call_count + scope_call_count += 1 + actions.append('preflight-scope' if scope_call_count == 1 else 'fresh-scope') + return [9] + + async def scan_pallet(*_args: Any) -> BizPallet: + actions.append('qr-snapshot-read') + return self.pallet + + async def rollback() -> None: + actions.append('rollback') async def lock_inventory(*_args: Any) -> BizPalletInventory: actions.append('inventory-lock') @@ -711,17 +728,16 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): ) -> None: if for_update: actions.append('site-mapping-lock') + else: + actions.append('site-access') async def get_usage_by_key( - _db: AsyncSession, - _idempotency_key: str, - _site_ids: list[int] | None = None, - *, - for_update: bool = False, + *_args: Any, **_kwargs: Any, ) -> None: - if for_update: - actions.append('usage-current-read') + nonlocal key_call_count + key_call_count += 1 + actions.append('usage-preflight-read' if key_call_count == 1 else 'usage-fresh-read') async def add_usage(_db: AsyncSession, usage: BizUsageRecord) -> BizUsageRecord: actions.append('usage-flush') @@ -745,18 +761,21 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): actions.append('commit') captured['usage'].__dict__.pop('usage_id', None) + self.db.rollback.side_effect = rollback self.db.commit.side_effect = commit + scope_read = AsyncMock(side_effect=read_site_scope) + key_read = AsyncMock(side_effect=get_usage_by_key) access = AsyncMock(side_effect=assert_site_access) update_filling = AsyncMock() with ( - patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object(AccessDao, 'list_user_site_ids', new=scope_read), patch.object(AccessService, 'assert_site_access', new=access), patch.object( OperationDao, 'get_usage_by_idempotency_key', - new=AsyncMock(side_effect=get_usage_by_key), + new=key_read, ), - patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(side_effect=scan_pallet)), patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(side_effect=lock_inventory)), patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=lock_pallets)), patch.object(OperationDao, 'get_active_filling_for_update', new=AsyncMock(side_effect=lock_filling)), @@ -772,10 +791,16 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): self.assertEqual( [ + 'preflight-scope', + 'usage-preflight-read', + 'qr-snapshot-read', + 'rollback', 'inventory-lock', + 'site-access', + 'fresh-scope', 'pallet-lock', 'filling-lock', - 'usage-current-read', + 'usage-fresh-read', 'site-lock', 'site-mapping-lock', 'line-lock', @@ -785,6 +810,11 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): ], actions, ) + self.assertEqual(2, scope_read.await_count) + self.assertEqual( + [{'site_ids': [9]}, {'site_ids': [9]}], + [call.kwargs for call in key_read.await_args_list], + ) self.assertEqual([False, True], [call.kwargs.get('for_update', False) for call in access.await_args_list]) self.assertEqual([9, 9], [call.args[2] for call in access.await_args_list]) self.assertEqual(4, self.inventory.current_line_id) @@ -829,7 +859,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): self.assertEqual('usage-key', event.idempotency_key) self.assertEqual(usage.request_hash, event.request_hash) self.db.commit.assert_awaited_once_with() - self.db.rollback.assert_not_awaited() + self.db.rollback.assert_awaited_once_with() self.assertTrue(result.is_success) self.assertEqual(61, result.result) @@ -869,7 +899,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): self.assertEqual(0, event.after_remaining_packages) self.assertEqual(4, event.after_line_id) self.db.commit.assert_awaited_once_with() - self.db.rollback.assert_not_awaited() + self.db.rollback.assert_awaited_once_with() mocks['add_usage'].assert_awaited_once() async def test_same_line_can_continue_but_different_line_is_rejected(self) -> None: @@ -900,7 +930,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): mocks['add_usage'].assert_not_awaited() mocks['add_event'].assert_not_awaited() self.db.commit.assert_not_awaited() - self.db.rollback.assert_awaited_once() + self.assertEqual(2, self.db.rollback.await_count) async def test_line_must_exist_be_enabled_and_belong_to_locked_site(self) -> None: for line, message in ( @@ -919,7 +949,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): mocks['add_usage'].assert_not_awaited() mocks['add_event'].assert_not_awaited() self.db.commit.assert_not_awaited() - self.db.rollback.assert_awaited_once() + self.assertEqual(2, self.db.rollback.await_count) async def test_only_site_usage_states_and_strict_current_quantity_are_allowed(self) -> None: cases = ( @@ -950,7 +980,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): mocks['add_usage'].assert_not_awaited() mocks['add_event'].assert_not_awaited() self.db.commit.assert_not_awaited() - self.db.rollback.assert_awaited_once() + self.assertEqual(2, self.db.rollback.await_count) async def test_initial_actual_site_denial_hides_qr_before_secondary_locks(self) -> None: access = AsyncMock(side_effect=PermissionException(data='', message='无权访问该客户现场')) @@ -965,7 +995,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): mocks['site_lock'].assert_not_awaited() mocks['line_lock'].assert_not_awaited() mocks['add_usage'].assert_not_awaited() - self.db.rollback.assert_awaited_once() + self.assertEqual(2, self.db.rollback.await_count) async def test_current_mapping_denial_wins_over_disabled_site_without_leaking_status(self) -> None: async def access_check( @@ -988,7 +1018,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): access.assert_awaited_with(self.db, self.user, 9, for_update=True) mocks['line_lock'].assert_not_awaited() mocks['add_usage'].assert_not_awaited() - self.db.rollback.assert_awaited_once() + self.assertEqual(2, self.db.rollback.await_count) async def test_admin_bypasses_mapping_but_gets_precise_site_errors(self) -> None: for site, message in ((None, '客户现场不存在'), (make_site(status='1'), '客户现场已停用')): @@ -1001,7 +1031,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): await UsageService.record_usage_services(self.db, self.page, make_user(1)) self.assertEqual(message, raised.exception.message) mocks['add_usage'].assert_not_awaited() - self.db.rollback.assert_awaited_once() + self.assertEqual(2, self.db.rollback.await_count) async def test_locked_qr_asset_and_active_filling_invariants_fail_before_usage_write(self) -> None: cases = ( @@ -1021,7 +1051,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): mocks['update_filling'].assert_not_awaited() mocks['add_event'].assert_not_awaited() self.db.commit.assert_not_awaited() - self.db.rollback.assert_awaited_once() + self.assertEqual(2, self.db.rollback.await_count) async def test_invalid_locked_inventory_version_fails_before_usage_write(self) -> None: inventory = make_inventory(version=-1) @@ -1032,7 +1062,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): self.assertEqual('托盘库存版本无效', raised.exception.message) mocks['add_usage'].assert_not_awaited() mocks['add_event'].assert_not_awaited() - self.db.rollback.assert_awaited_once() + self.assertEqual(2, self.db.rollback.await_count) async def test_record_hides_missing_and_non_site_qr_but_admin_gets_precise_error(self) -> None: with ( @@ -1061,7 +1091,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase): await UsageService.record_usage_services(self.db, self.page, self.user) self.assertEqual('无权访问该托盘', non_site.exception.message) mocks['pallet_lock'].assert_not_awaited() - self.db.rollback.assert_awaited_once() + self.assertEqual(2, self.db.rollback.await_count) @unittest.skipUnless(UsageService is not None, 'UsageService 尚未实现') @@ -1086,17 +1116,40 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase): *, inventory: BizPalletInventory, recovered: BizUsageRecord, - site_scope: list[int] | None = None, + scope_sequence: tuple[list[int], list[int]] | None = None, site_status: str = '0', revoke_mapping: bool = False, + revoke_initial_access: bool = False, ) -> tuple[ExitStack, dict[str, AsyncMock], list[str]]: actions: list[str] = [] locked_site = make_site(status=site_status) + scopes = scope_sequence or ([9], [9]) + scope_call_count = 0 + key_call_count = 0 + rollback_count = 0 + + async def read_site_scope(*_args: Any) -> list[int]: + nonlocal scope_call_count + scope_call_count += 1 + actions.append('preflight-scope' if scope_call_count == 1 else 'fresh-scope') + return scopes[scope_call_count - 1] async def read_key(*_args: Any, **kwargs: Any) -> BizUsageRecord | None: - for_update = kwargs.get('for_update', False) - actions.append('usage-current-read' if for_update else 'usage-snapshot-read') - return recovered if for_update else None + nonlocal key_call_count + key_call_count += 1 + actions.append('usage-preflight-read' if key_call_count == 1 else 'usage-fresh-read') + if key_call_count == 1: + return None + return recovered if kwargs.get('site_ids') == scopes[1] else None + + async def scan_pallet(*_args: Any) -> BizPallet: + actions.append('qr-snapshot-read') + return self.core.pallet + + async def rollback() -> None: + nonlocal rollback_count + rollback_count += 1 + actions.append('preflight-rollback' if rollback_count == 1 else 'result-rollback') async def lock_inventory(*_args: Any) -> BizPalletInventory: actions.append('inventory-lock') @@ -1119,18 +1172,24 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase): actions.append('site-mapping-lock') if revoke_mapping: raise PermissionException(data='', message='无权访问该客户现场') + else: + actions.append('site-access') + if revoke_initial_access: + raise PermissionException(data='', message='无权访问该客户现场') stack, mocks = self.core._patch_record_path( inventory=inventory, site=locked_site, - site_scope=site_scope, + site_scope=AsyncMock(side_effect=read_site_scope), key_read=AsyncMock(side_effect=read_key), access=AsyncMock(side_effect=check_access), ) + mocks['scan_pallet'].side_effect = scan_pallet mocks['inventory_lock'].side_effect = lock_inventory mocks['pallet_lock'].side_effect = lock_pallet mocks['filling_lock'].side_effect = lock_filling mocks['site_lock'].side_effect = lock_site + self.db.rollback.side_effect = rollback return stack, mocks, actions async def test_exact_retry_rechecks_current_mapping_and_returns_without_mutation(self) -> None: @@ -1156,7 +1215,6 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase): self.db, 'usage-key', site_ids=[9], - for_update=False, ) site_lock.assert_awaited_once_with(self.db, 9, for_update=True) access.assert_awaited_once_with(self.db, self.user, 9, for_update=True) @@ -1254,21 +1312,31 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase): self.fail(f'精确并发重试被旧库存状态提前拒绝：{exc.message}') expected_actions = [ - 'usage-snapshot-read', + 'preflight-scope', + 'usage-preflight-read', + 'qr-snapshot-read', + 'preflight-rollback', 'inventory-lock', + 'site-access', + 'fresh-scope', 'pallet-lock', ] if expect_filling_lock: expected_actions.append('filling-lock') expected_actions.extend( [ - 'usage-current-read', + 'usage-fresh-read', 'site-lock', 'site-mapping-lock', + 'result-rollback', ] ) self.assertEqual(expected_actions, actions) - self.assertEqual([False, True], [call.kwargs['for_update'] for call in key_read.await_args_list]) + self.assertEqual( + [{'site_ids': [9]}, {'site_ids': [9]}], + [call.kwargs for call in key_read.await_args_list], + ) + self.assertEqual(2, mocks['scope_read'].await_count) self.assertTrue(all(call.kwargs['site_ids'] == [9] for call in key_read.await_args_list)) if expect_filling_lock: mocks['filling_lock'].assert_awaited_once() @@ -1280,7 +1348,7 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase): mocks['add_event'].assert_not_awaited() mocks['update_filling'].assert_not_awaited() self.db.commit.assert_not_awaited() - self.db.rollback.assert_awaited_once() + self.assertEqual(2, self.db.rollback.await_count) self.assertEqual( inventory_before, ( @@ -1293,6 +1361,144 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase): ) self.assertEqual(61, result.result) + async def test_concurrent_retry_uses_fresh_scope_granted_during_inventory_wait(self) -> None: + inventory = make_inventory( + current_line_id=4, + active_filling_id=None, + remaining_packages=0, + flow_state=PalletFlowState.AT_SITE_EMPTY.value, + version=4, + ) + recovered = self._exact_usage() + stack, mocks, actions = self._patch_concurrent_retry_path( + inventory=inventory, + recovered=recovered, + scope_sequence=([], [9]), + ) + with stack: + try: + result = await UsageService.record_usage_services(self.db, self.page, self.user) + except ServiceException as exc: + self.fail(f'锁等待期间新增授权后未使用 fresh scope：{exc.message}') + + self.assertEqual( + [ + 'preflight-scope', + 'usage-preflight-read', + 'qr-snapshot-read', + 'preflight-rollback', + 'inventory-lock', + 'site-access', + 'fresh-scope', + 'pallet-lock', + 'usage-fresh-read', + 'site-lock', + 'site-mapping-lock', + 'result-rollback', + ], + actions, + ) + self.assertEqual( + [{'site_ids': []}, {'site_ids': [9]}], + [call.kwargs for call in mocks['key_read'].await_args_list], + ) + mocks['filling_lock'].assert_not_awaited() + mocks['line_lock'].assert_not_awaited() + mocks['add_usage'].assert_not_awaited() + mocks['add_event'].assert_not_awaited() + self.db.commit.assert_not_awaited() + self.assertEqual(2, self.db.rollback.await_count) + self.assertEqual(61, result.result) + + async def test_preflight_grant_revoked_before_fresh_scope_stays_hidden(self) -> None: + recovered = self._exact_usage() + stack, mocks, actions = self._patch_concurrent_retry_path( + inventory=make_inventory(), + recovered=recovered, + revoke_initial_access=True, + ) + with stack: + try: + await UsageService.record_usage_services(self.db, self.page, self.user) + except PermissionException as exc: + self.assertEqual('无权访问该托盘', exc.message) + else: + self.fail('锁事务开始前已撤权时不得继续 fresh scope 或返回成功') + + self.assertEqual( + [ + 'preflight-scope', + 'usage-preflight-read', + 'qr-snapshot-read', + 'preflight-rollback', + 'inventory-lock', + 'site-access', + 'result-rollback', + ], + actions, + ) + self.assertEqual(1, mocks['scope_read'].await_count) + self.assertEqual(1, mocks['key_read'].await_count) + mocks['pallet_lock'].assert_not_awaited() + mocks['filling_lock'].assert_not_awaited() + mocks['site_lock'].assert_not_awaited() + mocks['add_usage'].assert_not_awaited() + self.db.commit.assert_not_awaited() + self.assertEqual(2, self.db.rollback.await_count) + + async def test_preflight_rollback_expires_scanned_pallet_and_fresh_lock_rechecks_qr(self) -> None: + class ExpiringPalletProbe: + def __init__(self, pallet_id: int) -> None: + self.expired = False + self.pallet_id_reads = 0 + self._pallet_id = pallet_id + + @property + def pallet_id(self) -> int: + if self.expired: + raise AssertionError('rollback 后禁止读取旧 scanned pallet ID') + self.pallet_id_reads += 1 + return self._pallet_id + + def expire(self) -> None: + self.expired = True + + def __getattr__(self, name: str) -> NoReturn: + raise AssertionError(f'预检 scanned pallet 只允许读取纯整数 ID：{name}') + + scanned = ExpiringPalletProbe(pallet_id=2) + fresh_locked_pallet = make_pallet(qr_token='fresh-qr-mismatch') + scope = AsyncMock(side_effect=[[9], [9]]) + key_read = AsyncMock(return_value=None) + stack, mocks = self.core._patch_record_path( + scanned_pallet=scanned, + pallet=fresh_locked_pallet, + site_scope=scope, + key_read=key_read, + ) + + async def rollback() -> None: + scanned.expire() + + self.db.rollback.side_effect = rollback + with stack, self.assertRaises(ServiceException) as raised: + await UsageService.record_usage_services(self.db, self.page, self.user) + + self.assertEqual('托盘二维码与库存不一致', raised.exception.message) + self.assertTrue(scanned.expired) + self.assertEqual(1, scanned.pallet_id_reads) + mocks['inventory_lock'].assert_awaited_once_with(self.db, 2) + mocks['pallet_lock'].assert_awaited_once_with(self.db, [2]) + self.assertEqual(2, scope.await_count) + self.assertEqual( + [{'site_ids': [9]}, {'site_ids': [9]}], + [call.kwargs for call in key_read.await_args_list], + ) + mocks['add_usage'].assert_not_awaited() + mocks['add_event'].assert_not_awaited() + self.db.commit.assert_not_awaited() + self.assertEqual(2, self.db.rollback.await_count) + async def test_concurrent_exact_retry_rechecks_current_mapping_before_success(self) -> None: inventory = make_inventory( current_line_id=4, @@ -1319,7 +1525,11 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase): else: self.fail('current mapping 已撤权时不得返回幂等成功') - self.assertEqual([False, True], [call.kwargs['for_update'] for call in key_read.await_args_list]) + self.assertEqual( + [{'site_ids': [9]}, {'site_ids': [9]}], + [call.kwargs for call in key_read.await_args_list], + ) + self.assertEqual(2, mocks['scope_read'].await_count) mocks['site_lock'].assert_awaited_once_with(self.db, 9, for_update=True) self.assertEqual([False, True], [call.kwargs.get('for_update', False) for call in access.await_args_list]) mocks['line_lock'].assert_not_awaited() @@ -1327,9 +1537,9 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase): mocks['add_event'].assert_not_awaited() mocks['update_filling'].assert_not_awaited() self.db.commit.assert_not_awaited() - self.db.rollback.assert_awaited_once() + self.assertEqual(2, self.db.rollback.await_count) - async def test_concurrent_current_read_rejects_hash_or_actual_site_conflict(self) -> None: + async def test_concurrent_fresh_key_read_rejects_hash_or_actual_site_conflict(self) -> None: request_hash = UsageService._build_request_hash(self.page) conflicts = ( ('different-request-hash', make_usage(request_hash='different-hash')), @@ -1348,7 +1558,7 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase): stack, mocks, _ = self._patch_concurrent_retry_path( inventory=inventory, recovered=recovered, - site_scope=[9, 10], + scope_sequence=([9, 10], [9, 10]), ) key_read = mocks['key_read'] with stack: @@ -1362,7 +1572,11 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase): else: self.fail('锁后命中的摘要或实际现场冲突不得返回成功') - self.assertEqual([False, True], [call.kwargs['for_update'] for call in key_read.await_args_list]) + self.assertEqual( + [{'site_ids': [9, 10]}, {'site_ids': [9, 10]}], + [call.kwargs for call in key_read.await_args_list], + ) + self.assertEqual(2, mocks['scope_read'].await_count) self.assertTrue(all(call.kwargs['site_ids'] == [9, 10] for call in key_read.await_args_list)) mocks['site_lock'].assert_not_awaited() mocks['line_lock'].assert_not_awaited() @@ -1370,11 +1584,11 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase): mocks['add_event'].assert_not_awaited() mocks['update_filling'].assert_not_awaited() self.db.commit.assert_not_awaited() - self.db.rollback.assert_awaited_once() + self.assertEqual(2, self.db.rollback.await_count) async def test_integrity_error_recovers_only_fresh_scoped_exact_result(self) -> None: recovered = self._exact_usage() - scope = AsyncMock(side_effect=[[9], [9]]) + scope = AsyncMock(side_effect=[[9], [9], [9]]) key_read = AsyncMock(side_effect=[None, None, recovered]) add_usage = AsyncMock(side_effect=integrity_error()) stack, mocks = self.core._patch_record_path( @@ -1386,11 +1600,14 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase): result = await UsageService.record_usage_services(self.db, self.page, self.user) self.assertEqual(61, result.result) - self.assertEqual(2, scope.await_count) + self.assertEqual(3, scope.await_count) self.assertEqual(3, key_read.await_count) - self.assertEqual([False, True, False], [call.kwargs['for_update'] for call in key_read.await_args_list]) + self.assertEqual( + [{'site_ids': [9]}, {'site_ids': [9]}, {'site_ids': [9]}], + [call.kwargs for call in key_read.await_args_list], + ) self.assertEqual([9], key_read.await_args_list[2].kwargs['site_ids']) - self.assertEqual(2, self.db.rollback.await_count) + self.assertEqual(3, self.db.rollback.await_count) self.db.commit.assert_not_awaited() mocks['add_event'].assert_not_awaited() @@ -1406,11 +1623,15 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase): with stack, self.assertRaises(ServiceException) as raised: await UsageService.record_usage_services(self.db, self.page, self.user) self.assertTrue('幂等' in raised.exception.message or '冲突' in raised.exception.message) - self.assertEqual(2, self.db.rollback.await_count) + self.assertEqual( + [{'site_ids': [9]}, {'site_ids': [9]}, {'site_ids': [9]}], + [call.kwargs for call in key_read.await_args_list], + ) + self.assertEqual(3, self.db.rollback.await_count) self.db.commit.assert_not_awaited() async def test_integrity_recovery_rechecks_fresh_scope_and_rejects_revoked_site(self) -> None: - scope = AsyncMock(side_effect=[[9], []]) + scope = AsyncMock(side_effect=[[9], [9], []]) key_read = AsyncMock(return_value=None) stack, mocks = self.core._patch_record_path( site_scope=scope, @@ -1420,21 +1641,24 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase): with stack, self.assertRaises(PermissionException): await UsageService.record_usage_services(self.db, self.page, self.user) - self.assertEqual(2, scope.await_count) + self.assertEqual(3, scope.await_count) self.assertEqual(2, key_read.await_count) - self.assertEqual([False, True], [call.kwargs['for_update'] for call in key_read.await_args_list]) - self.assertEqual(2, self.db.rollback.await_count) + self.assertEqual( + [{'site_ids': [9]}, {'site_ids': [9]}], + [call.kwargs for call in key_read.await_args_list], + ) + self.assertEqual(3, self.db.rollback.await_count) mocks['add_event'].assert_not_awaited() self.db.commit.assert_not_awaited() - async def test_unexpected_write_failure_rolls_back_once_and_never_commits(self) -> None: + async def test_unexpected_write_failure_rolls_back_lock_transaction_and_never_commits(self) -> None: stack, _ = self.core._patch_record_path( add_event=AsyncMock(side_effect=RuntimeError('event failed')), ) with stack, self.assertRaises(RuntimeError): await UsageService.record_usage_services(self.db, self.page, self.user) - self.db.rollback.assert_awaited_once() + self.assertEqual(2, self.db.rollback.await_count) self.db.commit.assert_not_awaited() async def test_request_hash_failure_also_rolls_back(self) -> None: