396 lines
20 KiB
Python
396 lines
20 KiB
Python
from fastapi.testclient import TestClient
|
||
from uuid import uuid4
|
||
|
||
from app.main import app
|
||
from app.config import settings
|
||
|
||
|
||
def test_admin_login_me_and_logout():
|
||
with TestClient(app) as client:
|
||
login = client.post("/api/admin/auth/login", json={"username": "admin", "password": "admin123456"})
|
||
assert login.status_code == 200
|
||
assert login.json()["data"]["user"]["role_code"] == "super_admin"
|
||
assert client.get("/api/admin/auth/me").status_code == 200
|
||
assert client.get("/api/admin/auth/login-logs").status_code == 200
|
||
assert client.post("/api/admin/auth/logout").status_code == 200
|
||
assert client.get("/api/admin/auth/me").status_code == 401
|
||
|
||
|
||
def test_wrong_password_is_rejected():
|
||
with TestClient(app) as client:
|
||
response = client.post("/api/admin/auth/login", json={"username": "admin", "password": "wrong"})
|
||
assert response.status_code == 401
|
||
|
||
|
||
def test_operator_cannot_access_super_admin_apis():
|
||
with TestClient(app) as client:
|
||
login = client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"})
|
||
assert login.status_code == 200
|
||
assert client.get("/api/admin/auth/login-logs").status_code == 403
|
||
assert client.get("/api/admin/users").status_code == 403
|
||
assert client.get("/api/admin/settings").status_code == 403
|
||
|
||
|
||
def test_register_and_super_admin_user_crud():
|
||
username = f"tester_{uuid4().hex[:8]}"
|
||
with TestClient(app) as client:
|
||
registration = client.post("/api/admin/auth/register", json={"username": username, "password": "password123", "display_name": "测试账号"})
|
||
assert registration.status_code == 201
|
||
assert registration.json()["data"]["user"]["role_code"] == "operator"
|
||
assert client.post("/api/admin/auth/login", json={"username": "admin", "password": "admin123456"}).status_code == 200
|
||
records = client.get("/api/admin/users").json()["data"]
|
||
created = next(item for item in records if item["username"] == username)
|
||
assert client.get(f"/api/admin/users/{created['id']}").status_code == 200
|
||
update = client.patch(f"/api/admin/users/{created['id']}", json={"display_name": "已修改账号", "is_active": False})
|
||
assert update.status_code == 200
|
||
assert client.delete(f"/api/admin/users/{created['id']}").status_code == 200
|
||
assert client.get(f"/api/admin/users/{created['id']}").status_code == 404
|
||
|
||
|
||
def test_qa_generation_requires_product_and_model_key():
|
||
with TestClient(app) as client:
|
||
assert client.post("/api/admin/qa/generate", json={"question_count": 2}).status_code == 401
|
||
login = client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"})
|
||
assert login.status_code == 200
|
||
assert client.post("/api/admin/qa/generate", json={"question_count": 2}).status_code == 400
|
||
|
||
|
||
def test_accepted_qa_is_saved_to_library_without_duplicates():
|
||
unique_product = f"问答库测试商品-{uuid4().hex[:8]}"
|
||
payload = {
|
||
"product_name": unique_product,
|
||
"category": "测试品类",
|
||
"benefits": "测试功效",
|
||
"selling_points": "测试卖点",
|
||
"items": [{
|
||
"question": "这款商品适合什么人群？",
|
||
"answers": ["适合关注日常使用体验的人群。", "建议结合自身需求选择。"],
|
||
"source": "AI生成",
|
||
}],
|
||
}
|
||
with TestClient(app) as client:
|
||
assert client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"}).status_code == 200
|
||
created = client.post("/api/admin/qa-library", json=payload)
|
||
assert created.status_code == 200
|
||
assert created.json()["data"]["added_count"] == 1
|
||
duplicate = client.post("/api/admin/qa-library", json=payload)
|
||
assert duplicate.status_code == 200
|
||
assert duplicate.json()["data"]["added_count"] == 0
|
||
assert duplicate.json()["data"]["skipped_count"] == 1
|
||
records = client.get(f"/api/admin/qa-library?search={unique_product}")
|
||
assert records.status_code == 200
|
||
assert len(records.json()["data"]) == 1
|
||
assert records.json()["data"][0]["answers"] == payload["items"][0]["answers"]
|
||
entry_id = records.json()["data"][0]["id"]
|
||
edited = client.patch(
|
||
f"/api/admin/qa-library/{entry_id}",
|
||
json={
|
||
**payload,
|
||
"question": "编辑后的问题？",
|
||
"answers": ["编辑后的参考回答。"],
|
||
},
|
||
)
|
||
assert edited.status_code == 200
|
||
assert edited.json()["data"]["question"] == "编辑后的问题？"
|
||
assert edited.json()["data"]["answers"] == ["编辑后的参考回答。"]
|
||
|
||
|
||
def test_task_board_requires_login():
|
||
with TestClient(app) as client:
|
||
assert client.get("/api/admin/task-board").status_code == 401
|
||
assert client.get("/api/admin/task-board/999999").status_code == 401
|
||
assert client.post("/api/admin/task-board/999999/cancel").status_code == 401
|
||
|
||
|
||
def test_task_board_is_available_to_operator():
|
||
with TestClient(app) as client:
|
||
login = client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"})
|
||
assert login.status_code == 200
|
||
response = client.get("/api/admin/task-board")
|
||
assert response.status_code == 200
|
||
assert isinstance(response.json()["data"], list)
|
||
counts = client.get("/api/admin/review-pending-count")
|
||
assert counts.status_code == 200
|
||
count_data = counts.json()["data"]
|
||
assert count_data["total"] == count_data["ask"] + count_data["answer"]
|
||
|
||
|
||
def test_answer_review_rejection_requires_reason():
|
||
with TestClient(app) as client:
|
||
login = client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"})
|
||
assert login.status_code == 200
|
||
missing = client.post("/api/admin/answer-completion-reviews/999999/decision", json={"approved": False})
|
||
assert missing.status_code == 404
|
||
|
||
|
||
def test_answer_rejection_reason_syncs_to_public_user():
|
||
previous_key = settings.llm_api_key
|
||
settings.llm_api_key = ""
|
||
phone = f"19{uuid4().int % 1_000_000_000:09d}"
|
||
try:
|
||
with TestClient(app) as client:
|
||
login = client.post("/api/answer/auth/phone-login", json={"phone": phone})
|
||
assert login.status_code == 200
|
||
token = login.json()["data"]["token"]
|
||
created = client.post(
|
||
"/api/answer/me/submissions",
|
||
data={"token": token},
|
||
files=[("files", ("invited.png", b"test-image", "image/png"))],
|
||
)
|
||
assert created.status_code == 200
|
||
submission = created.json()["data"]
|
||
question_id = submission["questions"][0]["id"]
|
||
|
||
assert client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"}).status_code == 200
|
||
assert client.patch(f"/api/admin/answer-task-drafts/questions/{question_id}", json={"selected": True}).status_code == 200
|
||
assert client.post(f"/api/admin/answer-task-drafts/{submission['id']}/publish").status_code == 200
|
||
completed = client.post(
|
||
f"/api/answer/me/submissions/{submission['id']}/completion",
|
||
data={"token": token},
|
||
files=[("files", ("done.png", b"done-image", "image/png"))],
|
||
)
|
||
assert completed.status_code == 200
|
||
no_reason = client.post(f"/api/admin/answer-completion-reviews/{submission['id']}/decision", json={"approved": False})
|
||
assert no_reason.status_code == 400
|
||
rejected = client.post(f"/api/admin/answer-completion-reviews/{submission['id']}/decision", json={"approved": False, "reason_code": "数量不足", "reason_text": "应完成1条回答，截图未完整展示"})
|
||
assert rejected.status_code == 200
|
||
current = client.get(f"/api/answer/me/submissions?token={token}").json()["data"][0]
|
||
assert current["status"] == "未通过"
|
||
assert current["latest_review"]["reason_code"] == "数量不足"
|
||
assert "截图未完整展示" in current["latest_review"]["reason_text"]
|
||
finally:
|
||
settings.llm_api_key = previous_key
|
||
|
||
|
||
def test_ask_task_publish_appears_on_task_board():
|
||
with TestClient(app) as client:
|
||
assert client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"}).status_code == 200
|
||
missing_link = client.post("/api/admin/ask-tasks/generate", json={"product_name": "测试商品"})
|
||
assert missing_link.status_code == 400
|
||
generated = client.post("/api/admin/ask-tasks/generate", json={"product_link": "https://item.example.com/1", "product_name": "测试商品", "question_count": 3})
|
||
assert generated.status_code == 200
|
||
task = generated.json()["data"]
|
||
published = client.post(f"/api/admin/ask-tasks/{task['id']}/publish")
|
||
assert published.status_code == 200
|
||
board = client.get("/api/admin/task-board").json()["data"]
|
||
created = next(item for item in board if item["task_id"] == task["task_id"])
|
||
assert created["task_type"] == "ask"
|
||
assert created["task_type_label"] == "提问任务"
|
||
assert created["qa_count"] == 3
|
||
|
||
|
||
def test_ask_task_generation_honors_question_count():
|
||
with TestClient(app) as client:
|
||
assert client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"}).status_code == 200
|
||
generated = client.post(
|
||
"/api/admin/ask-tasks/generate",
|
||
json={
|
||
"product_link": "https://item.example.com/count-test",
|
||
"product_name": "数量测试商品",
|
||
"question_count": 25,
|
||
},
|
||
)
|
||
assert generated.status_code == 200
|
||
questions = generated.json()["data"]["selected_questions"]
|
||
assert len(questions) == 25
|
||
assert len({item["question"] for item in questions}) == 25
|
||
|
||
|
||
def test_public_ask_user_gets_stable_unique_five_question_assignment():
|
||
with TestClient(app) as client:
|
||
assert client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"}).status_code == 200
|
||
generated = client.post(
|
||
"/api/admin/ask-tasks/generate",
|
||
json={
|
||
"product_link": "https://item.example.com/public-ask",
|
||
"product_name": "公共提问测试商品",
|
||
"question_count": 8,
|
||
},
|
||
).json()["data"]
|
||
assert client.post(f"/api/admin/ask-tasks/{generated['id']}/publish").status_code == 200
|
||
|
||
phone = f"139{uuid4().int % 100_000_000:08d}"
|
||
session = client.post("/api/ask/auth/phone-login", json={"phone": phone}).json()["data"]
|
||
url = f"/api/ask/tasks/{generated['task_id']}?token={session['token']}"
|
||
first = client.get(url)
|
||
second = client.get(url)
|
||
assert first.status_code == 200
|
||
assert second.status_code == 200
|
||
first_questions = first.json()["data"]["questions"]
|
||
second_questions = second.json()["data"]["questions"]
|
||
assert len(first_questions) == 5
|
||
assert len({item["id"] for item in first_questions}) == 5
|
||
assert [item["id"] for item in first_questions] == [item["id"] for item in second_questions]
|
||
|
||
no_selection = client.post(
|
||
f"/api/ask/tasks/{generated['task_id']}/selection",
|
||
json={"token": session["token"], "question_ids": []},
|
||
)
|
||
assert no_selection.status_code == 400
|
||
chosen = [first_questions[0]["id"]]
|
||
selected = client.post(
|
||
f"/api/ask/tasks/{generated['task_id']}/selection",
|
||
json={"token": session["token"], "question_ids": chosen},
|
||
)
|
||
assert selected.status_code == 200
|
||
assert selected.json()["data"]["status"] == "进行中"
|
||
assert [item["id"] for item in selected.json()["data"]["questions"] if item["selected"]] == chosen
|
||
|
||
submitted = client.post(
|
||
f"/api/ask/tasks/{generated['task_id']}/submission",
|
||
data={"token": session["token"], "payee_name": "测试用户"},
|
||
files=[
|
||
("files", ("question.png", b"question-image", "image/png")),
|
||
("payment_file", ("payment.png", b"payment-image", "image/png")),
|
||
],
|
||
)
|
||
assert submitted.status_code == 200
|
||
assert submitted.json()["data"]["status"] == "待审核"
|
||
assert len(submitted.json()["data"]["question_screenshots"]) == 1
|
||
assert len(submitted.json()["data"]["payment_screenshots"]) == 1
|
||
|
||
reviews = client.get("/api/admin/ask-completion-reviews")
|
||
assert reviews.status_code == 200
|
||
review = next(item for item in reviews.json()["data"] if item["assignment_id"] == submitted.json()["data"]["assignment_id"])
|
||
assert review["status"] == "待审核"
|
||
decision = client.post(
|
||
f"/api/admin/ask-completion-reviews/{review['assignment_id']}/decision",
|
||
json={"approved": True},
|
||
)
|
||
assert decision.status_code == 200
|
||
assert decision.json()["data"]["status"] == "已通过"
|
||
board = client.get("/api/admin/task-board").json()["data"]
|
||
payout = next(
|
||
item
|
||
for item in board
|
||
if item.get("record_kind") == "ask_batch"
|
||
and item["id"] == review["assignment_id"]
|
||
)
|
||
assert payout["status_label"] == "待返款"
|
||
assert payout["task_type"] == "ask"
|
||
assert payout["qa_count"] == 1
|
||
payout_detail = client.get(payout["detail_endpoint"])
|
||
assert payout_detail.status_code == 200
|
||
assert len(payout_detail.json()["data"]["question_screenshots"]) == 1
|
||
assert len(payout_detail.json()["data"]["payment_screenshots"]) == 1
|
||
refunded = client.post(payout["refund_endpoint"])
|
||
assert refunded.status_code == 200
|
||
assert refunded.json()["data"]["status_label"] == "已完成"
|
||
assert refunded.json()["data"]["refund_endpoint"] is None
|
||
board_after_refund = client.get("/api/admin/task-board").json()["data"]
|
||
completed_payout = next(
|
||
item
|
||
for item in board_after_refund
|
||
if item.get("record_kind") == "ask_batch"
|
||
and item["id"] == review["assignment_id"]
|
||
)
|
||
assert completed_payout["status_label"] == "已完成"
|
||
audit_history = client.get("/api/admin/ask-completion-reviews").json()["data"]
|
||
assert not any(item["assignment_id"] == review["assignment_id"] for item in audit_history)
|
||
refreshed = client.get(url)
|
||
assert refreshed.json()["data"]["status"] == "已通过"
|
||
|
||
|
||
def test_approved_answer_appears_as_pending_refund_with_detail():
|
||
previous_key = settings.llm_api_key
|
||
settings.llm_api_key = ""
|
||
phone = f"19{uuid4().int % 1_000_000_000:09d}"
|
||
try:
|
||
with TestClient(app) as client:
|
||
login = client.post("/api/answer/auth/phone-login", json={"phone": phone})
|
||
token = login.json()["data"]["token"]
|
||
created = client.post(
|
||
"/api/answer/me/submissions",
|
||
data={"token": token},
|
||
files=[("files", ("invited.png", b"invited-image", "image/png"))],
|
||
).json()["data"]
|
||
question_id = created["questions"][0]["id"]
|
||
|
||
assert client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"}).status_code == 200
|
||
assert client.patch(f"/api/admin/answer-task-drafts/questions/{question_id}", json={"selected": True}).status_code == 200
|
||
assert client.post(f"/api/admin/answer-task-drafts/{created['id']}/publish").status_code == 200
|
||
completed = client.post(
|
||
f"/api/answer/me/submissions/{created['id']}/completion",
|
||
data={"token": token},
|
||
files=[
|
||
("files", ("done.png", b"done-image", "image/png")),
|
||
("payment_files", ("payment.png", b"payment-image", "image/png")),
|
||
],
|
||
)
|
||
assert completed.status_code == 200
|
||
approved = client.post(
|
||
f"/api/admin/answer-completion-reviews/{created['id']}/decision",
|
||
json={"approved": True},
|
||
)
|
||
assert approved.status_code == 200
|
||
|
||
board = client.get("/api/admin/task-board").json()["data"]
|
||
payout = next(
|
||
item
|
||
for item in board
|
||
if item.get("record_kind") == "answer_submission"
|
||
and item["id"] == created["id"]
|
||
)
|
||
assert payout["status_label"] == "待返款"
|
||
detail = client.get(f"/api/admin/task-board/{created['id']}")
|
||
assert detail.status_code == 200
|
||
assert len(detail.json()["data"]["completion_screenshots"]) == 1
|
||
assert len(detail.json()["data"]["payment_screenshots"]) == 1
|
||
refunded = client.post(payout["refund_endpoint"])
|
||
assert refunded.status_code == 200
|
||
assert refunded.json()["data"]["status_label"] == "已完成"
|
||
assert refunded.json()["data"]["refund_endpoint"] is None
|
||
board_after_refund = client.get("/api/admin/task-board").json()["data"]
|
||
completed_payout = next(
|
||
item
|
||
for item in board_after_refund
|
||
if item.get("record_kind") == "answer_submission"
|
||
and item["id"] == created["id"]
|
||
)
|
||
assert completed_payout["status_label"] == "已完成"
|
||
review_history = client.get("/api/admin/answer-completion-reviews").json()["data"]
|
||
assert not any(item["id"] == created["id"] for item in review_history)
|
||
finally:
|
||
settings.llm_api_key = previous_key
|
||
|
||
|
||
def test_public_ask_questions_are_unique_across_users_and_eventually_exhausted():
|
||
with TestClient(app) as client:
|
||
assert client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"}).status_code == 200
|
||
generated = client.post(
|
||
"/api/admin/ask-tasks/generate",
|
||
json={
|
||
"product_link": "https://item.example.com/unique-allocation",
|
||
"product_name": "跨用户唯一分配测试商品",
|
||
"question_count": 10,
|
||
},
|
||
).json()["data"]
|
||
assert client.post(f"/api/admin/ask-tasks/{generated['id']}/publish").status_code == 200
|
||
|
||
sessions = []
|
||
for _ in range(3):
|
||
phone = f"138{uuid4().int % 100_000_000:08d}"
|
||
sessions.append(client.post("/api/ask/auth/phone-login", json={"phone": phone}).json()["data"])
|
||
|
||
first_url = f"/api/ask/tasks/{generated['task_id']}?token={sessions[0]['token']}"
|
||
second_url = f"/api/ask/tasks/{generated['task_id']}?token={sessions[1]['token']}"
|
||
third_url = f"/api/ask/tasks/{generated['task_id']}?token={sessions[2]['token']}"
|
||
first = client.get(first_url)
|
||
second = client.get(second_url)
|
||
|
||
assert first.status_code == 200
|
||
assert second.status_code == 200
|
||
first_ids = {item["id"] for item in first.json()["data"]["questions"]}
|
||
second_ids = {item["id"] for item in second.json()["data"]["questions"]}
|
||
assert len(first_ids) == 5
|
||
assert len(second_ids) == 5
|
||
assert first_ids.isdisjoint(second_ids)
|
||
assert len(first_ids | second_ids) == 10
|
||
|
||
# Returning users keep their original allocation.
|
||
assert {item["id"] for item in client.get(first_url).json()["data"]["questions"]} == first_ids
|
||
|
||
exhausted = client.get(third_url)
|
||
assert exhausted.status_code == 409
|
||
assert exhausted.json()["detail"] == "本次任务已被领取完，请关注下一个任务！"
|