"""K1 端到端：Kimi Code OAuth device-flow（device/token/refresh **全 mock**）→ 真 pg。

证明 K1 闭环（DoD：device-flow 连接 → 加密 token 持久化(DB 真源,无明文) → 网关建出
KimiCodeAdapter(coding base + 伪造头 + bearer) → 建网关时按需刷新更新存量 token → 断开清除）。
**绝不联网**：device authorization / token / refresh 全经注入的 fake httpx 客户端，
auth.kimi.com / api.kimi.com 一次都不碰。

  用例 1 · 连接 device flow → token 加密持久化（DB 真源）
    `POST .../oauth/start` → 202 `{job_id, user_code, verification_uri, ...}`。fake
    `device_authorization` 返 device_code；fake `token` 端点先 `authorization_pending`
    两次、再成功 `{access_token, refresh_token, expires_in}`。后台 `run_job` 轮询循环跑在
    **真 e2e_sm session** 上（ASGITransport await background task；override
    `get_session_factory`→e2e_sm、monkeypatch 模块级 `_default_http_client`→fake、
    `asyncio.sleep`→no-op）。`GET /jobs/{id}` → done，result `{connected:true,
    provider:"kimi-code"}` 且**不含任何 token 串**。DB 真源：`provider_credentials` 行
    `auth_type="oauth"`、`oauth_enc` 非空、`api_key_enc` 为 None；解密 `oauth_enc` 回环
    access/refresh token（证明加密持久化、非明文）。`GET .../oauth/status` →
    `{connected:true, expires_at}` 无 token。

  用例 2 · 网关建出 Kimi Code 适配器（正确 token + 伪造头 + coding base）
    存 oauth 凭据 + `tier_routing` 行把某档位指向 `kimi-code:kimi-for-coding` →
    `build_gateway_for_tier(...,"writer")` → 断言适配器是 `KimiCodeAdapter`，其
    `AsyncOpenAI` 客户端 `base_url=https://api.kimi.com/coding/v1`、带伪造 `User-Agent`
    + 全 `X-Msh-*` 默认头、`api_key`(=bearer) = 存量 access token。

  用例 3 · 建网关时刷新（refresh-on-build）
    存一个 `expires_at` 临近过期的 oauth 包；fake token 端点对 `grant_type=refresh_token`
    返**新** token 组。`_build_provider_adapter(store,"kimi-code")`（经真 e2e_sm session）
    → 断言 refresh 被调、适配器用**新** access token、`oauth_enc`(DB 真源)已更新为新包。

  用例 4 · 断开（disconnect）
    `POST .../oauth/disconnect` → 凭据行被清除（DB 真源：之后无 oauth 凭据）。

零成本 & 确定性：fake httpx（device/token/refresh）+ 真 `SqlCredentialStore`/真 pg。无 pg → skip。

坑（见 memory/gotchas）：
- 后台 work 自建 http 走**模块级** `routers.kimi_oauth._default_http_client()`（非 dep）——
  须 monkeypatch 它指到同一 scripted fake（端点 device-auth call[0] + work poll call[1..]
  顺序共享）；端点 device-auth 走 `http` dep，再 override `_default_http_client` dep。
- 真 pg 持久化：**不** monkeypatch `SqlCredentialStore`（让真 store 落库），override
  `get_session_factory`→`e2e_sm`（同测试 engine/loop）使后台 work 用真 session 写库。
- monkeypatch `routers.kimi_oauth.asyncio.sleep`→no-op（不真等 interval 秒）。
- `_resolve_kimi_code_token` 刷新走 `project_deps.httpx.AsyncClient` + `project_deps.kimi_refresh`——
  用字符串 target monkeypatch（mypy strict 不静态校验字符串属性）。
- token 绝不进 job 结果/状态响应/start 响应——显式负向断言。
- get_sessionmaker engine 绑首个 loop → 每 DB 测清缓存重建、结束 dispose（同 M3/M4/M5）。
"""

from __future__ import annotations

import os
from collections.abc import AsyncIterator
from datetime import UTC, datetime, timedelta
from typing import Any

import httpx
import pytest
from asgi_lifespan import LifespanManager
from cryptography.fernet import Fernet
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from ww_db import get_sessionmaker
from ww_db.models import Job, ProviderCredential, TierRouting
from ww_llm_gateway.adapters.kimi_code import (
    KIMI_CODE_BASE_URL,
    KIMI_CODE_PROVIDER,
    KIMI_CODE_USER_AGENT,
    KimiCodeAdapter,
)

