2307 lines
68 KiB
Python
2307 lines
68 KiB
Python
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.models.base import Base
|
|
from app.models.models import (
|
|
BizAttachment,
|
|
BizCellSampleType,
|
|
BizCollectionHospital,
|
|
BizCollectionProtocol,
|
|
BizCollectionRegion,
|
|
BizCollectionRegionStaff,
|
|
BizPresignOrder,
|
|
BizPresignOrderSampleType,
|
|
BizPresignTransfer,
|
|
BizProtocolCollectionInfo,
|
|
BizProtocolPartyCustomer,
|
|
BizProtocolSample,
|
|
SysPost,
|
|
SysPostRole,
|
|
SysRole,
|
|
SysRoleDataScopeDetail,
|
|
SysUser,
|
|
SysUserPost,
|
|
SysUserRole,
|
|
)
|
|
from app.models.research_donation import (
|
|
BizResearchDonation,
|
|
BizResearchDonationSample,
|
|
)
|
|
|
|
|
|
_TEST_TABLES = [
|
|
BizCollectionRegionStaff.__table__,
|
|
BizCollectionRegion.__table__,
|
|
BizCollectionHospital.__table__,
|
|
BizPresignOrder.__table__,
|
|
BizPresignOrderSampleType.__table__,
|
|
BizPresignTransfer.__table__,
|
|
BizCellSampleType.__table__,
|
|
BizCollectionProtocol.__table__,
|
|
BizProtocolPartyCustomer.__table__,
|
|
BizProtocolCollectionInfo.__table__,
|
|
BizProtocolSample.__table__,
|
|
BizResearchDonation.__table__,
|
|
BizResearchDonationSample.__table__,
|
|
BizAttachment.__table__,
|
|
SysUser.__table__,
|
|
SysRole.__table__,
|
|
SysRoleDataScopeDetail.__table__,
|
|
SysUserRole.__table__,
|
|
SysPost.__table__,
|
|
SysPostRole.__table__,
|
|
SysUserPost.__table__,
|
|
]
|
|
|
|
|
|
def _session_factory(database_url="sqlite:///:memory:"):
|
|
engine = create_engine(database_url)
|
|
Base.metadata.create_all(
|
|
engine,
|
|
tables=_TEST_TABLES,
|
|
)
|
|
return sessionmaker(bind=engine)
|
|
|
|
|
|
def _session():
|
|
return _session_factory()()
|
|
|
|
|
|
def _order(
|
|
order_id,
|
|
*,
|
|
tenant_no="tenant-a",
|
|
region_id="region-a",
|
|
sales_user_id="sales-a",
|
|
manager_user_id="manager-a",
|
|
):
|
|
return BizPresignOrder(
|
|
id=order_id,
|
|
tenant_no=tenant_no,
|
|
region_id=region_id,
|
|
sales_user_id=sales_user_id,
|
|
region_manager_user_id=manager_user_id,
|
|
order_status="active",
|
|
create_by="creator-a",
|
|
create_dept_id="dept-a",
|
|
del_flag=0,
|
|
)
|
|
|
|
|
|
def _transfer(
|
|
transfer_id,
|
|
*,
|
|
order_id="own",
|
|
tenant_no="tenant-a",
|
|
pay_status="dfj",
|
|
):
|
|
return BizPresignTransfer(
|
|
id=transfer_id,
|
|
presign_order_id=order_id,
|
|
transfer_amount=100,
|
|
transfer_type="wechat",
|
|
pay_status=pay_status,
|
|
tenant_no=tenant_no,
|
|
create_by="creator-a",
|
|
create_dept_id="dept-a",
|
|
del_flag=0,
|
|
)
|
|
|
|
|
|
def _ctx(scope_kind, user_id="u1"):
|
|
from app.services.market_field_scope import MarketFieldContext
|
|
|
|
return MarketFieldContext("tenant-a", user_id, scope_kind)
|
|
|
|
|
|
def _user(user_id="u1"):
|
|
return {
|
|
"user_id": user_id,
|
|
"tenant_no": "tenant-a",
|
|
"role": "super_admin",
|
|
"dept_id": "dept-a",
|
|
}
|
|
|
|
|
|
def _real_auth_user(user_id="u1"):
|
|
return {
|
|
"user_id": user_id,
|
|
"tenant_no": "tenant-a",
|
|
"role": "super_admin",
|
|
"region_id": "dept-from-auth",
|
|
}
|
|
|
|
|
|
def test_converted_presign_rejects_update_and_convert_actions():
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import convert_presign_order, update_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
db.add(_order('terminal', sales_user_id='u1'))
|
|
db.commit()
|
|
row = db.query(BizPresignOrder).filter_by(id='terminal').one()
|
|
row.order_status = 'converted'
|
|
db.commit()
|
|
|
|
with pytest.raises(ValueError, match='不可|已转'):
|
|
update_presign_order(
|
|
db,
|
|
order_id='terminal',
|
|
body=MarketPresignOrderBody(remark='不应保存'),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
with pytest.raises(ValueError, match='不可|已转'):
|
|
convert_presign_order(
|
|
db,
|
|
order_id='terminal',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
|
|
def test_void_presign_allows_update_but_rejects_convert():
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import convert_presign_order, update_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
db.add(_order('void-order', sales_user_id='u1'))
|
|
db.commit()
|
|
row = db.query(BizPresignOrder).filter_by(id='void-order').one()
|
|
row.order_status = 'void'
|
|
db.commit()
|
|
|
|
data = update_presign_order(
|
|
db,
|
|
order_id='void-order',
|
|
body=MarketPresignOrderBody(remark='作废单仍可修订'),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert data['remark'] == '作废单仍可修订'
|
|
with pytest.raises(ValueError, match='不可|作废'):
|
|
convert_presign_order(
|
|
db,
|
|
order_id='void-order',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
|
|
def test_presign_create_rejects_unregistered_transfer_attachment_url():
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import create_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
body = MarketPresignOrderBody.model_validate(
|
|
{
|
|
'regionId': 'region-a',
|
|
'hospitalId': 'hospital-a',
|
|
'transfers': [
|
|
{
|
|
'transferAmount': 100,
|
|
'transferAttachments': 'https://evil.test/proof.pdf',
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
with pytest.raises(ValueError, match='附件'):
|
|
create_presign_order(
|
|
db,
|
|
body=body,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize('order_status', ['converted', 'void'])
|
|
def test_presign_create_rejects_embedded_transfer_for_non_writable_order(
|
|
order_status,
|
|
):
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import create_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
body = MarketPresignOrderBody.model_validate(
|
|
{
|
|
'regionId': 'region-a',
|
|
'hospitalId': 'hospital-a',
|
|
'orderStatus': order_status,
|
|
'transfers': [
|
|
{
|
|
'transferAmount': 100,
|
|
'transferType': 'bank',
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
with pytest.raises(ValueError, match='订单|状态|转账'):
|
|
create_presign_order(
|
|
db,
|
|
body=body,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert db.query(BizPresignOrder).count() == 0
|
|
assert db.query(BizPresignTransfer).count() == 0
|
|
|
|
|
|
def _add_master_data(db):
|
|
db.add_all(
|
|
[
|
|
BizCollectionRegion(
|
|
id="region-a",
|
|
region_name="A区",
|
|
tenant_no="tenant-a",
|
|
create_dept_id="",
|
|
del_flag=0,
|
|
),
|
|
BizCollectionRegion(
|
|
id="region-b",
|
|
region_name="B区",
|
|
tenant_no="tenant-a",
|
|
create_dept_id="",
|
|
del_flag=0,
|
|
),
|
|
BizCollectionRegion(
|
|
id="region-cross",
|
|
region_name="跨租户地区",
|
|
tenant_no="tenant-b",
|
|
create_dept_id="",
|
|
del_flag=0,
|
|
),
|
|
BizCollectionHospital(
|
|
id="hospital-a",
|
|
region_id="region-a",
|
|
hospital_name="A医院",
|
|
tenant_no="tenant-a",
|
|
create_dept_id="",
|
|
del_flag=0,
|
|
),
|
|
BizCollectionHospital(
|
|
id="hospital-b",
|
|
region_id="region-b",
|
|
hospital_name="B医院",
|
|
tenant_no="tenant-a",
|
|
create_dept_id="",
|
|
del_flag=0,
|
|
),
|
|
BizCellSampleType(
|
|
id="sample-a",
|
|
type_code="A",
|
|
type_name="样本A",
|
|
tenant_no="tenant-a",
|
|
create_dept_id="",
|
|
del_flag=0,
|
|
),
|
|
SysUser(
|
|
id="u1",
|
|
user_no="u1",
|
|
nick_name="当前用户",
|
|
status="1",
|
|
tenant_no="tenant-a",
|
|
create_dept_id="dept-a",
|
|
del_flag=0,
|
|
),
|
|
SysUser(
|
|
id="collector-a",
|
|
user_no="collector-a",
|
|
nick_name="有效采集员",
|
|
status="1",
|
|
tenant_no="tenant-a",
|
|
create_dept_id="dept-a",
|
|
del_flag=0,
|
|
),
|
|
SysUser(
|
|
id="manager-a",
|
|
user_no="manager-a",
|
|
nick_name="有效经理",
|
|
status="1",
|
|
tenant_no="tenant-a",
|
|
create_dept_id="dept-a",
|
|
del_flag=0,
|
|
),
|
|
SysUser(
|
|
id="cross-user",
|
|
user_no="cross-secret",
|
|
nick_name="跨租户秘密姓名",
|
|
status="1",
|
|
tenant_no="tenant-b",
|
|
create_dept_id="dept-b",
|
|
del_flag=0,
|
|
),
|
|
SysUser(
|
|
id="deleted-user",
|
|
user_no="deleted-secret",
|
|
nick_name="已删除秘密姓名",
|
|
status="1",
|
|
tenant_no="tenant-a",
|
|
create_dept_id="dept-a",
|
|
del_flag=1,
|
|
),
|
|
SysUser(
|
|
id="disabled-user",
|
|
user_no="disabled-secret",
|
|
nick_name="停用秘密姓名",
|
|
status="0",
|
|
tenant_no="tenant-a",
|
|
create_dept_id="dept-a",
|
|
del_flag=0,
|
|
),
|
|
BizCellSampleType(
|
|
id="sample-cross",
|
|
type_code="X",
|
|
type_name="跨租户样本",
|
|
tenant_no="tenant-b",
|
|
create_dept_id="",
|
|
del_flag=0,
|
|
),
|
|
]
|
|
)
|
|
db.commit()
|
|
|
|
|
|
def test_specialist_presign_list_only_returns_own_sales_orders():
|
|
from app.services.market_presign import list_presign_orders
|
|
|
|
db = _session()
|
|
db.add_all(
|
|
[
|
|
_order("own", sales_user_id="u1"),
|
|
_order("other", sales_user_id="u2"),
|
|
_order("cross-tenant", tenant_no="tenant-b", sales_user_id="u1"),
|
|
]
|
|
)
|
|
db.commit()
|
|
|
|
items, total = list_presign_orders(
|
|
db, ctx=_ctx("specialist"), user_info=_user()
|
|
)
|
|
|
|
assert total == 1
|
|
assert [item["id"] for item in items] == ["own"]
|
|
|
|
|
|
def test_manager_presign_list_uses_region_or_manager_union():
|
|
from app.services.market_presign import list_presign_orders
|
|
|
|
db = _session()
|
|
db.add(
|
|
BizCollectionRegionStaff(
|
|
id="leader",
|
|
region_id="region-led",
|
|
user_id="u1",
|
|
is_leader=1,
|
|
tenant_no="tenant-a",
|
|
create_dept_id="dept-a",
|
|
del_flag=0,
|
|
)
|
|
)
|
|
db.add_all(
|
|
[
|
|
_order("by-region", region_id="region-led", manager_user_id="u2"),
|
|
_order("by-manager", region_id="region-other", manager_user_id="u1"),
|
|
_order("outside", region_id="region-other", manager_user_id="u2"),
|
|
]
|
|
)
|
|
db.commit()
|
|
|
|
items, total = list_presign_orders(
|
|
db, ctx=_ctx("manager"), user_info=_user()
|
|
)
|
|
|
|
assert total == 2
|
|
assert {item["id"] for item in items} == {"by-region", "by-manager"}
|
|
|
|
|
|
def test_presign_list_batches_deposits_and_user_names_without_n_plus_one():
|
|
from app.services.market_presign import list_presign_orders
|
|
|
|
db = _session()
|
|
db.add(
|
|
SysUser(
|
|
id="u1",
|
|
user_no="u1",
|
|
nick_name="当前用户",
|
|
status="1",
|
|
tenant_no="tenant-a",
|
|
create_dept_id="dept-a",
|
|
del_flag=0,
|
|
)
|
|
)
|
|
db.add_all(
|
|
[_order(f"order-{index}", sales_user_id="u1") for index in range(3)]
|
|
)
|
|
db.commit()
|
|
statements = []
|
|
|
|
def capture_statement(*args):
|
|
statements.append(args[2])
|
|
|
|
event.listen(db.get_bind(), "before_cursor_execute", capture_statement)
|
|
try:
|
|
items, total = list_presign_orders(
|
|
db, ctx=_ctx("specialist"), user_info=_user(), page_size=10
|
|
)
|
|
finally:
|
|
event.remove(db.get_bind(), "before_cursor_execute", capture_statement)
|
|
|
|
assert total == 3
|
|
assert len(items) == 3
|
|
assert len(statements) <= 4
|
|
|
|
|
|
def test_presign_response_does_not_leak_invalid_or_cross_tenant_user_names():
|
|
from app.services.market_presign import list_presign_orders
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
db.add(
|
|
_order(
|
|
"own",
|
|
sales_user_id="u1",
|
|
manager_user_id="cross-user",
|
|
)
|
|
)
|
|
row = db.query(BizPresignOrder).filter_by(id="own").one_or_none()
|
|
if row is None:
|
|
db.flush()
|
|
row = db.query(BizPresignOrder).filter_by(id="own").one()
|
|
row.collector_user_id = "deleted-user"
|
|
db.commit()
|
|
|
|
items, _ = list_presign_orders(
|
|
db, ctx=_ctx("specialist"), user_info=_user()
|
|
)
|
|
|
|
assert items[0]["salesUserName"] == "当前用户"
|
|
assert items[0]["collectorUserName"] == ""
|
|
assert items[0]["regionManagerUserName"] == ""
|
|
assert "秘密姓名" not in str(items[0])
|
|
|
|
|
|
def test_presign_detail_rejects_guessed_out_of_scope_id():
|
|
from app.services.market_presign import get_presign_order
|
|
|
|
db = _session()
|
|
db.add(_order("other", sales_user_id="u2"))
|
|
db.commit()
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
get_presign_order(
|
|
db,
|
|
order_id="other",
|
|
ctx=_ctx("specialist"),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
|
|
|
|
def test_specialist_create_forces_current_sales_user_and_accepts_order_status():
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import create_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
body = MarketPresignOrderBody.model_validate(
|
|
{
|
|
"motherName": "测试客户",
|
|
"regionId": "region-a",
|
|
"hospitalId": "hospital-a",
|
|
"sampleTypeIds": ["sample-a"],
|
|
"salesUserId": "attacker-selected-user",
|
|
"orderStatus": "void",
|
|
}
|
|
)
|
|
|
|
data = create_presign_order(
|
|
db,
|
|
body=body,
|
|
ctx=_ctx("specialist"),
|
|
user_info=_user(),
|
|
)
|
|
row = db.query(BizPresignOrder).filter_by(id=data["id"]).one()
|
|
|
|
assert row.tenant_no == "tenant-a"
|
|
assert row.sales_user_id == "u1"
|
|
assert row.order_status == "void"
|
|
|
|
|
|
def test_create_rejects_client_requested_converted_status_without_writing():
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import create_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
|
|
with pytest.raises(ValueError, match='转正式|converted'):
|
|
create_presign_order(
|
|
db,
|
|
body=MarketPresignOrderBody.model_validate(
|
|
{
|
|
'motherName': '不可直转客户',
|
|
'regionId': 'region-a',
|
|
'hospitalId': 'hospital-a',
|
|
'orderStatus': 'converted',
|
|
}
|
|
),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert db.query(BizPresignOrder).count() == 0
|
|
|
|
|
|
@pytest.mark.parametrize('current_status', ['active', 'draft', 'void'])
|
|
def test_update_rejects_client_requested_converted_status_without_writing(
|
|
current_status,
|
|
):
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import update_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
order = _order('own', sales_user_id='u1')
|
|
order.order_status = current_status
|
|
order.remark = 'before'
|
|
db.add(order)
|
|
db.commit()
|
|
|
|
with pytest.raises(ValueError, match='转正式|converted'):
|
|
update_presign_order(
|
|
db,
|
|
order_id='own',
|
|
body=MarketPresignOrderBody.model_validate(
|
|
{'orderStatus': 'converted', 'remark': 'after'}
|
|
),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
db.expire_all()
|
|
unchanged = db.query(BizPresignOrder).filter_by(id='own').one()
|
|
assert unchanged.order_status == current_status
|
|
assert unchanged.remark == 'before'
|
|
|
|
|
|
def test_update_rolls_back_when_response_serialization_fails(monkeypatch):
|
|
from app.crud import presign_order as presign_crud
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import update_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
order = _order('own', sales_user_id='u1')
|
|
order.remark = 'before'
|
|
db.add(order)
|
|
db.commit()
|
|
|
|
monkeypatch.setattr(
|
|
presign_crud,
|
|
'order_to_dict',
|
|
lambda *args, **kwargs: (_ for _ in ()).throw(
|
|
RuntimeError('serialize failed')
|
|
),
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match='serialize failed'):
|
|
update_presign_order(
|
|
db,
|
|
order_id='own',
|
|
body=MarketPresignOrderBody(remark='after'),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
db.expire_all()
|
|
assert db.query(BizPresignOrder).filter_by(id='own').one().remark == 'before'
|
|
|
|
|
|
def test_create_rolls_back_order_children_and_transfers_when_data_scope_rejects():
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import create_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
restricted_user = {
|
|
"user_id": "u1",
|
|
"tenant_no": "tenant-a",
|
|
"role": "ordinary",
|
|
"dept_id": "dept-a",
|
|
}
|
|
|
|
with pytest.raises(HTTPException):
|
|
create_presign_order(
|
|
db,
|
|
body=MarketPresignOrderBody.model_validate(
|
|
{
|
|
"regionId": "region-a",
|
|
"hospitalId": "hospital-a",
|
|
"sampleTypeIds": ["sample-a"],
|
|
"transfers": [{"transferAmount": 100}],
|
|
}
|
|
),
|
|
ctx=_ctx("specialist"),
|
|
user_info=restricted_user,
|
|
)
|
|
|
|
assert db.query(BizPresignOrder).count() == 0
|
|
assert db.query(BizPresignOrderSampleType).count() == 0
|
|
assert db.query(BizPresignTransfer).count() == 0
|
|
|
|
|
|
def test_create_uses_real_auth_region_id_as_department_for_all_rows():
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import create_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
|
|
data = create_presign_order(
|
|
db,
|
|
body=MarketPresignOrderBody.model_validate(
|
|
{
|
|
"regionId": "region-a",
|
|
"hospitalId": "hospital-a",
|
|
"sampleTypeIds": ["sample-a"],
|
|
"transfers": [{"transferAmount": 100}],
|
|
}
|
|
),
|
|
ctx=_ctx("specialist"),
|
|
user_info=_real_auth_user(),
|
|
)
|
|
|
|
assert db.query(BizPresignOrder).filter_by(id=data["id"]).one().create_dept_id == "dept-from-auth"
|
|
assert db.query(BizPresignOrderSampleType).one().create_dept_id == "dept-from-auth"
|
|
assert db.query(BizPresignTransfer).one().create_dept_id == "dept-from-auth"
|
|
|
|
|
|
@pytest.mark.parametrize("user_id", ["cross-user", "deleted-user", "disabled-user"])
|
|
def test_create_rejects_cross_tenant_deleted_or_disabled_person(user_id):
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import create_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
|
|
with pytest.raises(ValueError, match="用户"):
|
|
create_presign_order(
|
|
db,
|
|
body=MarketPresignOrderBody.model_validate(
|
|
{"regionId": "region-a", "collectorUserId": user_id}
|
|
),
|
|
ctx=_ctx("specialist"),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert db.query(BizPresignOrder).count() == 0
|
|
|
|
|
|
def test_manager_create_validates_sales_user_even_when_assignment_is_valid():
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import create_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
|
|
with pytest.raises(ValueError, match="用户"):
|
|
create_presign_order(
|
|
db,
|
|
body=MarketPresignOrderBody.model_validate(
|
|
{
|
|
"regionId": "region-a",
|
|
"salesUserId": "cross-user",
|
|
"regionManagerUserId": "u1",
|
|
}
|
|
),
|
|
ctx=_ctx("manager"),
|
|
user_info=_user(),
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("payload", "message"),
|
|
[
|
|
({"regionId": "region-cross"}, "采集地区"),
|
|
({"regionId": "region-missing"}, "采集地区"),
|
|
(
|
|
{"regionId": "region-a", "hospitalId": "hospital-b"},
|
|
"医院",
|
|
),
|
|
(
|
|
{"regionId": "region-a", "sampleTypeIds": ["sample-cross"]},
|
|
"样本类型",
|
|
),
|
|
(
|
|
{"regionId": "region-a", "sampleTypeIds": ["sample-missing"]},
|
|
"样本类型",
|
|
),
|
|
],
|
|
)
|
|
def test_create_rejects_invalid_or_cross_tenant_master_data(payload, message):
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import create_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
|
|
with pytest.raises(ValueError, match=message):
|
|
create_presign_order(
|
|
db,
|
|
body=MarketPresignOrderBody.model_validate(payload),
|
|
ctx=_ctx("specialist"),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert db.query(BizPresignOrder).count() == 0
|
|
|
|
|
|
def test_update_validates_resulting_hospital_region_and_sample_types_before_write():
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import update_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
db.add(
|
|
_order(
|
|
"own",
|
|
region_id="region-a",
|
|
sales_user_id="u1",
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
with pytest.raises(ValueError, match="医院"):
|
|
update_presign_order(
|
|
db,
|
|
order_id="own",
|
|
body=MarketPresignOrderBody.model_validate(
|
|
{"hospitalId": "hospital-b", "sampleTypeIds": ["sample-cross"]}
|
|
),
|
|
ctx=_ctx("specialist"),
|
|
user_info=_user(),
|
|
)
|
|
|
|
row = db.query(BizPresignOrder).filter_by(id="own").one()
|
|
assert row.hospital_id is None
|
|
assert db.query(BizPresignOrderSampleType).count() == 0
|
|
|
|
|
|
def test_update_rebuilt_sample_types_keep_current_user_department():
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import update_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
db.add(_order("own", region_id="region-a", sales_user_id="u1"))
|
|
db.commit()
|
|
|
|
update_presign_order(
|
|
db,
|
|
order_id="own",
|
|
body=MarketPresignOrderBody.model_validate(
|
|
{"sampleTypeIds": ["sample-a"]}
|
|
),
|
|
ctx=_ctx("specialist"),
|
|
user_info=_real_auth_user(),
|
|
)
|
|
|
|
child = db.query(BizPresignOrderSampleType).one()
|
|
assert child.create_dept_id == "dept-from-auth"
|
|
|
|
|
|
def test_update_rejects_disabled_person_before_write():
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import update_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
db.add(_order("own", region_id="region-a", sales_user_id="u1"))
|
|
db.commit()
|
|
|
|
with pytest.raises(ValueError, match="用户"):
|
|
update_presign_order(
|
|
db,
|
|
order_id="own",
|
|
body=MarketPresignOrderBody.model_validate(
|
|
{"collectorUserId": "disabled-user"}
|
|
),
|
|
ctx=_ctx("specialist"),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert db.query(BizPresignOrder).filter_by(id="own").one().collector_user_id is None
|
|
|
|
|
|
def test_update_rejects_resulting_assignment_outside_manager_scope():
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import update_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
db.add(_order("managed", region_id="region-a", manager_user_id="u1"))
|
|
db.commit()
|
|
|
|
with pytest.raises(ValueError, match="地区经理范围"):
|
|
update_presign_order(
|
|
db,
|
|
order_id="managed",
|
|
body=MarketPresignOrderBody.model_validate(
|
|
{
|
|
"regionId": "region-b",
|
|
"regionManagerUserId": "u2",
|
|
}
|
|
),
|
|
ctx=_ctx("manager"),
|
|
user_info=_user(),
|
|
)
|
|
|
|
row = db.query(BizPresignOrder).filter_by(id="managed").one()
|
|
assert row.region_id == "region-a"
|
|
assert row.region_manager_user_id == "u1"
|
|
|
|
|
|
def test_convert_rechecks_scope_before_reusing_presign_conversion(monkeypatch):
|
|
from app.crud import presign_order as presign_crud
|
|
from app.services.market_presign import convert_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
db.add_all(
|
|
[
|
|
_order("own", sales_user_id="u1", manager_user_id="manager-a"),
|
|
_order("other", sales_user_id="u2", manager_user_id="manager-b"),
|
|
]
|
|
)
|
|
db.commit()
|
|
calls = []
|
|
|
|
def fake_convert(db, *, order_id, tenant_no, user_id, create_dept_id=""):
|
|
calls.append((order_id, tenant_no, user_id, create_dept_id))
|
|
return {"protocolId": "protocol-1"}
|
|
|
|
monkeypatch.setattr(presign_crud, "convert_to_protocol", fake_convert)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
convert_presign_order(
|
|
db,
|
|
order_id="other",
|
|
ctx=_ctx("specialist"),
|
|
user_info=_user(),
|
|
)
|
|
assert exc_info.value.status_code == 404
|
|
assert calls == []
|
|
|
|
assert convert_presign_order(
|
|
db,
|
|
order_id="own",
|
|
ctx=_ctx("specialist"),
|
|
user_info=_real_auth_user(),
|
|
) == {"protocolId": "protocol-1"}
|
|
assert calls == [("own", "tenant-a", "u1", "dept-from-auth")]
|
|
|
|
|
|
def test_convert_rolls_back_protocol_children_and_presign_status_on_midway_failure(
|
|
monkeypatch,
|
|
):
|
|
from app.crud import collection_protocol as protocol_crud
|
|
from app.services.market_presign import convert_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
order = _order(
|
|
"own",
|
|
region_id="region-a",
|
|
sales_user_id="u1",
|
|
manager_user_id="manager-a",
|
|
)
|
|
order.protocol_no = "P-ATOMIC-1"
|
|
order.hospital_id = "hospital-a"
|
|
db.add(order)
|
|
db.add(
|
|
BizPresignOrderSampleType(
|
|
id="presign-sample",
|
|
presign_order_id="own",
|
|
sample_type_id="sample-a",
|
|
tenant_no="tenant-a",
|
|
create_dept_id="dept-a",
|
|
del_flag=0,
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
def fail_after_collection(*args, **kwargs):
|
|
raise RuntimeError("injected sample failure")
|
|
|
|
monkeypatch.setattr(protocol_crud, "save_samples", fail_after_collection)
|
|
|
|
with pytest.raises(RuntimeError, match="injected sample failure"):
|
|
convert_presign_order(
|
|
db,
|
|
order_id="own",
|
|
ctx=_ctx("specialist"),
|
|
user_info=_user(),
|
|
)
|
|
|
|
db.expire_all()
|
|
unchanged = db.query(BizPresignOrder).filter_by(id="own").one()
|
|
assert unchanged.order_status == "active"
|
|
assert unchanged.converted_protocol_id is None
|
|
assert db.query(BizCollectionProtocol).count() == 0
|
|
assert db.query(BizProtocolPartyCustomer).count() == 0
|
|
assert db.query(BizProtocolCollectionInfo).count() == 0
|
|
assert db.query(BizProtocolSample).count() == 0
|
|
|
|
|
|
def test_convert_rejects_invalid_person_before_creating_protocol():
|
|
from app.services.market_presign import convert_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
order = _order("own", sales_user_id="u1")
|
|
order.protocol_no = "P-INVALID-USER"
|
|
order.collector_user_id = "cross-user"
|
|
db.add(order)
|
|
db.commit()
|
|
|
|
with pytest.raises(ValueError, match="用户"):
|
|
convert_presign_order(
|
|
db,
|
|
order_id="own",
|
|
ctx=_ctx("specialist"),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert db.query(BizCollectionProtocol).count() == 0
|
|
|
|
|
|
def test_convert_fetches_presign_row_with_for_update(monkeypatch):
|
|
from app.crud import presign_order as presign_crud
|
|
|
|
class QueryProbe:
|
|
def __init__(self):
|
|
self.locked = False
|
|
|
|
def filter(self, *args):
|
|
return self
|
|
|
|
def with_for_update(self):
|
|
self.locked = True
|
|
return self
|
|
|
|
def populate_existing(self):
|
|
return self
|
|
|
|
def first(self):
|
|
return object()
|
|
|
|
class DbProbe:
|
|
def __init__(self):
|
|
self.query_probe = QueryProbe()
|
|
|
|
def query(self, model):
|
|
return self.query_probe
|
|
|
|
db = DbProbe()
|
|
assert presign_crud._get_order_or_raise(
|
|
db, order_id="order-1", tenant_no="tenant-a", for_update=True
|
|
) is not None
|
|
assert db.query_probe.locked is True
|
|
|
|
|
|
def test_locked_convert_read_refreshes_preloaded_stale_identity(tmp_path):
|
|
from app.crud import presign_order as presign_crud
|
|
|
|
engine = create_engine(f"sqlite:///{tmp_path / 'stale.db'}")
|
|
Base.metadata.create_all(engine, tables=[BizPresignOrder.__table__])
|
|
Session = sessionmaker(bind=engine)
|
|
first = Session()
|
|
second = Session()
|
|
try:
|
|
first.add(_order("race", sales_user_id="u1"))
|
|
first.commit()
|
|
cached = first.query(BizPresignOrder).filter_by(id="race").one()
|
|
assert cached.order_status == "active"
|
|
|
|
concurrent = second.query(BizPresignOrder).filter_by(id="race").one()
|
|
concurrent.order_status = "converted"
|
|
concurrent.converted_protocol_id = "protocol-existing"
|
|
second.commit()
|
|
|
|
locked = presign_crud._get_order_or_raise(
|
|
first,
|
|
order_id="race",
|
|
tenant_no="tenant-a",
|
|
for_update=True,
|
|
)
|
|
assert locked is cached
|
|
assert locked.order_status == "converted"
|
|
assert locked.converted_protocol_id == "protocol-existing"
|
|
with pytest.raises(ValueError, match="已转正式"):
|
|
presign_crud.assert_presign_convertible(
|
|
order_status=locked.order_status
|
|
)
|
|
finally:
|
|
first.close()
|
|
second.close()
|
|
|
|
|
|
def test_convert_translates_unique_race_to_already_converted_domain_error(
|
|
monkeypatch,
|
|
):
|
|
from app.crud import collection_protocol as protocol_crud
|
|
from app.services.market_presign import convert_presign_order
|
|
|
|
db = _session()
|
|
_add_master_data(db)
|
|
order = _order("own", sales_user_id="u1", manager_user_id="manager-a")
|
|
order.protocol_no = "P-RACE"
|
|
db.add(order)
|
|
db.commit()
|
|
|
|
def duplicate_source(*args, **kwargs):
|
|
raise IntegrityError(
|
|
"INSERT INTO biz_collection_protocol",
|
|
{},
|
|
Exception("Duplicate entry for key 'uk_collection_protocol_presign_source'"),
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
protocol_crud, "create_collection_protocol", duplicate_source
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="已转正式"):
|
|
convert_presign_order(
|
|
db,
|
|
order_id="own",
|
|
ctx=_ctx("specialist"),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert db.query(BizPresignOrder).filter_by(id="own").one().order_status == "active"
|
|
|
|
|
|
def test_market_presign_bodies_forbid_unknown_fields_and_validate_transfer_limits():
|
|
from pydantic import ValidationError
|
|
|
|
from app.schemas.market_mobile import (
|
|
MarketPresignOrderBody,
|
|
MarketPresignTransferBody,
|
|
)
|
|
|
|
with pytest.raises(ValidationError, match='tenantNo'):
|
|
MarketPresignOrderBody.model_validate({'tenantNo': 'tenant-b'})
|
|
assert (
|
|
MarketPresignOrderBody.model_validate(
|
|
{'orderStatus': 'void'}
|
|
).order_status
|
|
== 'void'
|
|
)
|
|
|
|
valid = MarketPresignTransferBody.model_validate(
|
|
{
|
|
'transferAmount': '123.45',
|
|
'transferTime': '2026-10-11 12:34:56',
|
|
'transferType': 'bank',
|
|
'transferAttachments': 'a' * 4000,
|
|
'remark': 'r' * 512,
|
|
}
|
|
)
|
|
assert str(valid.transfer_amount) == '123.45'
|
|
assert valid.transfer_time == '2026-10-11 12:34:56'
|
|
assert (
|
|
MarketPresignTransferBody.model_validate(
|
|
{'transferTime': '2026-10-11T12:34'}
|
|
).transfer_time
|
|
== '2026-10-11T12:34'
|
|
)
|
|
for payload in (
|
|
{'unknown': 'value'},
|
|
{'transferAmount': 0},
|
|
{'transferTime': 'not-a-datetime'},
|
|
{'transferTime': '2026-10-11'},
|
|
{'transferTime': '2026-10-11 12:34'},
|
|
{'transferTime': '2026-10-11T12'},
|
|
{'transferTime': '2026-10-11T12:34+08:00'},
|
|
{'transferType': 'x' * 17},
|
|
{'transferAttachments': 'x' * 4001},
|
|
{'remark': 'x' * 513},
|
|
):
|
|
with pytest.raises(ValidationError):
|
|
MarketPresignTransferBody.model_validate(payload)
|
|
|
|
|
|
def test_market_presign_order_body_accepts_database_sized_values():
|
|
from decimal import Decimal
|
|
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
|
|
body = MarketPresignOrderBody.model_validate(
|
|
{
|
|
'orderDate': '20261011',
|
|
'protocolNo': 'P' * 64,
|
|
'motherName': '母' * 64,
|
|
'expectedDeliveryDate': '20261231',
|
|
'regionId': 'r' * 64,
|
|
'hospitalId': 'h' * 64,
|
|
'sampleCollectTime': '2026-10-11 12:34:56',
|
|
'payMode': 'x' * 8,
|
|
'contractAmount': '9999999999999999.99',
|
|
'collectorUserId': 'c' * 64,
|
|
'salesUserId': 's' * 64,
|
|
'regionManagerUserId': 'm' * 64,
|
|
'remark': '备' * 512,
|
|
'sampleTypeIds': [f'sample-{index}' for index in range(100)],
|
|
}
|
|
)
|
|
|
|
assert body.order_date == '20261011'
|
|
assert body.expected_delivery_date == '20261231'
|
|
assert body.contract_amount == Decimal('9999999999999999.99')
|
|
assert len(body.sample_type_ids or []) == 100
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'payload',
|
|
[
|
|
{'orderDate': '2026-10-11'},
|
|
{'orderDate': '20260230'},
|
|
{'expectedDeliveryDate': '20261301'},
|
|
{'sampleCollectTime': 'not-a-datetime'},
|
|
{'protocolNo': 'p' * 65},
|
|
{'motherName': 'm' * 65},
|
|
{'regionId': 'r' * 65},
|
|
{'hospitalId': 'h' * 65},
|
|
{'collectorUserId': 'c' * 65},
|
|
{'salesUserId': 's' * 65},
|
|
{'regionManagerUserId': 'm' * 65},
|
|
{'payMode': 'p' * 9},
|
|
{'remark': 'r' * 513},
|
|
{'sampleTypeIds': ['s' * 65]},
|
|
{'sampleTypeIds': [f'sample-{index}' for index in range(101)]},
|
|
{'contractAmount': 'NaN'},
|
|
{'contractAmount': 'Infinity'},
|
|
{'contractAmount': '-0.01'},
|
|
{'contractAmount': '10000000000000000.00'},
|
|
{'contractAmount': '1.001'},
|
|
],
|
|
)
|
|
def test_market_presign_order_body_rejects_values_outside_database_bounds(
|
|
payload,
|
|
):
|
|
from pydantic import ValidationError
|
|
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
|
|
with pytest.raises(ValidationError):
|
|
MarketPresignOrderBody.model_validate(payload)
|
|
|
|
|
|
@pytest.mark.parametrize('status', ['active', 'void'])
|
|
def test_specialist_can_soft_delete_own_non_converted_presign(status):
|
|
from app.services.market_presign import delete_presign_order
|
|
|
|
db = _session()
|
|
order = _order('own', sales_user_id='u1')
|
|
order.order_status = status
|
|
db.add(order)
|
|
db.commit()
|
|
|
|
delete_presign_order(
|
|
db,
|
|
order_id='own',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
deleted = db.query(BizPresignOrder).filter_by(id='own').one()
|
|
assert deleted.del_flag == 1
|
|
assert deleted.update_by == 'u1'
|
|
|
|
|
|
def test_specialist_cannot_delete_another_specialists_presign():
|
|
from app.services.market_presign import delete_presign_order
|
|
|
|
db = _session()
|
|
db.add(_order('other', sales_user_id='u2'))
|
|
db.commit()
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
delete_presign_order(
|
|
db,
|
|
order_id='other',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert db.query(BizPresignOrder).filter_by(id='other').one().del_flag == 0
|
|
|
|
|
|
def test_manager_delete_is_limited_to_manager_scope():
|
|
from app.services.market_presign import delete_presign_order
|
|
|
|
db = _session()
|
|
db.add(
|
|
BizCollectionRegionStaff(
|
|
id='manager-region',
|
|
region_id='region-led',
|
|
user_id='u1',
|
|
is_leader=1,
|
|
tenant_no='tenant-a',
|
|
create_dept_id='dept-a',
|
|
del_flag=0,
|
|
)
|
|
)
|
|
db.add_all(
|
|
[
|
|
_order(
|
|
'managed',
|
|
region_id='region-led',
|
|
sales_user_id='u2',
|
|
manager_user_id='u2',
|
|
),
|
|
_order(
|
|
'outside',
|
|
region_id='region-other',
|
|
sales_user_id='u2',
|
|
manager_user_id='u2',
|
|
),
|
|
]
|
|
)
|
|
db.commit()
|
|
|
|
delete_presign_order(
|
|
db,
|
|
order_id='managed',
|
|
ctx=_ctx('manager'),
|
|
user_info=_user(),
|
|
)
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
delete_presign_order(
|
|
db,
|
|
order_id='outside',
|
|
ctx=_ctx('manager'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert db.query(BizPresignOrder).filter_by(id='managed').one().del_flag == 1
|
|
assert db.query(BizPresignOrder).filter_by(id='outside').one().del_flag == 0
|
|
|
|
|
|
def test_converted_presign_cannot_be_deleted():
|
|
from app.services.market_presign import delete_presign_order
|
|
|
|
db = _session()
|
|
order = _order('converted', sales_user_id='u1')
|
|
order.order_status = 'converted'
|
|
db.add(order)
|
|
db.commit()
|
|
|
|
with pytest.raises(ValueError, match='已转|converted'):
|
|
delete_presign_order(
|
|
db,
|
|
order_id='converted',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert db.query(BizPresignOrder).filter_by(id='converted').one().del_flag == 0
|
|
|
|
|
|
@pytest.mark.parametrize('scope_kind', ['specialist', 'manager'])
|
|
@pytest.mark.parametrize(
|
|
('action', 'pay_status'),
|
|
[
|
|
('create', 'dfj'),
|
|
('update', 'dfj'),
|
|
('delete', 'dfj'),
|
|
('submit', 'dfj'),
|
|
('withdraw', 'jfqrz'),
|
|
],
|
|
)
|
|
def test_presign_transfer_writes_recheck_parent_order_scope(
|
|
scope_kind, action, pay_status
|
|
):
|
|
from app.schemas.market_mobile import MarketPresignTransferBody
|
|
from app.services.market_presign import (
|
|
create_presign_transfer,
|
|
delete_presign_transfer,
|
|
submit_presign_transfer,
|
|
update_presign_transfer,
|
|
withdraw_presign_transfer,
|
|
)
|
|
|
|
db = _session()
|
|
db.add(
|
|
_order(
|
|
'outside',
|
|
region_id='region-outside',
|
|
sales_user_id='u2',
|
|
manager_user_id='u2',
|
|
)
|
|
)
|
|
db.add(_transfer('outside-transfer', order_id='outside', pay_status=pay_status))
|
|
db.commit()
|
|
body = MarketPresignTransferBody(
|
|
transferAmount=200,
|
|
transferType='bank',
|
|
)
|
|
|
|
calls = {
|
|
'create': lambda: create_presign_transfer(
|
|
db,
|
|
order_id='outside',
|
|
body=body,
|
|
ctx=_ctx(scope_kind),
|
|
user_info=_user(),
|
|
),
|
|
'update': lambda: update_presign_transfer(
|
|
db,
|
|
transfer_id='outside-transfer',
|
|
body=body,
|
|
ctx=_ctx(scope_kind),
|
|
user_info=_user(),
|
|
),
|
|
'delete': lambda: delete_presign_transfer(
|
|
db,
|
|
transfer_id='outside-transfer',
|
|
ctx=_ctx(scope_kind),
|
|
user_info=_user(),
|
|
),
|
|
'submit': lambda: submit_presign_transfer(
|
|
db,
|
|
transfer_id='outside-transfer',
|
|
ctx=_ctx(scope_kind),
|
|
user_info=_user(),
|
|
),
|
|
'withdraw': lambda: withdraw_presign_transfer(
|
|
db,
|
|
transfer_id='outside-transfer',
|
|
ctx=_ctx(scope_kind),
|
|
user_info=_user(),
|
|
),
|
|
}
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
calls[action]()
|
|
|
|
assert exc_info.value.status_code == 404
|
|
row = db.query(BizPresignTransfer).filter_by(id='outside-transfer').one()
|
|
assert row.del_flag == 0
|
|
assert row.pay_status == pay_status
|
|
|
|
|
|
def test_dfj_presign_transfer_can_edit_submit_withdraw_and_soft_delete():
|
|
from app.schemas.market_mobile import MarketPresignTransferBody
|
|
from app.services.market_presign import (
|
|
create_presign_transfer,
|
|
delete_presign_transfer,
|
|
submit_presign_transfer,
|
|
update_presign_transfer,
|
|
withdraw_presign_transfer,
|
|
)
|
|
|
|
db = _session()
|
|
db.add(_order('own', sales_user_id='u1'))
|
|
db.commit()
|
|
|
|
created = create_presign_transfer(
|
|
db,
|
|
order_id='own',
|
|
body=MarketPresignTransferBody(
|
|
transferAmount='100.50',
|
|
transferType='bank',
|
|
),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_real_auth_user(),
|
|
)
|
|
transfer_id = created['id']
|
|
updated = update_presign_transfer(
|
|
db,
|
|
transfer_id=transfer_id,
|
|
body=MarketPresignTransferBody(transferAmount='200.25', remark='更新'),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
assert updated['transferAmount'] == '200.25'
|
|
assert updated['remark'] == '更新'
|
|
assert (
|
|
db.query(BizPresignTransfer).filter_by(id=transfer_id).one().create_dept_id
|
|
== 'dept-from-auth'
|
|
)
|
|
|
|
submitted = submit_presign_transfer(
|
|
db,
|
|
transfer_id=transfer_id,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
assert submitted['payStatus'] == 'jfqrz'
|
|
with pytest.raises(ValueError, match='状态|待缴费|已缴费'):
|
|
delete_presign_transfer(
|
|
db,
|
|
transfer_id=transfer_id,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
withdrawn = withdraw_presign_transfer(
|
|
db,
|
|
transfer_id=transfer_id,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
assert withdrawn['payStatus'] == 'dfj'
|
|
delete_presign_transfer(
|
|
db,
|
|
transfer_id=transfer_id,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
deleted = db.query(BizPresignTransfer).filter_by(id=transfer_id).one()
|
|
assert deleted.del_flag == 1
|
|
assert deleted.update_by == 'u1'
|
|
|
|
|
|
def test_jfqrz_presign_transfer_only_allows_withdraw():
|
|
from app.schemas.market_mobile import MarketPresignTransferBody
|
|
from app.services.market_presign import (
|
|
delete_presign_transfer,
|
|
submit_presign_transfer,
|
|
update_presign_transfer,
|
|
withdraw_presign_transfer,
|
|
)
|
|
|
|
db = _session()
|
|
db.add(_order('own', sales_user_id='u1'))
|
|
db.add(_transfer('pending', pay_status='jfqrz'))
|
|
db.commit()
|
|
|
|
for action in (
|
|
lambda: update_presign_transfer(
|
|
db,
|
|
transfer_id='pending',
|
|
body=MarketPresignTransferBody(remark='不允许'),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
lambda: delete_presign_transfer(
|
|
db,
|
|
transfer_id='pending',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
lambda: submit_presign_transfer(
|
|
db,
|
|
transfer_id='pending',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
):
|
|
with pytest.raises(ValueError):
|
|
action()
|
|
|
|
assert withdraw_presign_transfer(
|
|
db,
|
|
transfer_id='pending',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)['payStatus'] == 'dfj'
|
|
|
|
|
|
@pytest.mark.parametrize('action', ['update', 'delete', 'submit', 'withdraw'])
|
|
def test_yjf_presign_transfer_rejects_all_market_writes(action):
|
|
from app.schemas.market_mobile import MarketPresignTransferBody
|
|
from app.services.market_presign import (
|
|
delete_presign_transfer,
|
|
submit_presign_transfer,
|
|
update_presign_transfer,
|
|
withdraw_presign_transfer,
|
|
)
|
|
|
|
db = _session()
|
|
db.add(_order('own', sales_user_id='u1'))
|
|
db.add(_transfer('paid', pay_status='yjf'))
|
|
db.commit()
|
|
calls = {
|
|
'update': lambda: update_presign_transfer(
|
|
db,
|
|
transfer_id='paid',
|
|
body=MarketPresignTransferBody(remark='不允许'),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
'delete': lambda: delete_presign_transfer(
|
|
db,
|
|
transfer_id='paid',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
'submit': lambda: submit_presign_transfer(
|
|
db,
|
|
transfer_id='paid',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
'withdraw': lambda: withdraw_presign_transfer(
|
|
db,
|
|
transfer_id='paid',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
}
|
|
|
|
with pytest.raises(ValueError):
|
|
calls[action]()
|
|
|
|
|
|
@pytest.mark.parametrize('order_status', ['converted', 'void'])
|
|
@pytest.mark.parametrize('action', ['create', 'update', 'delete', 'submit', 'withdraw'])
|
|
def test_non_writable_presign_order_rejects_every_transfer_write(
|
|
order_status, action
|
|
):
|
|
from app.schemas.market_mobile import MarketPresignTransferBody
|
|
from app.services.market_presign import (
|
|
create_presign_transfer,
|
|
delete_presign_transfer,
|
|
submit_presign_transfer,
|
|
update_presign_transfer,
|
|
withdraw_presign_transfer,
|
|
)
|
|
|
|
db = _session()
|
|
order = _order('own', sales_user_id='u1')
|
|
order.order_status = order_status
|
|
db.add(order)
|
|
db.add(
|
|
_transfer(
|
|
'transfer',
|
|
pay_status='jfqrz' if action == 'withdraw' else 'dfj',
|
|
)
|
|
)
|
|
db.commit()
|
|
body = MarketPresignTransferBody(
|
|
transferAmount=100,
|
|
transferType='bank',
|
|
)
|
|
calls = {
|
|
'create': lambda: create_presign_transfer(
|
|
db,
|
|
order_id='own',
|
|
body=body,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
'update': lambda: update_presign_transfer(
|
|
db,
|
|
transfer_id='transfer',
|
|
body=body,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
'delete': lambda: delete_presign_transfer(
|
|
db,
|
|
transfer_id='transfer',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
'submit': lambda: submit_presign_transfer(
|
|
db,
|
|
transfer_id='transfer',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
'withdraw': lambda: withdraw_presign_transfer(
|
|
db,
|
|
transfer_id='transfer',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
}
|
|
|
|
with pytest.raises(ValueError, match='订单|状态|转账'):
|
|
calls[action]()
|
|
|
|
|
|
@pytest.mark.parametrize('action', ['create', 'update'])
|
|
def test_presign_transfer_rejects_unregistered_attachment_url(action):
|
|
from app.schemas.market_mobile import MarketPresignTransferBody
|
|
from app.services.market_presign import (
|
|
create_presign_transfer,
|
|
update_presign_transfer,
|
|
)
|
|
|
|
db = _session()
|
|
db.add(_order('own', sales_user_id='u1'))
|
|
db.add(_transfer('draft'))
|
|
db.commit()
|
|
body = MarketPresignTransferBody(
|
|
transferAttachments='https://evil.test/proof.pdf'
|
|
)
|
|
|
|
with pytest.raises(ValueError, match='附件'):
|
|
if action == 'create':
|
|
create_presign_transfer(
|
|
db,
|
|
order_id='own',
|
|
body=body,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
else:
|
|
update_presign_transfer(
|
|
db,
|
|
transfer_id='draft',
|
|
body=body,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
|
|
def test_presign_transfer_attachments_bind_to_parent_order():
|
|
from app.schemas.market_mobile import MarketPresignTransferBody
|
|
from app.services.market_presign import (
|
|
create_presign_transfer,
|
|
update_presign_transfer,
|
|
)
|
|
|
|
db = _session()
|
|
db.add(_order('own', sales_user_id='u1'))
|
|
db.add_all(
|
|
[
|
|
BizAttachment(
|
|
id='attachment-create',
|
|
business_type='market:specialist:presign:staging',
|
|
original_name='create.pdf',
|
|
file_url='presign/create.pdf',
|
|
create_by='u1',
|
|
create_dept_id='dept-a',
|
|
tenant_no='tenant-a',
|
|
del_flag=0,
|
|
),
|
|
BizAttachment(
|
|
id='attachment-update',
|
|
business_type='market:specialist:presign:staging',
|
|
original_name='update.pdf',
|
|
file_url='presign/update.pdf',
|
|
create_by='u1',
|
|
create_dept_id='dept-a',
|
|
tenant_no='tenant-a',
|
|
del_flag=0,
|
|
),
|
|
]
|
|
)
|
|
db.commit()
|
|
|
|
created = create_presign_transfer(
|
|
db,
|
|
order_id='own',
|
|
body=MarketPresignTransferBody(
|
|
transferAmount=100,
|
|
transferType='bank',
|
|
transferAttachments='attachment-create',
|
|
),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
updated = update_presign_transfer(
|
|
db,
|
|
transfer_id=created['id'],
|
|
body=MarketPresignTransferBody(
|
|
transferAttachments='attachment-create,attachment-update'
|
|
),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert updated['transferAttachments'] == (
|
|
'presign/create.pdf,presign/update.pdf'
|
|
)
|
|
attachments = db.query(BizAttachment).order_by(BizAttachment.id).all()
|
|
assert {row.business_id for row in attachments} == {'own'}
|
|
assert {row.business_type for row in attachments} == {
|
|
'market:specialist:presign:bound'
|
|
}
|
|
|
|
|
|
def test_stale_withdraw_does_not_overwrite_later_finance_confirmation(tmp_path):
|
|
from app.crud import presign_order as presign_crud
|
|
from app.services.market_presign import withdraw_presign_transfer
|
|
|
|
Session = _session_factory(f"sqlite:///{tmp_path / 'withdraw-race.db'}")
|
|
market_db = Session()
|
|
finance_db = Session()
|
|
try:
|
|
market_db.add(_order('own', sales_user_id='u1'))
|
|
market_db.add(_transfer('transfer', pay_status='jfqrz'))
|
|
market_db.commit()
|
|
stale = market_db.query(BizPresignTransfer).filter_by(id='transfer').one()
|
|
assert stale.pay_status == 'jfqrz'
|
|
|
|
presign_crud.confirm_receipt(
|
|
finance_db,
|
|
transfer_id='transfer',
|
|
tenant_no='tenant-a',
|
|
user_id='finance-user',
|
|
)
|
|
|
|
with pytest.raises(ValueError, match='状态|撤回'):
|
|
withdraw_presign_transfer(
|
|
market_db,
|
|
transfer_id='transfer',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
finance_db.expire_all()
|
|
current = finance_db.query(BizPresignTransfer).filter_by(id='transfer').one()
|
|
assert current.pay_status == 'yjf'
|
|
assert current.update_by == 'finance-user'
|
|
finally:
|
|
market_db.close()
|
|
finance_db.close()
|
|
|
|
|
|
def test_stale_finance_confirmation_does_not_overwrite_later_mobile_withdraw(
|
|
tmp_path,
|
|
):
|
|
from app.crud import presign_order as presign_crud
|
|
from app.services.market_presign import withdraw_presign_transfer
|
|
|
|
Session = _session_factory(
|
|
f"sqlite:///{tmp_path / 'finance-confirm-after-withdraw-race.db'}"
|
|
)
|
|
finance_db = Session()
|
|
market_db = Session()
|
|
try:
|
|
finance_db.add(_order('own', sales_user_id='u1'))
|
|
finance_db.add(_transfer('transfer', pay_status='jfqrz'))
|
|
finance_db.commit()
|
|
stale = finance_db.query(BizPresignTransfer).filter_by(id='transfer').one()
|
|
assert stale.pay_status == 'jfqrz'
|
|
|
|
withdrawn = withdraw_presign_transfer(
|
|
market_db,
|
|
transfer_id='transfer',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
assert withdrawn['payStatus'] == 'dfj'
|
|
|
|
with pytest.raises(ValueError, match='状态'):
|
|
presign_crud.confirm_receipt(
|
|
finance_db,
|
|
transfer_id='transfer',
|
|
tenant_no='tenant-a',
|
|
user_id='finance-user',
|
|
)
|
|
|
|
market_db.expire_all()
|
|
current = market_db.query(BizPresignTransfer).filter_by(id='transfer').one()
|
|
assert current.pay_status == 'dfj'
|
|
assert current.update_by == 'u1'
|
|
finally:
|
|
finance_db.close()
|
|
market_db.close()
|
|
|
|
|
|
@pytest.mark.parametrize('action', ['update', 'submit'])
|
|
def test_stale_web_draft_action_does_not_mutate_later_paid_transfer(
|
|
tmp_path, action
|
|
):
|
|
from app.crud import presign_order as presign_crud
|
|
|
|
Session = _session_factory(
|
|
f"sqlite:///{tmp_path / f'web-{action}-after-paid-race.db'}"
|
|
)
|
|
web_db = Session()
|
|
finance_db = Session()
|
|
try:
|
|
web_db.add(_order('own', sales_user_id='u1'))
|
|
transfer = _transfer('transfer', pay_status='dfj')
|
|
transfer.remark = 'original'
|
|
web_db.add(transfer)
|
|
web_db.commit()
|
|
stale = web_db.query(BizPresignTransfer).filter_by(id='transfer').one()
|
|
assert stale.pay_status == 'dfj'
|
|
|
|
presign_crud.submit_receipt_finance(
|
|
finance_db,
|
|
transfer_id='transfer',
|
|
tenant_no='tenant-a',
|
|
user_id='finance-user',
|
|
)
|
|
presign_crud.confirm_receipt(
|
|
finance_db,
|
|
transfer_id='transfer',
|
|
tenant_no='tenant-a',
|
|
user_id='finance-user',
|
|
)
|
|
|
|
calls = {
|
|
'update': lambda: presign_crud.update_transfer_draft(
|
|
web_db,
|
|
transfer_id='transfer',
|
|
tenant_no='tenant-a',
|
|
user_id='web-user',
|
|
fields={'remark': 'stale-write'},
|
|
),
|
|
'submit': lambda: presign_crud.submit_receipt_finance(
|
|
web_db,
|
|
transfer_id='transfer',
|
|
tenant_no='tenant-a',
|
|
user_id='web-user',
|
|
),
|
|
}
|
|
with pytest.raises(ValueError, match='状态'):
|
|
calls[action]()
|
|
|
|
finance_db.expire_all()
|
|
current = finance_db.query(BizPresignTransfer).filter_by(id='transfer').one()
|
|
assert current.pay_status == 'yjf'
|
|
assert current.remark == 'original'
|
|
assert current.update_by == 'finance-user'
|
|
finally:
|
|
web_db.close()
|
|
finance_db.close()
|
|
|
|
|
|
@pytest.mark.parametrize('action', ['withdraw', 'confirm', 'reject'])
|
|
def test_stale_shared_pending_action_does_not_mutate_later_draft_transfer(
|
|
tmp_path, action
|
|
):
|
|
from app.crud import presign_order as presign_crud
|
|
from app.services.market_presign import withdraw_presign_transfer
|
|
|
|
Session = _session_factory(
|
|
f"sqlite:///{tmp_path / f'shared-{action}-after-withdraw-race.db'}"
|
|
)
|
|
stale_db = Session()
|
|
market_db = Session()
|
|
try:
|
|
stale_db.add(_order('own', sales_user_id='u1'))
|
|
stale_db.add(_transfer('transfer', pay_status='jfqrz'))
|
|
stale_db.commit()
|
|
stale = stale_db.query(BizPresignTransfer).filter_by(id='transfer').one()
|
|
assert stale.pay_status == 'jfqrz'
|
|
|
|
withdraw_presign_transfer(
|
|
market_db,
|
|
transfer_id='transfer',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
calls = {
|
|
'withdraw': lambda: presign_crud.withdraw_receipt(
|
|
stale_db,
|
|
transfer_id='transfer',
|
|
tenant_no='tenant-a',
|
|
user_id='stale-user',
|
|
),
|
|
'confirm': lambda: presign_crud.confirm_receipt(
|
|
stale_db,
|
|
transfer_id='transfer',
|
|
tenant_no='tenant-a',
|
|
user_id='stale-user',
|
|
),
|
|
'reject': lambda: presign_crud.reject_receipt(
|
|
stale_db,
|
|
transfer_id='transfer',
|
|
tenant_no='tenant-a',
|
|
user_id='stale-user',
|
|
),
|
|
}
|
|
with pytest.raises(ValueError, match='状态'):
|
|
calls[action]()
|
|
|
|
market_db.expire_all()
|
|
current = market_db.query(BizPresignTransfer).filter_by(id='transfer').one()
|
|
assert current.pay_status == 'dfj'
|
|
assert current.update_by == 'u1'
|
|
finally:
|
|
stale_db.close()
|
|
market_db.close()
|
|
|
|
|
|
def test_mobile_transfer_update_serializes_before_commit(monkeypatch):
|
|
from app.crud import presign_order as presign_crud
|
|
from app.schemas.market_mobile import MarketPresignTransferBody
|
|
from app.services.market_presign import update_presign_transfer
|
|
|
|
db = _session()
|
|
db.add(_order('own', sales_user_id='u1'))
|
|
db.add(_transfer('transfer', pay_status='dfj'))
|
|
db.commit()
|
|
|
|
committed = False
|
|
original_commit = db.commit
|
|
original_to_dict = presign_crud._transfer_to_dict
|
|
|
|
def tracked_commit():
|
|
nonlocal committed
|
|
committed = True
|
|
original_commit()
|
|
|
|
def snapshot_before_commit(*args, **kwargs):
|
|
assert not committed
|
|
return original_to_dict(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(db, 'commit', tracked_commit)
|
|
monkeypatch.setattr(presign_crud, '_transfer_to_dict', snapshot_before_commit)
|
|
|
|
result = update_presign_transfer(
|
|
db,
|
|
transfer_id='transfer',
|
|
body=MarketPresignTransferBody(remark='updated'),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert committed
|
|
assert result['remark'] == 'updated'
|
|
|
|
|
|
@pytest.mark.parametrize('action', ['update', 'delete', 'submit'])
|
|
def test_stale_dfj_action_does_not_mutate_later_paid_transfer(tmp_path, action):
|
|
from app.crud import presign_order as presign_crud
|
|
from app.schemas.market_mobile import MarketPresignTransferBody
|
|
from app.services.market_presign import (
|
|
delete_presign_transfer,
|
|
submit_presign_transfer,
|
|
update_presign_transfer,
|
|
)
|
|
|
|
Session = _session_factory(
|
|
f"sqlite:///{tmp_path / f'transfer-{action}-race.db'}"
|
|
)
|
|
market_db = Session()
|
|
finance_db = Session()
|
|
try:
|
|
market_db.add(_order('own', sales_user_id='u1'))
|
|
transfer = _transfer('transfer', pay_status='dfj')
|
|
transfer.remark = 'original'
|
|
market_db.add(transfer)
|
|
market_db.commit()
|
|
stale = market_db.query(BizPresignTransfer).filter_by(id='transfer').one()
|
|
assert stale.pay_status == 'dfj'
|
|
|
|
presign_crud.submit_receipt_finance(
|
|
finance_db,
|
|
transfer_id='transfer',
|
|
tenant_no='tenant-a',
|
|
user_id='finance-user',
|
|
)
|
|
presign_crud.confirm_receipt(
|
|
finance_db,
|
|
transfer_id='transfer',
|
|
tenant_no='tenant-a',
|
|
user_id='finance-user',
|
|
)
|
|
calls = {
|
|
'update': lambda: update_presign_transfer(
|
|
market_db,
|
|
transfer_id='transfer',
|
|
body=MarketPresignTransferBody(remark='stale-write'),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
'delete': lambda: delete_presign_transfer(
|
|
market_db,
|
|
transfer_id='transfer',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
'submit': lambda: submit_presign_transfer(
|
|
market_db,
|
|
transfer_id='transfer',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
}
|
|
|
|
with pytest.raises(ValueError, match='状态|待缴费|提交'):
|
|
calls[action]()
|
|
|
|
finance_db.expire_all()
|
|
current = finance_db.query(BizPresignTransfer).filter_by(id='transfer').one()
|
|
assert current.pay_status == 'yjf'
|
|
assert current.del_flag == 0
|
|
assert current.remark == 'original'
|
|
finally:
|
|
market_db.close()
|
|
finance_db.close()
|
|
|
|
|
|
@pytest.mark.parametrize('action', ['update', 'delete'])
|
|
def test_stale_order_write_does_not_mutate_later_conversion(tmp_path, action):
|
|
from app.schemas.market_mobile import MarketPresignOrderBody
|
|
from app.services.market_presign import (
|
|
convert_presign_order,
|
|
delete_presign_order,
|
|
update_presign_order,
|
|
)
|
|
|
|
Session = _session_factory(
|
|
f"sqlite:///{tmp_path / f'order-{action}-race.db'}"
|
|
)
|
|
market_db = Session()
|
|
convert_db = Session()
|
|
try:
|
|
_add_master_data(market_db)
|
|
order = _order('own', sales_user_id='u1')
|
|
order.protocol_no = f'P-RACE-{action}'
|
|
market_db.add(order)
|
|
market_db.commit()
|
|
stale = market_db.query(BizPresignOrder).filter_by(id='own').one()
|
|
assert stale.order_status == 'active'
|
|
|
|
convert_presign_order(
|
|
convert_db,
|
|
order_id='own',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
calls = {
|
|
'update': lambda: update_presign_order(
|
|
market_db,
|
|
order_id='own',
|
|
body=MarketPresignOrderBody(remark='stale-write'),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
'delete': lambda: delete_presign_order(
|
|
market_db,
|
|
order_id='own',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
),
|
|
}
|
|
|
|
with pytest.raises(ValueError, match='不可|已转'):
|
|
calls[action]()
|
|
|
|
convert_db.expire_all()
|
|
current = convert_db.query(BizPresignOrder).filter_by(id='own').one()
|
|
assert current.order_status == 'converted'
|
|
assert current.converted_protocol_id
|
|
assert current.del_flag == 0
|
|
assert current.remark is None
|
|
finally:
|
|
market_db.close()
|
|
convert_db.close()
|
|
|
|
|
|
def test_conversion_unique_index_migration_is_mysql57_idempotent():
|
|
from pathlib import Path
|
|
|
|
sql = (
|
|
Path(__file__).parents[2]
|
|
/ "docs/sql/migrations/2026-10-10-presign-convert-unique.sql"
|
|
).read_text(encoding="utf-8")
|
|
normalized = " ".join(sql.lower().replace("`", "").split())
|
|
|
|
assert "information_schema.statistics" in normalized
|
|
assert "create procedure ensure_presign_convert_unique" in normalized
|
|
assert "call ensure_presign_convert_unique()" in normalized
|
|
assert "group by tenant_no, source_presign_order_id having count(*) > 1" in normalized
|
|
assert "signal sqlstate '45000'" in normalized
|
|
assert "duplicate" in normalized
|
|
assert "non_unique" in normalized
|
|
assert "group_concat(column_name order by seq_in_index" in normalized
|
|
assert "tenant_no,source_presign_order_id" in normalized
|
|
assert "wrong index" in normalized
|
|
assert "unique index uk_collection_protocol_presign_source" in normalized
|
|
|
|
|
|
def test_market_presign_routes_are_fixed_and_expose_scoped_transfer_actions():
|
|
from app.api.market_mobile import router
|
|
|
|
actual = {
|
|
(method, route.path)
|
|
for route in router.routes
|
|
if "/presign-orders" in route.path
|
|
for method in route.methods
|
|
}
|
|
expected = set()
|
|
for role in ("specialist", "manager"):
|
|
base = f"/mini/market/{role}/presign-orders"
|
|
expected.update(
|
|
{
|
|
("GET", base),
|
|
("POST", base),
|
|
("GET", f"{base}/form-options"),
|
|
("GET", f"{base}/{{order_id}}"),
|
|
("PUT", f"{base}/{{order_id}}"),
|
|
("DELETE", f"{base}/{{order_id}}"),
|
|
("POST", f"{base}/{{order_id}}/convert"),
|
|
("POST", f"{base}/{{order_id}}/transfers"),
|
|
}
|
|
)
|
|
|
|
assert actual == expected
|
|
|
|
transfer_actual = {
|
|
(method, route.path)
|
|
for route in router.routes
|
|
if "/presign-transfers" in route.path
|
|
for method in route.methods
|
|
}
|
|
transfer_expected = set()
|
|
for role in ("specialist", "manager"):
|
|
base = f"/mini/market/{role}/presign-transfers/{{transfer_id}}"
|
|
transfer_expected.update(
|
|
{
|
|
("PUT", base),
|
|
("DELETE", base),
|
|
("POST", f"{base}/submit"),
|
|
("POST", f"{base}/withdraw"),
|
|
}
|
|
)
|
|
assert transfer_actual == transfer_expected
|
|
|
|
|
|
def test_mobile_presign_money_dto_preserves_decimal_strings_for_presenter(
|
|
monkeypatch,
|
|
):
|
|
import json
|
|
import subprocess
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
|
|
from app.crud import presign_order as presign_crud
|
|
from app.services import market_presign
|
|
|
|
db = _session()
|
|
normal = _order('money-normal', sales_user_id='u1')
|
|
normal.contract_amount = Decimal('1234.50')
|
|
empty = _order('money-empty', sales_user_id='u1')
|
|
empty.contract_amount = None
|
|
paid = _transfer(
|
|
'money-paid', order_id=normal.id, pay_status='yjf'
|
|
)
|
|
paid.transfer_amount = Decimal('200.25')
|
|
draft = _transfer(
|
|
'money-draft', order_id=normal.id, pay_status='dfj'
|
|
)
|
|
draft.transfer_amount = Decimal('12.30')
|
|
db.add_all([normal, empty, paid, draft])
|
|
db.commit()
|
|
|
|
items, total = market_presign.list_presign_orders(
|
|
db,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
page=1,
|
|
page_size=10,
|
|
)
|
|
by_id = {item['id']: item for item in items}
|
|
assert total == 2
|
|
assert by_id['money-normal']['contractAmount'] == '1234.50'
|
|
assert by_id['money-normal']['confirmedDepositAmount'] == '200.25'
|
|
assert by_id['money-empty']['contractAmount'] is None
|
|
assert by_id['money-empty']['confirmedDepositAmount'] == '0.00'
|
|
|
|
normal_detail = market_presign.get_presign_order(
|
|
db,
|
|
order_id=normal.id,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
transfer_amounts = {
|
|
transfer['id']: transfer['transferAmount']
|
|
for transfer in normal_detail['transfers']
|
|
}
|
|
assert transfer_amounts == {
|
|
'money-paid': '200.25',
|
|
'money-draft': '12.30',
|
|
}
|
|
|
|
maximum = Decimal('9999999999999999.99')
|
|
maximum_order = _order('money-maximum', sales_user_id='u1')
|
|
maximum_order.contract_amount = maximum
|
|
monkeypatch.setattr(
|
|
market_presign,
|
|
'assert_presign_access',
|
|
lambda *_args, **_kwargs: maximum_order,
|
|
)
|
|
monkeypatch.setattr(
|
|
presign_crud,
|
|
'confirmed_deposit_amount',
|
|
lambda *_args, **_kwargs: maximum,
|
|
)
|
|
maximum_detail = market_presign.get_presign_order(
|
|
db,
|
|
order_id=maximum_order.id,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
assert maximum_detail['contractAmount'] == str(maximum)
|
|
assert maximum_detail['confirmedDepositAmount'] == str(maximum)
|
|
maximum_transfer = _transfer(
|
|
'money-maximum-transfer', order_id=maximum_order.id
|
|
)
|
|
maximum_transfer.transfer_amount = maximum
|
|
maximum_transfer_dto = market_presign._mobile_transfer_to_dict(
|
|
db, maximum_transfer, 'tenant-a'
|
|
)
|
|
assert maximum_transfer_dto['transferAmount'] == str(maximum)
|
|
|
|
serialized = json.dumps(
|
|
{
|
|
'maximum': maximum_detail,
|
|
'maximumTransfer': maximum_transfer_dto,
|
|
'normal': normal_detail,
|
|
'empty': by_id['money-empty'],
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
presenter_url = (
|
|
Path(__file__).resolve().parents[2]
|
|
/ 'mobile-oa-h5/src/market/presign/presign-presenter.js'
|
|
).as_uri()
|
|
script = """
|
|
import fs from 'node:fs'
|
|
const { formatPresignMoney } = await import(process.argv[1])
|
|
const dto = JSON.parse(fs.readFileSync(0, 'utf8'))
|
|
const values = [
|
|
dto.maximum.contractAmount,
|
|
dto.maximum.confirmedDepositAmount,
|
|
dto.maximumTransfer.transferAmount,
|
|
dto.normal.contractAmount,
|
|
dto.normal.transfers.find((item) => item.id === 'money-paid').transferAmount,
|
|
dto.empty.contractAmount,
|
|
dto.empty.confirmedDepositAmount
|
|
]
|
|
process.stdout.write(JSON.stringify(values.map(formatPresignMoney)))
|
|
"""
|
|
formatted = subprocess.run(
|
|
['node', '--input-type=module', '-e', script, presenter_url],
|
|
input=serialized,
|
|
text=True,
|
|
capture_output=True,
|
|
check=True,
|
|
)
|
|
assert json.loads(formatted.stdout) == [
|
|
'¥9,999,999,999,999,999.99',
|
|
'¥9,999,999,999,999,999.99',
|
|
'¥9,999,999,999,999,999.99',
|
|
'¥1,234.50',
|
|
'¥200.25',
|
|
'-',
|
|
'¥0.00',
|
|
]
|