from __future__ import annotations import asyncio from datetime import timedelta from types import SimpleNamespace import pytest from aiohttp import CookieJar from aiohttp.test_utils import TestClient, TestServer def _user(user_id: int, *, username: str | None = None, name: str = "测试用户"): return SimpleNamespace( id=user_id, username=username, first_name=name, last_name=None, is_bot=False, ) def _package(*, name: str = "基础护理", category: str = "护理"): return { "name": name, "category": category, "description": "标准线下服务套餐", "tags": ["线下", "护理"], "service_modes": ["at_store", "onsite"], "price_mode": "fixed", "currency": "CNY", "price_unit": "per_service", "min_price": "100", "max_price": "100", "duration_minutes": 60, "included_items": "基础服务", "excluded_items": "额外耗材", "addons": [ { "name": "延长服务", "unit": "次", "price": "20", "extra_minutes": 15, } ], "travel_fee": {"mode": "quoted"}, "service_radius_km": 20, "out_of_range_policy": "超区请重新询价", "preparation": "请提前沟通需求", } async def _technician(app_modules, service_db, *, user_id: int = 200, category: str = "护理"): directory_db = app_modules.load("wbb.utils.dbdirectory") await directory_db.upsert_directory_identity( user_id=user_id, username=f"technician{user_id}", first_name="认证", last_name="技师", ) await directory_db.save_directory_location( user_id=user_id, longitude=116.4074, latitude=39.9042, source="test", actor_id=user_id, actor_name="认证技师", ) await directory_db.upsert_directory_membership( bot_id="primary", chat_id=-100200, user_id=user_id, status="member", active=True, verified=True, ) await directory_db.submit_teacher_application( user_id=user_id, source="test", bot_id="primary", chat_id=-100200, ) await directory_db.decide_teacher_application( user_id=user_id, action="approve", actor_id="admin", actor_name="admin", source="test", ) await directory_db.set_teacher_state( user_id=user_id, action="list", actor_id=user_id, actor_name="认证技师", source="test", ) profile = await service_db.publish_technician_profile( user_id, { "headline": "专注线下护理", "bio": "提供到店和上门服务", "tags": ["护理"], "contact_hours": "10:00-20:00", "public_area_text": "北京朝阳区附近", "venue": {"name": "护理工作室", "address_hint": "测试路 100 号"}, "onsite_policy": {"description": "20 公里内可上门"}, "packages": [_package(category=category)], }, actor_id=user_id, ) return profile async def _accepted_customer(service_db, *, user_id: int = 300): customer = _user(user_id, username=None, name=f"顾客{user_id}") await service_db.observe_customer(customer, accepted_terms=True) return customer async def _order(service_db, customer, profile, *, service_mode: str = "onsite"): package = profile["service_profile"]["packages"][0] kwargs = { "customer": customer, "technician_id": profile["user_id"], "package_id": package["package_id"], "service_mode": service_mode, "requirements": "希望完成一次标准护理服务", } if service_mode == "onsite": kwargs.update( { "longitude": 116.4174, "latitude": 39.9042, "address_text": "测试街道 1 号 101 室", } ) else: kwargs.update({"longitude": 116.4174, "latitude": 39.9042}) return await service_db.create_service_order(**kwargs) async def test_public_customer_order_quote_address_and_completion_qr(app_modules): service_db = app_modules.load("wbb.utils.dbservice") profile = await _technician(app_modules, service_db) customer = await _accepted_customer(service_db) order = await _order(service_db, customer, profile) assert order["status"] == "requested" assert order["distance_band"] == "1 公里内" stored_address = await app_modules.wbb.control_db.service_order_addresses.find_one( {"order_id": order["order_id"]} ) assert stored_address["redact_after"] is None assert "测试街道" not in stored_address["encrypted_payload"] hidden = await service_db.get_order_for_actor( order["order_id"], customer.id, reveal_address=True ) assert "exact_address" not in hidden quoted_order, quote = await service_db.submit_service_quote( order["order_id"], profile["user_id"], { "base_amount": "100", "travel_fee": "10", "addons": [{"name": "延长服务", "quantity": 2, "unit_price": "20"}], "discount": "5", "currency": "CNY", "total_amount": "145", "note": "最终报价", }, ) assert quoted_order["status"] == "quoted" assert quote["total_amount"] == "145.00" confirmed = await service_db.confirm_service_quote(order["order_id"], customer.id) assert confirmed["status"] == "confirmed" revealed = await service_db.get_order_for_actor( order["order_id"], profile["user_id"], reveal_address=True ) assert revealed["exact_address"]["address_text"] == "测试街道 1 号 101 室" started = await service_db.start_service_order(order["order_id"], profile["user_id"]) assert started["status"] == "in_progress" qr, token = await service_db.issue_review_qr(order["order_id"], profile["user_id"]) assert qr["status"] == "issued" assert len(token) >= 22 assert token not in str(qr) with pytest.raises(service_db.ServiceDataError, match="不属于当前顾客"): await service_db.preview_review_qr(token, customer.id + 1) claim_results = await asyncio.gather( service_db.claim_review_qr(token, customer.id), service_db.claim_review_qr(token, customer.id), return_exceptions=True, ) successful_claims = [item for item in claim_results if isinstance(item, tuple)] failed_claims = [item for item in claim_results if isinstance(item, Exception)] assert len(successful_claims) == 1 assert len(failed_claims) == 1 claimed, completed = successful_claims[0] assert claimed["status"] == "claimed" assert completed["status"] == "completed" assert "已使用" in str(failed_claims[0]) resumable = await service_db.preview_review_qr(token, customer.id) assert resumable["status"] == "claimed" verified_review = await service_db.submit_review( customer=customer, technician_id=profile["user_id"], source="qr_verified", answers={ "overall_experience": "excellent", "service_highlights": ["professional", "communication"], "comment": "已完成服务单评价", }, anonymous=False, qr_id=claimed["qr_id"], ) assert verified_review["source"] == "qr_verified" assert verified_review["order_id"] == order["order_id"] with pytest.raises(service_db.ServiceDataError, match="无效"): await service_db.preview_review_qr(token, customer.id) with pytest.raises(service_db.ServiceDataError, match="已经提交评价"): await service_db.submit_review( customer=customer, technician_id=profile["user_id"], source="qr_verified", answers={ "overall_experience": "excellent", "service_highlights": ["professional"], }, anonymous=True, qr_id=claimed["qr_id"], ) address_after = await app_modules.wbb.control_db.service_order_addresses.find_one( {"order_id": order["order_id"]} ) assert address_after["redact_after"] - completed["closed_at"] == timedelta(days=7) await app_modules.wbb.control_db.service_order_addresses.update_one( {"order_id": order["order_id"]}, {"$set": {"redact_after": service_db.utc_now() - timedelta(seconds=1)}}, ) assert await service_db.redact_expired_addresses() == 1 redacted_order = await app_modules.wbb.control_db.service_orders.find_one( {"order_id": order["order_id"]} ) redacted_address = await app_modules.wbb.control_db.service_order_addresses.find_one( {"order_id": order["order_id"]} ) assert redacted_order["address_redacted"] is True assert "distance_meters" not in redacted_order assert "encrypted_payload" not in redacted_address async def test_state_machine_cancellation_and_template_snapshot(app_modules): service_db = app_modules.load("wbb.utils.dbservice") profile = await _technician(app_modules, service_db) customer = await _accepted_customer(service_db) template = await service_db.create_package_template( {"template_name": "护理模板", "package": _package(name="模板初版")}, actor_id="admin", ) selected = await service_db.instantiate_package_template(template["template_id"]) assert selected["source_template_version"] == 1 await service_db.update_package_template( template["template_id"], {"template_name": "护理模板", "package": _package(name="模板新版")}, actor_id="admin", ) assert selected["name"] == "模板初版" order = await _order(service_db, customer, profile, service_mode="at_store") assert order["distance_meters"] > 0 assert order["distance_band"] == "1 公里内" with pytest.raises(service_db.ServiceDataError, match="只有已确认"): await service_db.start_service_order(order["order_id"], profile["user_id"]) canceled = await service_db.cancel_service_order( order["order_id"], customer.id, "顾客计划调整" ) assert canceled["status"] == "canceled_customer" with pytest.raises(service_db.ServiceDataError, match="不能取消"): await service_db.cancel_service_order( order["order_id"], profile["user_id"], "重复取消" ) pending_order = await _order(service_db, customer, profile) await service_db.submit_service_quote( pending_order["order_id"], profile["user_id"], {"base_amount": "100", "travel_fee": "0", "discount": "0"}, ) await service_db.confirm_service_quote(pending_order["order_id"], customer.id) await service_db.start_service_order(pending_order["order_id"], profile["user_id"]) _, pending_token = await service_db.issue_review_qr( pending_order["order_id"], profile["user_id"] ) canceled_pending = await service_db.cancel_service_order( pending_order["order_id"], profile["user_id"], "服务现场无法继续" ) assert canceled_pending["status"] == "canceled_technician" with pytest.raises(service_db.ServiceDataError, match="无效"): await service_db.preview_review_qr(pending_token, customer.id) disputed_order = await _order(service_db, customer, profile) disputed = await service_db.dispute_service_order( disputed_order["order_id"], customer.id, "需要管理员协调" ) assert disputed["status"] == "disputed" resumed = await service_db.resolve_service_dispute( disputed_order["order_id"], actor_id="admin", action="resume", reason="双方已达成一致", ) assert resumed["status"] == "requested" async def test_order_requires_encryption_before_persisting(app_modules): service_db = app_modules.load("wbb.utils.dbservice") profile = await _technician(app_modules, service_db) customer = await _accepted_customer(service_db) original_key = app_modules.wbb.SERVICE_ADDRESS_ENCRYPTION_KEY app_modules.wbb.SERVICE_ADDRESS_ENCRYPTION_KEY = "" try: with pytest.raises(service_db.ServiceDataError, match="加密密钥"): await _order(service_db, customer, profile) finally: app_modules.wbb.SERVICE_ADDRESS_ENCRYPTION_KEY = original_key assert ( await app_modules.wbb.control_db.service_orders.count_documents( {"customer_id": customer.id} ) == 0 ) async def test_expired_review_qr_reopens_order_for_reissue(app_modules): service_db = app_modules.load("wbb.utils.dbservice") profile = await _technician(app_modules, service_db) customer = await _accepted_customer(service_db) order = await _order(service_db, customer, profile) await service_db.submit_service_quote( order["order_id"], profile["user_id"], {"base_amount": "100", "travel_fee": "0", "discount": "0"}, ) await service_db.confirm_service_quote(order["order_id"], customer.id) await service_db.start_service_order(order["order_id"], profile["user_id"]) first_qr, first_token = await service_db.issue_review_qr( order["order_id"], profile["user_id"] ) await app_modules.wbb.control_db.teacher_review_qr_records.update_one( {"qr_id": first_qr["qr_id"]}, {"$set": {"expires_at": service_db.utc_now() - timedelta(seconds=1)}}, ) with pytest.raises(service_db.ServiceDataError, match="已过期"): await service_db.preview_review_qr(first_token, customer.id) reopened = await app_modules.wbb.control_db.service_orders.find_one( {"order_id": order["order_id"]} ) assert reopened["status"] == "in_progress" assert "current_review_qr_id" not in reopened replacement, replacement_token = await service_db.issue_review_qr( order["order_id"], profile["user_id"] ) assert replacement["qr_id"] != first_qr["qr_id"] assert replacement_token != first_token async def test_reviews_same_weight_anonymous_and_bayesian_leaderboard(app_modules): service_db = app_modules.load("wbb.utils.dbservice") profile = await _technician(app_modules, service_db) customer = await _accepted_customer(service_db) answers = { "overall_experience": "excellent", "service_highlights": ["professional", "communication"], "comment": "服务很专业", } template = await service_db.get_active_review_template() draft = await service_db.begin_review_draft( customer_id=customer.id, technician_id=profile["user_id"], source="student_initiated", template=template, ) await service_db.save_review_draft_progress( customer.id, answers={"overall_experience": "excellent"}, question_index=1, ) resumed_draft = await service_db.begin_review_draft( customer_id=customer.id, technician_id=profile["user_id"], source="student_initiated", template=template, ) assert resumed_draft["draft_id"] == draft["draft_id"] assert resumed_draft["answers"] == {"overall_experience": "excellent"} await service_db.delete_review_draft(customer.id) review_ids = [] for _ in range(1): review = await service_db.submit_review( customer=customer, technician_id=profile["user_id"], source="student_initiated", answers=answers, anonymous=True, ) review_ids.append(review["review_id"]) assert "qr_id" not in review assert review["package_snapshot"]["name"] == "技师服务" assert review["category"] == "护理" assert len(set(review_ids)) == 1 for review_id in review_ids: await service_db.moderate_review( review_id, "approve", actor_id="admin" ) public, total = await service_db.list_public_reviews(profile["user_id"]) assert total == 1 assert all(item["customer_name"] == "匿名顾客" for item in public) assert all(item["source_label"] == "用户主动评价" for item in public) summary = await service_db.get_technician_review_summary(profile["user_id"]) assert summary["review_count"] == 1 pending, _ = await service_db.list_reviews( status="approved", source="student_initiated" ) assert all(item["similar_content_count"] == 1 for item in pending) leaderboard, total = await service_db.list_leaderboard(category="护理") assert total == 1 assert leaderboard[0]["review_count"] == 1 assert leaderboard[0]["eligible"] is True assert leaderboard[0]["rank_score"] == pytest.approx( leaderboard[0]["average_score"] ) async def test_direct_review_skips_package_selection(app_modules, monkeypatch): service_db = app_modules.load("wbb.utils.dbservice") service_module = app_modules.load("wbb.modules.technician_services") profile = await _technician(app_modules, service_db) customer = await _accepted_customer(service_db) captured = {} async def fake_begin_review(target, **values): captured.update(values) captured["customer_id"] = target.from_user.id monkeypatch.setattr(service_module, "_begin_review", fake_begin_review) await service_module._choose_review_package( SimpleNamespace(from_user=customer), profile["user_id"], ) assert captured == { "customer_id": customer.id, "technician_id": profile["user_id"], "source": "student_initiated", } async def test_review_moderation_only_allows_defined_state_transitions(app_modules): service_db = app_modules.load("wbb.utils.dbservice") profile = await _technician(app_modules, service_db) customer = await _accepted_customer(service_db) package_id = profile["service_profile"]["packages"][0]["package_id"] review = await service_db.submit_review( customer=customer, technician_id=profile["user_id"], source="student_initiated", answers={ "overall_experience": "excellent", "service_highlights": ["professional"], }, anonymous=True, package_id=package_id, ) approved = await service_db.moderate_review( review["review_id"], "approve", actor_id="admin" ) assert approved["status"] == "approved" with pytest.raises(service_db.ServiceDataError, match="不能执行"): await service_db.moderate_review( review["review_id"], "reject", actor_id="admin", reason="重复处理" ) voided = await service_db.moderate_review( review["review_id"], "void", actor_id="admin", reason="争议作废" ) assert voided["status"] == "voided" with pytest.raises(service_db.ServiceDataError, match="不能执行"): await service_db.moderate_review( review["review_id"], "approve", actor_id="admin" ) async def test_first_approved_review_creates_topic_and_is_published_once( app_modules, monkeypatch, ): service_db = app_modules.load("wbb.utils.dbservice") review_service = app_modules.load("wbb.services.technician_reviews") profile = await _technician(app_modules, service_db) customer = await _accepted_customer(service_db) package_id = profile["service_profile"]["packages"][0]["package_id"] await service_db.set_service_settings( { "review_forum_chat_id": "-1001234567890", "review_forum_username": "public_reviews", } ) calls = [] async def fake_bot_api(_token, method, payload): calls.append((method, payload)) if method == "createForumTopic": return {"message_thread_id": 321} return {"message_id": 900 + len(calls)} monkeypatch.setattr(review_service, "_bot_api", fake_bot_api) review = await service_db.submit_review( customer=customer, technician_id=profile["user_id"], source="student_initiated", answers={ "overall_experience": "excellent", "service_highlights": ["professional"], "comment": "服务很专业", }, anonymous=True, package_id=package_id, ) approved = await service_db.moderate_review( review["review_id"], "approve", actor_id="admin", ) published = await review_service.publish_approved_review( "123456:test-token", approved, ) assert published["status"] == "published" assert published["topic_url"] == "https://t.me/public_reviews/321" assert [method for method, _ in calls] == [ "createForumTopic", "sendMessage", "sendMessage", ] repeated = await review_service.publish_approved_review( "123456:test-token", approved, ) assert repeated["status"] == "published" assert len(calls) == 3 async def test_technician_requires_active_managed_group_membership(app_modules): service_db = app_modules.load("wbb.utils.dbservice") profile = await _technician(app_modules, service_db) await app_modules.wbb.control_db.directory_memberships.update_many( {"user_id": profile["user_id"]}, {"$set": {"active": False}} ) customer = await _accepted_customer(service_db) with pytest.raises(service_db.ServiceDataError, match="当前不满足接单资格"): await _order(service_db, customer, profile) async def test_technician_uses_public_area_without_precise_location(app_modules): service_db = app_modules.load("wbb.utils.dbservice") profile = await _technician(app_modules, service_db) await app_modules.wbb.control_db.directory_locations.delete_one( {"user_id": profile["user_id"]} ) republished = await service_db.publish_technician_profile( profile["user_id"], profile["service_profile"], actor_id=profile["user_id"], ) assert republished["service_profile"]["public_area_text"] == "北京朝阳区附近" technicians, total = await service_db.list_available_technicians() assert total == 1 assert technicians[0]["public_area_text"] == "北京朝阳区附近" assert technicians[0]["distance_band"] == "" async def test_service_admin_api_settings_templates_and_permissions(app_modules): admin_api = app_modules.load("wbb.admin.api") app_modules.wbb.BOT_PERMISSIONS = {"teacher_directory.manage"} application = admin_api.build_admin_application() await application["admin_api"].initialize() client = TestClient(TestServer(application), cookie_jar=CookieJar(unsafe=True)) await client.start_server() try: login = await client.post( "/api/admin/v1/auth/login", json={"username": "admin", "password": "qwe0.123456"}, ) login_data = (await login.json())["data"] changed = await client.put( "/api/admin/v1/auth/password", headers={"X-CSRF-Token": login_data["csrf_token"]}, json={ "current_password": "qwe0.123456", "new_password": "changed-pass-123", }, ) headers = {"X-CSRF-Token": (await changed.json())["data"]["csrf_token"]} unconfirmed = await client.put( "/api/admin/v1/directory/service-settings", headers=headers, json={"customer_max_open_orders": 4}, ) assert unconfirmed.status == 409 saved = await client.put( "/api/admin/v1/directory/service-settings", headers=headers, json={ "customer_max_open_orders": 4, "customer_daily_request_limit": 12, "review_qr_expiry_hours": 48, "confirm": True, }, ) assert saved.status == 200 assert (await saved.json())["data"]["review_qr_expiry_hours"] == 48 created = await client.post( "/api/admin/v1/directory/package-templates", headers=headers, json={ "template_name": "后台护理模板", "package": _package(), "confirm": True, }, ) assert created.status == 201 created_data = (await created.json())["data"] assert created_data["version"] == 1 duplicated = await client.post( f"/api/admin/v1/directory/package-templates/{created_data['template_id']}/actions", headers=headers, json={"action": "duplicate", "confirm": True}, ) assert duplicated.status == 200 assert (await duplicated.json())["data"]["status"] == "disabled" review_template = await client.get( "/api/admin/v1/directory/review-template" ) assert review_template.status == 200 template_data = (await review_template.json())["data"] assert {item["type"] for item in template_data["questions"]} == { "single_choice", "multiple_choice", "text", } assert sum( item.get("weight", 0) for item in template_data["questions"] if item["type"] == "single_choice" ) == 100 metrics = await client.get( "/api/admin/v1/directory/fulfillment-metrics" ) assert metrics.status == 200 assert (await metrics.json())["data"]["scope_note"].startswith( "仅统计平台服务单" ) app_modules.wbb.BOT_PERMISSIONS = set() denied = await client.get( "/api/admin/v1/directory/fulfillment-metrics" ) assert denied.status == 403 finally: await client.close()