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_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"] == "" 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 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 missing_link = client.post("/api/admin/ask-tasks/generate", json={"product_name": "测试商品"}) assert missing_link.status_code == 400 missing_product_access = client.post( "/api/admin/ask-tasks/generate", json={"shop_name": "测试店铺", "product_name": "测试商品", "question_count": 3}, ) assert missing_product_access.status_code == 400 missing_shop = client.post("/api/admin/ask-tasks/generate", json={"product_link": "https://item.example.com/1", "product_name": "测试商品", "question_count": 3}) assert missing_shop.status_code == 400 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_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_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