"""b5-1 task-detail persistence and API contract tests."""

from __future__ import annotations

from types import SimpleNamespace

import pytest
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient
from pydantic import ValidationError
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from starlette.requests import Request

from app.gateway import auth_middleware, csrf_middleware
from app.gateway.routers import task_deeplink_details as router
from deerflow.persistence.task_deeplink_details import (
    MemoryTaskDeeplinkDetailStore,
    TaskDeeplinkDetailRepository,
)
from deerflow.persistence.task_deeplink_details.model import TaskDeeplinkDetailRow


def _request(store, *, role: str = "admin", external: bool = False, anonymous: bool = False):
    state = SimpleNamespace(
        user=None
        if anonymous
        else SimpleNamespace(id="admin-1", system_role=role)
    )
    if external:
        state.task_deeplink_external_writer = True
    return SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(task_deeplink_detail_store=store)), state=state)


def _http_request(method: str, path: str, headers: dict[str, str] | None = None) -> Request:
    return Request(
        {
            "type": "http",
            "http_version": "1.1",
            "method": method,
            "scheme": "http",
            "path": path,
            "query_string": b"",
            "headers": [
                (key.lower().encode("latin-1"), value.encode("latin-1"))
                for key, value in (headers or {}).items()
            ],
            "client": ("test", 1234),
            "server": ("test", 80),
        }
    )


@pytest.mark.asyncio
async def test_memory_store_skips_identical_snapshot_writes_and_versions_changes() -> None:
    store = MemoryTaskDeeplinkDetailStore()
    first, created, changed = await store.upsert_detail(
        "b5-1",
        "task-1",
        task_name="任务甲",
        task_content="第一版详情",
        overview="概述",
        extra={"source": "external"},
        updated_by="tester",
    )
    assert created is True
    assert changed is True
    assert first["version"] == 1

    second, created, changed = await store.upsert_detail(
        "b5-1",
        "task-1",
        task_name="任务甲",
        task_content="第二版详情",
        overview="",
        extra={"priority": 3},
        updated_by="tester-2",
    )
    assert created is False
    assert changed is True
    assert second["version"] == 2
    assert second["extra"] == {"priority": 3}

    same, created, changed = await store.upsert_detail(
        "b5-1",
        "task-1",
        task_name="任务甲",
        task_content="第二版详情",
        overview="",
        extra={"priority": 3},
        updated_by="retry-writer",
    )
    assert created is False
    assert changed is False
    assert same["version"] == 2
    assert same["updated_at"] == second["updated_at"]
    assert await store.get_detail("b5-1", "task-2") is None


@pytest.mark.asyncio
async def test_memory_store_partitions_same_id_by_sub_type() -> None:
    store = MemoryTaskDeeplinkDetailStore()
    _, created_a, _ = await store.upsert_detail(
        "b5-1",
        "task-9",
        sub_type="yuqing",
        task_name="舆情任务",
        task_content="舆情详情",
        overview="",
        extra={},
        updated_by="writer",
    )
    _, created_b, _ = await store.upsert_detail(
        "b5-1",
        "task-9",
        sub_type="other",
        task_name="其它任务",
        task_content="其它详情",
        overview="",
        extra={},
        updated_by="writer",
    )
    assert created_a is True and created_b is True

    yuqing = await store.get_detail("b5-1", "task-9", sub_type="yuqing")
    other = await store.get_detail("b5-1", "task-9", sub_type="other")
    untyped = await store.get_detail("b5-1", "task-9")
    assert yuqing is not None and yuqing["task_name"] == "舆情任务" and yuqing["sub_type"] == "yuqing"
    assert other is not None and other["task_name"] == "其它任务"
    assert untyped is None

    # 同类型重试幂等；跨类型互不覆盖。
    _, created, changed = await store.upsert_detail(
        "b5-1",
        "task-9",
        sub_type="yuqing",
        task_name="舆情任务",
        task_content="舆情详情",
        overview="",
        extra={},
        updated_by="retry",
    )
    assert (created, changed) == (False, False)
    assert (await store.get_detail("b5-1", "task-9", sub_type="other"))["version"] == 1


