119 lines
6.7 KiB
Python
119 lines
6.7 KiB
Python
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app import main
|
|
from app.database import Base, get_db
|
|
from app.models import (
|
|
AdminUser, AnswerUser, AnswerSubmission, AnswerRecognizedQuestion,
|
|
AnswerScreenshot, AskTask, AskQuestion, AskAssignment,
|
|
AskAssignmentQuestion, AskAssignmentBatch, RefundRecipient, RefundPaymentAccount,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def setup(monkeypatch):
|
|
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as db:
|
|
owner = AdminUser(username="owner", display_name="运营", password_hash="unused", answer_public_token="owner-link")
|
|
person = AnswerUser(public_id="buyer", phone="13800000000", taobao_id="buyer", auth_token=main.PUBLIC_USER_SESSION_PREFIX + "buyer-token")
|
|
db.add_all([owner, person, RefundPaymentAccount(name="公司账户", alipay_account="payer@example.com", balance_fen=10000, single_limit_fen=50000)])
|
|
db.flush()
|
|
owner.refund_payment_account_id = db.scalar(select(RefundPaymentAccount.id))
|
|
record = AnswerSubmission(submission_no="ANS-test", user_id=person.id, owner_id=owner.id, status="已通过")
|
|
db.add(record)
|
|
db.flush()
|
|
for i, selected in enumerate([True, True, False]):
|
|
db.add(AnswerRecognizedQuestion(submission_id=record.id, sequence_no=i + 1, product_title="商品", question_text=f"问题{i}", selected=selected))
|
|
db.add(RefundRecipient(target_kind="answer_submission", target_id=record.id, account="buyer@example.com", payee_name="张三"))
|
|
db.commit()
|
|
main.app.dependency_overrides[get_db] = lambda: db
|
|
main.app.dependency_overrides[main.current_user] = lambda: owner
|
|
# Exercise uploads without touching production files or image conversion.
|
|
monkeypatch.setattr(main, "screenshot_fields", lambda data, filename, content_type, folder: dict(image_data=data, filename=filename, content_type=content_type))
|
|
with TestClient(main.app, base_url="https://testserver") as client:
|
|
yield db, client, owner, person, record
|
|
main.app.dependency_overrides.clear()
|
|
engine.dispose()
|
|
|
|
|
|
def test_amount_default_override_reset_and_scope(setup):
|
|
db, client, owner, _, record = setup
|
|
data = main._task_board_data(record, include_images=True)
|
|
assert data["refund_amount_fen"] == 100
|
|
assert data["alipay_account"] == "buyer@example.com"
|
|
assert data["payee_name"] == "张三"
|
|
payload = dict(target_kind="answer_submission", target_id=record.id, amount="2.35")
|
|
response = client.patch("/api/admin/task-board/refund-amount", json=payload)
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["data"]["refund_amount_fen"] == 235
|
|
db.expire_all()
|
|
assert main._refund_recipient_data(record, "answer_submission")["refund_amount_fen"] == 235
|
|
response = client.patch("/api/admin/task-board/refund-amount", json={**payload, "amount": None})
|
|
assert response.json()["data"]["refund_amount_fen"] == 100
|
|
stranger = AdminUser(username="stranger", display_name="其他运营", password_hash="unused")
|
|
db.add(stranger)
|
|
db.commit()
|
|
main.app.dependency_overrides[main.current_user] = lambda: stranger
|
|
assert client.patch("/api/admin/task-board/refund-amount", json=payload).status_code == 404
|
|
|
|
|
|
@pytest.mark.parametrize("amount", ["0", "-1", "0.001", "NaN", "Infinity", "1000000", "abc"])
|
|
def test_reject_invalid_amount(setup, amount):
|
|
_, client, _, _, record = setup
|
|
response = client.patch("/api/admin/task-board/refund-amount", json=dict(target_kind="answer_submission", target_id=record.id, amount=amount))
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_confirm_requires_account_without_proof_and_saves_final_amount(setup):
|
|
db, client, _, person, record = setup
|
|
url = f"/api/admin/task-board/{record.id}/confirm-refund"
|
|
assert client.post(url, data={"refund_amount": "1.25"}).status_code == 422
|
|
db.rollback()
|
|
payer_id = db.scalar(select(RefundPaymentAccount.id))
|
|
response = client.post(url, data={"refund_amount": "1.25", "payer_account_id": payer_id})
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["data"]["refund_amount_fen"] == 125
|
|
assert record.status == "已完成"
|
|
assert client.post(url, data={"refund_amount": "2", "payer_account_id": payer_id}).status_code == 409
|
|
assert client.patch("/api/admin/task-board/refund-amount", json=dict(target_kind="answer_submission", target_id=record.id, amount="2")).status_code == 409
|
|
|
|
|
|
def test_answer_completion_without_collection_qr(setup):
|
|
db, client, _, person, record = setup
|
|
record.status = "待用户完成"
|
|
db.commit()
|
|
response = client.post(f"/api/answer/me/submissions/{record.id}/completion", data=dict(token=person.auth_token, owner_token="owner-link", alipay_account="new@example.com", payee_name="李四"), files={"files": ("task.png", b"proof", "image/png")})
|
|
assert response.status_code == 200, response.text
|
|
db.expire_all()
|
|
assert record.status == "待审核"
|
|
assert [image.stage for image in record.screenshots] == ["completion"]
|
|
assert main._refund_recipient_data(record, "answer_submission")["payee_name"] == "李四"
|
|
|
|
|
|
def test_ask_completion_without_collection_qr_and_batch_amount(setup):
|
|
db, client, owner, person, _ = setup
|
|
task = AskTask(task_no="TASK-test", product_link="https://example.com", product_name="商品", owner_id=owner.id, status="已发布")
|
|
db.add(task)
|
|
db.flush()
|
|
assignment = AskAssignment(task_id=task.id, user_id=person.id, status="进行中")
|
|
db.add(assignment)
|
|
db.flush()
|
|
for i in range(2):
|
|
question = AskQuestion(task_id=task.id, sequence_no=i + 1, question_text=f"问题{i}")
|
|
db.add(question)
|
|
db.flush()
|
|
db.add(AskAssignmentQuestion(assignment_id=assignment.id, question_id=question.id, selected=True))
|
|
db.commit()
|
|
response = client.post("/api/ask/tasks/TASK-test/submission", data=dict(token=person.auth_token, alipay_account="ask@example.com", payee_name="王五"), files={"files": ("task.png", b"proof", "image/png")})
|
|
assert response.status_code == 200, response.text
|
|
batch = db.scalar(select(AskAssignmentBatch).where(AskAssignmentBatch.assignment_id == assignment.id))
|
|
assert [image.stage for image in batch.screenshots] == ["question"]
|
|
data = main._refund_recipient_data(batch, "ask_batch")
|
|
assert data["refund_amount_fen"] == 100
|
|
assert data["alipay_account"] == "ask@example.com"
|
|
assert main._refund_recipient_data(assignment, "ask_assignment")["refund_default_amount_fen"] == 100
|