1346 lines
63 KiB
Python
1346 lines
63 KiB
Python
import base64
|
||
from datetime import datetime, timedelta
|
||
from io import BytesIO
|
||
|
||
from fastapi.testclient import TestClient
|
||
from openpyxl import Workbook
|
||
from uuid import uuid4
|
||
|
||
import app.main as main_module
|
||
from app.main import app
|
||
from app.config import settings
|
||
|
||
|
||
def fake_ask_questions(payload, count, reference_text):
|
||
assert "shop_name" not in payload
|
||
return [f"本地测试问题{i + 1}？" for i in range(count)]
|
||
|
||
|
||
def answer_owner_token(client: TestClient, username: str = "operator") -> str:
|
||
password = "operator123456" if username == "operator" else "admin123456"
|
||
assert client.post("/api/admin/auth/login", json={"username": username, "password": password}).status_code == 200
|
||
response = client.get("/api/admin/auth/answer-link")
|
||
assert response.status_code == 200
|
||
return response.json()["data"]["token"]
|
||
|
||
|
||
def task_board_items(client: TestClient, **params):
|
||
response = client.get("/api/admin/task-board", params=params)
|
||
assert response.status_code == 200
|
||
parents = response.json()["data"]["items"]
|
||
return [
|
||
row
|
||
for parent in parents
|
||
for row in [parent, *(parent.get("claimed_tasks") or [])]
|
||
]
|
||
|
||
|
||
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 == 403
|
||
assert client.post("/api/admin/auth/login", json={"username": "admin", "password": "admin123456"}).status_code == 200
|
||
created_response = client.post(
|
||
"/api/admin/users",
|
||
json={"username": username, "password": "password123", "display_name": "测试账号", "role_code": "operator"},
|
||
)
|
||
assert created_response.status_code == 201
|
||
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_imported_questions_can_generate_configured_answer_count(monkeypatch):
|
||
captured = {}
|
||
|
||
def fake_bulk_answers(payload, questions):
|
||
captured["questions"] = questions
|
||
captured["answers_per_question"] = payload.answers_per_question
|
||
return [
|
||
{
|
||
"id": f"Q{index:03d}",
|
||
"question": question,
|
||
"answers": [f"回答{answer_index}" for answer_index in range(1, payload.answers_per_question + 1)],
|
||
"source": "批量导入（AI回答）",
|
||
"status": "待采纳",
|
||
"accepted": False,
|
||
}
|
||
for index, question in enumerate(questions, start=1)
|
||
]
|
||
|
||
monkeypatch.setattr(main_module, "generate_answers_for_questions", fake_bulk_answers)
|
||
payload = {
|
||
"product_name": "批量回答测试商品",
|
||
"questions": ["导入问题一？", "导入问题二？", "导入问题一？"],
|
||
"answers_per_question": 3,
|
||
}
|
||
with TestClient(app) as client:
|
||
assert client.post("/api/admin/qa/generate-answers", json=payload).status_code == 401
|
||
assert client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"}).status_code == 200
|
||
assert client.post(
|
||
"/api/admin/qa/generate-answers",
|
||
json={"questions": ["缺少商品信息？"], "answers_per_question": 2},
|
||
).status_code == 400
|
||
response = client.post("/api/admin/qa/generate-answers", json=payload)
|
||
assert response.status_code == 200
|
||
assert captured == {
|
||
"questions": ["导入问题一？", "导入问题二？"],
|
||
"answers_per_question": 3,
|
||
}
|
||
items = response.json()["data"]["items"]
|
||
assert [item["question"] for item in items] == captured["questions"]
|
||
assert all(len(item["answers"]) == 3 for item in items)
|
||
|
||
|
||
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"]["items"], list)
|
||
assert response.json()["data"]["page"] == 1
|
||
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_taobao_id_login_is_shared_by_both_user_portals():
|
||
taobao_id = f"tb{uuid4().hex[:14]}"
|
||
with TestClient(app) as client:
|
||
owner_token = answer_owner_token(client)
|
||
answer_login = client.post(
|
||
"/api/answer/auth/taobao-login",
|
||
json={"taobao_id": taobao_id, "owner_token": owner_token},
|
||
)
|
||
assert answer_login.status_code == 200
|
||
answer_session = answer_login.json()["data"]
|
||
assert answer_session["user_id"] == taobao_id
|
||
assert answer_session["taobao_id"] == taobao_id
|
||
assert answer_session["account"] == taobao_id
|
||
assert answer_session["phone"] == ""
|
||
assert answer_session["token"].startswith("tb2_")
|
||
|
||
ask_login = client.post(
|
||
"/api/ask/auth/taobao-login",
|
||
json={"taobao_id": taobao_id},
|
||
)
|
||
assert ask_login.status_code == 200
|
||
assert ask_login.json()["data"]["token"] == answer_session["token"]
|
||
|
||
invalid = client.post(
|
||
"/api/ask/auth/taobao-login",
|
||
json={"taobao_id": "不合法 ID"},
|
||
)
|
||
assert invalid.status_code == 400
|
||
|
||
chinese_taobao_id = f"淘宝用户{uuid4().hex[:6]}"
|
||
chinese_login = client.post(
|
||
"/api/ask/auth/taobao-login",
|
||
json={"taobao_id": chinese_taobao_id},
|
||
)
|
||
assert chinese_login.status_code == 200
|
||
assert chinese_login.json()["data"]["taobao_id"] == chinese_taobao_id
|
||
|
||
answer_chinese_login = client.post(
|
||
"/api/answer/auth/taobao-login",
|
||
json={"taobao_id": chinese_taobao_id, "owner_token": owner_token},
|
||
)
|
||
assert answer_chinese_login.status_code == 200
|
||
assert answer_chinese_login.json()["data"]["token"] == chinese_login.json()["data"]["token"]
|
||
|
||
invalid_symbol = client.post(
|
||
"/api/ask/auth/taobao-login",
|
||
json={"taobao_id": "淘宝用户@测试"},
|
||
)
|
||
assert invalid_symbol.status_code == 400
|
||
|
||
|
||
def test_legacy_public_user_sessions_are_invalidated_and_rotated_on_taobao_login():
|
||
taobao_id = f"tb{uuid4().hex[:14]}"
|
||
legacy_token = uuid4().hex
|
||
with main_module.SessionLocal() as db:
|
||
user = main_module.AnswerUser(
|
||
public_id=f"U{uuid4().int % 10_000_000:07d}",
|
||
phone=f"tb_{uuid4().hex[:16]}",
|
||
taobao_id=taobao_id,
|
||
auth_token=legacy_token,
|
||
)
|
||
db.add(user)
|
||
db.commit()
|
||
|
||
with TestClient(app) as client:
|
||
expired = client.get(f"/api/ask/me/assignments?token={legacy_token}")
|
||
assert expired.status_code == 401
|
||
assert "淘宝ID重新登录" in expired.json()["detail"]
|
||
|
||
renewed = client.post("/api/ask/auth/taobao-login", json={"taobao_id": taobao_id})
|
||
assert renewed.status_code == 200
|
||
renewed_token = renewed.json()["data"]["token"]
|
||
assert renewed_token.startswith("tb2_")
|
||
assert renewed_token != legacy_token
|
||
assert client.get(f"/api/ask/me/assignments?token={renewed_token}").status_code == 200
|
||
|
||
|
||
def test_answer_task_publish_remembers_operator_shop_name():
|
||
unique = uuid4().hex[:10]
|
||
remembered_shop = f"记忆店铺-{unique}"
|
||
with TestClient(app) as client:
|
||
assert client.post(
|
||
"/api/admin/auth/login",
|
||
json={"username": "operator", "password": "operator123456"},
|
||
).status_code == 200
|
||
with main_module.SessionLocal() as db:
|
||
owner = db.scalar(
|
||
main_module.select(main_module.AdminUser).where(main_module.AdminUser.username == "operator")
|
||
)
|
||
previous_shop_name = owner.last_answer_shop_name
|
||
public_user = main_module.AnswerUser(
|
||
public_id=f"U{unique[:7]}",
|
||
phone=f"136{uuid4().int % 100_000_000:08d}",
|
||
auth_token=uuid4().hex,
|
||
)
|
||
db.add(public_user)
|
||
db.flush()
|
||
submission = main_module.AnswerSubmission(
|
||
submission_no=f"ANS{unique.upper()}",
|
||
user_id=public_user.id,
|
||
owner_id=owner.id,
|
||
status="待筛选",
|
||
)
|
||
db.add(submission)
|
||
db.flush()
|
||
db.add(main_module.AnswerRecognizedQuestion(
|
||
submission_id=submission.id,
|
||
sequence_no=1,
|
||
product_title="记忆测试商品",
|
||
question_text="店铺名会被记住吗？",
|
||
answer_text="会。",
|
||
selected=True,
|
||
))
|
||
db.commit()
|
||
submission_id = submission.id
|
||
|
||
published = client.post(
|
||
f"/api/admin/answer-task-drafts/{submission_id}/publish",
|
||
json={"shop_name": remembered_shop},
|
||
)
|
||
assert published.status_code == 200
|
||
link_settings = client.get("/api/admin/auth/answer-link")
|
||
assert link_settings.status_code == 200
|
||
assert link_settings.json()["data"]["last_shop_name"] == remembered_shop
|
||
with main_module.SessionLocal() as db:
|
||
owner = db.scalar(
|
||
main_module.select(main_module.AdminUser).where(main_module.AdminUser.username == "operator")
|
||
)
|
||
owner.last_answer_shop_name = previous_shop_name
|
||
db.commit()
|
||
|
||
|
||
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_review_detail_loads_screenshots_from_protected_endpoint():
|
||
unique = uuid4().hex[:10]
|
||
with TestClient(app) as client:
|
||
assert client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"}).status_code == 200
|
||
with main_module.SessionLocal() as db:
|
||
owner = db.scalar(
|
||
main_module.select(main_module.AdminUser).where(main_module.AdminUser.username == "operator")
|
||
)
|
||
public_user = main_module.AnswerUser(
|
||
public_id=f"U{unique[:7]}",
|
||
phone=f"139{uuid4().int % 100_000_000:08d}",
|
||
auth_token=uuid4().hex,
|
||
)
|
||
db.add(public_user)
|
||
db.flush()
|
||
submission = main_module.AnswerSubmission(
|
||
submission_no=f"ANS{unique.upper()}",
|
||
user_id=public_user.id,
|
||
owner_id=owner.id,
|
||
shop_name="审核性能测试店铺",
|
||
status="待审核",
|
||
)
|
||
db.add(submission)
|
||
db.flush()
|
||
screenshot = main_module.AnswerScreenshot(
|
||
submission_id=submission.id,
|
||
stage="completion",
|
||
filename="completion.png",
|
||
content_type="image/png",
|
||
image_data=b"review-image-content",
|
||
)
|
||
db.add_all([
|
||
screenshot,
|
||
main_module.AnswerRecognizedQuestion(
|
||
submission_id=submission.id,
|
||
sequence_no=1,
|
||
product_title="审核性能测试商品",
|
||
question_text="测试问题？",
|
||
answer_text="测试回答。",
|
||
selected=True,
|
||
),
|
||
])
|
||
db.commit()
|
||
submission_id = submission.id
|
||
|
||
detail = client.get(f"/api/admin/answer-completion-reviews/{submission_id}")
|
||
assert detail.status_code == 200
|
||
screenshot_url = detail.json()["data"]["completion_screenshots"][0]["data_url"]
|
||
assert screenshot_url.startswith("/api/admin/answer-screenshots/")
|
||
assert "base64" not in detail.text
|
||
image = client.get(screenshot_url)
|
||
assert image.status_code == 200
|
||
assert image.headers["content-type"] == "image/png"
|
||
assert image.content == b"review-image-content"
|
||
|
||
|
||
def test_ask_review_detail_loads_screenshots_from_protected_endpoint():
|
||
unique = uuid4().hex[:10]
|
||
with TestClient(app) as client:
|
||
assert client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"}).status_code == 200
|
||
with main_module.SessionLocal() as db:
|
||
owner = db.scalar(
|
||
main_module.select(main_module.AdminUser).where(main_module.AdminUser.username == "operator")
|
||
)
|
||
public_user = main_module.AnswerUser(
|
||
public_id=f"U{unique[:7]}",
|
||
phone=f"tb_{unique}",
|
||
taobao_id=f"tb{unique}",
|
||
auth_token=uuid4().hex,
|
||
)
|
||
task = main_module.AskTask(
|
||
task_no=f"TASK{unique.upper()}",
|
||
public_token=uuid4().hex,
|
||
owner_id=owner.id,
|
||
product_link="TEST-TOKEN",
|
||
shop_name="审核性能测试店铺",
|
||
product_name="审核性能测试商品",
|
||
status="已发布",
|
||
)
|
||
db.add_all([public_user, task])
|
||
db.flush()
|
||
question = main_module.AskQuestion(
|
||
task_id=task.id,
|
||
sequence_no=1,
|
||
question_text="提问详情是否轻量？",
|
||
selected=True,
|
||
)
|
||
assignment = main_module.AskAssignment(
|
||
task_id=task.id,
|
||
user_id=public_user.id,
|
||
status="待审核",
|
||
)
|
||
db.add_all([question, assignment])
|
||
db.flush()
|
||
db.add(main_module.AskAssignmentQuestion(
|
||
assignment_id=assignment.id,
|
||
question_id=question.id,
|
||
selected=True,
|
||
))
|
||
batch = main_module.AskAssignmentBatch(
|
||
assignment_id=assignment.id,
|
||
status="pending",
|
||
selected_question_ids=str(question.id),
|
||
)
|
||
db.add(batch)
|
||
db.flush()
|
||
db.add_all([
|
||
main_module.AskAssignmentBatchScreenshot(
|
||
batch_id=batch.id,
|
||
stage="question",
|
||
filename="question.png",
|
||
content_type="image/png",
|
||
image_data=b"ask-question-image",
|
||
),
|
||
main_module.AskAssignmentBatchScreenshot(
|
||
batch_id=batch.id,
|
||
stage="payment",
|
||
filename="payment.png",
|
||
content_type="image/png",
|
||
image_data=b"ask-payment-image",
|
||
),
|
||
])
|
||
db.commit()
|
||
batch_id = batch.id
|
||
|
||
detail = client.get(f"/api/admin/ask-batch-reviews/{batch_id}")
|
||
assert detail.status_code == 200
|
||
data = detail.json()["data"]
|
||
assert "base64" not in detail.text
|
||
question_url = data["question_screenshots"][0]["data_url"]
|
||
payment_url = data["payment_screenshots"][0]["data_url"]
|
||
assert question_url.startswith("/api/admin/ask-batch-screenshots/")
|
||
assert client.get(question_url).content == b"ask-question-image"
|
||
assert client.get(payment_url).content == b"ask-payment-image"
|
||
|
||
decision = client.post(
|
||
f"/api/admin/ask-batch-reviews/{batch_id}/decision",
|
||
json={"approved": True},
|
||
)
|
||
assert decision.status_code == 200
|
||
assert "base64" not in decision.text
|
||
assert decision.json()["data"]["latest_review"]["approved"] is True
|
||
|
||
|
||
def test_failed_answer_recognition_has_no_fake_questions_and_can_retry(monkeypatch):
|
||
unique = uuid4().hex[:10]
|
||
with TestClient(app) as client:
|
||
assert client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"}).status_code == 200
|
||
with main_module.SessionLocal() as db:
|
||
owner = db.scalar(
|
||
main_module.select(main_module.AdminUser).where(main_module.AdminUser.username == "operator")
|
||
)
|
||
public_user = main_module.AnswerUser(
|
||
public_id=f"U{unique[:7]}",
|
||
phone=f"137{uuid4().int % 100_000_000:08d}",
|
||
auth_token=uuid4().hex,
|
||
)
|
||
db.add(public_user)
|
||
db.flush()
|
||
submission = main_module.AnswerSubmission(
|
||
submission_no=f"ANS{unique.upper()}",
|
||
user_id=public_user.id,
|
||
owner_id=owner.id,
|
||
status="待筛选",
|
||
)
|
||
db.add(submission)
|
||
db.flush()
|
||
db.add(main_module.AnswerScreenshot(
|
||
submission_id=submission.id,
|
||
stage="invited",
|
||
filename="invited.png",
|
||
content_type="image/png",
|
||
image_data=b"image-content",
|
||
))
|
||
db.commit()
|
||
submission_id = submission.id
|
||
|
||
monkeypatch.setattr(main_module, "extract_questions_from_images", lambda images: [])
|
||
main_module._process_answer_submission(submission_id)
|
||
with main_module.SessionLocal() as db:
|
||
failed = db.get(main_module.AnswerSubmission, submission_id)
|
||
assert failed.status == "识别失败"
|
||
assert failed.questions == []
|
||
|
||
monkeypatch.setattr(
|
||
main_module,
|
||
"extract_questions_from_images",
|
||
lambda images: [{"product_title": "识别商品", "question": "重新识别成功了吗？", "source_index": 0}],
|
||
)
|
||
monkeypatch.setattr(main_module, "generate_answer_for_question", lambda product, question: "重新识别成功。")
|
||
retried = client.post(f"/api/admin/answer-task-drafts/{submission_id}/retry-recognition")
|
||
assert retried.status_code == 200
|
||
with main_module.SessionLocal() as db:
|
||
completed = db.get(main_module.AnswerSubmission, submission_id)
|
||
assert completed.status == "待筛选"
|
||
assert [question.question_text for question in completed.questions] == ["重新识别成功了吗？"]
|
||
question_id = completed.questions[0].id
|
||
|
||
cleared = client.patch(
|
||
f"/api/admin/answer-task-drafts/questions/{question_id}",
|
||
json={"answer": ""},
|
||
)
|
||
assert cleared.status_code == 200
|
||
assert cleared.json()["data"]["answer"] == ""
|
||
with main_module.SessionLocal() as db:
|
||
assert db.get(main_module.AnswerRecognizedQuestion, question_id).answer_text == ""
|
||
|
||
|
||
def test_answer_rejection_reason_syncs_to_public_user(monkeypatch):
|
||
previous_key = settings.llm_api_key
|
||
settings.llm_api_key = ""
|
||
monkeypatch.setattr(
|
||
main_module,
|
||
"extract_questions_from_images",
|
||
lambda images: [{"product_title": "测试商品", "question": "测试问题？", "source_index": 0}],
|
||
)
|
||
taobao_id = f"tb{uuid4().hex[:14]}"
|
||
try:
|
||
with TestClient(app) as client:
|
||
owner_token = answer_owner_token(client)
|
||
login = client.post("/api/answer/auth/taobao-login", json={"taobao_id": taobao_id, "owner_token": owner_token})
|
||
assert login.status_code == 200
|
||
token = login.json()["data"]["token"]
|
||
created = client.post(
|
||
"/api/answer/me/submissions",
|
||
data={"token": token, "owner_token": owner_token},
|
||
files=[("files", ("invited.png", b"test-image", "image/png"))],
|
||
)
|
||
assert created.status_code == 200
|
||
submission = created.json()["data"]
|
||
submission = client.get(
|
||
f"/api/answer/me/submissions/{submission['id']}?token={token}&owner_token={owner_token}"
|
||
).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",
|
||
json={"shop_name": "测试店铺"},
|
||
).status_code == 200
|
||
completed = client.post(
|
||
f"/api/answer/me/submissions/{submission['id']}/completion",
|
||
data={"token": token, "owner_token": owner_token},
|
||
files=[
|
||
("files", ("done.png", b"done-image", "image/png")),
|
||
("payment_files", ("payment.png", b"payment-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}&owner_token={owner_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(monkeypatch):
|
||
monkeypatch.setattr(main_module, "generate_questions", fake_ask_questions)
|
||
with TestClient(app) as client:
|
||
assert client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"}).status_code == 200
|
||
generation_without_publish_fields = client.post(
|
||
"/api/admin/ask-tasks/generate",
|
||
json={"product_name": "测试商品", "question_count": 3},
|
||
)
|
||
assert generation_without_publish_fields.status_code == 200
|
||
assert generation_without_publish_fields.json()["data"]["status"] == "草稿"
|
||
generated = client.post("/api/admin/ask-tasks/generate", json={"shop_name": "测试店铺", "product_link": "https://item.example.com/1", "product_name": "测试商品", "question_count": 3})
|
||
assert generated.status_code == 200
|
||
task = generated.json()["data"]
|
||
assert task["public_token"] is None
|
||
assert task["shop_name"] == "测试店铺"
|
||
published = client.post(f"/api/admin/ask-tasks/{task['id']}/publish")
|
||
assert published.status_code == 200
|
||
public_token = published.json()["data"]["public_token"]
|
||
assert public_token
|
||
assert public_token != task["task_id"]
|
||
assert len(public_token) >= 24
|
||
board = task_board_items(client, page_size=100)
|
||
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
|
||
assert created["public_token"] == public_token
|
||
assert created["shop_name"] == "测试店铺"
|
||
detail = client.get(created["detail_endpoint"])
|
||
assert detail.status_code == 200
|
||
assert detail.json()["data"]["shop_name"] == "测试店铺"
|
||
|
||
|
||
def test_imported_questions_can_publish_without_ai_generation():
|
||
with TestClient(app) as client:
|
||
assert client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"}).status_code == 200
|
||
published = client.post(
|
||
"/api/admin/ask-tasks/publish",
|
||
json={
|
||
"shop_name": "直接发布测试店铺",
|
||
"product_link": "TOKEN-DIRECT-PUBLISH",
|
||
"product_name": "直接发布测试商品",
|
||
"claim_questions_per_user": 20,
|
||
"questions": [
|
||
{"id": "new-import-1", "question": "批量导入问题一？", "source": "批量导入"},
|
||
{"id": "new-manual-2", "question": "自主添加问题二？", "source": "自主添加"},
|
||
],
|
||
},
|
||
)
|
||
assert published.status_code == 200
|
||
data = published.json()["data"]
|
||
assert data["status"] == "已发布"
|
||
assert data["shop_name"] == "直接发布测试店铺"
|
||
assert data["claim_questions_per_user"] == 20
|
||
assert [item["question"] for item in data["selected_questions"]] == ["批量导入问题一？", "自主添加问题二？"]
|
||
assert [item["source"] for item in data["selected_questions"]] == ["批量导入", "自主添加"]
|
||
|
||
missing_fields = client.post(
|
||
"/api/admin/ask-tasks/publish",
|
||
json={"questions": [{"question": "不能缺少发布信息？", "source": "批量导入"}]},
|
||
)
|
||
assert missing_fields.status_code == 400
|
||
|
||
over_limit = client.post(
|
||
"/api/admin/ask-tasks/publish",
|
||
json={
|
||
"shop_name": "领取上限测试店铺",
|
||
"product_link": "TOKEN-CLAIM-LIMIT",
|
||
"claim_questions_per_user": 21,
|
||
"questions": [{"question": "超过领取上限？", "source": "自主添加"}],
|
||
},
|
||
)
|
||
assert over_limit.status_code == 400
|
||
assert over_limit.json()["detail"] == "每个用户可领取的问题数量必须在 2-20 条之间"
|
||
|
||
|
||
def test_ask_task_generation_honors_question_count(monkeypatch):
|
||
monkeypatch.setattr(main_module, "generate_questions", fake_ask_questions)
|
||
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={
|
||
"shop_name": "数量测试店铺",
|
||
"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_ask_task_accepts_qr_image_without_token_and_excludes_it_from_ai(monkeypatch):
|
||
monkeypatch.setattr(main_module, "generate_questions", fake_ask_questions)
|
||
qr_data_url = "data:image/png;base64," + base64.b64encode(b"local-qr-image").decode()
|
||
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={
|
||
"shop_name": "二维码测试店铺",
|
||
"product_link": "",
|
||
"product_qr_image": qr_data_url,
|
||
"product_qr_image_name": "商品二维码.png",
|
||
"product_name": "二维码测试商品",
|
||
"question_count": 2,
|
||
},
|
||
)
|
||
assert generated.status_code == 200
|
||
task = generated.json()["data"]
|
||
assert task["product_link"] == ""
|
||
assert task["product_qr"]["content_type"] == "image/png"
|
||
assert task["product_qr"]["name"] == "商品二维码.png"
|
||
assert task["product_qr"]["data_url"] == qr_data_url
|
||
published = client.post(f"/api/admin/ask-tasks/{task['id']}/publish")
|
||
assert published.status_code == 200
|
||
assert published.json()["data"]["status"] == "已发布"
|
||
board = task_board_items(client, page_size=100)
|
||
created = next(item for item in board if item["task_id"] == task["task_id"])
|
||
detail = client.get(created["detail_endpoint"])
|
||
assert detail.status_code == 200
|
||
assert detail.json()["data"]["product_qr"]["data_url"] == qr_data_url
|
||
public_token = published.json()["data"]["public_token"]
|
||
assert public_token
|
||
|
||
|
||
def test_ask_task_regeneration_appends_to_existing_questions(monkeypatch):
|
||
monkeypatch.setattr(main_module, "generate_questions", fake_ask_questions)
|
||
with TestClient(app) as client:
|
||
assert client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"}).status_code == 200
|
||
first = client.post("/api/admin/ask-tasks/generate", json={"shop_name": "追加测试店铺", "product_link": "https://item.example.com/append", "product_name": "追加测试商品", "question_count": 2})
|
||
assert first.status_code == 200
|
||
task = first.json()["data"]
|
||
assert len(task["selected_questions"]) == 2
|
||
second = client.post("/api/admin/ask-tasks/generate", json={"task_id": task["id"], "shop_name": "追加测试店铺", "product_link": "https://item.example.com/append", "product_name": "追加测试商品", "question_count": 2})
|
||
assert second.status_code == 200
|
||
questions = second.json()["data"]["selected_questions"]
|
||
assert len(questions) == 4
|
||
assert [item["sequence_no"] for item in questions] == [1, 2, 3, 4]
|
||
|
||
|
||
def test_public_ask_assignment_expires_after_selection_timeout(monkeypatch):
|
||
monkeypatch.setattr(main_module, "generate_questions", fake_ask_questions)
|
||
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={"shop_name": "超时测试店铺", "product_link": "TOKEN-TIMEOUT-1", "product_name": "超时测试商品", "question_count": 2},
|
||
).json()["data"]
|
||
assert client.post(f"/api/admin/ask-tasks/{generated['id']}/publish").status_code == 200
|
||
taobao_id = f"tb{uuid4().hex[:14]}"
|
||
token = client.post("/api/ask/auth/taobao-login", json={"taobao_id": taobao_id}).json()["data"]["token"]
|
||
task_url = f"/api/ask/tasks/{generated['task_id']}?token={token}"
|
||
first = client.get(task_url)
|
||
assert first.status_code == 200
|
||
first_data = first.json()["data"]
|
||
assignment_id = first_data["assignment_id"]
|
||
with main_module.SessionLocal() as db:
|
||
assignment = db.get(main_module.AskAssignment, assignment_id)
|
||
assignment.reservation_started_at = datetime.utcnow() - timedelta(minutes=2)
|
||
db.commit()
|
||
expired = client.get(task_url)
|
||
assert expired.status_code == 409
|
||
assert expired.json()["detail"] == "提问任务已失效，请点击刷新重新分配任务"
|
||
refreshed = client.get(task_url)
|
||
assert refreshed.status_code == 200
|
||
assert refreshed.json()["data"]["created_at"] != first_data["created_at"]
|
||
|
||
|
||
def test_public_ask_assignment_expires_after_upload_timeout(monkeypatch):
|
||
monkeypatch.setattr(main_module, "generate_questions", fake_ask_questions)
|
||
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={"shop_name": "超时测试店铺二", "product_link": "TOKEN-TIMEOUT-2", "product_name": "超时测试商品二", "question_count": 2},
|
||
).json()["data"]
|
||
assert client.post(f"/api/admin/ask-tasks/{generated['id']}/publish").status_code == 200
|
||
taobao_id = f"tb{uuid4().hex[:14]}"
|
||
token = client.post("/api/ask/auth/taobao-login", json={"taobao_id": taobao_id}).json()["data"]["token"]
|
||
task_url = f"/api/ask/tasks/{generated['task_id']}?token={token}"
|
||
assignment = client.get(task_url).json()["data"]
|
||
selected = assignment["questions"][0]["id"]
|
||
confirmed = client.post(f"/api/ask/tasks/{generated['task_id']}/selection", json={"token": token, "question_ids": [selected]})
|
||
assert confirmed.status_code == 200
|
||
with main_module.SessionLocal() as db:
|
||
stored = db.get(main_module.AskAssignment, assignment["assignment_id"])
|
||
stored.reservation_started_at = datetime.utcnow() - timedelta(minutes=31)
|
||
db.commit()
|
||
expired = client.get(task_url)
|
||
assert expired.status_code == 409
|
||
assert expired.json()["detail"] == "提问任务已失效，请点击刷新重新分配任务"
|
||
refreshed = client.get(task_url)
|
||
assert refreshed.status_code == 200
|
||
assert refreshed.json()["data"]["status"] == "待选择"
|
||
|
||
|
||
def test_ask_task_added_question_is_idempotent_and_deletable(monkeypatch):
|
||
monkeypatch.setattr(main_module, "generate_questions", fake_ask_questions)
|
||
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/ask-tasks/generate",
|
||
json={
|
||
"shop_name": f"去重测试店铺-{uuid4().hex[:8]}",
|
||
"product_link": "TOKEN-IDEMPOTENT",
|
||
"product_name": "去重测试商品",
|
||
"question_count": 1,
|
||
},
|
||
)
|
||
assert created.status_code == 200
|
||
task_id = created.json()["data"]["id"]
|
||
first = client.post(f"/api/admin/ask-tasks/{task_id}/questions", json={"question": "问答库重复问题？", "source": "问答库"})
|
||
second = client.post(f"/api/admin/ask-tasks/{task_id}/questions", json={"question": "问答库重复问题？", "source": "问答库"})
|
||
assert first.status_code == 200 and second.status_code == 200
|
||
assert first.json()["data"]["id"] == second.json()["data"]["id"]
|
||
assert second.json()["data"]["source"] == "问答库"
|
||
question_id = first.json()["data"]["id"]
|
||
removed = client.delete(f"/api/admin/ask-tasks/{task_id}/questions/{question_id}")
|
||
assert removed.status_code == 200
|
||
refreshed = client.post(
|
||
"/api/admin/ask-tasks/generate",
|
||
json={
|
||
"task_id": task_id,
|
||
"shop_name": "去重测试店铺",
|
||
"product_link": "TOKEN-IDEMPOTENT",
|
||
"product_name": "去重测试商品",
|
||
"question_count": 1,
|
||
},
|
||
)
|
||
assert refreshed.status_code == 200
|
||
assert all(item["question"] != "问答库重复问题？" for item in refreshed.json()["data"]["selected_questions"])
|
||
|
||
|
||
def test_ask_question_file_import_parses_supported_text_formats():
|
||
with TestClient(app) as client:
|
||
assert client.post("/api/admin/auth/login", json={"username": "operator", "password": "operator123456"}).status_code == 200
|
||
|
||
text_import = client.post(
|
||
"/api/admin/ask-questions/import",
|
||
files={"file": ("questions.txt", "Q1：第一条问题？\nA1：参考回答\nQ2：第二条问题？\nQ1：第一条问题？".encode("utf-8"), "text/plain")},
|
||
)
|
||
assert text_import.status_code == 200
|
||
assert text_import.json()["data"]["questions"] == ["第一条问题？", "第二条问题？"]
|
||
|
||
csv_import = client.post(
|
||
"/api/admin/ask-questions/import",
|
||
files={"file": ("questions.csv", "问题,答案\nCSV问题一？,参考回答\nCSV问题二？,参考回答".encode("utf-8"), "text/csv")},
|
||
)
|
||
assert csv_import.status_code == 200
|
||
assert csv_import.json()["data"]["questions"] == ["CSV问题一？", "CSV问题二？"]
|
||
|
||
workbook = Workbook()
|
||
sheet = workbook.active
|
||
sheet.append(["问题", "答案"])
|
||
sheet.append(["表格问题一？", "参考回答"])
|
||
sheet.append(["表格问题二？", "参考回答"])
|
||
workbook_file = BytesIO()
|
||
workbook.save(workbook_file)
|
||
workbook.close()
|
||
xlsx_import = client.post(
|
||
"/api/admin/ask-questions/import",
|
||
files={"file": ("questions.xlsx", workbook_file.getvalue(), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")},
|
||
)
|
||
assert xlsx_import.status_code == 200
|
||
assert xlsx_import.json()["data"]["questions"] == ["表格问题一？", "表格问题二？"]
|
||
|
||
json_import = client.post(
|
||
"/api/admin/ask-questions/import",
|
||
files={"file": ("questions.json", '[{"question":"JSON问题一？"},{"问题":"JSON问题二？"}]'.encode("utf-8"), "application/json")},
|
||
)
|
||
assert json_import.status_code == 200
|
||
assert json_import.json()["data"]["questions"] == ["JSON问题一？", "JSON问题二？"]
|
||
|
||
|
||
def test_ask_question_bulk_import_appends_and_deduplicates(monkeypatch):
|
||
monkeypatch.setattr(main_module, "generate_questions", fake_ask_questions)
|
||
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/ask-tasks/generate",
|
||
json={
|
||
"shop_name": f"批量导入测试店铺-{uuid4().hex[:8]}",
|
||
"product_link": "TOKEN-BULK-IMPORT",
|
||
"product_name": "批量导入测试商品",
|
||
"question_count": 1,
|
||
},
|
||
)
|
||
assert created.status_code == 200
|
||
task_id = created.json()["data"]["id"]
|
||
imported = client.post(
|
||
f"/api/admin/ask-tasks/{task_id}/questions/bulk",
|
||
json={"questions": ["批量问题一？", "批量问题二？", "批量问题一？", "本地测试问题1？"]},
|
||
)
|
||
assert imported.status_code == 200
|
||
result = imported.json()["data"]
|
||
assert result["added_count"] == 2
|
||
assert result["skipped_count"] == 2
|
||
assert [item["question"] for item in result["items"]] == ["批量问题一？", "批量问题二？"]
|
||
assert all(item["source"] == "批量导入" for item in result["items"])
|
||
|
||
|
||
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={
|
||
"shop_name": "公共提问测试店铺",
|
||
"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
|
||
|
||
taobao_id = f"tb{uuid4().hex[:14]}"
|
||
session = client.post("/api/ask/auth/taobao-login", json={"taobao_id": taobao_id}).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"] == "待审核"
|
||
batch_detail = client.get(
|
||
f"/api/admin/ask-batch-reviews/{submitted.json()['data']['batch_id']}"
|
||
)
|
||
assert batch_detail.status_code == 200
|
||
assert len(batch_detail.json()["data"]["question_screenshots"]) == 1
|
||
assert len(batch_detail.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-batch-reviews/{review['batch_id']}/decision",
|
||
json={"approved": True},
|
||
)
|
||
assert decision.status_code == 200
|
||
assert decision.json()["data"]["status"] == "已通过"
|
||
assert "base64" not in decision.text
|
||
board = task_board_items(client, page_size=100)
|
||
payout = next(
|
||
item
|
||
for item in board
|
||
if item.get("record_kind") == "ask_batch"
|
||
and item["id"] == review["batch_id"]
|
||
)
|
||
assert payout["status_label"] == "待返款"
|
||
assert payout["task_type"] == "ask"
|
||
assert payout["qa_count"] == 1
|
||
assert payout["completed_count"] == 0
|
||
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
|
||
question_image_url = payout_detail.json()["data"]["question_screenshots"][0]["data_url"]
|
||
payment_image_url = payout_detail.json()["data"]["payment_screenshots"][0]["data_url"]
|
||
assert question_image_url.startswith("/api/admin/ask-")
|
||
assert payment_image_url.startswith("/api/admin/ask-")
|
||
assert "base64" not in payout_detail.text
|
||
assert client.get(question_image_url).content == b"question-image"
|
||
assert client.get(payment_image_url).content == b"payment-image"
|
||
refunded = client.post(
|
||
payout["refund_endpoint"],
|
||
files={"refund_file": ("refund.png", b"refund-image", "image/png")},
|
||
)
|
||
assert refunded.status_code == 200
|
||
assert refunded.json()["data"]["status_label"] == "已完成"
|
||
assert refunded.json()["data"]["refund_endpoint"] is None
|
||
board_after_refund = task_board_items(client, page_size=100)
|
||
completed_payout = next(
|
||
item
|
||
for item in board_after_refund
|
||
if item.get("record_kind") == "ask_batch"
|
||
and item["id"] == review["batch_id"]
|
||
)
|
||
assert completed_payout["status_label"] == "已完成"
|
||
assert completed_payout["completed_count"] == 1
|
||
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(monkeypatch):
|
||
previous_key = settings.llm_api_key
|
||
settings.llm_api_key = ""
|
||
monkeypatch.setattr(
|
||
main_module,
|
||
"extract_questions_from_images",
|
||
lambda images: [{"product_title": "测试商品", "question": "测试问题？", "source_index": 0}],
|
||
)
|
||
taobao_id = f"tb{uuid4().hex[:14]}"
|
||
try:
|
||
with TestClient(app) as client:
|
||
owner_token = answer_owner_token(client)
|
||
login = client.post("/api/answer/auth/taobao-login", json={"taobao_id": taobao_id, "owner_token": owner_token})
|
||
token = login.json()["data"]["token"]
|
||
created = client.post(
|
||
"/api/answer/me/submissions",
|
||
data={"token": token, "owner_token": owner_token},
|
||
files=[("files", ("invited.png", b"invited-image", "image/png"))],
|
||
).json()["data"]
|
||
created = client.get(
|
||
f"/api/answer/me/submissions/{created['id']}?token={token}&owner_token={owner_token}"
|
||
).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",
|
||
json={"shop_name": "测试店铺"},
|
||
).status_code == 200
|
||
completed = client.post(
|
||
f"/api/answer/me/submissions/{created['id']}/completion",
|
||
data={"token": token, "owner_token": owner_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 = task_board_items(client, page_size=100)
|
||
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"],
|
||
files={"refund_file": ("refund.png", b"refund-image", "image/png")},
|
||
)
|
||
assert refunded.status_code == 200
|
||
assert refunded.json()["data"]["status_label"] == "已完成"
|
||
assert refunded.json()["data"]["refund_endpoint"] is None
|
||
user_detail = client.get(
|
||
f"/api/answer/me/submissions/{created['id']}?token={token}&owner_token={owner_token}"
|
||
)
|
||
assert user_detail.status_code == 200
|
||
assert user_detail.json()["data"]["status"] == "已完成"
|
||
assert len(user_detail.json()["data"]["refund_screenshots"]) == 1
|
||
board_after_refund = task_board_items(client, page_size=100)
|
||
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={
|
||
"shop_name": "唯一分配测试店铺",
|
||
"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):
|
||
taobao_id = f"tb{uuid4().hex[:14]}"
|
||
sessions.append(client.post("/api/ask/auth/taobao-login", json={"taobao_id": taobao_id}).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"] == "本次任务已被领取完，请关注下一个任务！"
|
||
|
||
|
||
def test_unselected_ask_questions_are_released_immediately(monkeypatch):
|
||
monkeypatch.setattr(main_module, "generate_questions", fake_ask_questions)
|
||
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={
|
||
"shop_name": "未领取释放测试店铺",
|
||
"product_link": "TOKEN-RELEASE-UNSELECTED",
|
||
"product_name": "未领取释放测试商品",
|
||
"question_count": 8,
|
||
},
|
||
).json()["data"]
|
||
assert client.post(f"/api/admin/ask-tasks/{generated['id']}/publish").status_code == 200
|
||
|
||
first_session = client.post(
|
||
"/api/ask/auth/taobao-login",
|
||
json={"taobao_id": f"tb{uuid4().hex[:14]}"},
|
||
).json()["data"]
|
||
second_session = client.post(
|
||
"/api/ask/auth/taobao-login",
|
||
json={"taobao_id": f"tb{uuid4().hex[:14]}"},
|
||
).json()["data"]
|
||
first_url = f"/api/ask/tasks/{generated['task_id']}?token={first_session['token']}"
|
||
first_assignment = client.get(first_url).json()["data"]
|
||
first_ids = {item["id"] for item in first_assignment["questions"]}
|
||
assert len(first_ids) == 5
|
||
|
||
selected_ids = set(list(first_ids)[:3])
|
||
confirmed = client.post(
|
||
f"/api/ask/tasks/{generated['task_id']}/selection",
|
||
json={"token": first_session["token"], "question_ids": list(selected_ids)},
|
||
)
|
||
assert confirmed.status_code == 200
|
||
confirmed_ids = {item["id"] for item in confirmed.json()["data"]["questions"]}
|
||
assert confirmed_ids == selected_ids
|
||
|
||
second_url = f"/api/ask/tasks/{generated['task_id']}?token={second_session['token']}"
|
||
second_assignment = client.get(second_url)
|
||
assert second_assignment.status_code == 200
|
||
second_ids = {item["id"] for item in second_assignment.json()["data"]["questions"]}
|
||
assert len(second_ids) == 5
|
||
assert second_ids == {
|
||
item["id"]
|
||
for item in generated["selected_questions"]
|
||
} - selected_ids
|
||
|
||
|
||
def test_answer_submission_requires_owner_link_and_allows_at_most_two_images():
|
||
with TestClient(app) as client:
|
||
missing_owner = client.post(
|
||
"/api/answer/auth/taobao-login",
|
||
json={"taobao_id": f"tb{uuid4().hex[:14]}"},
|
||
)
|
||
assert missing_owner.status_code == 400
|
||
|
||
owner_token = answer_owner_token(client)
|
||
login = client.post(
|
||
"/api/answer/auth/taobao-login",
|
||
json={"taobao_id": f"tb{uuid4().hex[:14]}", "owner_token": owner_token},
|
||
)
|
||
assert login.status_code == 200
|
||
token = login.json()["data"]["token"]
|
||
too_many = client.post(
|
||
"/api/answer/me/submissions",
|
||
data={"token": token, "owner_token": owner_token},
|
||
files=[
|
||
("files", (f"shot-{index}.png", b"image", "image/png"))
|
||
for index in range(3)
|
||
],
|
||
)
|
||
assert too_many.status_code == 400
|
||
assert too_many.json()["detail"] == "一次最多上传2张截图"
|
||
|
||
|
||
def test_ask_batches_survive_reservation_timeout_and_parent_completes_only_after_all_paid(monkeypatch):
|
||
monkeypatch.setattr(main_module, "generate_questions", fake_ask_questions)
|
||
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={
|
||
"shop_name": "分批保留测试店铺",
|
||
"product_link": "https://item.example.com/batch-history",
|
||
"product_name": "分批保留测试商品",
|
||
"question_count": 2,
|
||
"claim_questions_per_user": 2,
|
||
},
|
||
).json()["data"]
|
||
assert client.post(f"/api/admin/ask-tasks/{generated['id']}/publish").status_code == 200
|
||
|
||
session = client.post(
|
||
"/api/ask/auth/taobao-login",
|
||
json={"taobao_id": f"tb{uuid4().hex[:14]}"},
|
||
).json()["data"]
|
||
task_url = f"/api/ask/tasks/{generated['task_id']}?token={session['token']}"
|
||
assignment = client.get(task_url).json()["data"]
|
||
first_question_id = assignment["questions"][0]["id"]
|
||
assert client.post(
|
||
f"/api/ask/tasks/{generated['task_id']}/selection",
|
||
json={"token": session["token"], "question_ids": [first_question_id]},
|
||
).status_code == 200
|
||
first_batch = client.post(
|
||
f"/api/ask/tasks/{generated['task_id']}/submission",
|
||
data={"token": session["token"], "payee_name": "测试用户"},
|
||
files=[
|
||
("files", ("question-1.png", b"question-1", "image/png")),
|
||
("payment_file", ("payment-1.png", b"payment-1", "image/png")),
|
||
],
|
||
).json()["data"]
|
||
assert client.post(
|
||
f"/api/admin/ask-batch-reviews/{first_batch['batch_id']}/decision",
|
||
json={"approved": True},
|
||
).status_code == 200
|
||
assert client.post(
|
||
f"/api/admin/task-board/ask-batches/{first_batch['batch_id']}/confirm-refund",
|
||
files={"refund_file": ("refund-1.png", b"refund-1", "image/png")},
|
||
).status_code == 200
|
||
|
||
board = task_board_items(client, task_type="ask", page_size=100)
|
||
parent = next(item for item in board if item.get("record_kind") == "ask_task" and item["id"] == generated["id"])
|
||
assert parent["status_label"] == "进行中"
|
||
assert parent["completed_count"] == 1
|
||
|
||
next_assignment = client.get(f"{task_url}&refresh=true").json()["data"]
|
||
assert next_assignment["status"] == "待选择"
|
||
with main_module.SessionLocal() as db:
|
||
stored = db.get(main_module.AskAssignment, next_assignment["assignment_id"])
|
||
stored.reservation_started_at = datetime.utcnow() - timedelta(minutes=2)
|
||
db.commit()
|
||
expired = client.get(task_url)
|
||
assert expired.status_code == 409
|
||
|
||
with main_module.SessionLocal() as db:
|
||
stored = db.get(main_module.AskAssignment, next_assignment["assignment_id"])
|
||
assert len(stored.batches) == 1
|
||
assert stored.batches[0].status == "paid"
|
||
|
||
final_assignment = client.get(f"{task_url}&refresh=true").json()["data"]
|
||
remaining_question_id = next(
|
||
item["id"] for item in final_assignment["questions"] if item["id"] != first_question_id
|
||
)
|
||
assert client.post(
|
||
f"/api/ask/tasks/{generated['task_id']}/selection",
|
||
json={"token": session["token"], "question_ids": [remaining_question_id]},
|
||
).status_code == 200
|
||
second_batch = client.post(
|
||
f"/api/ask/tasks/{generated['task_id']}/submission",
|
||
data={"token": session["token"], "payee_name": "测试用户"},
|
||
files=[
|
||
("files", ("question-2.png", b"question-2", "image/png")),
|
||
("payment_file", ("payment-2.png", b"payment-2", "image/png")),
|
||
],
|
||
).json()["data"]
|
||
assert client.post(
|
||
f"/api/admin/ask-batch-reviews/{second_batch['batch_id']}/decision",
|
||
json={"approved": True},
|
||
).status_code == 200
|
||
assert client.post(
|
||
f"/api/admin/task-board/ask-batches/{second_batch['batch_id']}/confirm-refund",
|
||
files={"refund_file": ("refund-2.png", b"refund-2", "image/png")},
|
||
).status_code == 200
|
||
|
||
completed_board = task_board_items(client, task_type="ask", page_size=100)
|
||
completed_parent = next(
|
||
item for item in completed_board
|
||
if item.get("record_kind") == "ask_task" and item["id"] == generated["id"]
|
||
)
|
||
assert completed_parent["status_label"] == "已完成"
|
||
assert completed_parent["completed_count"] == 2
|
||
assert len(completed_parent["claimed_tasks"]) == 2
|
||
parent_detail = client.get(completed_parent["detail_endpoint"])
|
||
assert parent_detail.status_code == 200
|
||
assert {
|
||
question["user_id"]
|
||
for question in parent_detail.json()["data"]["selected_questions"]
|
||
} == {session["user_id"]}
|
||
|
||
completed_link = client.get(task_url)
|
||
assert completed_link.status_code == 409
|
||
assert completed_link.json()["detail"] == "本次提问任务已全部完成，可查看历史状态"
|
||
history = client.get(
|
||
f"/api/ask/me/assignments?token={session['token']}"
|
||
)
|
||
assert history.status_code == 200
|
||
completed_batches = [
|
||
item for item in history.json()["data"]
|
||
if item["task_id"] == generated["task_id"] and item["status"] == "已完成"
|
||
]
|
||
assert len(completed_batches) == 2
|
||
|
||
other_session = client.post(
|
||
"/api/ask/auth/taobao-login",
|
||
json={"taobao_id": f"tb{uuid4().hex[:14]}"},
|
||
).json()["data"]
|
||
unrelated_user = client.get(
|
||
f"/api/ask/tasks/{generated['task_id']}?token={other_session['token']}"
|
||
)
|
||
assert unrelated_user.status_code == 404
|