niuniu/ShiErFeng
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复：限定幂等键精确业务域

Browse Source
This commit is contained in:
焦龙言
2026-07-27 18:54:50 +08:00
parent 4a321098f8
commit b6f3b1e1f0
18 changed files with 730 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shierfeng-fastapi-backend/module_business/dao/operation_dao.py
+8 -6
View File
@@ -431,15 +431,16 @@ class OperationDao:
cls,
db: AsyncSession,
idempotency_key: str,
site_ids: list[int] | None = None,
warehouse_ids: list[int] | None = None,
source_site_id: int,
target_warehouse_id: int,
) -> BizReturnOrder | None:
return await cls._get_by_key(
db,
BizReturnOrder,
BizReturnOrder.request_idempotency_key,
idempotency_key,
*cls._return_scope_conditions(site_ids, warehouse_ids),
BizReturnOrder.source_site_id == source_site_id,
BizReturnOrder.target_warehouse_id == target_warehouse_id,
)
@classmethod
@@ -447,15 +448,16 @@ class OperationDao:
cls,
db: AsyncSession,
idempotency_key: str,
site_ids: list[int] | None = None,
warehouse_ids: list[int] | None = None,
source_site_id: int,
target_warehouse_id: int,
) -> BizReturnOrder | None:
return await cls._get_by_key(
db,
BizReturnOrder,
BizReturnOrder.confirm_idempotency_key,
idempotency_key,
*cls._return_scope_conditions(site_ids, warehouse_ids),
BizReturnOrder.source_site_id == source_site_id,
BizReturnOrder.target_warehouse_id == target_warehouse_id,
)
@classmethod
shierfeng-fastapi-backend/module_business/entity/do/operation_do.py
+35 -6
View File
@@ -12,6 +12,11 @@ class BizInboundOrder(Base):
__tablename__ = 'biz_inbound_order'
__table_args__ = (
UniqueConstraint(
'warehouse_id',
'confirm_idempotency_key',
name='uq_biz_inbound_order_warehouse_confirm_idempotency',
),
CheckConstraint("status IN ('DRAFT', 'CONFIRMED')", name='ck_biz_inbound_order_status'),
{'comment': '首次入库单'},
)
@@ -25,7 +30,7 @@ class BizInboundOrder(Base):
comment='仓库ID',
)
status = Column(String(32), nullable=False, server_default='DRAFT', comment='单据状态')
confirm_idempotency_key = Column(String(64), nullable=True, unique=True, comment='确认幂等键')
confirm_idempotency_key = Column(String(64), nullable=True, comment='确认幂等键')
confirm_request_hash = Column(String(64), nullable=True, comment='确认请求摘要')
confirmed_by = Column(BigInteger, ForeignKey('sys_user.user_id'), nullable=True, comment='确认人ID')
confirmed_at = Column(DateTime, nullable=True, comment='确认时间')
@@ -73,6 +78,11 @@ class BizFillingRecord(Base):
__tablename__ = 'biz_filling_record'
__table_args__ = (
UniqueConstraint(
'warehouse_id',
'idempotency_key',
name='uq_biz_filling_record_warehouse_idempotency',
),
UniqueConstraint('pallet_id', 'active_slot', name='uq_biz_filling_record_pallet_active_slot'),
CheckConstraint('initial_packages > 0', name='ck_biz_filling_record_initial_packages'),
CheckConstraint(
@@ -99,7 +109,7 @@ class BizFillingRecord(Base):
initial_packages = Column(Integer, nullable=False, comment='初始包数')
active_slot = Column(String(16), nullable=True, comment='活动装填槽')
closed_at = Column(DateTime, nullable=True, comment='装填周期关闭时间')
idempotency_key = Column(String(64), nullable=False, unique=True, comment='幂等键')
idempotency_key = Column(String(64), nullable=False, comment='幂等键')
request_hash = Column(String(64), nullable=False, comment='请求摘要')
operator_id = Column(BigInteger, ForeignKey('sys_user.user_id'), nullable=False, comment='操作人ID')
operated_at = Column(DateTime, nullable=False, comment='操作时间')
@@ -120,6 +130,12 @@ class BizShipmentOrder(Base):
__tablename__ = 'biz_shipment_order'
__table_args__ = (
UniqueConstraint(
'source_warehouse_id',
'target_site_id',
'confirm_idempotency_key',
name='uq_biz_shipment_order_route_confirm_idempotency',
),
CheckConstraint("status IN ('DRAFT', 'CONFIRMED')", name='ck_biz_shipment_order_status'),
{'comment': '整托发货单'},
)
@@ -139,7 +155,7 @@ class BizShipmentOrder(Base):
comment='目标现场ID',
)
status = Column(String(32), nullable=False, server_default='DRAFT', comment='单据状态')
confirm_idempotency_key = Column(String(64), nullable=True, unique=True, comment='确认幂等键')
confirm_idempotency_key = Column(String(64), nullable=True, comment='确认幂等键')
confirm_request_hash = Column(String(64), nullable=True, comment='确认请求摘要')
confirmed_by = Column(BigInteger, ForeignKey('sys_user.user_id'), nullable=True, comment='确认人ID')
confirmed_at = Column(DateTime, nullable=True, comment='确认时间')
@@ -198,6 +214,7 @@ class BizUsageRecord(Base):
__tablename__ = 'biz_usage_record'
__table_args__ = (
UniqueConstraint('site_id', 'idempotency_key', name='uq_biz_usage_record_site_idempotency'),
CheckConstraint('used_packages > 0', name='ck_biz_usage_record_used_packages'),
CheckConstraint(
'remaining_packages_after >= 0',
@@ -219,7 +236,7 @@ class BizUsageRecord(Base):
line_id = Column(BigInteger, ForeignKey('biz_production_line.line_id'), nullable=False, comment='产线ID')
used_packages = Column(Integer, nullable=False, comment='本次使用包数')
remaining_packages_after = Column(Integer, nullable=False, comment='使用后剩余包数')
idempotency_key = Column(String(64), nullable=False, unique=True, comment='幂等键')
idempotency_key = Column(String(64), nullable=False, comment='幂等键')
request_hash = Column(String(64), nullable=False, comment='请求摘要')
operator_id = Column(BigInteger, ForeignKey('sys_user.user_id'), nullable=False, comment='操作人ID')
occurred_at = Column(DateTime, nullable=False, comment='发生时间')
@@ -230,6 +247,18 @@ class BizReturnOrder(Base):
__tablename__ = 'biz_return_order'
__table_args__ = (
UniqueConstraint(
'source_site_id',
'target_warehouse_id',
'request_idempotency_key',
name='uq_biz_return_order_route_request_idempotency',
),
UniqueConstraint(
'source_site_id',
'target_warehouse_id',
'confirm_idempotency_key',
name='uq_biz_return_order_route_confirm_idempotency',
),
CheckConstraint("status IN ('REQUESTED', 'CONFIRMED')", name='ck_biz_return_order_status'),
{'comment': '托盘退回单'},
)
@@ -249,9 +278,9 @@ class BizReturnOrder(Base):
comment='目标仓库ID',
)
status = Column(String(32), nullable=False, server_default='REQUESTED', comment='退回状态')
request_idempotency_key = Column(String(64), nullable=False, unique=True, comment='申请幂等键')
request_idempotency_key = Column(String(64), nullable=False, comment='申请幂等键')
request_hash = Column(String(64), nullable=False, comment='申请请求摘要')
confirm_idempotency_key = Column(String(64), nullable=True, unique=True, comment='确认幂等键')
confirm_idempotency_key = Column(String(64), nullable=True, comment='确认幂等键')
confirm_request_hash = Column(String(64), nullable=True, comment='确认请求摘要')
requested_by = Column(BigInteger, ForeignKey('sys_user.user_id'), nullable=False, comment='申请人ID')
requested_at = Column(DateTime, nullable=False, comment='申请时间')
shierfeng-fastapi-backend/module_business/service/filling_service.py
+2 -2
View File
@@ -300,7 +300,7 @@ class FillingService:
recovered = await OperationDao.get_filling_by_idempotency_key(
query_db,
page_object.idempotency_key,
warehouse_ids=warehouse_ids,
warehouse_ids=[page_object.warehouse_id],
)
if recovered is None:
raise ServiceException(message='装填失败，数据冲突')
@@ -333,7 +333,7 @@ class FillingService:
existing = await OperationDao.get_filling_by_idempotency_key(
query_db,
page_object.idempotency_key,
warehouse_ids=warehouse_ids,
warehouse_ids=[page_object.warehouse_id],
)
if existing is not None:
await AccessService.assert_warehouse_access(
shierfeng-fastapi-backend/module_business/service/inbound_service.py
+9 -2
View File
@@ -513,10 +513,17 @@ class InboundService:
) -> CrudResponseModel:
try:
warehouse_ids = await cls._warehouse_scope(query_db, current_user)
current = await OperationDao.get_inbound_by_id(
query_db,
inbound_id,
warehouse_ids=warehouse_ids,
)
if current is None:
cls._raise_inbound_not_accessible(current_user)
recovered = await OperationDao.get_inbound_by_confirm_idempotency_key(
query_db,
idempotency_key,
warehouse_ids=warehouse_ids,
warehouse_ids=[current.warehouse_id],
)
if recovered is None:
raise ServiceException(message='确认入库失败，幂等冲突')
@@ -575,7 +582,7 @@ class InboundService:
existing = await OperationDao.get_inbound_by_confirm_idempotency_key(
query_db,
idempotency_key,
warehouse_ids=warehouse_ids,
warehouse_ids=[inbound.warehouse_id],
)
if existing is not None:
await AccessService.assert_warehouse_access(
shierfeng-fastapi-backend/module_business/service/return_service.py
+35 -27
View File
@@ -586,12 +586,14 @@ class ReturnService:
try:
site_ids = await cls._site_scope(query_db, current_user)
request_warehouse_ids = None if site_ids is None else []
recovered = await OperationDao.get_return_by_request_idempotency_key(
query_db,
page_object.idempotency_key,
site_ids=site_ids,
warehouse_ids=request_warehouse_ids,
)
recovered = None
if site_ids is None or page_object.source_site_id in site_ids:
recovered = await OperationDao.get_return_by_request_idempotency_key(
query_db,
page_object.idempotency_key,
source_site_id=page_object.source_site_id,
target_warehouse_id=page_object.target_warehouse_id,
)
if recovered is None or not cls._is_exact_request_result(recovered, page_object, request_hash):
raise ServiceException(message='申请退回失败，幂等冲突')
locked = await OperationDao.get_return_for_update(
@@ -640,8 +642,8 @@ class ReturnService:
existing = await OperationDao.get_return_by_request_idempotency_key(
query_db,
page_object.idempotency_key,
site_ids=preflight_site_ids,
warehouse_ids=request_warehouse_ids,
source_site_id=page_object.source_site_id,
target_warehouse_id=page_object.target_warehouse_id,
)
if existing is not None and not cls._is_exact_request_result(existing, page_object, request_hash):
raise ServiceException(message='申请退回失败，幂等键已被其他请求使用')
@@ -681,12 +683,11 @@ class ReturnService:
fresh_site_ids = await cls._site_scope(query_db, current_user)
if fresh_site_ids is not None and page_object.source_site_id not in fresh_site_ids:
cls._raise_pallet_not_accessible(current_user)
fresh_warehouse_ids = None if fresh_site_ids is None else []
fresh_existing = await OperationDao.get_return_by_request_idempotency_key(
query_db,
page_object.idempotency_key,
site_ids=fresh_site_ids,
warehouse_ids=fresh_warehouse_ids,
source_site_id=page_object.source_site_id,
target_warehouse_id=page_object.target_warehouse_id,
)
if fresh_existing is not None:
if not cls._is_exact_request_result(fresh_existing, page_object, request_hash):
@@ -905,11 +906,19 @@ class ReturnService:
try:
warehouse_ids = await cls._warehouse_scope(query_db, current_user)
confirm_site_ids = None if warehouse_ids is None else []
route = await OperationDao.get_return_by_id(
query_db,
return_id,
site_ids=confirm_site_ids,
warehouse_ids=warehouse_ids,
)
if route is None:
cls._raise_return_not_accessible(current_user)
recovered = await OperationDao.get_return_by_confirm_idempotency_key(
query_db,
page_object.idempotency_key,
site_ids=confirm_site_ids,
warehouse_ids=warehouse_ids,
source_site_id=route.source_site_id,
target_warehouse_id=route.target_warehouse_id,
)
if recovered is None:
raise ServiceException(message='确认退回失败，幂等冲突')
@@ -955,20 +964,6 @@ class ReturnService:
request_hash = cls._build_confirm_request_hash(return_id, page_object)
warehouse_ids = await cls._warehouse_scope(query_db, current_user)
confirm_site_ids = None if warehouse_ids is None else []
existing = await OperationDao.get_return_by_confirm_idempotency_key(
query_db,
page_object.idempotency_key,
site_ids=confirm_site_ids,
warehouse_ids=warehouse_ids,
)
if existing is not None and not cls._is_exact_confirm_result(
existing,
return_id,
page_object,
request_hash,
):
raise ServiceException(message='确认退回失败，幂等键已被其他请求使用')
return_order = await OperationDao.get_return_by_id(
query_db,
return_id,
@@ -985,6 +980,19 @@ class ReturnService:
)
except PermissionException:
cls._raise_return_not_accessible(current_user)
existing = await OperationDao.get_return_by_confirm_idempotency_key(
query_db,
page_object.idempotency_key,
source_site_id=return_order.source_site_id,
target_warehouse_id=return_order.target_warehouse_id,
)
if existing is not None and not cls._is_exact_confirm_result(
existing,
return_id,
page_object,
request_hash,
):
raise ServiceException(message='确认退回失败，幂等键已被其他请求使用')
await query_db.rollback()
locked = await OperationDao.get_return_for_update(
shierfeng-fastapi-backend/module_business/service/shipment_service.py
+12 -4
View File
@@ -750,11 +750,19 @@ class ShipmentService:
) -> CrudResponseModel:
try:
warehouse_ids, site_ids = await cls._scopes(query_db, current_user)
current = await OperationDao.get_shipment_by_id(
query_db,
shipment_id,
warehouse_ids=warehouse_ids,
site_ids=site_ids,
)
if current is None:
cls._raise_shipment_not_accessible(current_user)
recovered = await OperationDao.get_shipment_by_confirm_idempotency_key(
query_db,
idempotency_key,
warehouse_ids=warehouse_ids,
site_ids=site_ids,
warehouse_ids=[current.source_warehouse_id],
site_ids=[current.target_site_id],
)
if recovered is None:
raise ServiceException(message='确认发货失败，幂等冲突')
@@ -815,8 +823,8 @@ class ShipmentService:
existing = await OperationDao.get_shipment_by_confirm_idempotency_key(
query_db,
idempotency_key,
warehouse_ids=warehouse_ids,
site_ids=site_ids,
warehouse_ids=[shipment.source_warehouse_id],
site_ids=[shipment.target_site_id],
)
if existing is not None and not cls._is_exact_confirmed(
existing,
shierfeng-fastapi-backend/module_business/service/usage_service.py
+16 -7
View File
@@ -179,7 +179,7 @@ class UsageService:
recovered = await OperationDao.get_usage_by_idempotency_key(
query_db,
page_object.idempotency_key,
site_ids=site_ids,
site_ids=[expected_site_id],
)
if recovered is None:
raise ServiceException(message='使用失败，数据冲突')
@@ -351,11 +351,18 @@ class UsageService:
request_hash = cls._build_request_hash(page_object)
user_id, user_name = cls._current_identity(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=preflight_site_ids,
)
# 产线归属是请求内稳定路由；锁事务仍以库存中的实际现场复核。
preflight_line = await MasterDao.get_line(query_db, page_object.line_id)
preflight_site_id = preflight_line.site_id if preflight_line is not None else None
existing = None
if preflight_site_id is not None and (
preflight_site_ids is None or preflight_site_id in preflight_site_ids
):
existing = await OperationDao.get_usage_by_idempotency_key(
query_db,
page_object.idempotency_key,
site_ids=[preflight_site_id],
)
if existing is not None:
if not cls._is_exact_idempotent_result(existing, page_object, request_hash):
raise ServiceException(message='使用失败，幂等键已被其他请求使用')
@@ -390,6 +397,8 @@ class UsageService:
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)
if locked_site_ids is not None and inventory.site_id not in locked_site_ids:
cls._raise_pallet_not_accessible(current_user)
locked_pallets = await InventoryDao.get_pallets_by_ids(
query_db,
@@ -410,7 +419,7 @@ class UsageService:
locked_existing = await OperationDao.get_usage_by_idempotency_key(
query_db,
page_object.idempotency_key,
site_ids=locked_site_ids,
site_ids=[inventory.site_id],
)
if locked_existing is not None:
if locked_existing.site_id != inventory.site_id or not cls._is_exact_idempotent_result(
shierfeng-fastapi-backend/sql/patches/20260724_pallet_workflow_mysql.sql
+6 -6
View File
@@ -148,7 +148,7 @@ create table if not exists biz_inbound_order (
remark varchar(500) default null comment '备注',
primary key (inbound_id),
constraint uq_biz_inbound_order_no unique (inbound_no),
constraint uq_biz_inbound_order_confirm_idempotency unique (confirm_idempotency_key),
constraint uq_biz_inbound_order_warehouse_confirm_idempotency unique (warehouse_id, confirm_idempotency_key),
constraint fk_biz_inbound_order_warehouse foreign key (warehouse_id) references biz_warehouse (warehouse_id),
constraint fk_biz_inbound_order_confirmed_by foreign key (confirmed_by) references sys_user (user_id),
constraint ck_biz_inbound_order_status check (status in ('DRAFT', 'CONFIRMED'))
@@ -192,7 +192,7 @@ create table if not exists biz_filling_record (
remark varchar(500) default null comment '备注',
primary key (filling_id),
constraint uq_biz_filling_record_no unique (filling_no),
constraint uq_biz_filling_record_idempotency unique (idempotency_key),
constraint uq_biz_filling_record_warehouse_idempotency unique (warehouse_id, idempotency_key),
constraint uq_biz_filling_record_pallet_active_slot unique (pallet_id, active_slot),
constraint fk_biz_filling_record_pallet foreign key (pallet_id) references biz_pallet (pallet_id),
constraint fk_biz_filling_record_warehouse foreign key (warehouse_id) references biz_warehouse (warehouse_id),
@@ -220,7 +220,7 @@ create table if not exists biz_shipment_order (
remark varchar(500) default null comment '备注',
primary key (shipment_id),
constraint uq_biz_shipment_order_no unique (shipment_no),
constraint uq_biz_shipment_order_confirm_idempotency unique (confirm_idempotency_key),
constraint uq_biz_shipment_order_route_confirm_idempotency unique (source_warehouse_id, target_site_id, confirm_idempotency_key),
constraint fk_biz_shipment_order_warehouse foreign key (source_warehouse_id) references biz_warehouse (warehouse_id),
constraint fk_biz_shipment_order_site foreign key (target_site_id) references biz_customer_site (site_id),
constraint fk_biz_shipment_order_confirmed_by foreign key (confirmed_by) references sys_user (user_id),
@@ -260,7 +260,7 @@ create table if not exists biz_usage_record (
occurred_at datetime not null comment '发生时间',
primary key (usage_id),
constraint uq_biz_usage_record_no unique (usage_no),
constraint uq_biz_usage_record_idempotency unique (idempotency_key),
constraint uq_biz_usage_record_site_idempotency unique (site_id, idempotency_key),
constraint fk_biz_usage_record_pallet foreign key (pallet_id) references biz_pallet (pallet_id),
constraint fk_biz_usage_record_filling foreign key (active_filling_id) references biz_filling_record (filling_id),
constraint fk_biz_usage_record_site foreign key (site_id) references biz_customer_site (site_id),
@@ -292,8 +292,8 @@ create table if not exists biz_return_order (
remark varchar(500) default null comment '备注',
primary key (return_id),
constraint uq_biz_return_order_no unique (return_no),
constraint uq_biz_return_order_request_idempotency unique (request_idempotency_key),
constraint uq_biz_return_order_confirm_idempotency unique (confirm_idempotency_key),
constraint uq_biz_return_order_route_request_idempotency unique (source_site_id, target_warehouse_id, request_idempotency_key),
constraint uq_biz_return_order_route_confirm_idempotency unique (source_site_id, target_warehouse_id, confirm_idempotency_key),
constraint fk_biz_return_order_site foreign key (source_site_id) references biz_customer_site (site_id),
constraint fk_biz_return_order_warehouse foreign key (target_warehouse_id) references biz_warehouse (warehouse_id),
constraint fk_biz_return_order_requested_by foreign key (requested_by) references sys_user (user_id),
shierfeng-fastapi-backend/sql/shierfeng-fastapi-pg.sql
+6 -6
View File
@@ -1211,7 +1211,7 @@ create table biz_inbound_order (
remark varchar(500) default null,
primary key (inbound_id),
constraint uq_biz_inbound_order_no unique (inbound_no),
constraint uq_biz_inbound_order_confirm_key unique (confirm_idempotency_key),
constraint uq_biz_inbound_order_warehouse_confirm_idempotency unique (warehouse_id, confirm_idempotency_key),
constraint ck_biz_inbound_order_status check (status in ('DRAFT', 'CONFIRMED')),
constraint fk_biz_inbound_order_warehouse foreign key (warehouse_id) references biz_warehouse(warehouse_id),
constraint fk_biz_inbound_order_confirmer foreign key (confirmed_by) references sys_user(user_id)
@@ -1253,7 +1253,7 @@ create table biz_filling_record (
remark varchar(500) default null,
primary key (filling_id),
constraint uq_biz_filling_record_no unique (filling_no),
constraint uq_biz_filling_record_idempotency unique (idempotency_key),
constraint uq_biz_filling_record_warehouse_idempotency unique (warehouse_id, idempotency_key),
constraint uq_biz_filling_record_pallet_active_slot unique (pallet_id, active_slot),
constraint ck_biz_filling_record_initial_packages check (initial_packages > 0),
constraint ck_biz_filling_record_active_slot check (active_slot is null or active_slot = 'ACTIVE'),
@@ -1280,7 +1280,7 @@ create table biz_shipment_order (
remark varchar(500) default null,
primary key (shipment_id),
constraint uq_biz_shipment_order_no unique (shipment_no),
constraint uq_biz_shipment_order_confirm_key unique (confirm_idempotency_key),
constraint uq_biz_shipment_order_route_confirm_idempotency unique (source_warehouse_id, target_site_id, confirm_idempotency_key),
constraint ck_biz_shipment_order_status check (status in ('DRAFT', 'CONFIRMED')),
constraint fk_biz_shipment_order_warehouse foreign key (source_warehouse_id) references biz_warehouse(warehouse_id),
constraint fk_biz_shipment_order_site foreign key (target_site_id) references biz_customer_site(site_id),
@@ -1318,7 +1318,7 @@ create table biz_usage_record (
occurred_at timestamp without time zone not null,
primary key (usage_id),
constraint uq_biz_usage_record_no unique (usage_no),
constraint uq_biz_usage_record_idempotency unique (idempotency_key),
constraint uq_biz_usage_record_site_idempotency unique (site_id, idempotency_key),
constraint ck_biz_usage_record_used_packages check (used_packages > 0),
constraint ck_biz_usage_record_remaining_packages_after check (remaining_packages_after >= 0),
constraint fk_biz_usage_record_pallet foreign key (pallet_id) references biz_pallet(pallet_id),
@@ -1349,8 +1349,8 @@ create table biz_return_order (
remark varchar(500) default null,
primary key (return_id),
constraint uq_biz_return_order_no unique (return_no),
constraint uq_biz_return_order_request_key unique (request_idempotency_key),
constraint uq_biz_return_order_confirm_key unique (confirm_idempotency_key),
constraint uq_biz_return_order_route_request_idempotency unique (source_site_id, target_warehouse_id, request_idempotency_key),
constraint uq_biz_return_order_route_confirm_idempotency unique (source_site_id, target_warehouse_id, confirm_idempotency_key),
constraint ck_biz_return_order_status check (status in ('REQUESTED', 'CONFIRMED')),
constraint fk_biz_return_order_site foreign key (source_site_id) references biz_customer_site(site_id),
constraint fk_biz_return_order_warehouse foreign key (target_warehouse_id) references biz_warehouse(warehouse_id),
shierfeng-fastapi-backend/sql/shierfeng-fastapi.sql
+6 -6
View File
@@ -955,7 +955,7 @@ create table biz_inbound_order (
remark varchar(500) default null comment '备注',
primary key (inbound_id),
constraint uq_biz_inbound_order_no unique (inbound_no),
constraint uq_biz_inbound_order_confirm_idempotency unique (confirm_idempotency_key),
constraint uq_biz_inbound_order_warehouse_confirm_idempotency unique (warehouse_id, confirm_idempotency_key),
constraint fk_biz_inbound_order_warehouse foreign key (warehouse_id) references biz_warehouse (warehouse_id),
constraint fk_biz_inbound_order_confirmed_by foreign key (confirmed_by) references sys_user (user_id),
constraint ck_biz_inbound_order_status check (status in ('DRAFT', 'CONFIRMED'))
@@ -999,7 +999,7 @@ create table biz_filling_record (
remark varchar(500) default null comment '备注',
primary key (filling_id),
constraint uq_biz_filling_record_no unique (filling_no),
constraint uq_biz_filling_record_idempotency unique (idempotency_key),
constraint uq_biz_filling_record_warehouse_idempotency unique (warehouse_id, idempotency_key),
constraint uq_biz_filling_record_pallet_active_slot unique (pallet_id, active_slot),
constraint fk_biz_filling_record_pallet foreign key (pallet_id) references biz_pallet (pallet_id),
constraint fk_biz_filling_record_warehouse foreign key (warehouse_id) references biz_warehouse (warehouse_id),
@@ -1027,7 +1027,7 @@ create table biz_shipment_order (
remark varchar(500) default null comment '备注',
primary key (shipment_id),
constraint uq_biz_shipment_order_no unique (shipment_no),
constraint uq_biz_shipment_order_confirm_idempotency unique (confirm_idempotency_key),
constraint uq_biz_shipment_order_route_confirm_idempotency unique (source_warehouse_id, target_site_id, confirm_idempotency_key),
constraint fk_biz_shipment_order_warehouse foreign key (source_warehouse_id) references biz_warehouse (warehouse_id),
constraint fk_biz_shipment_order_site foreign key (target_site_id) references biz_customer_site (site_id),
constraint fk_biz_shipment_order_confirmed_by foreign key (confirmed_by) references sys_user (user_id),
@@ -1067,7 +1067,7 @@ create table biz_usage_record (
occurred_at datetime not null comment '发生时间',
primary key (usage_id),
constraint uq_biz_usage_record_no unique (usage_no),
constraint uq_biz_usage_record_idempotency unique (idempotency_key),
constraint uq_biz_usage_record_site_idempotency unique (site_id, idempotency_key),
constraint fk_biz_usage_record_pallet foreign key (pallet_id) references biz_pallet (pallet_id),
constraint fk_biz_usage_record_filling foreign key (active_filling_id) references biz_filling_record (filling_id),
constraint fk_biz_usage_record_site foreign key (site_id) references biz_customer_site (site_id),
@@ -1099,8 +1099,8 @@ create table biz_return_order (
remark varchar(500) default null comment '备注',
primary key (return_id),
constraint uq_biz_return_order_no unique (return_no),
constraint uq_biz_return_order_request_idempotency unique (request_idempotency_key),
constraint uq_biz_return_order_confirm_idempotency unique (confirm_idempotency_key),
constraint uq_biz_return_order_route_request_idempotency unique (source_site_id, target_warehouse_id, request_idempotency_key),
constraint uq_biz_return_order_route_confirm_idempotency unique (source_site_id, target_warehouse_id, confirm_idempotency_key),
constraint fk_biz_return_order_site foreign key (source_site_id) references biz_customer_site (site_id),
constraint fk_biz_return_order_warehouse foreign key (target_warehouse_id) references biz_warehouse (warehouse_id),
constraint fk_biz_return_order_requested_by foreign key (requested_by) references sys_user (user_id),
shierfeng-fastapi-backend/tests/module_business/test_filling_service.py
+24 -1
View File
@@ -729,6 +729,25 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase):
self.assertTrue(result.is_success)
self.assertEqual(31, result.result)
async def test_idempotency_lookup_uses_requested_warehouse_for_multi_scope_and_admin(self) -> None:
for user, scope in ((self.user, [9, 7, 9]), (make_user(1), None)):
with self.subTest(admin=bool(user.user and user.user.admin)):
self.db.reset_mock()
existing = make_filling()
get_key = AsyncMock(return_value=existing)
scope_read = AsyncMock(return_value=scope)
with (
patch.object(AccessDao, 'list_user_warehouse_ids', new=scope_read),
patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()),
patch.object(OperationDao, 'get_filling_by_idempotency_key', new=get_key),
):
result = await FillingService.fill_pallet_services(self.db, self.page, user)
get_key.assert_awaited_once_with(self.db, 'fill-key', warehouse_ids=[7])
self.assertEqual(31, result.result)
if user.user and user.user.admin:
scope_read.assert_not_awaited()
async def test_same_idempotency_key_with_different_hash_is_rejected(self) -> None:
existing = make_filling(request_hash='different-hash')
with (
@@ -913,7 +932,7 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase):
key_reads = AsyncMock(side_effect=[None, recovered])
access = AsyncMock()
with (
patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])),
patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[9, 7, 9])),
patch.object(AccessService, 'assert_warehouse_access', new=access),
patch.object(OperationDao, 'get_filling_by_idempotency_key', new=key_reads),
patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)),
@@ -926,6 +945,10 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase):
result = await FillingService.fill_pallet_services(self.db, self.page, self.user)
self.assertEqual(2, key_reads.await_count)
self.assertEqual(
[{'warehouse_ids': [7]}, {'warehouse_ids': [7]}],
[call.kwargs for call in key_reads.await_args_list],
)
self.assertEqual(2, self.db.rollback.await_count)
access.assert_any_await(self.db, self.user, 7)
self.assertTrue(result.is_success)
shierfeng-fastapi-backend/tests/module_business/test_inbound_service.py
+130 -1
View File
@@ -9,7 +9,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
from fastapi import FastAPI, Request
from fastapi.routing import APIRoute
from sqlalchemy.exc import IntegrityError
from sqlalchemy.exc import IntegrityError, MultipleResultsFound
from sqlalchemy.ext.asyncio import AsyncSession
from common.aspect.interface_auth import CheckUserInterfaceAuth
@@ -1407,6 +1407,65 @@ class TestInboundConfirmService(unittest.IsolatedAsyncioTestCase):
self.db.rollback.assert_awaited_once()
self.assertTrue(result.is_success)
async def test_same_key_retry_uses_exact_warehouse_when_user_can_access_duplicate_keys(self) -> None:
target = make_inbound(
status='CONFIRMED',
idempotency_key='confirm-key',
request_hash=self.request_hash,
)
other_domain = make_inbound(
inbound_id=12,
warehouse_id=8,
status='CONFIRMED',
idempotency_key='confirm-key',
request_hash=InboundService._build_confirm_request_hash(12),
)
async def read_key(
_db: AsyncSession,
_key: str,
*,
warehouse_ids: list[int] | None = None,
) -> BizInboundOrder | None:
matches = [
inbound
for inbound in (target, other_domain)
if warehouse_ids is None or inbound.warehouse_id in warehouse_ids
]
if len(matches) > 1:
raise MultipleResultsFound('广域幂等键查询返回多条入库单')
return matches[0] if matches else None
get_detail = AsyncMock(return_value=target)
key_read = AsyncMock(side_effect=read_key)
with (
patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 8])),
patch.object(OperationDao, 'get_inbound_by_id', new=get_detail),
patch.object(OperationDao, 'get_inbound_by_confirm_idempotency_key', new=key_read),
patch.object(OperationDao, 'get_inbound_for_update', new=AsyncMock(return_value=target)),
patch.object(
MasterDao,
'get_warehouse',
new=AsyncMock(return_value=SimpleNamespace(warehouse_id=7, status='0')),
),
patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()),
):
result = await InboundService.confirm_inbound_services(
self.db,
11,
self.command,
self.user,
)
self.assertTrue(result.is_success)
get_detail.assert_awaited_once_with(self.db, 11, warehouse_ids=[7, 8])
key_read.assert_awaited_once_with(
self.db,
'confirm-key',
warehouse_ids=[7],
)
self.db.rollback.assert_awaited_once()
async def test_same_key_retry_hides_document_when_current_mapping_is_revoked(self) -> None:
confirmed = make_inbound(
status='CONFIRMED',
@@ -1569,6 +1628,76 @@ class TestInboundConfirmService(unittest.IsolatedAsyncioTestCase):
self.assertEqual(2, self.db.rollback.await_count)
self.assertTrue(result.is_success)
async def test_integrity_recovery_uses_exact_warehouse_with_duplicate_visible_keys(self) -> None:
initial = make_inbound(status='DRAFT')
recovered = make_inbound(
status='CONFIRMED',
idempotency_key='confirm-key',
request_hash=self.request_hash,
)
other_domain = make_inbound(
inbound_id=12,
warehouse_id=8,
status='CONFIRMED',
idempotency_key='confirm-key',
request_hash=InboundService._build_confirm_request_hash(12),
)
read_count = 0
async def read_key(
_db: AsyncSession,
_key: str,
*,
warehouse_ids: list[int] | None = None,
) -> BizInboundOrder | None:
nonlocal read_count
read_count += 1
if read_count == 1:
return None
matches = [
inbound
for inbound in (recovered, other_domain)
if warehouse_ids is None or inbound.warehouse_id in warehouse_ids
]
if len(matches) > 1:
raise MultipleResultsFound('广域幂等键恢复查询返回多条入库单')
return matches[0] if matches else None
warehouse_scope = AsyncMock(side_effect=[[7, 8], [7, 8]])
key_read = AsyncMock(side_effect=read_key)
parent_lock = AsyncMock(side_effect=[integrity_error(), recovered])
with (
patch.object(AccessDao, 'list_user_warehouse_ids', new=warehouse_scope),
patch.object(OperationDao, 'get_inbound_by_id', new=AsyncMock(return_value=initial)),
patch.object(OperationDao, 'get_inbound_by_confirm_idempotency_key', new=key_read),
patch.object(OperationDao, 'get_inbound_for_update', new=parent_lock),
patch.object(
MasterDao,
'get_warehouse',
new=AsyncMock(return_value=SimpleNamespace(warehouse_id=7, status='0')),
),
patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()),
):
result = await InboundService.confirm_inbound_services(
self.db,
11,
self.command,
self.user,
)
self.assertTrue(result.is_success)
self.assertEqual(2, warehouse_scope.await_count)
self.assertEqual(2, parent_lock.await_count)
self.assertEqual(
[
unittest.mock.call(self.db, 'confirm-key', warehouse_ids=[7]),
unittest.mock.call(self.db, 'confirm-key', warehouse_ids=[7]),
],
key_read.await_args_list,
)
self.assertEqual(2, self.db.rollback.await_count)
self.db.commit.assert_not_awaited()
async def test_integrity_recovery_locks_parent_and_rechecks_revoked_current_mapping(self) -> None:
inbound = make_inbound(pallet_ids=(2,))
recovered = make_inbound(
shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py
+25 -10
View File
@@ -141,14 +141,14 @@ class TestDaoPublicContract(unittest.TestCase):
'get_return_by_request_idempotency_key': (
'db',
'idempotency_key',
'site_ids',
'warehouse_ids',
'source_site_id',
'target_warehouse_id',
),
'get_return_by_confirm_idempotency_key': (
'db',
'idempotency_key',
'site_ids',
'warehouse_ids',
'source_site_id',
'target_warehouse_id',
),
'get_inbound_by_id': ('db', 'inbound_id', 'warehouse_ids'),
'add_inbound': ('db', 'inbound'),
@@ -249,8 +249,6 @@ class TestDaoPublicContract(unittest.TestCase):
self.assertIs(True, return_list_signature.parameters['is_page'].default)
for method_name in (
'get_return_for_update',
'get_return_by_request_idempotency_key',
'get_return_by_confirm_idempotency_key',
'get_return_by_id',
'get_return_list',
):
@@ -259,6 +257,15 @@ class TestDaoPublicContract(unittest.TestCase):
self.assertIsNone(signature.parameters['site_ids'].default)
self.assertIsNone(signature.parameters['warehouse_ids'].default)
for method_name in (
'get_return_by_request_idempotency_key',
'get_return_by_confirm_idempotency_key',
):
with self.subTest(method=method_name):
signature = inspect.signature(getattr(OperationDao, method_name))
self.assertIs(inspect.Parameter.empty, signature.parameters['source_site_id'].default)
self.assertIs(inspect.Parameter.empty, signature.parameters['target_warehouse_id'].default)
def test_dao_source_never_controls_transaction_boundary(self) -> None:
for module in (inventory_dao_module, operation_dao_module):
with self.subTest(module=module.__name__):
@@ -609,26 +616,30 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
OperationDao.get_inbound_by_confirm_idempotency_key,
BizInboundOrder,
'confirm_idempotency_key',
{},
),
(
OperationDao.get_shipment_by_confirm_idempotency_key,
BizShipmentOrder,
'confirm_idempotency_key',
{},
),
(OperationDao.get_filling_by_idempotency_key, BizFillingRecord, 'idempotency_key'),
(OperationDao.get_usage_by_idempotency_key, BizUsageRecord, 'idempotency_key'),
(OperationDao.get_filling_by_idempotency_key, BizFillingRecord, 'idempotency_key', {}),
(OperationDao.get_usage_by_idempotency_key, BizUsageRecord, 'idempotency_key', {}),
(
OperationDao.get_return_by_request_idempotency_key,
BizReturnOrder,
'request_idempotency_key',
{'source_site_id': 9, 'target_warehouse_id': 7},
),
(
OperationDao.get_return_by_confirm_idempotency_key,
BizReturnOrder,
'confirm_idempotency_key',
{'source_site_id': 9, 'target_warehouse_id': 7},
),
)
for method, model, key_column in cases:
for method, model, key_column, kwargs in cases:
with self.subTest(method=method.__name__):
self.db.reset_mock()
record = SimpleNamespace()
@@ -636,13 +647,17 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
result.scalar_one_or_none.return_value = record
self.db.execute.return_value = result
actual = await method(self.db, 'same-key')
actual = await method(self.db, 'same-key', **kwargs)
self.assertIs(record, actual)
statement = self.db.execute.await_args.args[0]
sql = compile_sql(statement)
self.assertTrue(statement.get_final_froms()[0].compare(model.__table__))
self.assertIn(f"{model.__tablename__}.{key_column} = 'same-key'", sql)
if model is BizReturnOrder:
self.assertIn('biz_return_order.source_site_id = 9', sql)
self.assertIn('biz_return_order.target_warehouse_id = 7', sql)
self.assertNotIn(' OR ', sql)
self.assertNotIn('FOR UPDATE', sql)
async def test_inbound_confirm_key_pushes_warehouse_scope_into_sql(self) -> None:
shierfeng-fastapi-backend/tests/module_business/test_model_contract.py
+6 -6
View File
@@ -862,27 +862,27 @@ EXPECTED_UNIQUE_SETS = {
'biz_user_site': set(),
'biz_inbound_order': {
frozenset({'inbound_no'}),
frozenset({'confirm_idempotency_key'}),
frozenset({'warehouse_id', 'confirm_idempotency_key'}),
},
'biz_inbound_item': {frozenset({'inbound_id', 'pallet_id'})},
'biz_filling_record': {
frozenset({'filling_no'}),
frozenset({'idempotency_key'}),
frozenset({'warehouse_id', 'idempotency_key'}),
frozenset({'pallet_id', 'active_slot'}),
},
'biz_shipment_order': {
frozenset({'shipment_no'}),
frozenset({'confirm_idempotency_key'}),
frozenset({'source_warehouse_id', 'target_site_id', 'confirm_idempotency_key'}),
},
'biz_shipment_item': {frozenset({'shipment_id', 'pallet_id'})},
'biz_usage_record': {
frozenset({'usage_no'}),
frozenset({'idempotency_key'}),
frozenset({'site_id', 'idempotency_key'}),
},
'biz_return_order': {
frozenset({'return_no'}),
frozenset({'request_idempotency_key'}),
frozenset({'confirm_idempotency_key'}),
frozenset({'source_site_id', 'target_warehouse_id', 'request_idempotency_key'}),
frozenset({'source_site_id', 'target_warehouse_id', 'confirm_idempotency_key'}),
},
'biz_return_item': {frozenset({'return_id', 'pallet_id'})},
'biz_pallet_inventory': set(),
shierfeng-fastapi-backend/tests/module_business/test_return_service.py
+145 -30
View File
@@ -188,22 +188,23 @@ class ReturnDaoScopeRedTest(unittest.IsolatedAsyncioTestCase):
OperationDao.get_return_by_request_idempotency_key: (
'db',
'idempotency_key',
'site_ids',
'warehouse_ids',
'source_site_id',
'target_warehouse_id',
),
OperationDao.get_return_by_confirm_idempotency_key: (
'db',
'idempotency_key',
'site_ids',
'warehouse_ids',
'source_site_id',
'target_warehouse_id',
),
OperationDao.get_return_by_id: ('db', 'return_id', 'site_ids', 'warehouse_ids'),
}
for method, parameters in expected.items():
with self.subTest(method=method.__name__):
self.assertEqual(parameters, tuple(inspect.signature(method).parameters))
self.assertIsNone(inspect.signature(method).parameters['site_ids'].default)
self.assertIsNone(inspect.signature(method).parameters['warehouse_ids'].default)
if 'site_ids' in parameters:
self.assertIsNone(inspect.signature(method).parameters['site_ids'].default)
self.assertIsNone(inspect.signature(method).parameters['warehouse_ids'].default)
list_signature = inspect.signature(OperationDao.get_return_list)
self.assertEqual(
@@ -278,25 +279,40 @@ class ReturnDaoScopeRedTest(unittest.IsolatedAsyncioTestCase):
self.assertIn('biz_return_order.target_warehouse_id IN (7)', sql)
self.assertIn(' OR ', sql)
async def test_request_key_read_is_scoped_and_never_locks_missing_unique_key(self) -> None:
async def test_return_key_reads_use_exact_route_and_never_lock_missing_unique_key(self) -> None:
result = MagicMock()
result.scalar_one_or_none.return_value = None
self.db.execute.return_value = result
await OperationDao.get_return_by_request_idempotency_key(
self.db,
'request-key',
site_ids=[9],
warehouse_ids=[],
key_reads = (
(OperationDao.get_return_by_request_idempotency_key, 'request-key', 'request_idempotency_key'),
(OperationDao.get_return_by_confirm_idempotency_key, 'confirm-key', 'confirm_idempotency_key'),
)
for method, key, column in key_reads:
for source_site_id, target_warehouse_id in ((9, 7), (9, 8), (10, 7)):
with self.subTest(
method=method.__name__,
source_site_id=source_site_id,
target_warehouse_id=target_warehouse_id,
):
self.db.reset_mock()
self.db.execute.return_value = result
await method(
self.db,
key,
source_site_id=source_site_id,
target_warehouse_id=target_warehouse_id,
)
statement = self.db.execute.await_args.args[0]
self.assertTrue(statement.get_final_froms()[0].compare(BizReturnOrder.__table__))
for dialect in (mysql.dialect(), postgresql.dialect()):
sql = compile_sql(statement, dialect)
self.assertIn("biz_return_order.request_idempotency_key = 'request-key'", sql)
self.assertIn('biz_return_order.source_site_id IN (9)', sql)
self.assertNotIn('FOR UPDATE', sql)
statement = self.db.execute.await_args.args[0]
self.assertTrue(statement.get_final_froms()[0].compare(BizReturnOrder.__table__))
for dialect in (mysql.dialect(), postgresql.dialect()):
sql = compile_sql(statement, dialect)
self.assertIn(f"biz_return_order.{column} = '{key}'", sql)
self.assertIn(f'biz_return_order.source_site_id = {source_site_id}', sql)
self.assertIn(f'biz_return_order.target_warehouse_id = {target_warehouse_id}', sql)
self.assertNotIn(' OR ', sql)
self.assertNotIn('FOR UPDATE', sql)
@unittest.skipUnless(ReturnService is not None, 'ReturnService 尚未实现')
@@ -1021,8 +1037,8 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
self.assertEqual(13, result.result)
self.assertEqual(
[
{'site_ids': [9], 'warehouse_ids': []},
{'site_ids': [9, 10], 'warehouse_ids': []},
{'source_site_id': 9, 'target_warehouse_id': 7},
{'source_site_id': 9, 'target_warehouse_id': 7},
],
[call.kwargs for call in key_read.await_args_list],
)
@@ -1276,7 +1292,10 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
self.assertEqual(2, scope_read.await_count)
self.assertEqual(2, key_read.await_count)
self.assertEqual(
[{'site_ids': [9], 'warehouse_ids': []}, {'site_ids': [9], 'warehouse_ids': []}],
[
{'source_site_id': 9, 'target_warehouse_id': 7},
{'source_site_id': 9, 'target_warehouse_id': 7},
],
[call.kwargs for call in key_read.await_args_list],
)
locks.assert_awaited_once_with(self.db, [2, 5])
@@ -1447,7 +1466,41 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
inventory_locks.assert_not_awaited()
self.db.rollback.assert_awaited_once()
async def test_admin_request_idempotency_scope_is_unrestricted(self) -> None:
async def test_request_integrity_recovery_key_lookup_uses_exact_route(self) -> None:
request_hash = ReturnService._build_request_hash(self.page)
exact = make_return_order(request_hash=request_hash)
key_read = AsyncMock(return_value=exact)
order_lock = AsyncMock(return_value=exact)
with (
patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[10, 9])),
patch.object(OperationDao, 'get_return_by_request_idempotency_key', new=key_read),
patch.object(OperationDao, 'get_return_for_update', new=order_lock),
patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0'))),
patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
):
result = await ReturnService._recover_request_integrity_error(
self.db,
self.page,
request_hash,
self.user,
)
self.assertEqual(13, result.result)
key_read.assert_awaited_once_with(
self.db,
'request-key',
source_site_id=9,
target_warehouse_id=7,
)
order_lock.assert_awaited_once_with(
self.db,
13,
site_ids=[9, 10],
warehouse_ids=[],
)
self.db.rollback.assert_awaited_once()
async def test_admin_request_idempotency_lookup_still_uses_exact_route(self) -> None:
admin = make_user(1)
request_hash = ReturnService._build_request_hash(self.page)
exact = make_return_order(request_hash=request_hash)
@@ -1464,8 +1517,8 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
get_by_key.assert_awaited_once_with(
self.db,
'request-key',
site_ids=None,
warehouse_ids=None,
source_site_id=9,
target_warehouse_id=7,
)
@@ -1558,7 +1611,7 @@ class ReturnConfirmTransactionRedTest(unittest.IsolatedAsyncioTestCase):
self.assertEqual([None, None], [event.after_site_id for event in events])
self.db.commit.assert_awaited_once()
async def test_confirm_preflights_idempotency_key_before_return_id(self) -> None:
async def test_confirm_resolves_authorized_return_route_before_key_lookup(self) -> None:
call_order: list[str] = []
order = make_return_order()
@@ -1569,9 +1622,10 @@ class ReturnConfirmTransactionRedTest(unittest.IsolatedAsyncioTestCase):
call_order.append('return')
return order
key_read = AsyncMock(side_effect=get_by_key)
with (
patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])),
patch.object(OperationDao, 'get_return_by_confirm_idempotency_key', new=AsyncMock(side_effect=get_by_key)),
patch.object(OperationDao, 'get_return_by_confirm_idempotency_key', new=key_read),
patch.object(OperationDao, 'get_return_by_id', new=AsyncMock(side_effect=get_by_id)),
patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()),
patch.object(OperationDao, 'get_return_for_update', new=AsyncMock(return_value=order)),
@@ -1583,7 +1637,13 @@ class ReturnConfirmTransactionRedTest(unittest.IsolatedAsyncioTestCase):
except ServiceException:
pass
self.assertEqual(['key', 'return'], call_order[:2])
self.assertEqual(['return', 'key'], call_order[:2])
key_read.assert_awaited_once_with(
self.db,
'confirm-key',
source_site_id=9,
target_warehouse_id=7,
)
async def test_confirm_requires_exact_scan_set_before_inventory_locks(self) -> None:
order = make_return_order()
@@ -1688,7 +1748,7 @@ class ReturnConfirmTransactionRedTest(unittest.IsolatedAsyncioTestCase):
self.assertIn('幂等键已被其他请求使用', raised.exception.message)
conflict_db.rollback.assert_awaited()
async def test_admin_confirm_scope_is_unrestricted(self) -> None:
async def test_admin_confirm_idempotency_lookup_still_uses_exact_route(self) -> None:
admin = make_user(1)
confirm_hash = ReturnService._build_confirm_request_hash(13, self.page)
exact = make_return_order(
@@ -1697,12 +1757,13 @@ class ReturnConfirmTransactionRedTest(unittest.IsolatedAsyncioTestCase):
confirm_hash=confirm_hash,
)
get_order = AsyncMock(return_value=exact)
get_by_key = AsyncMock(return_value=exact)
with (
patch.object(OperationDao, 'get_return_by_id', new=get_order),
patch.object(
OperationDao,
'get_return_by_confirm_idempotency_key',
new=AsyncMock(return_value=exact),
new=get_by_key,
),
patch.object(OperationDao, 'get_return_for_update', new=AsyncMock(return_value=exact)),
patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()),
@@ -1717,6 +1778,60 @@ class ReturnConfirmTransactionRedTest(unittest.IsolatedAsyncioTestCase):
site_ids=None,
warehouse_ids=None,
)
get_by_key.assert_awaited_once_with(
self.db,
'confirm-key',
source_site_id=9,
target_warehouse_id=7,
)
async def test_confirm_integrity_recovery_resolves_route_before_exact_key_lookup(self) -> None:
confirm_hash = ReturnService._build_confirm_request_hash(13, self.page)
exact = make_return_order(
status=ReturnStatus.CONFIRMED.value,
confirm_key='confirm-key',
confirm_hash=confirm_hash,
)
call_order: list[str] = []
async def read_route(*_args: Any, **_kwargs: Any) -> BizReturnOrder:
call_order.append('return')
return exact
async def read_key(*_args: Any, **_kwargs: Any) -> BizReturnOrder:
call_order.append('key')
return exact
route_read = AsyncMock(side_effect=read_route)
key_read = AsyncMock(side_effect=read_key)
order_lock = AsyncMock(return_value=exact)
with (
patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[8, 7])),
patch.object(OperationDao, 'get_return_by_id', new=route_read),
patch.object(OperationDao, 'get_return_by_confirm_idempotency_key', new=key_read),
patch.object(OperationDao, 'get_return_for_update', new=order_lock),
patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=SimpleNamespace(status='0'))),
patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()),
):
result = await ReturnService._recover_confirm_integrity_error(
self.db,
13,
self.page,
confirm_hash,
self.user,
)
self.assertEqual(13, result.result)
self.assertEqual(['return', 'key'], call_order)
route_read.assert_awaited_once_with(self.db, 13, site_ids=[], warehouse_ids=[7, 8])
key_read.assert_awaited_once_with(
self.db,
'confirm-key',
source_site_id=9,
target_warehouse_id=7,
)
order_lock.assert_awaited_once_with(self.db, 13, site_ids=[], warehouse_ids=[7, 8])
self.db.rollback.assert_awaited_once()
async def test_confirm_rechecks_warehouse_mapping_before_business_locks(self) -> None:
order = make_return_order()
shierfeng-fastapi-backend/tests/module_business/test_shipment_service.py
+198 -1
View File
@@ -11,7 +11,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
from fastapi import FastAPI
from fastapi.routing import APIRoute
from sqlalchemy.dialects import postgresql
from sqlalchemy.exc import IntegrityError
from sqlalchemy.exc import IntegrityError, MultipleResultsFound
from sqlalchemy.ext.asyncio import AsyncSession
from common.aspect.interface_auth import CheckUserInterfaceAuth
@@ -2181,6 +2181,96 @@ class ShipmentConfirmServiceTest(unittest.IsolatedAsyncioTestCase):
parent_lock.assert_not_awaited()
self.db.rollback.assert_awaited_once()
async def test_same_key_retry_uses_exact_route_when_user_can_access_duplicate_keys(self) -> None:
target = make_shipment(
status='CONFIRMED',
idempotency_key='confirm-key',
request_hash=self.request_hash,
)
other_domain = make_shipment(
13,
source_warehouse_id=8,
target_site_id=10,
status='CONFIRMED',
idempotency_key='confirm-key',
request_hash=ShipmentService._build_confirm_request_hash(13),
)
async def read_key(
_db: AsyncSession,
_key: str,
*,
warehouse_ids: list[int] | None = None,
site_ids: list[int] | None = None,
) -> BizShipmentOrder | None:
matches = [
shipment
for shipment in (target, other_domain)
if (warehouse_ids is None or shipment.source_warehouse_id in warehouse_ids)
and (site_ids is None or shipment.target_site_id in site_ids)
]
if len(matches) > 1:
raise MultipleResultsFound('广域幂等键查询返回多条发货单')
return matches[0] if matches else None
get_detail = AsyncMock(return_value=target)
key_read = AsyncMock(side_effect=read_key)
with (
patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 8])),
patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 10])),
patch.object(OperationDao, 'get_shipment_by_id', new=get_detail),
patch.object(OperationDao, 'get_shipment_by_confirm_idempotency_key', new=key_read),
patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=target)),
patch.object(
MasterDao,
'get_warehouse',
new=AsyncMock(
return_value=BizWarehouse(
warehouse_id=7,
warehouse_code='W-7',
warehouse_name='七号仓',
status='0',
)
),
),
patch.object(
MasterDao,
'get_site',
new=AsyncMock(
return_value=BizCustomerSite(
site_id=9,
customer_id=1,
site_code='S-9',
site_name='九号现场',
status='0',
)
),
),
patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()),
patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
):
result = await ShipmentService.confirm_shipment_services(
self.db,
12,
self.command,
self.user,
)
self.assertTrue(result.is_success)
get_detail.assert_awaited_once_with(
self.db,
12,
warehouse_ids=[7, 8],
site_ids=[9, 10],
)
key_read.assert_awaited_once_with(
self.db,
'confirm-key',
warehouse_ids=[7],
site_ids=[9],
)
self.db.rollback.assert_awaited_once()
async def test_integrity_recovery_nonexact_result_is_stable_conflict_not_success(self) -> None:
initial = make_shipment()
nonexact = make_shipment(
@@ -2413,6 +2503,113 @@ class ShipmentConfirmServiceTest(unittest.IsolatedAsyncioTestCase):
self.assertEqual(2, self.db.rollback.await_count)
self.db.commit.assert_not_awaited()
async def test_integrity_recovery_uses_exact_route_with_duplicate_visible_keys(self) -> None:
initial = make_shipment(status='DRAFT')
recovered = make_shipment(
status='CONFIRMED',
idempotency_key='confirm-key',
request_hash=self.request_hash,
)
other_domain = make_shipment(
13,
source_warehouse_id=8,
target_site_id=10,
status='CONFIRMED',
idempotency_key='confirm-key',
request_hash=ShipmentService._build_confirm_request_hash(13),
)
read_count = 0
async def read_key(
_db: AsyncSession,
_key: str,
*,
warehouse_ids: list[int] | None = None,
site_ids: list[int] | None = None,
) -> BizShipmentOrder | None:
nonlocal read_count
read_count += 1
if read_count == 1:
return None
matches = [
shipment
for shipment in (recovered, other_domain)
if (warehouse_ids is None or shipment.source_warehouse_id in warehouse_ids)
and (site_ids is None or shipment.target_site_id in site_ids)
]
if len(matches) > 1:
raise MultipleResultsFound('广域幂等键恢复查询返回多条发货单')
return matches[0] if matches else None
warehouse_scope = AsyncMock(side_effect=[[7, 8], [7, 8]])
site_scope = AsyncMock(side_effect=[[9, 10], [9, 10]])
key_read = AsyncMock(side_effect=read_key)
parent_lock = AsyncMock(side_effect=[integrity_error(), recovered])
with (
patch.object(AccessDao, 'list_user_warehouse_ids', new=warehouse_scope),
patch.object(AccessDao, 'list_user_site_ids', new=site_scope),
patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=initial)),
patch.object(OperationDao, 'get_shipment_by_confirm_idempotency_key', new=key_read),
patch.object(OperationDao, 'get_shipment_for_update', new=parent_lock),
patch.object(
MasterDao,
'get_warehouse',
new=AsyncMock(
return_value=BizWarehouse(
warehouse_id=7,
warehouse_code='W-7',
warehouse_name='七号仓',
status='0',
)
),
),
patch.object(
MasterDao,
'get_site',
new=AsyncMock(
return_value=BizCustomerSite(
site_id=9,
customer_id=1,
site_code='S-9',
site_name='九号现场',
status='0',
)
),
),
patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()),
patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
):
result = await ShipmentService.confirm_shipment_services(
self.db,
12,
self.command,
self.user,
)
self.assertTrue(result.is_success)
self.assertEqual(2, warehouse_scope.await_count)
self.assertEqual(2, site_scope.await_count)
self.assertEqual(2, parent_lock.await_count)
self.assertEqual(
[
unittest.mock.call(
self.db,
'confirm-key',
warehouse_ids=[7],
site_ids=[9],
),
unittest.mock.call(
self.db,
'confirm-key',
warehouse_ids=[7],
site_ids=[9],
),
],
key_read.await_args_list,
)
self.assertEqual(2, self.db.rollback.await_count)
self.db.commit.assert_not_awaited()
async def test_revoked_mapping_after_locks_hides_document_and_mutates_nothing(self) -> None:
shipment = make_shipment(pallet_ids=(2,))
inventory = make_inventory(2, version=4)
shierfeng-fastapi-backend/tests/module_business/test_sql_contract.py
+16 -5
View File
@@ -218,13 +218,24 @@ EXPECTED_UNIQUE_KEYS = {
'biz_pallet': {('pallet_code',), ('qr_token',)},
'biz_user_warehouse': set(),
'biz_user_site': set(),
'biz_inbound_order': {('inbound_no',), ('confirm_idempotency_key',)},
'biz_inbound_order': {('inbound_no',), ('warehouse_id', 'confirm_idempotency_key')},
'biz_inbound_item': {('inbound_id', 'pallet_id')},
'biz_filling_record': {('filling_no',), ('idempotency_key',), ('pallet_id', 'active_slot')},
'biz_shipment_order': {('shipment_no',), ('confirm_idempotency_key',)},
'biz_filling_record': {
('filling_no',),
('warehouse_id', 'idempotency_key'),
('pallet_id', 'active_slot'),
},
'biz_shipment_order': {
('shipment_no',),
('source_warehouse_id', 'target_site_id', 'confirm_idempotency_key'),
},
'biz_shipment_item': {('shipment_id', 'pallet_id')},
'biz_usage_record': {('usage_no',), ('idempotency_key',)},
'biz_return_order': {('return_no',), ('request_idempotency_key',), ('confirm_idempotency_key',)},
'biz_usage_record': {('usage_no',), ('site_id', 'idempotency_key')},
'biz_return_order': {
('return_no',),
('source_site_id', 'target_warehouse_id', 'request_idempotency_key'),
('source_site_id', 'target_warehouse_id', 'confirm_idempotency_key'),
},
'biz_return_item': {('return_id', 'pallet_id')},
'biz_pallet_inventory': set(),
'biz_pallet_event': {('pallet_id', 'event_seq'), ('pallet_id', 'idempotency_key')},
shierfeng-fastapi-backend/tests/module_business/test_usage_service.py
+51 -9
View File
@@ -732,8 +732,19 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
pallet_lock = AsyncMock(return_value=[selected_pallet])
filling_lock = AsyncMock(return_value=selected_filling)
site_lock = AsyncMock(return_value=selected_site)
preflight_line_read = AsyncMock(return_value=selected_line)
line_lock = AsyncMock(return_value=selected_line)
async def read_line(
db: AsyncSession,
line_id: int,
*,
for_update: bool = False,
) -> BizProductionLine | None:
if for_update:
return await line_lock(db, line_id, for_update=True)
return await preflight_line_read(db, line_id)
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))
@@ -742,7 +753,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
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))
stack.enter_context(patch.object(MasterDao, 'get_site', new=site_lock))
stack.enter_context(patch.object(MasterDao, 'get_line', new=line_lock))
stack.enter_context(patch.object(MasterDao, 'get_line', new=AsyncMock(side_effect=read_line)))
stack.enter_context(patch.object(OperationDao, 'add_usage_record', new=usage_write))
stack.enter_context(patch.object(OperationDao, 'update_filling_record', new=filling_write))
stack.enter_context(patch.object(InventoryDao, 'add_event', new=event_write))
@@ -757,6 +768,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
'pallet_lock': pallet_lock,
'filling_lock': filling_lock,
'site_lock': site_lock,
'preflight_line_read': preflight_line_read,
'line_lock': line_lock,
'add_usage': usage_write,
'update_filling': filling_write,
@@ -877,6 +889,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
self.assertEqual(
[
'preflight-scope',
'line-lock',
'usage-preflight-read',
'qr-snapshot-read',
'rollback',
@@ -1044,6 +1057,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
self.db.reset_mock()
stack, mocks = self._patch_record_path(line=line)
if line is None:
mocks['preflight_line_read'].return_value = None
mocks['line_lock'].return_value = None
with stack, self.assertRaises(ServiceException) as raised:
await UsageService.record_usage_services(self.db, self.page, self.user)
@@ -1169,6 +1183,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
async def test_record_hides_missing_and_non_site_qr_but_admin_gets_precise_error(self) -> None:
with (
patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
patch.object(MasterDao, 'get_line', new=AsyncMock(return_value=make_line())),
patch.object(OperationDao, 'get_usage_by_idempotency_key', new=AsyncMock(return_value=None)),
patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=None)),
self.assertRaises(PermissionException) as normal_missing,
@@ -1179,6 +1194,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
self.db.reset_mock()
with (
patch.object(MasterDao, 'get_line', new=AsyncMock(return_value=make_line())),
patch.object(OperationDao, 'get_usage_by_idempotency_key', new=AsyncMock(return_value=None)),
patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=None)),
self.assertRaises(ServiceException) as admin_missing,
@@ -1239,10 +1255,11 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
async def read_key(*_args: Any, **kwargs: Any) -> BizUsageRecord | 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:
is_locked_phase = 'inventory-lock' in actions
actions.append('usage-fresh-read' if is_locked_phase else 'usage-preflight-read')
if not is_locked_phase:
return None
return recovered if kwargs.get('site_ids') == scopes[1] else None
return recovered if kwargs.get('site_ids') == [inventory.site_id] else None
async def scan_pallet(*_args: Any) -> BizPallet:
actions.append('qr-snapshot-read')
@@ -1301,6 +1318,7 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
access = AsyncMock()
with (
patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
patch.object(MasterDao, 'get_line', new=AsyncMock(return_value=make_line())),
patch.object(
OperationDao,
'get_usage_by_idempotency_key',
@@ -1334,10 +1352,34 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
result.result.model_dump(by_alias=True),
)
async def test_preflight_key_lookup_uses_line_site_for_multi_scope_and_admin(self) -> None:
for user, scope in ((self.user, [10, 9, 10]), (make_user(1), None)):
with self.subTest(admin=bool(user.user and user.user.admin)):
self.db.reset_mock()
existing = self._exact_usage()
line_read = AsyncMock(return_value=make_line(site_id=9))
get_key = AsyncMock(return_value=existing)
scope_read = AsyncMock(return_value=scope)
with (
patch.object(AccessDao, 'list_user_site_ids', new=scope_read),
patch.object(MasterDao, 'get_line', new=line_read),
patch.object(OperationDao, 'get_usage_by_idempotency_key', new=get_key),
patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=make_site())),
patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
):
result = await UsageService.record_usage_services(self.db, self.page, user)
line_read.assert_awaited_once_with(self.db, 4)
get_key.assert_awaited_once_with(self.db, 'usage-key', site_ids=[9])
self.assertEqual(61, result.result.usage_id)
if user.user and user.user.admin:
scope_read.assert_not_awaited()
async def test_same_key_with_different_hash_is_stable_conflict(self) -> None:
existing = self._exact_usage(request_hash='different-hash')
with (
patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
patch.object(MasterDao, 'get_line', new=AsyncMock(return_value=make_line())),
patch.object(OperationDao, 'get_usage_by_idempotency_key', new=AsyncMock(return_value=existing)),
patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock()) as inventory_lock,
self.assertRaises(ServiceException) as raised,
@@ -1353,6 +1395,7 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
access = AsyncMock(side_effect=PermissionException(data='', message='无权访问该客户现场'))
with (
patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
patch.object(MasterDao, 'get_line', new=AsyncMock(return_value=make_line())),
patch.object(OperationDao, 'get_usage_by_idempotency_key', new=AsyncMock(return_value=existing)),
patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=make_site())),
patch.object(AccessService, 'assert_site_access', new=access),
@@ -1505,7 +1548,6 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
self.assertEqual(
[
'preflight-scope',
'usage-preflight-read',
'qr-snapshot-read',
'preflight-rollback',
'inventory-lock',
@@ -1520,7 +1562,7 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
actions,
)
self.assertEqual(
[{'site_ids': []}, {'site_ids': [9]}],
[{'site_ids': [9]}],
[call.kwargs for call in mocks['key_read'].await_args_list],
)
mocks['filling_lock'].assert_not_awaited()
@@ -1702,11 +1744,11 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
self.fail('锁后命中的摘要或实际现场冲突不得返回成功')
self.assertEqual(
[{'site_ids': [9, 10]}, {'site_ids': [9, 10]}],
[{'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, 10] for call in key_read.await_args_list))
self.assertTrue(all(call.kwargs['site_ids'] == [9] for call in key_read.await_args_list))
mocks['site_lock'].assert_not_awaited()
mocks['line_lock'].assert_not_awaited()
mocks['add_usage'].assert_not_awaited()
@@ -1717,7 +1759,7 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
async def test_integrity_error_recovers_only_fresh_scoped_exact_result(self) -> None:
recovered = self._exact_usage()
scope = AsyncMock(side_effect=[[9], [9], [9]])
scope = AsyncMock(side_effect=[[10, 9], [10, 9], [10, 9]])
key_read = AsyncMock(side_effect=[None, None, recovered])
add_usage = AsyncMock(side_effect=integrity_error())
stack, mocks = self.core._patch_record_path(