TelegramBotAdmin/tests/test_technician_mini_app.py

from __future__ import annotations

import hashlib
import hmac
import json
from datetime import UTC, datetime, timedelta
from urllib.parse import urlencode

from aiohttp.test_utils import TestClient, TestServer


def _signed_init_data(
    token: str,
    user_id: int,
    *,
    auth_date: datetime | None = None,
) -> str:
    values = {
        "auth_date": str(int((auth_date or datetime.now(UTC)).timestamp())),
        "query_id": "AAE-test-query",
        "user": json.dumps(
            {
                "id": user_id,
                "first_name": "认证",
                "last_name": "技师",
                "username": f"technician{user_id}",
                "language_code": "zh-hans",
            },
            ensure_ascii=False,
            separators=(",", ":"),
        ),
    }
    data_check_string = "\n".join(
        f"{key}={value}" for key, value in sorted(values.items())
    )
    secret_key = hmac.new(
        b"WebAppData",
        token.encode(),
        hashlib.sha256,
    ).digest()
    values["hash"] = hmac.new(
        secret_key,
        data_check_string.encode(),
        hashlib.sha256,
    ).hexdigest()
    return urlencode(values)


async def _approved_technician(app_modules, user_id: int) -> None:
    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.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",
    )


async def test_builtin_templates_are_seeded_without_overwriting_admin_changes(
    app_modules,
):
    service_db = app_modules.load("wbb.utils.dbservice")
    first = await service_db.ensure_builtin_package_templates()
    assert [item["template_id"] for item in first] == [
        "builtin_quick_at_store",
        "builtin_quick_onsite",
        "builtin_quick_hourly",
    ]
    await app_modules.wbb.control_db.service_package_templates.update_one(
        {"template_id": "builtin_quick_at_store"},
        {"$set": {"name": "管理员修改后的到店模板", "status": "disabled"}},
    )
    await service_db.ensure_builtin_package_templates()
    stored = await app_modules.wbb.control_db.service_package_templates.find_one(
        {"template_id": "builtin_quick_at_store"}
    )
    assert stored["name"] == "管理员修改后的到店模板"
    assert stored["status"] == "disabled"


async def test_technician_mini_app_uses_signed_identity_without_admin_login(
    app_modules,
):
    user_id = 700
    await _approved_technician(app_modules, user_id)
    bot_config = app_modules.load("wbb.admin.bot_config")
    token = "123456:abcdefghijklmnopqrstuvwxyzABCDE"
    bot = bot_config.create_bot_profile(
        app_modules.wbb.BOT_PROFILES_PATH,
        {
            "label": "技师服务 Bot",
            "bot_token": token,
            "enabled": True,
        },
    )
    admin_api = app_modules.load("wbb.admin.api")
    application = admin_api.build_admin_application()
    await application["admin_api"].initialize()
    client = TestClient(TestServer(application))
    await client.start_server()
    headers = {
        "X-Telegram-Bot-Id": bot["bot_id"],
        "X-Telegram-Init-Data": _signed_init_data(token, user_id),
    }
    try:
        invalid = await client.get(
            "/api/technician/v1/bootstrap",
            headers={**headers, "X-Telegram-Init-Data": f"{headers['X-Telegram-Init-Data']}x"},
        )
        assert invalid.status == 401
        assert (await invalid.json())["error"]["code"] == "mini_app_auth_failed"

        expired = await client.get(
            "/api/technician/v1/bootstrap",
            headers={
                **headers,
                "X-Telegram-Init-Data": _signed_init_data(
                    token,
                    user_id,
                    auth_date=datetime.now(UTC) - timedelta(hours=2),
                ),
            },
        )
        assert expired.status == 401

        loaded = await client.get("/api/technician/v1/bootstrap", headers=headers)
        assert loaded.status == 200
        payload = (await loaded.json())["data"]
        assert payload["technician"]["user_id"] == str(user_id)
        assert payload["technician"]["eligibility"]["ready"] is True
        assert [item["name"] for item in payload["templates"]] == [
            "快速到店服务",
            "快速上门服务",
            "快速按小时服务",
        ]
        assert all("admin_note" not in item for item in payload["templates"])

        package = payload["templates"][0]["package"]
        package.update(
            {
                "package_id": "self_service_package",
                "source_template_id": payload["templates"][0]["template_id"],
                "source_template_version": payload["templates"][0]["version"],
            }
        )
        published = await client.put(
            "/api/technician/v1/profile",
            headers=headers,
            json={
                "user_id": "999999",
                "service_profile": {
                    "headline": "一分钟发布套餐",
                    "bio": package["description"],
                    "tags": package["tags"],
                    "contact_hours": "请通过 Telegram 私聊确认",
                    "public_area_text": "测试城区及附近",
                    "venue": {
                        "name": "测试工作地点",
                        "address_hint": "测试路 100 号",
                    },
                    "onsite_policy": {"description": ""},
                    "packages": [package],
                    "accepting_requests": True,
                },
            },
        )
        assert published.status == 200
        published_data = (await published.json())["data"]
        assert published_data["user_id"] == str(user_id)
        assert published_data["service_profile"]["packages"][0]["price_unit"] == "per_service"
        assert (
            await app_modules.wbb.control_db.directory_profiles.find_one(
                {"user_id": 999999}
            )
        ) is None

        paused = await client.patch(
            "/api/technician/v1/availability",
            headers=headers,
            json={"accepting_requests": False},
        )
        assert paused.status == 200
        assert (
            (await paused.json())["data"]["service_profile"]["accepting_requests"]
            is False
        )
    finally:
        await client.close()