@pytest.mark.asyncio
async def test_sql_store_round_trips_detail_snapshot() -> None:
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    try:
        async with engine.begin() as connection:
            await connection.run_sync(TaskDeeplinkDetailRow.__table__.create)
        store = TaskDeeplinkDetailRepository(async_sessionmaker(engine, expire_on_commit=False))
        saved, created, changed = await store.upsert_detail(
            "b5-1",
            "task-42",
            task_name="SQL 任务",
            task_content="详情",
            overview="概述",
            extra={"tags": ["a", "b"]},
            updated_by="admin",
        )
        assert created is True
        assert changed is True
        assert saved["detail_id"] == "task-42"
        assert (await store.get_detail("b5-1", "task-42"))["extra"] == {"tags": ["a", "b"]}

        same, created, changed = await store.upsert_detail(
            "b5-1",
            "task-42",
            task_name="SQL 任务",
            task_content="详情",
            overview="概述",
            extra={"tags": ["a", "b"]},
            updated_by="retry-writer",
        )
        assert created is False
        assert changed is False
        assert same["version"] == 1
    finally:
        await engine.dispose()


@pytest.mark.asyncio
async def test_router_returns_camel_case_contract_and_allows_open_writes() -> None:
    store = MemoryTaskDeeplinkDetailStore()
    request = _request(store)
    body = router.TaskDeeplinkDetailUpsertRequest.model_validate(
        {
            "taskName": "任务分析名称",
            "taskContent": "这是任务详情",
            "overview": "任务概述",
            "extra": {"origin": "b5", "severity": 2},
        }
    )
    saved = await router.upsert_task_deeplink_detail(request, body, "b5-1", "id-7")
    assert saved.created is True
    assert saved.changed is True
    assert saved.data.task_name == "任务分析名称"
    assert saved.data.extra == {"origin": "b5", "severity": 2}

    fetched = await router.get_task_deeplink_detail(request, "b5-1", "id-7", type="")
    assert fetched.data.detail_id == "id-7"
    assert fetched.data.model_dump(by_alias=True)["taskName"] == "任务分析名称"
    assert fetched.data.model_dump(by_alias=True)["type"] == "b5-1"
    assert fetched.data.model_dump(by_alias=True)["taskType"] == ""

    retry = await router.upsert_task_deeplink_detail(request, body, "b5-1", "id-7")
    assert retry.created is False
    assert retry.changed is False
    assert retry.data.version == 1

    # 写入端点对所有人开放：普通用户与未登录调用方都可直接写。
    by_user = await router.upsert_task_deeplink_detail(
        _request(store, role="user"), body, "b5-1", "id-8"
    )
    assert by_user.created is True

    by_anonymous = await router.upsert_task_deeplink_detail(
        _request(store, anonymous=True), body, "b5-1", "id-anonymous"
    )
    assert by_anonymous.created is True
    stored = await store.get_detail("b5-1", "id-anonymous")
    assert stored is not None and stored["updated_by"] == "external:task-deeplink"


