"""Gateway contract for the dedicated task deep-link history endpoint."""

from __future__ import annotations

from types import SimpleNamespace

import pytest

from app.gateway.routers.task_deeplink_threads import list_task_deeplink_threads
from deerflow.persistence.task_deeplink_threads.base import (
    MAX_TASK_DEEPLINK_SUB_TYPE,
    task_deeplink_binding,
)


class _HistoryStore:
    def __init__(self) -> None:
        self.calls: list[tuple[str, str, int, str | None]] = []

    async def list_threads(
        self,
        task_id: str,
        go_path: str,
        *,
        limit: int,
        sub_type: str | None = None,
    ):
        self.calls.append((task_id, go_path, limit, sub_type))
        return [
            {
                "thread_id": "thread-1",
                "status": "idle",
                "created_at": "2026-10-19T08:00:00+00:00",
                "updated_at": "2026-10-19T08:01:00+00:00",
                "metadata": {
                    "taskId": "108",
                    "taskRecordId": "108",
                    "taskGoPath": "rwfx",
                },
                "values": {"title": "任务分析"},
                "interrupts": {},
            }
        ]


def _request_with(store: _HistoryStore) -> SimpleNamespace:
    return SimpleNamespace(
        app=SimpleNamespace(
            state=SimpleNamespace(task_deeplink_thread_store=store),
        )
    )


@pytest.mark.asyncio
async def test_list_endpoint_trims_task_id_and_preserves_thread_shape():
    store = _HistoryStore()

    response = await list_task_deeplink_threads(
        _request_with(store),
        task_id=" 108 ",
        go_path="rwfx",
        limit=25,
        sub_type=None,
    )

    assert store.calls == [("108", "rwfx", 25, None)]
    assert response.model_dump() == {
        "threads": [
            {
                "thread_id": "thread-1",
                "status": "idle",
                "created_at": "2026-10-19T08:00:00+00:00",
                "updated_at": "2026-10-19T08:01:00+00:00",
                "metadata": {
                    "taskId": "108",
                    "taskRecordId": "108",
                    "taskGoPath": "rwfx",
                },
                "values": {"title": "任务分析"},
                "interrupts": {},
            }
        ]
    }


@pytest.mark.asyncio
async def test_list_endpoint_passes_sub_type_through():
    store = _HistoryStore()

    await list_task_deeplink_threads(
        _request_with(store),
        task_id="task-1",
        go_path="b5-1",
        limit=100,
        sub_type=" yuqing ",
    )
    # Explicitly empty sub_type must survive as "" (the no-type partition),
    # not collapse to None (which would mean "every sub-type").
    await list_task_deeplink_threads(
        _request_with(store),
        task_id="task-1",
        go_path="b5-1",
        limit=100,
        sub_type="",
    )

    assert store.calls == [
        ("task-1", "b5-1", 100, "yuqing"),
        ("task-1", "b5-1", 100, ""),
    ]


@pytest.mark.asyncio
async def test_list_endpoint_accepts_a_custom_mapping_go_path() -> None:
    store = _HistoryStore()

    await list_task_deeplink_threads(
        _request_with(store),
        task_id="task-1",
        go_path="区域研判",
        limit=100,
        sub_type=None,
    )

    assert store.calls == [("task-1", "区域研判", 100, None)]


def test_binding_extracts_b51_task_type_as_sub_type():
    binding = task_deeplink_binding(
        {
            "taskId": "task-1",
            "taskRecordId": "task-1",
            "taskGoPath": "b5-1",
            "taskType": "yuqing",
        }
    )

    assert binding is not None
    assert binding["sub_type"] == "yuqing"


def test_binding_defaults_missing_sub_type_to_empty_string():
    for go_path in ("3qfx", "rwfx", "xdfx", "b5-1"):
        binding = task_deeplink_binding(
            {
                "taskId": "task-1",
                "taskRecordId": "task-1",
                "taskGoPath": go_path,
            }
        )
        assert binding is not None
        assert binding["sub_type"] == ""


def test_binding_caps_sub_type_to_column_width():
    binding = task_deeplink_binding(
        {
            "taskId": "task-1",
            "taskRecordId": "task-1",
            "taskGoPath": "b5-1",
            "taskType": "x" * (MAX_TASK_DEEPLINK_SUB_TYPE + 10),
        }
    )

    assert binding is not None
    assert len(binding["sub_type"]) == MAX_TASK_DEEPLINK_SUB_TYPE


def test_binding_accepts_custom_mapping_go_path_but_not_its_a_suffix():
    binding = task_deeplink_binding(
        {
            "taskId": "task-1",
            "taskRecordId": "task-1",
            "taskGoPath": "区域研判",
        }
    )

    assert binding is not None
    assert binding["go_path"] == "区域研判"
    assert task_deeplink_binding(
        {
            "taskId": "task-1",
            "taskRecordId": "task-1",
            "taskGoPath": "区域研判-A",
        }
    ) is None