57 lines
3.6 KiB
Python
57 lines
3.6 KiB
Python
from uuid import uuid4
|
|
|
|
from fastapi.testclient import TestClient
|
|
import app.main as main
|
|
|
|
|
|
def test_refund_submission_is_atomic_idempotent_and_does_not_mark_paid():
|
|
with TestClient(main.app) as client:
|
|
client.post('/api/admin/auth/login', json={'username': 'operator', 'password': 'operator123456'})
|
|
with main.SessionLocal() as db:
|
|
owner = db.scalar(main.select(main.AdminUser).where(main.AdminUser.username == 'operator'))
|
|
user = main.AnswerUser(public_id=uuid4().hex[:10], phone=uuid4().hex[:20], auth_token=uuid4().hex)
|
|
db.add(user)
|
|
db.flush()
|
|
records = [main.AnswerSubmission(submission_no=uuid4().hex, user_id=user.id, owner_id=owner.id, status='已通过') for _ in range(2)]
|
|
db.add_all(records)
|
|
db.flush()
|
|
ids = [record.id for record in records]
|
|
main._save_refund_recipient(db, 'answer_submission', ids[0], 'payer@example.com', '测试收款人')
|
|
db.commit()
|
|
detail = client.get(f'/api/admin/task-board/{ids[0]}')
|
|
assert detail.status_code == 200
|
|
assert detail.json()['data']['alipay_account'] == 'payer@example.com'
|
|
assert detail.json()['data']['payee_name'] == '测试收款人'
|
|
items = [{'target_kind': 'answer_submission', 'target_id': target_id, 'amount': amount, 'account': 'payer@example.com', 'payee_name': '测试收款人'} for target_id, amount in zip(ids, ['1.25', '2.36'])]
|
|
bad = [items[0], {**items[1], 'amount': 'NaN'}]
|
|
assert client.post('/api/admin/alipay-refunds', json={'items': bad}).status_code == 400
|
|
with main.SessionLocal() as db:
|
|
assert db.scalar(main.select(main.AlipayRefundOrder).where(main.AlipayRefundOrder.target_id.in_(ids))) is None
|
|
assert client.post('/api/admin/alipay-refunds', json={'items': items}).status_code == 200
|
|
assert client.post('/api/admin/alipay-refunds', json={'items': items}).status_code == 409
|
|
with main.SessionLocal() as db:
|
|
assert [db.get(main.AnswerSubmission, target_id).status for target_id in ids] == ['已通过', '已通过']
|
|
orders = db.scalars(main.select(main.AlipayRefundOrder).where(main.AlipayRefundOrder.target_id.in_(ids))).all()
|
|
assert sorted(order.amount_fen for order in orders) == [125, 236]
|
|
assert all(order.status == 'pending' for order in orders)
|
|
listed = client.get('/api/admin/alipay-refunds').json()['data']
|
|
assert any(row['target_id'] == ids[0] and row['submitted'] for row in listed)
|
|
|
|
|
|
def test_refund_requires_login_and_task_access():
|
|
with TestClient(main.app) as client:
|
|
assert client.get('/api/admin/alipay-refunds').status_code == 401
|
|
client.post('/api/admin/auth/login', json={'username': 'operator', 'password': 'operator123456'})
|
|
with main.SessionLocal() as db:
|
|
admin = db.scalar(main.select(main.AdminUser).where(main.AdminUser.username == 'admin'))
|
|
user = main.AnswerUser(public_id=uuid4().hex[:10], phone=uuid4().hex[:20], auth_token=uuid4().hex)
|
|
db.add(user)
|
|
db.flush()
|
|
record = main.AnswerSubmission(submission_no=uuid4().hex, user_id=user.id, owner_id=admin.id, status='已通过')
|
|
db.add(record)
|
|
db.flush()
|
|
target_id = record.id
|
|
db.commit()
|
|
response = client.post('/api/admin/alipay-refunds', json={'items': [{'target_kind': 'answer_submission', 'target_id': target_id, 'amount': '1.00', 'account': 'test@example.com', 'payee_name': '测试'}]})
|
|
assert response.status_code == 404
|