eryao/backend/tests/integration/test_locale_timezone_bootstrap.py

from __future__ import annotations

import time

import httpx
import pytest

from core.db.session import AsyncSessionLocal
from models.profile import Profile
from sqlalchemy import select


async def _create_email_session(
    client: httpx.AsyncClient,
    *,
    email: str,
    code: str,
    language: str | None = None,
    timezone: str | None = None,
) -> dict:
    payload: dict = {"email": email, "token": code}
    if language is not None:
        payload["language"] = language
    if timezone is not None:
        payload["timezone"] = timezone
    resp = await client.post(
        "/api/v1/auth/email-session",
        json=payload,
    )
    resp.raise_for_status()
    return resp.json()


async def _get_profile(
    client: httpx.AsyncClient,
    *,
    access_token: str,
) -> dict:
    headers = {"Authorization": f"Bearer {access_token}"}
    resp = await client.get("/api/v1/users/me/profile", headers=headers)
    resp.raise_for_status()
    return resp.json()


@pytest.mark.integration
async def test_new_user_registration_with_language_timezone(
    api_client: httpx.AsyncClient,
    test_identity: dict[str, str],
    db_cleanup: list[str],
) -> None:
    """场景1: 新用户首次打开App，注册时语言和时区写入后端"""
    email = test_identity["email"]
    code = test_identity["code"]
    db_cleanup.append(email)

    session_data = await _create_email_session(
        api_client,
        email=email,
        code=code,
        language="en-US",
        timezone="America/New_York",
    )
    access_token = str(session_data["access_token"])

    profile = await _get_profile(api_client, access_token=access_token)
    preferences = profile["settings"]["preferences"]

    assert preferences["language"] == "en-US", f"Expected en-US, got {preferences['language']}"
    assert preferences["timezone"] == "America/New_York", f"Expected America/New_York, got {preferences['timezone']}"


@pytest.mark.integration
async def test_new_user_registration_with_zh_hant(
    api_client: httpx.AsyncClient,
    test_identity: dict[str, str],
    db_cleanup: list[str],
) -> None:
    """场景1变体: 新用户注册时使用繁体中文"""
    email = test_identity["email"]
    code = test_identity["code"]
    db_cleanup.append(email)

    session_data = await _create_email_session(
        api_client,
        email=email,
        code=code,
        language="zh-Hant",
        timezone="Asia/Taipei",
    )
    access_token = str(session_data["access_token"])

    profile = await _get_profile(api_client, access_token=access_token)
    preferences = profile["settings"]["preferences"]

    assert preferences["language"] == "zh-Hant"
    assert preferences["timezone"] == "Asia/Taipei"


@pytest.mark.integration
async def test_new_user_registration_without_language_timezone(
    api_client: httpx.AsyncClient,
    test_identity: dict[str, str],
    db_cleanup: list[str],
) -> None:
    """场景2: 新用户注册时不传语言时区，使用后端默认值"""
    email = test_identity["email"]
    code = test_identity["code"]
    db_cleanup.append(email)

    session_data = await _create_email_session(
        api_client,
        email=email,
        code=code,
    )
    access_token = str(session_data["access_token"])

    profile = await _get_profile(api_client, access_token=access_token)
    preferences = profile["settings"]["preferences"]

    assert preferences["language"] == "zh-CN", "Default language should be zh-CN"
    assert preferences["timezone"] == "Asia/Shanghai", "Default timezone should be Asia/Shanghai"


@pytest.mark.integration
async def test_existing_user_login_sync_from_server(
    api_client: httpx.AsyncClient,
    test_identity: dict[str, str],
    db_cleanup: list[str],
) -> None:
    """场景2: 已有用户在新设备登录，从服务器同步语言时区"""
    email = test_identity["email"]
    code = test_identity["code"]
    db_cleanup.append(email)

    session1 = await _create_email_session(
        api_client,
        email=email,
        code=code,
        language="en-US",
        timezone="Europe/London",
    )
    access_token1 = str(session1["access_token"])

    profile1 = await _get_profile(api_client, access_token=access_token1)
    preferences1 = profile1["settings"]["preferences"]
    assert preferences1["language"] == "en-US"
    assert preferences1["timezone"] == "Europe/London"

    time.sleep(1)

    session2 = await _create_email_session(
        api_client,
        email=email,
        code=code,
    )
    access_token2 = str(session2["access_token"])

    profile2 = await _get_profile(api_client, access_token=access_token2)
    preferences2 = profile2["settings"]["preferences"]

    assert preferences2["language"] == "en-US", "Should sync language from server"
    assert preferences2["timezone"] == "Europe/London", "Should sync timezone from server"


@pytest.mark.integration
async def test_partial_language_only(
    api_client: httpx.AsyncClient,
    test_identity: dict[str, str],
    db_cleanup: list[str],
) -> None:
    """边界情况: 只传语言不传时区"""
    email = test_identity["email"]
    code = test_identity["code"]
    db_cleanup.append(email)

    session_data = await _create_email_session(
        api_client,
        email=email,
        code=code,
        language="zh-Hant",
    )
    access_token = str(session_data["access_token"])

    profile = await _get_profile(api_client, access_token=access_token)
    preferences = profile["settings"]["preferences"]

    assert preferences["language"] == "zh-Hant"
    assert preferences["timezone"] == "Asia/Shanghai", "Timezone should use default"


@pytest.mark.integration
async def test_partial_timezone_only(
    api_client: httpx.AsyncClient,
    test_identity: dict[str, str],
    db_cleanup: list[str],
) -> None:
    """边界情况: 只传时区不传语言"""
    email = test_identity["email"]
    code = test_identity["code"]
    db_cleanup.append(email)

    session_data = await _create_email_session(
        api_client,
        email=email,
        code=code,
        timezone="Australia/Sydney",
    )
    access_token = str(session_data["access_token"])

    profile = await _get_profile(api_client, access_token=access_token)
    preferences = profile["settings"]["preferences"]

    assert preferences["language"] == "zh-CN", "Language should use default"
    assert preferences["timezone"] == "Australia/Sydney"


@pytest.mark.integration
async def test_profile_settings_persisted_in_database(
    api_client: httpx.AsyncClient,
    test_identity: dict[str, str],
    db_cleanup: list[str],
) -> None:
    """验证语言时区正确持久化到数据库"""
    email = test_identity["email"]
    code = test_identity["code"]
    db_cleanup.append(email)

    session_data = await _create_email_session(
        api_client,
        email=email,
        code=code,
        language="en-US",
        timezone="America/Los_Angeles",
    )
    user_id = str(session_data["user"]["id"])

    async with AsyncSessionLocal() as db_session:
        stmt = select(Profile).where(Profile.id == user_id)
        result = await db_session.execute(stmt)
        profile = result.scalar_one_or_none()

        assert profile is not None, "Profile should exist in database"
        settings = profile.settings
        assert settings is not None
        preferences = settings.get("preferences", {})
        assert preferences.get("language") == "en-US"
        assert preferences.get("timezone") == "America/Los_Angeles"