_ACCESS_TOKEN = "kimi-acc-e2e-not-real-12345"
_REFRESH_TOKEN = "kimi-ref-e2e-not-real-67890"
_NEW_ACCESS_TOKEN = "kimi-acc-e2e-refreshed-abcde"
_NEW_REFRESH_TOKEN = "kimi-ref-e2e-refreshed-fghij"

_OAUTH_BASE = "/settings/providers/kimi-code/oauth"
_START_URL = f"{_OAUTH_BASE}/start"
_DISCONNECT_URL = f"{_OAUTH_BASE}/disconnect"
_STATUS_URL = f"{_OAUTH_BASE}/status"

_REQUIRED_HEADER_KEYS = (
    "User-Agent",
    "X-Msh-Platform",
    "X-Msh-Version",
    "X-Msh-Device-Name",
    "X-Msh-Device-Model",
    "X-Msh-Os-Version",
    "X-Msh-Device-Id",
)


class _FakeResponse:
    """`HttpResponse` Protocol 最小实现（status_code + json()）。"""

    def __init__(self, status_code: int, body: dict[str, Any]) -> None:
        self.status_code = status_code
        self._body = body

    def json(self) -> Any:
        return self._body


class _ScriptedHttp:
    """按调用序号返回预设响应（device auth → poll attempts）；记录每次调用。aclose 无操作。

    端点的 device authorization 调用是 call[0]，后台 work 的 token 轮询是 call[1..]。
    同一实例供两者顺序共享（须 monkeypatch 模块级 `_default_http_client` 指到它）。
    """

    def __init__(self, responses: list[_FakeResponse]) -> None:
        self._responses = responses
        self.calls: list[tuple[str, dict[str, str]]] = []

    async def post(self, url: str, *, data: dict[str, str]) -> _FakeResponse:
        idx = len(self.calls)
        self.calls.append((url, data))
        return self._responses[idx]

    async def aclose(self) -> None:
        return None


def _device_auth_response() -> _FakeResponse:
    return _FakeResponse(
        200,
        {
            "device_code": "dev-e2e-1",
            "user_code": "ABCD-1234",
            "verification_uri": "https://auth.kimi.com/device",
            "verification_uri_complete": "https://auth.kimi.com/device?code=ABCD-1234",
            "expires_in": 600,
            "interval": 5,
        },
    )


def _pending_response() -> _FakeResponse:
    return _FakeResponse(400, {"error": "authorization_pending"})


def _token_success_response() -> _FakeResponse:
    return _FakeResponse(
        200,
        {
            "access_token": _ACCESS_TOKEN,
            "refresh_token": _REFRESH_TOKEN,
            "expires_in": 900,
        },
    )


def _enc_key() -> str:
    """设置 `CREDENTIAL_ENC_KEY` 为有效 Fernet 密钥 + 清 settings 缓存，返回该密钥。"""
    key = Fernet.generate_key().decode()
    os.environ["CREDENTIAL_ENC_KEY"] = key
    from ww_config import get_settings

    get_settings.cache_clear()
    return key


@pytest.fixture
async def e2e_sm() -> AsyncIterator[async_sessionmaker[AsyncSession]]:
    """真实 DB session 工厂；无 pg 时跳过（每测试清缓存重建 engine、结束 dispose）。"""
    get_sessionmaker.cache_clear()
    maker = get_sessionmaker()
    try:
        async with maker() as probe:
            await probe.execute(select(1))
    except Exception:
        pytest.skip("postgres not reachable")
    yield maker
    await maker.kw["bind"].dispose()
    get_sessionmaker.cache_clear()


async def _cleanup_credentials(e2e_sm: async_sessionmaker[AsyncSession]) -> None:
    """清理本测试族写入的 kimi-code 凭据 + kimi_oauth job 行（单用户 stub owner）。"""
    from ww_api.services.credentials import STUB_OWNER_ID

    async with e2e_sm() as cleanup:
        await cleanup.execute(
            delete(ProviderCredential).where(
                ProviderCredential.owner_id == STUB_OWNER_ID,
                ProviderCredential.provider == KIMI_CODE_PROVIDER,
            )
        )
        await cleanup.execute(delete(Job).where(Job.kind == "kimi_oauth"))
        await cleanup.execute(delete(TierRouting).where(TierRouting.project_id.is_(None)))
        await cleanup.commit()


