862 lines
27 KiB
Python
862 lines
27 KiB
Python
from datetime import date, datetime, timedelta
|
|
from decimal import Decimal
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from fastapi.encoders import jsonable_encoder
|
|
from pydantic import ValidationError
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import sessionmaker
|
|
from starlette.responses import JSONResponse
|
|
|
|
from app.models.base import Base
|
|
from app.models.models import (
|
|
BizAttachment,
|
|
BizCellSampleType,
|
|
BizCollectionProtocol,
|
|
BizCollectionRegionStaff,
|
|
BizProtocolCollectionInfo,
|
|
BizProtocolPartyCustomer,
|
|
BizProtocolPaymentSchedule,
|
|
BizProtocolPaymentTransfer,
|
|
BizProtocolSample,
|
|
BizProtocolSamplePayment,
|
|
SysPost,
|
|
SysPostRole,
|
|
SysRole,
|
|
SysRoleDataScopeDetail,
|
|
SysUser,
|
|
SysUserPost,
|
|
SysUserRole,
|
|
)
|
|
|
|
|
|
def _session():
|
|
engine = create_engine('sqlite:///:memory:')
|
|
Base.metadata.create_all(
|
|
engine,
|
|
tables=[
|
|
BizCollectionProtocol.__table__,
|
|
BizProtocolCollectionInfo.__table__,
|
|
BizProtocolPartyCustomer.__table__,
|
|
BizProtocolSamplePayment.__table__,
|
|
BizProtocolPaymentSchedule.__table__,
|
|
BizProtocolPaymentTransfer.__table__,
|
|
BizProtocolSample.__table__,
|
|
BizCellSampleType.__table__,
|
|
BizAttachment.__table__,
|
|
BizCollectionRegionStaff.__table__,
|
|
SysUser.__table__,
|
|
SysRole.__table__,
|
|
SysRoleDataScopeDetail.__table__,
|
|
SysUserRole.__table__,
|
|
SysPost.__table__,
|
|
SysPostRole.__table__,
|
|
SysUserPost.__table__,
|
|
],
|
|
)
|
|
return sessionmaker(bind=engine)()
|
|
|
|
|
|
def _ctx(scope_kind='specialist', 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',
|
|
'region_id': 'dept-from-auth',
|
|
}
|
|
|
|
|
|
def _payment_tree(
|
|
db,
|
|
key,
|
|
*,
|
|
sales='u1',
|
|
manager='manager-a',
|
|
region='region-a',
|
|
tenant='tenant-a',
|
|
protocol_status='sxz',
|
|
payment_model_version=2,
|
|
due_date=None,
|
|
due_amount=100,
|
|
schedule_status='open',
|
|
deleted=0,
|
|
transfer_statuses=(),
|
|
transfer_dept='',
|
|
register_date='20261001',
|
|
storage_years=20,
|
|
sample_price=80,
|
|
sample_original_price=100,
|
|
):
|
|
protocol = BizCollectionProtocol(
|
|
id=f'p-{key}',
|
|
protocol_no=f'P-{key}',
|
|
protocol_status=protocol_status,
|
|
payment_model_version=payment_model_version,
|
|
register_date=register_date,
|
|
tenant_no=tenant,
|
|
create_dept_id='dept-a',
|
|
create_time=datetime.now(),
|
|
del_flag=deleted,
|
|
)
|
|
collection = BizProtocolCollectionInfo(
|
|
id=f'ci-{key}',
|
|
collection_protocol_id=protocol.id,
|
|
region_id=region,
|
|
sales_user_id=sales,
|
|
region_manager_user_id=manager,
|
|
tenant_no=tenant,
|
|
create_dept_id='dept-a',
|
|
del_flag=deleted,
|
|
)
|
|
party = BizProtocolPartyCustomer(
|
|
id=f'party-{key}',
|
|
collection_protocol_id=protocol.id,
|
|
mother_name=f'客户-{key}',
|
|
tenant_no=tenant,
|
|
create_dept_id='dept-a',
|
|
del_flag=deleted,
|
|
)
|
|
payment = BizProtocolSamplePayment(
|
|
id=f'sp-{key}',
|
|
collection_protocol_id=protocol.id,
|
|
sample_code=f'S-{key}',
|
|
pay_mode='fq',
|
|
contract_amount=due_amount,
|
|
tenant_no=tenant,
|
|
create_dept_id='dept-a',
|
|
create_time=datetime.now(),
|
|
del_flag=deleted,
|
|
)
|
|
schedule = BizProtocolPaymentSchedule(
|
|
id=f'sch-{key}',
|
|
sample_payment_id=payment.id,
|
|
collection_protocol_id=protocol.id,
|
|
sample_code=payment.sample_code,
|
|
schedule_no=1,
|
|
due_amount=due_amount,
|
|
due_date=due_date or date.today().strftime('%Y%m%d'),
|
|
schedule_status=schedule_status,
|
|
tenant_no=tenant,
|
|
create_dept_id='dept-a',
|
|
del_flag=deleted,
|
|
)
|
|
sample_type = BizCellSampleType(
|
|
id=f'type-{key}',
|
|
type_code=f'TYPE-{key}',
|
|
type_name=f'样本类型-{key}',
|
|
tenant_no=tenant,
|
|
create_dept_id='dept-a',
|
|
del_flag=deleted,
|
|
)
|
|
sample = BizProtocolSample(
|
|
id=f'sample-{key}',
|
|
collection_protocol_id=protocol.id,
|
|
sample_type_id=sample_type.id,
|
|
sample_code=payment.sample_code,
|
|
storage_years=storage_years,
|
|
price=sample_price,
|
|
original_price=sample_original_price,
|
|
tenant_no=tenant,
|
|
create_dept_id='dept-a',
|
|
del_flag=deleted,
|
|
)
|
|
db.add_all([protocol, collection, party, payment, schedule, sample_type, sample])
|
|
for index, status in enumerate(transfer_statuses):
|
|
db.add(
|
|
BizProtocolPaymentTransfer(
|
|
id=f't-{key}-{index}',
|
|
schedule_id=schedule.id,
|
|
sample_payment_id=payment.id,
|
|
collection_protocol_id=protocol.id,
|
|
sample_code=payment.sample_code,
|
|
transfer_amount=20,
|
|
transfer_time=datetime.now(),
|
|
transfer_type='bank',
|
|
pay_status=status,
|
|
tenant_no=tenant,
|
|
create_dept_id=transfer_dept,
|
|
del_flag=deleted,
|
|
)
|
|
)
|
|
db.commit()
|
|
return protocol, payment, schedule
|
|
|
|
|
|
def test_chase_is_overdue_union_next_30_days_with_parent_scope():
|
|
from app.services.market_payment import list_payment_chase
|
|
|
|
db = _session()
|
|
today = date(2026, 10, 11)
|
|
_payment_tree(db, 'overdue', due_date=(today - timedelta(days=1)).strftime('%Y%m%d'))
|
|
_payment_tree(db, 'today', due_date=today.strftime('%Y%m%d'))
|
|
_payment_tree(db, 'day30', due_date=(today + timedelta(days=30)).strftime('%Y%m%d'))
|
|
_payment_tree(db, 'day31', due_date=(today + timedelta(days=31)).strftime('%Y%m%d'))
|
|
_payment_tree(db, 'other', sales='u2', due_date=today.strftime('%Y%m%d'))
|
|
_payment_tree(db, 'void', protocol_status='yzf', due_date=today.strftime('%Y%m%d'))
|
|
_payment_tree(db, 'bad-date', due_date='not-a-date')
|
|
|
|
items, total = list_payment_chase(
|
|
db, ctx=_ctx(), user_info=_user(), today=today
|
|
)
|
|
|
|
assert total == 3
|
|
assert {item['scheduleId'] for item in items} == {
|
|
'sch-overdue',
|
|
'sch-today',
|
|
'sch-day30',
|
|
}
|
|
|
|
|
|
def test_payment_chase_uses_historical_nickname_without_user_no_fallback():
|
|
from app.services.market_payment import list_payment_chase
|
|
|
|
db = _session()
|
|
db.add_all(
|
|
[
|
|
SysUser(id='u1', user_no='deleted-sales-no', nick_name='历史删除专员', status='0', tenant_no='tenant-a', create_dept_id='', del_flag=1),
|
|
SysUser(id='blank-sales', user_no='raw-sales-no', nick_name='', status='0', tenant_no='tenant-a', create_dept_id='', del_flag=1),
|
|
]
|
|
)
|
|
_payment_tree(db, 'historical-sales', sales='u1')
|
|
_payment_tree(db, 'blank-sales', sales='blank-sales')
|
|
|
|
historical_items, historical_total = list_payment_chase(
|
|
db, ctx=_ctx(user_id='u1'), user_info=_user('u1')
|
|
)
|
|
blank_items, blank_total = list_payment_chase(
|
|
db,
|
|
ctx=_ctx(user_id='blank-sales'),
|
|
user_info=_user('blank-sales'),
|
|
)
|
|
|
|
assert historical_total == 1
|
|
assert historical_items[0]['salesUserName'] == '历史删除专员'
|
|
assert blank_total == 1
|
|
assert blank_items[0]['salesUserName'] == ''
|
|
|
|
|
|
def test_chase_counts_only_yjf_and_excludes_fully_confirmed():
|
|
from app.services.market_payment import list_payment_chase
|
|
|
|
db = _session()
|
|
_payment_tree(
|
|
db,
|
|
'part-paid',
|
|
due_amount=100,
|
|
transfer_statuses=('dfj', 'jfqrz', 'yjf'),
|
|
)
|
|
_payment_tree(
|
|
db,
|
|
'fully-paid',
|
|
due_amount=60,
|
|
transfer_statuses=('yjf', 'yjf', 'yjf'),
|
|
)
|
|
|
|
items, total = list_payment_chase(db, ctx=_ctx(), user_info=_user())
|
|
|
|
assert total == 1
|
|
assert items[0]['scheduleId'] == 'sch-part-paid'
|
|
assert items[0]['paidAmount'] == '20.00'
|
|
assert items[0]['arrearsAmount'] == '80.00'
|
|
|
|
|
|
def test_chase_filters_counts_orders_and_pages_in_sql_before_materialization():
|
|
from app.services.market_payment import list_payment_chase
|
|
|
|
db = _session()
|
|
today = date(2026, 10, 11)
|
|
for index in range(35):
|
|
_payment_tree(
|
|
db,
|
|
f'history-{index:02d}',
|
|
due_date=(today - timedelta(days=index + 1)).strftime('%Y%m%d'),
|
|
transfer_statuses=('dfj',),
|
|
)
|
|
_payment_tree(db, 'invalid-day', due_date='20260230')
|
|
statements = []
|
|
|
|
def capture(_conn, _cursor, statement, _parameters, _context, _many):
|
|
statements.append(statement)
|
|
|
|
event.listen(db.get_bind(), 'before_cursor_execute', capture)
|
|
try:
|
|
items, total = list_payment_chase(
|
|
db,
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
today=today,
|
|
page=3,
|
|
page_size=5,
|
|
)
|
|
finally:
|
|
event.remove(db.get_bind(), 'before_cursor_execute', capture)
|
|
|
|
assert total == 35
|
|
assert len(items) == 5
|
|
assert [item['scheduleId'] for item in items] == [
|
|
'sch-history-24',
|
|
'sch-history-23',
|
|
'sch-history-22',
|
|
'sch-history-21',
|
|
'sch-history-20',
|
|
]
|
|
chase_selects = [sql for sql in statements if 'biz_protocol_payment_schedule' in sql]
|
|
assert len(chase_selects) == 2
|
|
assert ' LIMIT ' in chase_selects[-1].upper()
|
|
assert 'SUM(CASE' in chase_selects[-1].upper()
|
|
assert not any(
|
|
'FROM biz_protocol_payment_transfer' in sql and 'SUM(CASE' not in sql.upper()
|
|
for sql in statements
|
|
)
|
|
|
|
|
|
def test_payment_list_is_v2_only_and_uses_parent_scope_not_transfer_department():
|
|
from app.services.market_payment import list_payments
|
|
|
|
db = _session()
|
|
_payment_tree(db, 'v2', transfer_statuses=('dfj',), transfer_dept='')
|
|
_payment_tree(db, 'v1', payment_model_version=1)
|
|
_payment_tree(db, 'other', sales='u2')
|
|
_payment_tree(db, 'cross', tenant='tenant-b')
|
|
_payment_tree(db, 'deleted', deleted=1)
|
|
|
|
items, total = list_payments(db, ctx=_ctx(), user_info=_user())
|
|
|
|
assert total == 1
|
|
assert items[0]['id'] == 'sp-v2'
|
|
assert items[0]['paymentModelVersion'] == 2
|
|
|
|
|
|
def test_manager_scope_is_fixed_by_route_context_and_client_filter_cannot_expand():
|
|
from app.services.market_payment import list_payment_chase
|
|
|
|
db = _session()
|
|
_payment_tree(db, 'manager', manager='manager-a', sales='u2')
|
|
_payment_tree(db, 'outside', manager='manager-b', sales='u1')
|
|
|
|
items, total = list_payment_chase(
|
|
db,
|
|
ctx=_ctx('manager', 'manager-a'),
|
|
user_info=_user('manager-a'),
|
|
sales_user_id='u1',
|
|
)
|
|
|
|
assert total == 0
|
|
assert items == []
|
|
|
|
|
|
def test_payment_detail_rejects_out_of_scope_sample():
|
|
from app.services.market_payment import get_payment_detail
|
|
|
|
db = _session()
|
|
_payment_tree(db, 'other', sales='u2')
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
get_payment_detail(
|
|
db, payment_id='sp-other', ctx=_ctx(), user_info=_user()
|
|
)
|
|
|
|
assert exc.value.status_code == 404
|
|
|
|
|
|
def test_payment_detail_identifies_legacy_v1_without_reading_wrong_tables():
|
|
from app.services.market_payment import get_payment_detail
|
|
|
|
db = _session()
|
|
_payment_tree(db, 'legacy', payment_model_version=1)
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
get_payment_detail(
|
|
db, payment_id='sp-legacy', ctx=_ctx(), user_info=_user()
|
|
)
|
|
|
|
assert exc.value.status_code == 409
|
|
assert '旧版' in str(exc.value.detail)
|
|
|
|
|
|
def test_market_can_save_submit_and_withdraw_transfer_atomically():
|
|
from app.services.market_payment import (
|
|
create_transfer,
|
|
submit_transfer,
|
|
update_transfer,
|
|
withdraw_transfer,
|
|
)
|
|
|
|
db = _session()
|
|
_, _, schedule = _payment_tree(db, 'own')
|
|
db.add(
|
|
BizAttachment(
|
|
id='att-pay',
|
|
business_type='market:specialist:payment:staging',
|
|
original_name='a.jpg',
|
|
file_ext='jpg',
|
|
file_url='a.jpg',
|
|
file_size=10,
|
|
file_type='image/jpeg',
|
|
tenant_no='tenant-a',
|
|
create_by='u1',
|
|
create_dept_id='',
|
|
del_flag=0,
|
|
)
|
|
)
|
|
db.commit()
|
|
created = create_transfer(
|
|
db,
|
|
schedule_id=schedule.id,
|
|
fields={
|
|
'transferAmount': 50,
|
|
'transferTime': '2026-10-11 10:00:00',
|
|
'transferType': 'bank',
|
|
'transferAttachments': 'a.jpg',
|
|
},
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
assert created['payStatus'] == 'dfj'
|
|
row = db.get(BizProtocolPaymentTransfer, created['id'])
|
|
assert row.create_dept_id == 'dept-from-auth'
|
|
attachment = db.get(BizAttachment, 'att-pay')
|
|
assert (attachment.business_type, attachment.business_id) == (
|
|
'market:specialist:payment:bound',
|
|
created['id'],
|
|
)
|
|
|
|
updated = update_transfer(
|
|
db,
|
|
transfer_id=created['id'],
|
|
fields={'remark': '补充说明'},
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
assert updated['remark'] == '补充说明'
|
|
assert submit_transfer(
|
|
db, transfer_id=created['id'], ctx=_ctx(), user_info=_user()
|
|
)['payStatus'] == 'jfqrz'
|
|
assert withdraw_transfer(
|
|
db, transfer_id=created['id'], ctx=_ctx(), user_info=_user()
|
|
)['payStatus'] == 'dfj'
|
|
|
|
|
|
def test_transfer_create_rejects_external_attachment_reference():
|
|
from app.services.market_payment import create_transfer
|
|
|
|
db = _session()
|
|
_, _, schedule = _payment_tree(db, 'attachment-reject')
|
|
with pytest.raises(ValueError, match='附件'):
|
|
create_transfer(
|
|
db,
|
|
schedule_id=schedule.id,
|
|
fields={
|
|
'transferAmount': 10,
|
|
'transferType': 'bank',
|
|
'transferAttachments': 'https://evil.test/a.jpg',
|
|
},
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
|
|
|
|
def test_transfer_create_rollback_restores_staging_attachment_on_business_failure(
|
|
monkeypatch,
|
|
):
|
|
from app.services.market_payment import create_transfer
|
|
|
|
db = _session()
|
|
_, _, schedule = _payment_tree(db, 'attachment-rollback')
|
|
attachment = BizAttachment(
|
|
id='att-rollback',
|
|
business_type='market:specialist:payment:staging',
|
|
original_name='rollback.jpg',
|
|
file_ext='jpg',
|
|
file_url='rollback.jpg',
|
|
file_size=10,
|
|
file_type='image/jpeg',
|
|
tenant_no='tenant-a',
|
|
create_by='u1',
|
|
create_dept_id='',
|
|
del_flag=0,
|
|
)
|
|
db.add(attachment)
|
|
db.commit()
|
|
monkeypatch.setattr(
|
|
db, 'commit', lambda: (_ for _ in ()).throw(RuntimeError('db down'))
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match='db down'):
|
|
create_transfer(
|
|
db,
|
|
schedule_id=schedule.id,
|
|
fields={
|
|
'transferAmount': 10,
|
|
'transferType': 'bank',
|
|
'transferAttachments': 'att-rollback',
|
|
},
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
|
|
db.expire_all()
|
|
restored = db.get(BizAttachment, 'att-rollback')
|
|
assert (restored.business_type, restored.business_id) == (
|
|
'market:specialist:payment:staging',
|
|
None,
|
|
)
|
|
assert db.query(BizProtocolPaymentTransfer).count() == 0
|
|
|
|
|
|
def test_out_of_scope_write_is_checked_before_any_mutation():
|
|
from app.services.market_payment import update_transfer
|
|
|
|
db = _session()
|
|
_payment_tree(db, 'other', sales='u2', transfer_statuses=('dfj',))
|
|
before = db.get(BizProtocolPaymentTransfer, 't-other-0').remark
|
|
|
|
with pytest.raises(HTTPException):
|
|
update_transfer(
|
|
db,
|
|
transfer_id='t-other-0',
|
|
fields={'remark': 'must-not-write'},
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
|
|
db.expire_all()
|
|
assert db.get(BizProtocolPaymentTransfer, 't-other-0').remark == before
|
|
|
|
|
|
@pytest.mark.parametrize('protocol_status', ['yzf', 'sdfq', 'zky'])
|
|
def test_terminal_protocol_rejects_direct_transfer_create_without_mutation(protocol_status):
|
|
from app.services.market_payment import create_transfer
|
|
|
|
db = _session()
|
|
_, _, schedule = _payment_tree(
|
|
db, f'terminal-create-{protocol_status}', protocol_status=protocol_status
|
|
)
|
|
|
|
with pytest.raises(HTTPException):
|
|
create_transfer(
|
|
db,
|
|
schedule_id=schedule.id,
|
|
fields={'transferAmount': 10, 'transferType': 'bank'},
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert db.query(BizProtocolPaymentTransfer).count() == 0
|
|
|
|
|
|
@pytest.mark.parametrize('schedule_status', ['cleared', 'void'])
|
|
def test_non_open_schedule_rejects_direct_transfer_create_without_mutation(schedule_status):
|
|
from app.services.market_payment import create_transfer
|
|
|
|
db = _session()
|
|
_, _, schedule = _payment_tree(
|
|
db, f'closed-create-{schedule_status}', schedule_status=schedule_status
|
|
)
|
|
|
|
with pytest.raises(HTTPException):
|
|
create_transfer(
|
|
db,
|
|
schedule_id=schedule.id,
|
|
fields={'transferAmount': 10, 'transferType': 'bank'},
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert db.query(BizProtocolPaymentTransfer).count() == 0
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
('protocol_status', 'schedule_status', 'pay_status', 'operation'),
|
|
[
|
|
('yzf', 'open', 'dfj', 'update'),
|
|
('sdfq', 'open', 'dfj', 'submit'),
|
|
('zky', 'open', 'jfqrz', 'withdraw'),
|
|
('sxz', 'cleared', 'dfj', 'update'),
|
|
('sxz', 'void', 'dfj', 'submit'),
|
|
],
|
|
)
|
|
def test_ineligible_parent_rejects_direct_transfer_mutations_without_mutation(
|
|
protocol_status, schedule_status, pay_status, operation
|
|
):
|
|
from app.services.market_payment import submit_transfer, update_transfer, withdraw_transfer
|
|
|
|
db = _session()
|
|
key = f'ineligible-{protocol_status}-{schedule_status}-{operation}'
|
|
_payment_tree(
|
|
db,
|
|
key,
|
|
protocol_status=protocol_status,
|
|
schedule_status=schedule_status,
|
|
transfer_statuses=(pay_status,),
|
|
)
|
|
transfer_id = f't-{key}-0'
|
|
before = db.get(BizProtocolPaymentTransfer, transfer_id)
|
|
before_values = (before.pay_status, before.remark, before.update_by, before.update_time)
|
|
|
|
with pytest.raises(HTTPException):
|
|
if operation == 'update':
|
|
update_transfer(
|
|
db,
|
|
transfer_id=transfer_id,
|
|
fields={'remark': 'must-not-write'},
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
elif operation == 'submit':
|
|
submit_transfer(db, transfer_id=transfer_id, ctx=_ctx(), user_info=_user())
|
|
else:
|
|
withdraw_transfer(db, transfer_id=transfer_id, ctx=_ctx(), user_info=_user())
|
|
|
|
db.expire_all()
|
|
after = db.get(BizProtocolPaymentTransfer, transfer_id)
|
|
assert (after.pay_status, after.remark, after.update_by, after.update_time) == before_values
|
|
|
|
|
|
def test_chase_excludes_all_terminal_protocols_and_non_open_schedules():
|
|
from app.services.market_payment import list_payment_chase
|
|
|
|
db = _session()
|
|
_payment_tree(db, 'eligible-chase')
|
|
for status in ('yzf', 'sdfq', 'zky'):
|
|
_payment_tree(db, f'terminal-chase-{status}', protocol_status=status)
|
|
_payment_tree(db, 'cleared-chase', schedule_status='cleared')
|
|
_payment_tree(db, 'void-chase', schedule_status='void')
|
|
|
|
items, total = list_payment_chase(db, ctx=_ctx(), user_info=_user())
|
|
|
|
assert total == 1
|
|
assert [item['scheduleId'] for item in items] == ['sch-eligible-chase']
|
|
|
|
|
|
def test_yjf_is_immutable_and_double_submit_fails_closed():
|
|
from app.services.market_payment import submit_transfer, update_transfer, withdraw_transfer
|
|
|
|
db = _session()
|
|
_payment_tree(db, 'draft', transfer_statuses=('dfj',))
|
|
submit_transfer(db, transfer_id='t-draft-0', ctx=_ctx(), user_info=_user())
|
|
with pytest.raises(ValueError, match='状态'):
|
|
submit_transfer(db, transfer_id='t-draft-0', ctx=_ctx(), user_info=_user())
|
|
|
|
_payment_tree(db, 'confirmed', transfer_statuses=('yjf',))
|
|
with pytest.raises(ValueError):
|
|
update_transfer(
|
|
db,
|
|
transfer_id='t-confirmed-0',
|
|
fields={'remark': 'no'},
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
with pytest.raises(ValueError):
|
|
withdraw_transfer(
|
|
db,
|
|
transfer_id='t-confirmed-0',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
|
|
|
|
def test_transfer_dto_forbids_authorization_status_and_finance_fields():
|
|
from app.schemas.market_mobile import MarketPaymentTransferBody
|
|
|
|
forbidden = (
|
|
{'tenantNo': 'tenant-b'},
|
|
{'scope': 'manager'},
|
|
{'payStatus': 'yjf'},
|
|
{'financeRemark': 'confirmed'},
|
|
{'scheduleStatus': 'void'},
|
|
)
|
|
for extra in forbidden:
|
|
with pytest.raises(ValidationError):
|
|
MarketPaymentTransferBody.model_validate(
|
|
{'transferAmount': 10, 'transferType': 'bank', **extra}
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'amount',
|
|
[float('inf'), float('-inf'), float('nan'), '0.001', '10000000000000000.00'],
|
|
)
|
|
def test_transfer_amount_rejects_non_finite_over_precision_and_overflow(amount):
|
|
from app.schemas.market_mobile import MarketPaymentTransferBody
|
|
|
|
with pytest.raises(ValidationError):
|
|
MarketPaymentTransferBody.model_validate({'transferAmount': amount})
|
|
|
|
|
|
@pytest.mark.parametrize('amount', ['0.01', '9999999999999999.99'])
|
|
def test_transfer_amount_accepts_decimal_18_2_boundaries_exactly(amount):
|
|
from app.schemas.market_mobile import MarketPaymentTransferBody
|
|
|
|
body = MarketPaymentTransferBody.model_validate({'transferAmount': amount})
|
|
|
|
assert body.transfer_amount == Decimal(amount)
|
|
|
|
|
|
def test_task5_money_is_fixed_two_decimal_string_in_actual_json_response():
|
|
from app.core.response import success_response
|
|
from app.schemas.market_mobile import MarketPaymentTransferCreateBody
|
|
from app.services.market_payment import _payment_dict, create_transfer
|
|
|
|
db = _session()
|
|
_, _, schedule = _payment_tree(
|
|
db, 'money-json', due_amount=Decimal('9999999999999999.99')
|
|
)
|
|
body = MarketPaymentTransferCreateBody.model_validate(
|
|
{
|
|
'scheduleId': schedule.id,
|
|
'transferAmount': '9999999999999999.99',
|
|
'transferType': 'bank',
|
|
}
|
|
)
|
|
payload = body.model_dump(by_alias=True, exclude_unset=True)
|
|
schedule_id = payload.pop('scheduleId')
|
|
transfer = create_transfer(
|
|
db,
|
|
schedule_id=schedule_id,
|
|
fields=payload,
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
max_money = Decimal('9999999999999999.99')
|
|
payment = _payment_dict(
|
|
SimpleNamespace(
|
|
id='sp-json',
|
|
sample_code='S-json',
|
|
pay_mode='fq',
|
|
contract_amount=max_money,
|
|
remark=None,
|
|
),
|
|
SimpleNamespace(id='p-json', protocol_no='P-json', protocol_status='sxz'),
|
|
None,
|
|
[
|
|
SimpleNamespace(
|
|
id='sch-json',
|
|
schedule_no=1,
|
|
due_amount=max_money,
|
|
due_date='20261011',
|
|
schedule_status='open',
|
|
)
|
|
],
|
|
{},
|
|
)
|
|
|
|
encoded = jsonable_encoder(
|
|
success_response({'transfer': transfer, 'payment': payment})
|
|
)
|
|
response = JSONResponse(encoded)
|
|
decoded = response.body.decode('utf-8')
|
|
|
|
assert transfer['transferAmount'] == '9999999999999999.99'
|
|
assert payment['contractAmount'] == '9999999999999999.99'
|
|
assert payment['schedules'][0]['dueAmount'] == '9999999999999999.99'
|
|
assert '"9999999999999999.99"' in decoded
|
|
|
|
|
|
def test_task5_normal_money_is_always_serialized_with_two_decimals():
|
|
from app.services.market_payment import list_payment_chase
|
|
|
|
db = _session()
|
|
_payment_tree(db, 'normal-money', due_amount=Decimal('100.00'))
|
|
|
|
items, _ = list_payment_chase(db, ctx=_ctx(), user_info=_user())
|
|
|
|
assert items[0]['dueAmount'] == '100.00'
|
|
assert items[0]['paidAmount'] == '0.00'
|
|
assert items[0]['arrearsAmount'] == '100.00'
|
|
|
|
|
|
def test_market_routes_expose_only_allowed_payment_mutations():
|
|
from app.api.market_mobile import router
|
|
|
|
paths = {route.path for route in router.routes}
|
|
for scope in ('specialist', 'manager'):
|
|
prefix = f'/mini/market/{scope}'
|
|
assert f'{prefix}/payment-chase' in paths
|
|
assert f'{prefix}/payment-chases' not in paths
|
|
assert f'{prefix}/payments' in paths
|
|
assert f'{prefix}/payment-transfers' in paths
|
|
assert f'{prefix}/payment-transfers/{{transfer_id}}/submit' in paths
|
|
assert f'{prefix}/payment-transfers/{{transfer_id}}/withdraw' in paths
|
|
assert f'{prefix}/payment-transfers/{{transfer_id}}/confirm' not in paths
|
|
assert f'{prefix}/payment-transfers/{{transfer_id}}/reject' not in paths
|
|
assert f'{prefix}/payment-transfers/{{transfer_id}}/refund' not in paths
|
|
assert f'{prefix}/payment-transfers/{{transfer_id}}/invoice' not in paths
|
|
|
|
|
|
def test_list_queries_batch_children_without_n_plus_one():
|
|
from app.services.market_payment import list_payments
|
|
|
|
db = _session()
|
|
for index in range(8):
|
|
_payment_tree(db, f'row-{index}', transfer_statuses=('dfj', 'yjf'))
|
|
statements = []
|
|
|
|
def capture(_conn, _cursor, statement, _parameters, _context, _many):
|
|
statements.append(statement)
|
|
|
|
event.listen(db.get_bind(), 'before_cursor_execute', capture)
|
|
try:
|
|
items, total = list_payments(
|
|
db, ctx=_ctx(), user_info=_user(), page=1, page_size=5
|
|
)
|
|
finally:
|
|
event.remove(db.get_bind(), 'before_cursor_execute', capture)
|
|
|
|
assert total == 8
|
|
assert len(items) == 5
|
|
assert len(statements) <= 5
|
|
|
|
|
|
def test_payment_list_contains_card_summary_totals():
|
|
from app.services.market_payment import list_payments
|
|
|
|
db = _session()
|
|
_payment_tree(db, 'summary', due_amount=Decimal('100.00'), transfer_statuses=('yjf',))
|
|
|
|
items, total = list_payments(db, ctx=_ctx(), user_info=_user())
|
|
|
|
assert total == 1
|
|
assert items[0]['paidAmount'] == '20.00'
|
|
assert items[0]['remainAmount'] == '80.00'
|
|
assert items[0]['activeScheduleCount'] == 1
|
|
assert items[0]['clearedScheduleCount'] == 0
|
|
assert items[0]['paymentProgress'] == '部分缴费'
|
|
assert items[0]['signDate'] == '2026-10-01'
|
|
assert items[0]['storageYears'] == 20
|
|
assert items[0]['sampleTypeName'] == '样本类型-summary'
|
|
assert items[0]['discountAmount'] == '20.00'
|
|
assert items[0]['receivableAmount'] == '100.00'
|
|
assert items[0]['dueRemainAmount'] == '80.00'
|
|
assert len(items[0]['periods']) == 6
|
|
assert items[0]['periods'][0]['scheduleNo'] == 1
|
|
|
|
|
|
def test_payment_sample_meta_uses_tenant_and_sample_code_after_protocol_change():
|
|
from app.services.market_payment import list_payments
|
|
|
|
db = _session()
|
|
_, payment, _ = _payment_tree(db, 'protocol-changed')
|
|
sample = (
|
|
db.query(BizProtocolSample)
|
|
.filter(BizProtocolSample.sample_code == payment.sample_code)
|
|
.one()
|
|
)
|
|
sample.collection_protocol_id = 'p-previous-protocol'
|
|
db.commit()
|
|
|
|
items, total = list_payments(db, ctx=_ctx(), user_info=_user())
|
|
|
|
assert total == 1
|
|
assert items[0]['storageYears'] == 20
|
|
assert items[0]['sampleTypeName'] == '样本类型-protocol-changed'
|
|
assert items[0]['discountAmount'] == '20.00'
|