@pytest.mark.asyncio
async def test_router_partitions_same_id_by_request_type() -> None:
    store = MemoryTaskDeeplinkDetailStore()
    request = _request(store)

    typed_body = router.TaskDeeplinkDetailUpsertRequest.model_validate(
        {"taskName": "舆情任务", "taskContent": "舆情详情", "type": "yuqing"}
    )
    saved = await router.upsert_task_deeplink_detail(request, typed_body, "b5-1", "task-x")
    assert saved.created is True
    assert saved.data.detail_sub_type == "yuqing"
    assert saved.data.model_dump(by_alias=True)["taskType"] == "yuqing"

    other_body = router.TaskDeeplinkDetailUpsertRequest.model_validate(
        {"taskName": "行动任务", "taskContent": "行动详情", "type": "action"}
    )
    await router.upsert_task_deeplink_detail(request, other_body, "b5-1", "task-x")

    fetched = await router.get_task_deeplink_detail(request, "b5-1", "task-x", type="yuqing")
    assert fetched.data.task_name == "舆情任务"
    fetched_other = await router.get_task_deeplink_detail(request, "b5-1", "task-x", type="action")
    assert fetched_other.data.task_name == "行动任务"

    # 该 id 未写无类型分区：不带 ?type= 读不到（不会回落到任意类型）。
    with pytest.raises(HTTPException) as error:
        await router.get_task_deeplink_detail(request, "b5-1", "task-x", type="")
    assert error.value.status_code == 404


def test_router_rejects_blank_required_fields_and_non_object_extra() -> None:
    with pytest.raises(ValidationError):
        router.TaskDeeplinkDetailUpsertRequest.model_validate(
            {"taskName": "   ", "taskContent": "详情"}
        )
    with pytest.raises(ValidationError):
        router.TaskDeeplinkDetailUpsertRequest.model_validate(
            {"taskName": "名称", "taskContent": "详情", "extra": ["not", "an", "object"]}
        )


def test_b5_put_is_credential_free_and_bypasses_csrf() -> None:
    # 无 key、无登录态：PUT 直接放行并打上外部写入标记（供审计）。
    bare = _http_request("PUT", "/api/task-deeplink-details/b5-1/id-9")
    assert auth_middleware._is_task_deeplink_external_write(bare, bare.url.path)
    assert auth_middleware._authorize_task_deeplink_external_write(bare) is None
    assert bare.state.task_deeplink_external_writer is True
    assert csrf_middleware.should_check_csrf(bare) is False

    # 旧集成方带的 X-Integration-Key 不再校验（只记指纹），错误 key 也放行。
    stale_key = _http_request(
        "PUT",
        "/api/task-deeplink-details/b5-1/id-9",
        {"X-Integration-Key": "anything"},
    )
    assert auth_middleware._authorize_task_deeplink_external_write(stale_key) is None
    assert stale_key.state.task_deeplink_external_key_fingerprint

    # 通道只覆盖 PUT；其它方法仍走正常登录校验。
    post_request = _http_request("POST", "/api/task-deeplink-details/b5-1/id-9")
    assert not auth_middleware._is_task_deeplink_external_write(post_request, post_request.url.path)
    assert csrf_middleware.should_check_csrf(post_request) is True


def test_unauthenticated_put_succeeds_but_read_still_requires_login() -> None:
    app = FastAPI()
    app.state.task_deeplink_detail_store = MemoryTaskDeeplinkDetailStore()
    app.include_router(router.router)
    app.add_middleware(auth_middleware.AuthMiddleware)

    client = TestClient(app)
    written = client.put(
        "/api/task-deeplink-details/b5-1/task-open",
        json={"taskName": "开放写入任务", "taskContent": "这是任务详情"},
    )
    assert written.status_code == 200
    assert written.json()["created"] is True

    # 同一 id 再写一个带类型的快照，互不覆盖（端到端）。
    typed = client.put(
        "/api/task-deeplink-details/b5-1/task-open",
        json={"taskName": "同id不同类型", "taskContent": "类型详情", "type": "yuqing"},
    )
    assert typed.status_code == 200
    assert typed.json()["data"]["taskType"] == "yuqing"
    assert typed.json()["created"] is True

    # 读取仍走正常登录校验：未登录 GET（无论带不带 type）→ 401，
    # 不因写入开放而连带放开。
    denied = client.get("/api/task-deeplink-details/b5-1/task-open")
    assert denied.status_code == 401
    denied_typed = client.get("/api/task-deeplink-details/b5-1/task-open?type=yuqing")
    assert denied_typed.status_code == 401