1576 lines
49 KiB
Python
1576 lines
49 KiB
Python
import asyncio
|
|
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,
|
|
BizCollectionHospital,
|
|
BizCollectionRegion,
|
|
BizItemUnit,
|
|
BizProtocolAudit,
|
|
BizProtocolCollectionInfo,
|
|
BizProtocolFullmoonTodo,
|
|
BizProtocolPayment,
|
|
BizProtocolPaymentDetail,
|
|
BizProtocolPartyCustomer,
|
|
BizProtocolPaymentSchedule,
|
|
BizProtocolPaymentTransfer,
|
|
BizProtocolSample,
|
|
BizProtocolSamplePayment,
|
|
BizProtocolSampleTransport,
|
|
BizProtocolStorageCert,
|
|
BizPresignOrder,
|
|
BizPresignTransfer,
|
|
BizSampleFreight,
|
|
BizSampleFreightItem,
|
|
BizSampleReception,
|
|
SysDictData,
|
|
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__,
|
|
BizProtocolPayment.__table__,
|
|
BizProtocolPaymentDetail.__table__,
|
|
BizProtocolSample.__table__,
|
|
BizProtocolSampleTransport.__table__,
|
|
BizProtocolAudit.__table__,
|
|
BizProtocolStorageCert.__table__,
|
|
BizProtocolFullmoonTodo.__table__,
|
|
BizPresignOrder.__table__,
|
|
BizPresignTransfer.__table__,
|
|
BizSampleFreight.__table__,
|
|
BizSampleFreightItem.__table__,
|
|
BizSampleReception.__table__,
|
|
BizCellSampleType.__table__,
|
|
BizAttachment.__table__,
|
|
BizCollectionRegionStaff.__table__,
|
|
BizCollectionRegion.__table__,
|
|
BizCollectionHospital.__table__,
|
|
BizItemUnit.__table__,
|
|
SysDictData.__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 == 4
|
|
assert {item['scheduleId'] for item in items} == {
|
|
'sch-overdue',
|
|
'sch-today',
|
|
'sch-day30',
|
|
'sch-void',
|
|
}
|
|
|
|
|
|
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_payment_list_filters_progress_from_schedule_status_and_due_date_range():
|
|
from app.services.market_payment import list_payments
|
|
|
|
db = _session()
|
|
_payment_tree(db, 'unpaid', due_date='20261010', due_amount=100)
|
|
_payment_tree(
|
|
db,
|
|
'partial',
|
|
due_date='20261011',
|
|
due_amount=100,
|
|
transfer_statuses=('yjf',),
|
|
)
|
|
_payment_tree(
|
|
db,
|
|
'paid',
|
|
due_date='20261012',
|
|
due_amount=20,
|
|
schedule_status='cleared',
|
|
transfer_statuses=('yjf',),
|
|
)
|
|
_payment_tree(db, 'outside', due_date='20261101', due_amount=100)
|
|
|
|
partial, partial_total = list_payments(
|
|
db,
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
progress_status='bfyj',
|
|
installment_due_date_start='2026-10-01',
|
|
installment_due_date_end='2026-10-31',
|
|
)
|
|
paid, paid_total = list_payments(
|
|
db,
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
progress_status='yjq',
|
|
installment_due_date_start='2026-10-01',
|
|
installment_due_date_end='2026-10-31',
|
|
)
|
|
|
|
assert partial_total == 1
|
|
assert [item['id'] for item in partial] == ['sp-partial']
|
|
assert paid_total == 1
|
|
assert [item['id'] for item in paid] == ['sp-paid']
|
|
|
|
|
|
def test_payment_progress_infers_cleared_when_confirmed_paid_covers_due():
|
|
from app.services.market_payment import list_payments
|
|
|
|
db = _session()
|
|
_payment_tree(
|
|
db,
|
|
'amount-covered-but-open',
|
|
due_amount=20,
|
|
schedule_status='open',
|
|
transfer_statuses=('yjf',),
|
|
)
|
|
_payment_tree(
|
|
db,
|
|
'status-cleared-without-transfer',
|
|
due_amount=100,
|
|
schedule_status='cleared',
|
|
)
|
|
_payment_tree(
|
|
db,
|
|
'partial-open',
|
|
due_amount=100,
|
|
schedule_status='open',
|
|
transfer_statuses=('yjf',),
|
|
)
|
|
|
|
partial, partial_total = list_payments(
|
|
db, ctx=_ctx(), user_info=_user(), progress_status='bfyj'
|
|
)
|
|
cleared, cleared_total = list_payments(
|
|
db, ctx=_ctx(), user_info=_user(), progress_status='yjq'
|
|
)
|
|
|
|
assert partial_total == 1
|
|
assert partial[0]['id'] == 'sp-partial-open'
|
|
assert partial[0]['paymentProgress'] == '部分已缴'
|
|
assert partial[0]['clearedScheduleCount'] == 0
|
|
assert cleared_total == 2
|
|
assert {item['id'] for item in cleared} == {
|
|
'sp-amount-covered-but-open',
|
|
'sp-status-cleared-without-transfer',
|
|
}
|
|
covered = next(item for item in cleared if item['id'] == 'sp-amount-covered-but-open')
|
|
assert covered['paymentProgress'] == '已缴清'
|
|
assert covered['clearedScheduleCount'] == 1
|
|
assert covered['schedules'][0]['scheduleStatus'] == 'cleared'
|
|
|
|
|
|
def test_progress_filter_pages_in_sql_before_loading_children(monkeypatch):
|
|
from app.services import market_payment
|
|
|
|
db = _session()
|
|
for key in ('partial-a', 'partial-b', 'partial-c'):
|
|
_payment_tree(
|
|
db,
|
|
key,
|
|
due_amount=100,
|
|
transfer_statuses=('yjf',),
|
|
)
|
|
|
|
calls = []
|
|
original = market_payment._load_children
|
|
|
|
def tracked_load_children(db_arg, payment_ids, tenant_no):
|
|
calls.append(list(payment_ids))
|
|
return original(db_arg, payment_ids, tenant_no)
|
|
|
|
monkeypatch.setattr(market_payment, '_load_children', tracked_load_children)
|
|
|
|
items, total = market_payment.list_payments(
|
|
db,
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
progress_status='bfyj',
|
|
page=2,
|
|
page_size=1,
|
|
)
|
|
|
|
assert total == 3
|
|
assert len(items) == 1
|
|
assert calls == [[items[0]['id']]]
|
|
|
|
|
|
def test_progress_counts_confirmed_transfers_from_void_schedules_like_web():
|
|
from app.services.market_payment import list_payments
|
|
|
|
db = _session()
|
|
_payment_tree(db, 'void-unpaid', schedule_status='void', due_amount=100)
|
|
_payment_tree(
|
|
db,
|
|
'void-confirmed',
|
|
schedule_status='void',
|
|
due_amount=20,
|
|
transfer_statuses=('yjf',),
|
|
)
|
|
|
|
unpaid, unpaid_total = list_payments(
|
|
db,
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
progress_status='dfj',
|
|
)
|
|
partial, partial_total = list_payments(
|
|
db,
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
progress_status='bfyj',
|
|
)
|
|
cleared, cleared_total = list_payments(
|
|
db,
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
progress_status='yjq',
|
|
)
|
|
|
|
assert unpaid_total == 1
|
|
assert [item['id'] for item in unpaid] == ['sp-void-unpaid']
|
|
assert unpaid[0]['progressStatus'] == 'dfj'
|
|
assert unpaid[0]['paymentProgress'] == '待缴费'
|
|
assert unpaid[0]['paidAmount'] == '0.00'
|
|
assert unpaid[0]['activeScheduleCount'] == 0
|
|
assert partial_total == 1
|
|
assert [item['id'] for item in partial] == ['sp-void-confirmed']
|
|
assert partial[0]['progressStatus'] == 'bfyj'
|
|
assert partial[0]['paymentProgress'] == '部分已缴'
|
|
assert partial[0]['paidAmount'] == '20.00'
|
|
assert partial[0]['activeScheduleCount'] == 0
|
|
assert cleared_total == 0
|
|
assert cleared == []
|
|
|
|
|
|
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_bound_payment_detail_shows_only_requested_schedule():
|
|
from app.services.market_payment import get_payment_chase_detail
|
|
|
|
db = _session()
|
|
_, payment, first_schedule = _payment_tree(
|
|
db,
|
|
'chase-detail',
|
|
due_amount=100,
|
|
transfer_statuses=('yjf',),
|
|
)
|
|
second_schedule = BizProtocolPaymentSchedule(
|
|
id='sch-chase-detail-2',
|
|
sample_payment_id=payment.id,
|
|
collection_protocol_id='p-chase-detail',
|
|
sample_code=payment.sample_code,
|
|
schedule_no=2,
|
|
due_amount=50,
|
|
due_date=date.today().strftime('%Y%m%d'),
|
|
schedule_status='cleared',
|
|
tenant_no='tenant-a',
|
|
create_dept_id='dept-a',
|
|
del_flag=0,
|
|
)
|
|
second_transfer = BizProtocolPaymentTransfer(
|
|
id='t-chase-detail-2',
|
|
schedule_id=second_schedule.id,
|
|
sample_payment_id=payment.id,
|
|
collection_protocol_id='p-chase-detail',
|
|
sample_code=payment.sample_code,
|
|
transfer_amount=50,
|
|
transfer_time=datetime.now(),
|
|
transfer_type='bank',
|
|
pay_status='yjf',
|
|
tenant_no='tenant-a',
|
|
create_dept_id='dept-a',
|
|
del_flag=0,
|
|
)
|
|
db.add_all([second_schedule, second_transfer])
|
|
db.commit()
|
|
|
|
detail = get_payment_chase_detail(
|
|
db,
|
|
schedule_id=first_schedule.id,
|
|
view='payment',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert detail['chaseView'] == 'payment'
|
|
assert detail['targetScheduleId'] == first_schedule.id
|
|
assert [row['id'] for row in detail['schedules']] == [first_schedule.id]
|
|
assert detail['receivableAmount'] == '100.00'
|
|
assert detail['paidAmount'] == '20.00'
|
|
assert detail['dueRemainAmount'] == '80.00'
|
|
assert detail['remainAmount'] == '80.00'
|
|
assert detail['clearedScheduleCount'] == 0
|
|
assert detail['activeScheduleCount'] == 1
|
|
assert detail['progressStatus'] == 'bfyj'
|
|
assert detail['paymentProgress'] == '部分已缴'
|
|
assert [
|
|
period['scheduleId']
|
|
for period in detail['periods']
|
|
if period is not None
|
|
] == [first_schedule.id]
|
|
assert detail['allowedActions'] == []
|
|
|
|
|
|
def test_chase_bound_protocol_detail_is_read_only(monkeypatch):
|
|
from app.services.market_payment import get_payment_chase_detail
|
|
from app.services import market_protocol
|
|
|
|
db = _session()
|
|
_, _, schedule = _payment_tree(db, 'chase-protocol')
|
|
resolved_parents = []
|
|
|
|
def fake_cell_detail(db_arg, protocol, tenant_no):
|
|
resolved_parents.append((db_arg, protocol.id, tenant_no))
|
|
return {
|
|
'source': 'cell_storage',
|
|
'basic': {'protocolNo': protocol.protocol_no},
|
|
'collection': None,
|
|
'customer': None,
|
|
'samples': [],
|
|
'payment': None,
|
|
'paymentModelVersion': protocol.payment_model_version,
|
|
'samplePayments': [],
|
|
'paymentDetails': [],
|
|
'depositTransfers': [],
|
|
'presignDepositHint': None,
|
|
'storageCert': None,
|
|
'fullmoonTodo': None,
|
|
'audits': [],
|
|
'attachments': {},
|
|
}
|
|
|
|
monkeypatch.setattr(market_protocol, '_cell_detail', fake_cell_detail)
|
|
|
|
detail = get_payment_chase_detail(
|
|
db,
|
|
schedule_id=schedule.id,
|
|
view='protocol',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert detail['chaseView'] == 'protocol'
|
|
assert detail['targetScheduleId'] == schedule.id
|
|
assert detail['source'] == 'cell_storage'
|
|
assert detail['basic']['protocolNo'] == 'P-chase-protocol'
|
|
assert detail['allowedActions'] == []
|
|
assert resolved_parents == [(db, 'p-chase-protocol', 'tenant-a')]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
('key', 'tree_kwargs', 'ctx', 'user'),
|
|
[
|
|
('outside-scope', {'sales': 'u2'}, _ctx(), _user()),
|
|
('cross-tenant', {'tenant': 'tenant-b'}, _ctx(), _user()),
|
|
('void-schedule', {'schedule_status': 'void'}, _ctx(), _user()),
|
|
('invalid-due-date', {'due_date': '20260230'}, _ctx(), _user()),
|
|
(
|
|
'outside-chase-window',
|
|
{'due_date': (date.today() + timedelta(days=31)).strftime('%Y%m%d')},
|
|
_ctx(),
|
|
_user(),
|
|
),
|
|
('no-arrears', {'transfer_statuses': ('yjf',) * 5}, _ctx(), _user()),
|
|
],
|
|
)
|
|
def test_chase_bound_detail_rejects_guessed_non_candidate_schedule(
|
|
key, tree_kwargs, ctx, user
|
|
):
|
|
from app.services.market_payment import get_payment_chase_detail
|
|
|
|
db = _session()
|
|
_, _, schedule = _payment_tree(db, key, **tree_kwargs)
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
get_payment_chase_detail(
|
|
db,
|
|
schedule_id=schedule.id,
|
|
view='payment',
|
|
ctx=ctx,
|
|
user_info=user,
|
|
)
|
|
|
|
assert exc.value.status_code == 404
|
|
|
|
|
|
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_includes_legacy_models_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 == 2
|
|
assert {item['id'] for item in items} == {'sp-v2', 'sp-v1'}
|
|
assert {item['paymentModelVersion'] for item in items} == {1, 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_reads_legacy_model_like_web():
|
|
from app.services.market_payment import get_payment_detail
|
|
|
|
db = _session()
|
|
_payment_tree(db, 'legacy', payment_model_version=1)
|
|
|
|
detail = get_payment_detail(
|
|
db, payment_id='sp-legacy', ctx=_ctx(), user_info=_user()
|
|
)
|
|
|
|
assert detail['id'] == 'sp-legacy'
|
|
assert detail['paymentModelVersion'] == 1
|
|
|
|
|
|
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'
|
|
|
|
|
|
@pytest.mark.parametrize('operation', ['confirm', 'reject', 'withdraw'])
|
|
def test_web_finance_transitions_lock_schedule_before_transfer(
|
|
monkeypatch, operation
|
|
):
|
|
from sqlalchemy.orm import Query
|
|
|
|
from app.crud import protocol_sample_payment as payment_crud
|
|
|
|
db = _session()
|
|
_payment_tree(db, f'finance-lock-{operation}', transfer_statuses=('jfqrz',))
|
|
transfer_id = f't-finance-lock-{operation}-0'
|
|
locked_entities = []
|
|
original = Query.with_for_update
|
|
|
|
def record_lock(query, *args, **kwargs):
|
|
entity = query.column_descriptions[0].get('entity')
|
|
if entity in {BizProtocolPaymentSchedule, BizProtocolPaymentTransfer}:
|
|
locked_entities.append(entity)
|
|
return original(query, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(Query, 'with_for_update', record_lock)
|
|
function = {
|
|
'confirm': payment_crud.confirm_receipt,
|
|
'reject': payment_crud.reject_receipt,
|
|
'withdraw': payment_crud.withdraw_receipt,
|
|
}[operation]
|
|
function(
|
|
db,
|
|
transfer_id=transfer_id,
|
|
tenant_no='tenant-a',
|
|
user_id='finance-user',
|
|
)
|
|
|
|
assert locked_entities == [
|
|
BizProtocolPaymentSchedule,
|
|
BizProtocolPaymentTransfer,
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
('operation', 'initial_status'),
|
|
[
|
|
('update', 'dfj'),
|
|
('submit', 'dfj'),
|
|
('withdraw', 'jfqrz'),
|
|
('confirm', 'jfqrz'),
|
|
('reject', 'jfqrz'),
|
|
('cross-tenant', 'dfj'),
|
|
],
|
|
)
|
|
def test_web_payment_transfer_writes_reject_out_of_scope_before_commit(
|
|
operation, initial_status
|
|
):
|
|
from app.api import protocol_sample_payment as payment_api
|
|
|
|
db = _session()
|
|
engine = db.get_bind()
|
|
protocol, _, _ = _payment_tree(
|
|
db,
|
|
f'web-scope-{operation}',
|
|
tenant='tenant-b' if operation == 'cross-tenant' else 'tenant-a',
|
|
transfer_statuses=(initial_status,),
|
|
)
|
|
protocol.create_by = 'owner-user'
|
|
db.add_all(
|
|
[
|
|
protocol,
|
|
SysUser(
|
|
id='denied-user',
|
|
dept_id='denied-dept',
|
|
user_no='denied-user',
|
|
nick_name='越权用户',
|
|
email='',
|
|
phone_no='',
|
|
password='',
|
|
status='1',
|
|
tenant_no='tenant-a',
|
|
create_dept_id='denied-dept',
|
|
del_flag=0,
|
|
),
|
|
SysRole(
|
|
id='self-only-role',
|
|
role_name='仅本人',
|
|
data_scope='4',
|
|
status='1',
|
|
tenant_no='tenant-a',
|
|
create_dept_id='denied-dept',
|
|
del_flag=0,
|
|
),
|
|
SysUserRole(
|
|
id='denied-user-role',
|
|
user_id='denied-user',
|
|
role_id='self-only-role',
|
|
tenant_no='tenant-a',
|
|
create_dept_id='denied-dept',
|
|
del_flag=0,
|
|
),
|
|
]
|
|
)
|
|
db.commit()
|
|
transfer_id = f't-web-scope-{operation}-0'
|
|
user = {
|
|
'user_id': 'denied-user',
|
|
'role': 'admin',
|
|
'dept_id': 'denied-dept',
|
|
}
|
|
if operation in {'update', 'cross-tenant'}:
|
|
response = asyncio.run(
|
|
payment_api.update_payment_transfer(
|
|
transfer_id,
|
|
payment_api.TransferDraftBody(transferAmount=99),
|
|
user=user,
|
|
db=db,
|
|
)
|
|
)
|
|
elif operation == 'submit':
|
|
response = asyncio.run(
|
|
payment_api.submit_payment_transfer_finance(
|
|
transfer_id,
|
|
payment_api.TransferDraftBody(transferAmount=99),
|
|
user=user,
|
|
db=db,
|
|
)
|
|
)
|
|
elif operation == 'withdraw':
|
|
response = asyncio.run(
|
|
payment_api.withdraw_payment_transfer(
|
|
transfer_id,
|
|
user=user,
|
|
db=db,
|
|
)
|
|
)
|
|
elif operation == 'confirm':
|
|
response = asyncio.run(
|
|
payment_api.confirm_payment_transfer(
|
|
transfer_id,
|
|
payment_api.FinanceConfirmBody(financeRemark='must-not-write'),
|
|
user=user,
|
|
db=db,
|
|
)
|
|
)
|
|
else:
|
|
response = asyncio.run(
|
|
payment_api.reject_payment_transfer(
|
|
transfer_id,
|
|
user=user,
|
|
db=db,
|
|
)
|
|
)
|
|
|
|
assert response['code'] == 403
|
|
db.close()
|
|
verify_db = sessionmaker(bind=engine)()
|
|
persisted = verify_db.get(BizProtocolPaymentTransfer, transfer_id)
|
|
assert persisted.pay_status == initial_status
|
|
assert persisted.transfer_amount == Decimal('20')
|
|
assert persisted.finance_remark is None
|
|
verify_db.close()
|
|
|
|
|
|
def test_market_withdraw_locks_schedule_before_transfer(monkeypatch):
|
|
from sqlalchemy.orm import Query
|
|
|
|
from app.services.market_payment import withdraw_transfer
|
|
|
|
db = _session()
|
|
_payment_tree(db, 'market-withdraw-lock', transfer_statuses=('jfqrz',))
|
|
locked_entities = []
|
|
original = Query.with_for_update
|
|
|
|
def record_lock(query, *args, **kwargs):
|
|
entity = query.column_descriptions[0].get('entity')
|
|
if entity in {BizProtocolPaymentSchedule, BizProtocolPaymentTransfer}:
|
|
locked_entities.append(entity)
|
|
return original(query, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(Query, 'with_for_update', record_lock)
|
|
withdraw_transfer(
|
|
db,
|
|
transfer_id='t-market-withdraw-lock-0',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert locked_entities == [
|
|
BizProtocolPaymentSchedule,
|
|
BizProtocolPaymentTransfer,
|
|
]
|
|
|
|
|
|
def test_market_can_soft_delete_only_scoped_draft_transfer():
|
|
from app.services.market_payment import delete_transfer
|
|
|
|
db = _session()
|
|
_payment_tree(db, 'draft-delete', transfer_statuses=('dfj',))
|
|
_payment_tree(db, 'pending-delete', transfer_statuses=('jfqrz',))
|
|
_payment_tree(db, 'outside-delete', sales='u2', transfer_statuses=('dfj',))
|
|
|
|
result = delete_transfer(
|
|
db,
|
|
transfer_id='t-draft-delete-0',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
assert result == {'id': 't-draft-delete-0'}
|
|
assert db.get(BizProtocolPaymentTransfer, 't-draft-delete-0').del_flag == 1
|
|
with pytest.raises(ValueError, match='状态'):
|
|
delete_transfer(
|
|
db,
|
|
transfer_id='t-pending-delete-0',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
with pytest.raises(HTTPException):
|
|
delete_transfer(
|
|
db,
|
|
transfer_id='t-outside-delete-0',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
|
|
|
|
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
|
|
|
|
|
|
def test_transfer_create_rejects_schedule_with_mismatched_payment_parent():
|
|
from app.services.market_payment import create_transfer
|
|
|
|
db = _session()
|
|
_, _, schedule = _payment_tree(db, 'create-parent-a')
|
|
_, other_payment, _ = _payment_tree(db, 'create-parent-b')
|
|
schedule.sample_payment_id = other_payment.id
|
|
db.add(schedule)
|
|
db.commit()
|
|
|
|
before = db.query(BizProtocolPaymentTransfer).count()
|
|
with pytest.raises(HTTPException) as exc:
|
|
create_transfer(
|
|
db,
|
|
schedule_id=schedule.id,
|
|
fields={'transferAmount': 10, 'transferType': 'bank'},
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert exc.value.status_code == 404
|
|
assert db.query(BizProtocolPaymentTransfer).count() == before
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
('parent_field', 'other_parent_id'),
|
|
[
|
|
('schedule_id', 'sch-transfer-parent-b'),
|
|
('sample_payment_id', 'sp-transfer-parent-b'),
|
|
],
|
|
)
|
|
def test_transfer_mutation_rejects_mismatched_parent_links(
|
|
parent_field, other_parent_id
|
|
):
|
|
from app.services.market_payment import update_transfer
|
|
|
|
db = _session()
|
|
_payment_tree(db, 'transfer-parent-a', transfer_statuses=('dfj',))
|
|
_payment_tree(db, 'transfer-parent-b')
|
|
transfer = db.get(BizProtocolPaymentTransfer, 't-transfer-parent-a-0')
|
|
setattr(transfer, parent_field, other_parent_id)
|
|
db.add(transfer)
|
|
db.commit()
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
update_transfer(
|
|
db,
|
|
transfer_id=transfer.id,
|
|
fields={'remark': 'must-not-write'},
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert exc.value.status_code == 404
|
|
db.expire_all()
|
|
assert db.get(BizProtocolPaymentTransfer, transfer.id).remark is None
|
|
|
|
|
|
@pytest.mark.parametrize('protocol_status', ['yzf', 'sdfq', 'zky'])
|
|
def test_terminal_protocol_does_not_block_transfer_create_when_schedule_has_arrears(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
|
|
)
|
|
|
|
created = create_transfer(
|
|
db,
|
|
schedule_id=schedule.id,
|
|
fields={'transferAmount': 10, 'transferType': 'bank'},
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert created['scheduleId'] == schedule.id
|
|
|
|
|
|
def test_cleared_schedule_with_arrears_can_create_transfer_like_web():
|
|
from app.services.market_payment import create_transfer
|
|
|
|
db = _session()
|
|
_, _, schedule = _payment_tree(db, 'cleared-create', schedule_status='cleared')
|
|
|
|
created = create_transfer(
|
|
db,
|
|
schedule_id=schedule.id,
|
|
fields={'transferAmount': 10, 'transferType': 'bank'},
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert created['scheduleId'] == schedule.id
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
('schedule_status', 'confirmed_count'),
|
|
[('void', 0), ('open', 5)],
|
|
)
|
|
def test_transfer_create_rejects_void_or_no_arrears_after_locked_recheck(
|
|
schedule_status, confirmed_count
|
|
):
|
|
from app.services.market_payment import create_transfer
|
|
|
|
db = _session()
|
|
_, _, schedule = _payment_tree(
|
|
db,
|
|
f'create-reject-{schedule_status}-{confirmed_count}',
|
|
schedule_status=schedule_status,
|
|
transfer_statuses=('yjf',) * confirmed_count,
|
|
)
|
|
|
|
before = db.query(BizProtocolPaymentTransfer).count()
|
|
with pytest.raises((HTTPException, ValueError), match='应缴|欠缴'):
|
|
create_transfer(
|
|
db,
|
|
schedule_id=schedule.id,
|
|
fields={'transferAmount': 10, 'transferType': 'bank'},
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert db.query(BizProtocolPaymentTransfer).count() == before
|
|
|
|
|
|
@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_existing_transfer_mutations_depend_on_transfer_status_not_parent_state(
|
|
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'
|
|
if operation == 'update':
|
|
result = update_transfer(
|
|
db,
|
|
transfer_id=transfer_id,
|
|
fields={'remark': 'allowed'},
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
)
|
|
assert result['remark'] == 'allowed'
|
|
elif operation == 'submit':
|
|
result = submit_transfer(
|
|
db, transfer_id=transfer_id, ctx=_ctx(), user_info=_user()
|
|
)
|
|
assert result['payStatus'] == 'jfqrz'
|
|
else:
|
|
result = withdraw_transfer(
|
|
db, transfer_id=transfer_id, ctx=_ctx(), user_info=_user()
|
|
)
|
|
assert result['payStatus'] == 'dfj'
|
|
|
|
|
|
def test_chase_includes_terminal_and_cleared_non_void_schedules_like_web():
|
|
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 == 5
|
|
assert {item['scheduleId'] for item in items} == {
|
|
'sch-eligible-chase',
|
|
'sch-terminal-chase-yzf',
|
|
'sch-terminal-chase-sdfq',
|
|
'sch-terminal-chase-zky',
|
|
'sch-cleared-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-chase/{{schedule_id}}' 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_chase_detail_route_accepts_chase_only_menu_and_denies_missing_menu(monkeypatch):
|
|
from app.api.market_mobile import router
|
|
from app.core import mobile_menu_guard
|
|
|
|
route = next(
|
|
item
|
|
for item in router.routes
|
|
if item.path == '/mini/market/specialist/payment-chase/{schedule_id}'
|
|
)
|
|
guard = next(
|
|
dependency.call
|
|
for dependency in route.dependant.dependencies
|
|
if dependency.name == 'user'
|
|
)
|
|
requested = []
|
|
monkeypatch.setattr(
|
|
mobile_menu_guard,
|
|
'resolve_web_role_for_user',
|
|
lambda *_args, **_kwargs: 'market_specialist',
|
|
)
|
|
monkeypatch.setattr(
|
|
mobile_menu_guard,
|
|
'user_has_tenant_menu_perm',
|
|
lambda _db, **kwargs: requested.append(kwargs['perm_code'])
|
|
or kwargs['perm_code'] == 'mini:market:specialist:payment-chase',
|
|
)
|
|
|
|
assert asyncio.run(guard(user=_user(), db=object())) == _user()
|
|
assert requested == ['mini:market:specialist:payment-chase']
|
|
|
|
monkeypatch.setattr(
|
|
mobile_menu_guard,
|
|
'user_has_tenant_menu_perm',
|
|
lambda *_args, **_kwargs: False,
|
|
)
|
|
with pytest.raises(HTTPException) as exc:
|
|
asyncio.run(guard(user=_user(), db=object()))
|
|
assert exc.value.status_code == 403
|
|
|
|
|
|
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',))
|
|
transfer = db.get(BizProtocolPaymentTransfer, 't-summary-0')
|
|
transfer.finance_remark = '财务已确认'
|
|
db.commit()
|
|
|
|
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
|
|
assert items[0]['periods'][0]['displayPaidAmount'] == 20.0
|
|
assert items[0]['periods'][0]['payDate']
|
|
assert items[0]['periods'][0]['transfers'][0]['financeRemark'] == '财务已确认'
|
|
assert items[0]['schedules'][0]['displayPaidAmount'] == '20.00'
|
|
assert items[0]['schedules'][0]['payDate']
|
|
|
|
|
|
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'
|