async def _seed_oauth_credential(e2e_sm: async_sessionmaker[AsyncSession], blob: bytes) -> None:
    """直接经真 store 在真 pg 写一行 oauth 凭据（用例 2/3 前置）。"""
    from ww_api.services.credentials import STUB_OWNER_ID, SqlCredentialStore

    async with e2e_sm() as session:
        store = SqlCredentialStore(session)
        await store.upsert_oauth_credential(STUB_OWNER_ID, KIMI_CODE_PROVIDER, blob)
        # store 只 flush；测试前置须显式提交，否则独立 session（建网关/校验）看不到（新契约）。
        await session.commit()


async def test_k1_connect_device_flow_persists_encrypted_token(
    e2e_sm: async_sessionmaker[AsyncSession],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """用例 1：device flow 连接 → 后台轮询(pending×2→成功) → 加密 token 真落 pg → status 已连接。

    全程零联网（fake httpx），token 绝不进 job 结果/状态响应/start 响应（负向断言）。
    """
    import ww_api.routers.kimi_oauth as kimi_mod
    from ww_api.main import create_app
    from ww_api.services.credentials import (
        AUTH_TYPE_OAUTH,
        STUB_OWNER_ID,
        SqlCredentialStore,
    )
    from ww_api.services.kimi_oauth import decrypt_oauth_bundle
    from ww_api.services.project_deps import get_session_factory

    enc_key = _enc_key()
    # device auth(call0) → poll pending(call1) → poll pending(call2) → poll success(call3)。
    http = _ScriptedHttp(
        [
            _device_auth_response(),
            _pending_response(),
            _pending_response(),
            _token_success_response(),
        ]
    )

    # 后台 work 不真等 interval 秒。
    async def _no_sleep(_seconds: float) -> None:
        return None

    monkeypatch.setattr("ww_api.routers.kimi_oauth.asyncio.sleep", _no_sleep)
    # 后台 work 自建 http 走模块级 `_default_http_client()`（非 dep）→ 指到同一 scripted fake。
    monkeypatch.setattr(kimi_mod, "_default_http_client", lambda: http)

    app = create_app()
    # 端点 device-auth 走 http dep → 也指到同一 fake（保证顺序共享）。
    app.dependency_overrides[kimi_mod._default_http_client] = lambda: http
    # 后台 work 用 run_job 自建独立 session → 真 e2e_sm（同测试 engine/loop），真 store 落库。
    app.dependency_overrides[get_session_factory] = lambda: e2e_sm

    transport = httpx.ASGITransport(app=app)
    try:
        await _cleanup_credentials(e2e_sm)
        async with LifespanManager(app):
            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
                # 1) 发起 device flow → 202 + user_code/verification_uri（ASGITransport 等后台）。
                start_resp = await client.post(_START_URL)
                assert start_resp.status_code == 202
                start_body = start_resp.json()
                assert start_body["user_code"] == "ABCD-1234"
                assert start_body["verification_uri"] == "https://auth.kimi.com/device"
                assert start_body["interval"] == 5
                job_id = start_body["job_id"]
                # start 响应绝不含 token。
                assert _ACCESS_TOKEN not in start_resp.text
                assert _REFRESH_TOKEN not in start_resp.text

                # 2) 轮询 job → done，result 只 {connected, provider}，无 token。
                job_resp = await client.get(f"/jobs/{job_id}")
                assert job_resp.status_code == 200
                job = job_resp.json()
                assert job["status"] == "done"
                assert job["kind"] == "kimi_oauth"
                assert job["result"] == {"connected": True, "provider": KIMI_CODE_PROVIDER}
                # job 整体（含 result）绝不泄露 token。
                assert _ACCESS_TOKEN not in job_resp.text
                assert _REFRESH_TOKEN not in job_resp.text

                # 3) status → 已连接 + expires_at，无 token 本体。
                status_resp = await client.get(_STATUS_URL)
                assert status_resp.status_code == 200
                status_body = status_resp.json()
                assert status_body["connected"] is True
                assert status_body["expires_at"] is not None
                assert "access_token" not in status_body
                assert "refresh_token" not in status_body
                assert _ACCESS_TOKEN not in status_resp.text
                assert _REFRESH_TOKEN not in status_resp.text

        # 后台经历两次 pending 才成功（证明 pending 路径真走）：device + 3 token = 4 调用。
        assert len(http.calls) == 4

        # DB 真源断言：oauth 凭据行加密持久化（非明文），解密回环 token。
        async with e2e_sm() as verify:
            row = (
                await verify.execute(
                    select(ProviderCredential).where(
                        ProviderCredential.owner_id == STUB_OWNER_ID,
                        ProviderCredential.provider == KIMI_CODE_PROVIDER,
                    )
                )
            ).scalar_one()
            assert row.auth_type == AUTH_TYPE_OAUTH
            assert row.oauth_enc is not None
            assert row.api_key_enc is None
            # 密文不是明文（token 不出现在原始字节里）。
            assert _ACCESS_TOKEN.encode() not in row.oauth_enc
            assert _REFRESH_TOKEN.encode() not in row.oauth_enc
            # 解密回环：证明真的是这套 token 被加密持久化。
            token = decrypt_oauth_bundle(row.oauth_enc, key=enc_key)
            assert token.access_token == _ACCESS_TOKEN
            assert token.refresh_token == _REFRESH_TOKEN

            # 经真 store 读 status 也是已连接（store 层一致）。
            store = SqlCredentialStore(verify)
            cred = await store.get_credential(STUB_OWNER_ID, KIMI_CODE_PROVIDER)
            assert cred is not None
            assert cred.auth_type == AUTH_TYPE_OAUTH
    finally:
        await _cleanup_credentials(e2e_sm)


async def test_k1_gateway_builds_kimi_code_adapter_with_headers_and_bearer(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """用例 2：存 oauth + 档位路由指向 kimi-code → 建网关 → 适配器带 coding base/伪造头/bearer。"""
    from ww_api.services.credentials import SqlCredentialStore, StoredRouting
    from ww_api.services.kimi_oauth import TokenSet, encrypt_oauth_bundle
    from ww_api.services.project_deps import build_gateway_for_tier

    enc_key = _enc_key()
    fresh = TokenSet(
        access_token=_ACCESS_TOKEN,
        refresh_token=_REFRESH_TOKEN,
        expires_at=datetime.now(UTC) + timedelta(seconds=900),  # 远未过期 → 不刷新
    )
    blob = encrypt_oauth_bundle(fresh, key=enc_key)

    try:
        await _cleanup_credentials(e2e_sm)
        await _seed_oauth_credential(e2e_sm, blob)
        # 档位路由：writer → kimi-code:kimi-for-coding（DB 真源）。
        async with e2e_sm() as seed:
            store = SqlCredentialStore(seed)
            await store.upsert_routing(
                StoredRouting(
                    tier="writer",
                    provider=KIMI_CODE_PROVIDER,
                    model="kimi-for-coding",
                    fallback=[],
                )
            )
            # store 只 flush；显式提交使独立的建网关 session 可见该路由（新契约）。
            await seed.commit()

        async with e2e_sm() as session:
            store = SqlCredentialStore(session)
            gateway = await build_gateway_for_tier(session, store, "writer")

        adapter = gateway._adapters[KIMI_CODE_PROVIDER]
        assert isinstance(adapter, KimiCodeAdapter)
        client = adapter._client
        # coding base。
        assert str(client.base_url).rstrip("/") == KIMI_CODE_BASE_URL.rstrip("/")
        # 伪造头：UA + 全 X-Msh-*。
        for key in _REQUIRED_HEADER_KEYS:
            assert key in client.default_headers
        assert client.default_headers["User-Agent"] == KIMI_CODE_USER_AGENT
        # bearer = 存量 access token（SDK 据 api_key 自动发 Authorization: Bearer）。
        assert client.api_key == _ACCESS_TOKEN
        # 未刷新 → DB oauth 凭据未变（仍是原 token）。
        async with e2e_sm() as verify:
            from ww_api.services.kimi_oauth import decrypt_oauth_bundle

            row = (
                await verify.execute(
                    select(ProviderCredential).where(
                        ProviderCredential.provider == KIMI_CODE_PROVIDER
                    )
                )
            ).scalar_one()
            assert row.oauth_enc is not None
            assert decrypt_oauth_bundle(row.oauth_enc, key=enc_key).access_token == _ACCESS_TOKEN
    finally:
        await _cleanup_credentials(e2e_sm)


async def test_k1_refresh_on_build_updates_stored_token(
    e2e_sm: async_sessionmaker[AsyncSession],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """用例 3：存量 token 临近过期 → 建适配器触发刷新 → 用新 token + DB oauth_enc 更新为新包。

    刷新经 fake httpx + fake `kimi_refresh`（不联网）；新包真落 pg（DB 真源）。
    """
    from ww_api.services import project_deps
    from ww_api.services.credentials import STUB_OWNER_ID, SqlCredentialStore
    from ww_api.services.kimi_oauth import (
        TokenSet,
        decrypt_oauth_bundle,
        encrypt_oauth_bundle,
    )
    from ww_api.services.project_deps import _build_provider_adapter

    enc_key = _enc_key()
    near = TokenSet(
        access_token=_ACCESS_TOKEN,
        refresh_token=_REFRESH_TOKEN,
        expires_at=datetime.now(UTC) + timedelta(seconds=30),  # 临近过期 → 触发刷新
    )
    blob = encrypt_oauth_bundle(near, key=enc_key)

    refreshed = TokenSet(
        access_token=_NEW_ACCESS_TOKEN,
        refresh_token=_NEW_REFRESH_TOKEN,
        expires_at=datetime.now(UTC) + timedelta(seconds=900),
    )
    refresh_calls: list[str] = []

    async def _fake_refresh(_http: Any, refresh_token: str) -> TokenSet:
        # 证明用存量 refresh token 刷新。
        refresh_calls.append(refresh_token)
        return refreshed

    class _FakeAsyncClient:
        async def __aenter__(self) -> _FakeAsyncClient:
            return self

        async def __aexit__(self, *exc: object) -> None:
            return None

    monkeypatch.setattr(project_deps, "kimi_refresh", _fake_refresh)
    monkeypatch.setattr(
        "ww_api.services.project_deps.httpx.AsyncClient", lambda *a, **k: _FakeAsyncClient()
    )

    try:
        await _cleanup_credentials(e2e_sm)
        await _seed_oauth_credential(e2e_sm, blob)

        async with e2e_sm() as session:
            store = SqlCredentialStore(session)
            adapter = await _build_provider_adapter(store, KIMI_CODE_PROVIDER)

        # refresh 被调，且用存量 refresh token。
        assert refresh_calls == [_REFRESH_TOKEN]
        # 适配器用新 access token。
        assert isinstance(adapter, KimiCodeAdapter)
        assert adapter._client.api_key == _NEW_ACCESS_TOKEN

        # DB 真源：oauth_enc 已更新为新包（下次建网关复用刷新结果）。
        async with e2e_sm() as verify:
            row = (
                await verify.execute(
                    select(ProviderCredential).where(
                        ProviderCredential.owner_id == STUB_OWNER_ID,
                        ProviderCredential.provider == KIMI_CODE_PROVIDER,
                    )
                )
            ).scalar_one()
            assert row.oauth_enc is not None
            stored = decrypt_oauth_bundle(row.oauth_enc, key=enc_key)
            assert stored.access_token == _NEW_ACCESS_TOKEN
            assert stored.refresh_token == _NEW_REFRESH_TOKEN
    finally:
        await _cleanup_credentials(e2e_sm)


async def test_k1_disconnect_clears_credential(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """用例 4：disconnect → 凭据行被清除（DB 真源：之后无 oauth 凭据）。"""
    from ww_api.main import create_app
    from ww_api.services.credentials import STUB_OWNER_ID
    from ww_api.services.kimi_oauth import TokenSet, encrypt_oauth_bundle

    enc_key = _enc_key()
    token = TokenSet(
        access_token=_ACCESS_TOKEN,
        refresh_token=_REFRESH_TOKEN,
        expires_at=datetime.now(UTC) + timedelta(seconds=900),
    )
    blob = encrypt_oauth_bundle(token, key=enc_key)

    app = create_app()
    transport = httpx.ASGITransport(app=app)
    try:
        await _cleanup_credentials(e2e_sm)
        await _seed_oauth_credential(e2e_sm, blob)
        # 前置：凭据确实在库。
        async with e2e_sm() as pre:
            assert (
                await pre.execute(
                    select(ProviderCredential).where(
                        ProviderCredential.owner_id == STUB_OWNER_ID,
                        ProviderCredential.provider == KIMI_CODE_PROVIDER,
                    )
                )
            ).scalar_one_or_none() is not None

        async with LifespanManager(app):
            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
                resp = await client.post(_DISCONNECT_URL)
                assert resp.status_code == 200
                assert resp.json()["disconnected"] is True

        # DB 真源：凭据行已清除。
        async with e2e_sm() as verify:
            assert (
                await verify.execute(
                    select(ProviderCredential).where(
                        ProviderCredential.owner_id == STUB_OWNER_ID,
                        ProviderCredential.provider == KIMI_CODE_PROVIDER,
                    )
                )
            ).scalar_one_or_none() is None
    finally:
        await _cleanup_credentials(e2e_sm)