from __future__ import annotations

import json

import httpx
import pytest
from fastapi import HTTPException
from starlette.datastructures import Headers, QueryParams
from starlette.requests import Request

from deerflow.config.agents_config import AgentConfig
from deerflow.config.llmwiki_config import (
    LlmWikiConfig,
    LlmWikiRuntimeOverride,
    resolve_llmwiki_runtime,
)
from deerflow.integrations.weknora.client import WeKnoraClient, WeKnoraError
from deerflow.persistence.llmwiki import MemoryLlmWikiStore
from deerflow.tools.builtins.llmwiki_search_tool import (
    _sanitize_conversation_deposit_value as sanitize_tool_deposit_value,
    selected_knowledge_base_ids,
)


def _request(
    path: str,
    *,
    query: str = "",
    cookies: dict[str, str] | None = None,
    request_headers: dict[str, str] | None = None,
) -> Request:
    headers: list[tuple[bytes, bytes]] = [
        (key.lower().encode(), value.encode()) for key, value in (request_headers or {}).items()
    ]
    if cookies:
        headers.append(("cookie".encode(), "; ".join(f"{key}={value}" for key, value in cookies.items()).encode()))
    return Request(
        {
            "type": "http",
            "method": "GET",
            "path": path,
            "query_string": query.encode(),
            "headers": headers,
            "scheme": "http",
            "server": ("testserver", 80),
            "client": ("testclient", 50000),
        }
    )


def _first_gateway_match(method: str, path: str) -> str:
    from app.gateway.app import app as gateway_app

    for route in gateway_app.router.routes:
        route_contexts = getattr(route, "effective_route_contexts", None)
        candidates = route_contexts() if callable(route_contexts) else [route]
        for candidate in candidates:
            methods = getattr(candidate, "methods", None)
            path_regex = getattr(candidate, "path_regex", None)
            if methods is not None and method in methods and path_regex is not None and path_regex.match(path):
                return str(getattr(candidate, "path", ""))
    raise AssertionError(f"No route matched {method} {path}")


def test_empty_weknora_address_keeps_legacy_mode() -> None:
    runtime = resolve_llmwiki_runtime(LlmWikiConfig())

    assert runtime.provider == "legacy"
    assert runtime.api_base_url == ""


def test_configured_weknora_address_enables_weknora_mode() -> None:
    config = LlmWikiConfig.model_validate(
        {
            "weknora": {
                "api_base_url": "http://weknora-app:8080/",
                "web_base_url": "https://weknora.example.test/",
                "admin_email": "admin@example.test",
                "admin_password": "secret",
            }
        }
    )

    runtime = resolve_llmwiki_runtime(config)

    assert runtime.provider == "weknora"
    assert runtime.api_base_url == "http://weknora-app:8080"
    assert runtime.web_base_url == "https://weknora.example.test"
    assert runtime.admin_email == "admin@example.test"
    assert runtime.admin_password == "secret"


def test_runtime_override_can_force_legacy_or_reset_to_file() -> None:
    config = LlmWikiConfig.model_validate({"weknora": {"api_base_url": "http://weknora:8080"}})

    forced_legacy = resolve_llmwiki_runtime(
        config,
        LlmWikiRuntimeOverride(enabled=True, api_base_url=""),
    )
    reset_to_file = resolve_llmwiki_runtime(
        config,
        LlmWikiRuntimeOverride(enabled=False, api_base_url=""),
    )

    assert forced_legacy.provider == "legacy"
    assert reset_to_file.provider == "weknora"


def test_weknora_url_rejects_credentials_and_non_http_schemes() -> None:
    with pytest.raises(ValueError):
        LlmWikiConfig.model_validate({"weknora": {"api_base_url": "ftp://weknora.local"}})
    with pytest.raises(ValueError):
        LlmWikiConfig.model_validate({"weknora": {"api_base_url": "http://user:pass@weknora.local"}})


def test_agent_config_tracks_allowed_llmwiki_knowledge_bases() -> None:
    config = AgentConfig(
        name="Research agent",
        llmwiki_knowledge_base_ids=["kb-one", "kb-two", "kb-one"],
    )

    assert config.llmwiki_knowledge_base_ids == ["kb-one", "kb-two"]


def test_conversation_selection_is_deduplicated_and_absence_is_distinct() -> None:
    assert selected_knowledge_base_ids({}) is None
    assert selected_knowledge_base_ids({"llmwiki_knowledge_base_ids": []}) == []
    assert selected_knowledge_base_ids({"llmwiki_knowledge_base_ids": ["kb-one", "kb-one", "", "kb-two"]}) == ["kb-one", "kb-two"]


def test_weknora_frame_bootstrap_accepts_query_token_only_for_frame_route() -> None:
    from app.gateway.deps import get_request_access_token

    frame_request = _request(
        "/api/llmwiki/knowledge-bases/local-kb/weknora/frame",
        query="access_token=deerflow-token",
    )
    normal_request = _request("/api/models", query="access_token=deerflow-token")

    assert get_request_access_token(frame_request) == "deerflow-token"
    assert get_request_access_token(normal_request) is None


def test_weknora_embed_cookie_bypasses_only_proxy_paths() -> None:
    from app.gateway import auth_middleware, csrf_middleware
    from app.gateway.weknora_embed import sign_weknora_embed_payload

    raw_cookie = sign_weknora_embed_payload(
        {
            "mapping_id": "local-kb",
            "weknora_id": "remote-kb",
            "user_id": "user-a",
            "is_admin": False,
            "can_write": False,
            "exp": 4_102_444_800,
        }
    )
    cookies = {"deerflow_weknora_embed": raw_cookie}

    assert auth_middleware._is_weknora_embed_proxy_request(_request("/platform/knowledge-bases/kb", cookies=cookies))
    assert auth_middleware._is_weknora_embed_proxy_request(_request("/assets/index.js", cookies=cookies))
    assert auth_middleware._is_weknora_embed_proxy_request(_request("/config.js", cookies=cookies))
    assert auth_middleware._is_weknora_embed_proxy_request(
        _request("/tdesign-icons/0.4.1/fonts/index.js", cookies=cookies)
    )
    assert auth_middleware._is_weknora_embed_proxy_request(_request("/api/v1/knowledge-search", cookies=cookies))
    assert auth_middleware._is_weknora_embed_proxy_request(
        _request("/api/llmwiki/weknora-embed/api/v1/auth/me", cookies=cookies)
    )
    assert csrf_middleware.is_weknora_embed_proxy_request(_request("/api/v1/knowledge-search", cookies=cookies))
    assert not auth_middleware._is_weknora_embed_proxy_request(
        _request("/api/v1/knowledge-search", cookies={"deerflow_weknora_embed": "signed"})
    )
    assert not csrf_middleware.is_weknora_embed_proxy_request(
        _request("/api/v1/knowledge-search", cookies={"deerflow_weknora_embed": "signed"})
    )
    assert not auth_middleware._is_weknora_embed_proxy_request(_request("/api/v1/auth/me", cookies=cookies))
    assert not auth_middleware._is_weknora_embed_proxy_request(_request("/api/models", cookies=cookies))
    assert not auth_middleware._is_weknora_embed_proxy_request(_request("/platform/knowledge-bases/kb"))


@pytest.mark.asyncio
async def test_weknora_session_endpoint_sets_cookie_and_returns_platform_path(monkeypatch: pytest.MonkeyPatch) -> None:
    from app.gateway.routers import llmwiki as llmwiki_router

    async def authorized_mapping(*_: object, **__: object) -> tuple[dict[str, str], str, bool]:
        return {"id": "local-kb", "weknora_id": "remote-kb", "name": "KB"}, "user-a", False

    class Store:
        async def get_authorized(self, *_: object, **__: object) -> dict[str, str]:
            return {"id": "local-kb", "weknora_id": "remote-kb"}

    monkeypatch.setattr(llmwiki_router, "_runtime_for_iframe_proxy", lambda _: object())
    monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized_mapping)
    monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store())

    response = await llmwiki_router.create_weknora_detail_session(
        _request(
            "/api/llmwiki/knowledge-bases/local-kb/weknora/session",
            request_headers={"X-Forwarded-Proto": "https"},
        ),
        "local-kb",
        tab="wiki",
    )

    payload = json.loads(response.body)
    assert payload["frame_path"] == "/platform/knowledge-bases/remote-kb?tab=wiki"
    cookie = response.headers["set-cookie"]
    assert "deerflow_weknora_embed=" in cookie
    assert "Secure" in cookie
    assert "SameSite=none" in cookie
    raw = cookie.split("deerflow_weknora_embed=", 1)[1].split(";", 1)[0]
    ctx = llmwiki_router._verify_embed_cookie(raw)
    assert ctx.mapping_id == "local-kb"
    assert ctx.weknora_id == "remote-kb"
    assert ctx.can_write is True


@pytest.mark.asyncio
async def test_weknora_client_sends_no_credentials_and_normalizes_search() -> None:
    requests: list[httpx.Request] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        assert "X-API-Key" not in request.headers
        assert "Authorization" not in request.headers
        assert json.loads(request.content) == {
            "query": "deployment guide",
            "knowledge_base_ids": ["kb-one", "kb-two"],
        }
        return httpx.Response(
            200,
            json={
                "success": True,
                "data": [
                    {
                        "id": "chunk-one",
                        "content": "matched passage",
                        "knowledge_id": "doc-one",
                        "knowledge_base_id": "kb-one",
                        "knowledge_title": "Guide",
                        "score": 0.93,
                    }
                ],
            },
        )

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(
            base_url="http://weknora.local:8080",
            http_client=http_client,
        )
        results = await client.search("deployment guide", ["kb-one", "kb-two"])

    assert requests[0].url.path == "/api/v1/knowledge-search"
    assert results[0]["chunk_id"] == "chunk-one"
    assert results[0]["title"] == "Guide"


@pytest.mark.asyncio
async def test_weknora_client_merges_required_name_for_description_update() -> None:
    requests: list[httpx.Request] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        if request.method == "GET":
            return httpx.Response(
                200,
                json={"success": True, "data": {"id": "kb-one", "name": "对话沉淀"}},
            )
        assert request.method == "PUT"
        assert json.loads(request.content) == {"name": "对话沉淀", "description": "cmzs description"}
        return httpx.Response(
            200,
            json={"success": True, "data": {"id": "kb-one", "name": "对话沉淀", "description": "cmzs description"}},
        )

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
        updated = await client.update_knowledge_base("kb-one", {"description": "cmzs description"})

    assert updated["description"] == "cmzs description"
    assert [request.method for request in requests] == ["GET", "PUT"]


@pytest.mark.asyncio
async def test_weknora_client_resolves_default_embedding_model_inside_weknora() -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        if request.method == "GET" and request.url.path == "/api/v1/models":
            return httpx.Response(
                200,
                json={
                    "success": True,
                    "data": [
                        {"id": "embed-default", "type": "Embedding", "is_default": True},
                    ],
                },
            )
        assert request.method == "POST"
        assert request.url.path == "/api/v1/knowledge-bases"
        assert json.loads(request.content)["embedding_model_id"] == "embed-default"
        return httpx.Response(201, json={"success": True, "data": {"id": "kb-created", "name": "Docs"}})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(
            base_url="http://weknora.local:8080",
            http_client=http_client,
        )
        created = await client.create_knowledge_base(name="Docs")

    assert created["id"] == "kb-created"


@pytest.mark.asyncio
async def test_weknora_client_creates_wiki_with_weknora_owned_models() -> None:
    model_reads = 0

    async def handler(request: httpx.Request) -> httpx.Response:
        nonlocal model_reads
        if request.method == "GET" and request.url.path == "/api/v1/models":
            model_reads += 1
            return httpx.Response(
                200,
                json={
                    "data": [
                        {"id": "embed-one", "type": "Embedding", "is_default": True},
                        {"id": "qa-one", "type": "KnowledgeQA", "status": "active"},
                    ]
                },
            )
        payload = json.loads(request.content)
        assert payload["embedding_model_id"] == "embed-one"
        assert payload["summary_model_id"] == "qa-one"
        assert payload["indexing_strategy"] == {
            "vector_enabled": True,
            "keyword_enabled": True,
            "wiki_enabled": True,
            "graph_enabled": False,
        }
        assert payload["wiki_config"]["synthesis_model_id"] == "qa-one"
        return httpx.Response(201, json={"data": {"id": "wiki-one", "name": "Wiki"}})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
        created = await client.create_knowledge_base(name="Wiki", wiki_enabled=True)

    assert created["id"] == "wiki-one"
    assert model_reads == 2


@pytest.mark.asyncio
async def test_weknora_client_imports_url_manual_content_and_lists_wiki_pages() -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path.endswith("/knowledge/url"):
            assert json.loads(request.content) == {"url": "https://example.test/guide"}
            return httpx.Response(201, json={"data": {"id": "url-one"}})
        if request.url.path.endswith("/knowledge/manual"):
            assert json.loads(request.content) == {
                "title": "Guide",
                "content": "# Guide",
                "status": "publish",
            }
            return httpx.Response(201, json={"data": {"id": "manual-one"}})
        assert request.url.path == "/api/v1/knowledgebase/kb-one/wiki/pages"
        assert request.url.params["query"] == "guide"
        return httpx.Response(
            200,
            json={"pages": [{"slug": "guide", "title": "Guide"}], "total": 1, "page": 1, "page_size": 50},
        )

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
        await client.import_document_url("kb-one", url="https://example.test/guide")
        await client.create_manual_document("kb-one", title="Guide", content="# Guide")
        pages = await client.list_wiki_pages("kb-one", query="guide")

    assert pages["total"] == 1
    assert pages["pages"][0]["slug"] == "guide"


@pytest.mark.asyncio
async def test_weknora_client_reads_document_chunks_and_wiki_graph() -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/api/v1/chunks/doc-one":
            assert request.url.params["page"] == "2"
            assert request.url.params["page_size"] == "10"
            return httpx.Response(
                200,
                json={
                    "success": True,
                    "data": [{"id": "chunk-one", "content": "parsed content", "chunk_index": 10}],
                    "total": 31,
                    "page": 2,
                    "page_size": 10,
                },
            )
        assert request.url.path == "/api/v1/knowledgebase/kb-one/wiki/graph"
        assert request.url.params["mode"] == "overview"
        assert request.url.params["limit"] == "50"
        return httpx.Response(
            200,
            json={
                "success": True,
                "data": {
                    "nodes": [{"slug": "entity/one", "title": "Entity One", "page_type": "entity"}],
                    "edges": [],
                    "meta": {"mode": "overview", "total": 1, "returned": 1},
                },
            },
        )

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
        chunks = await client.list_chunks("doc-one", page=2, page_size=10)
        graph = await client.get_wiki_graph("kb-one", limit=50)

    assert chunks["total"] == 31
    assert chunks["items"][0]["content"] == "parsed content"
    assert graph["nodes"][0]["slug"] == "entity/one"


@pytest.mark.asyncio
async def test_weknora_client_proxies_original_document_preview_without_credentials() -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/api/v1/knowledge/doc-one/preview"
        assert "X-API-Key" not in request.headers
        assert "Authorization" not in request.headers
        return httpx.Response(200, content=b"preview", headers={"Content-Type": "text/plain; charset=utf-8"})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
        content, media_type = await client.get_document_preview("doc-one")

    assert content == b"preview"
    assert media_type == "text/plain"


@pytest.mark.asyncio
async def test_memory_store_enforces_owner_and_public_visibility() -> None:
    store = MemoryLlmWikiStore()
    created = await store.create_mapping(
        weknora_id="wk-private",
        owner_user_id="user-a",
        name="Private notes",
        description="",
        kb_type="document",
    )

    assert [row["id"] for row in await store.list_visible("user-a", scope="personal")] == [created["id"]]
    assert await store.list_visible("user-b", scope="personal") == []
    assert await store.get_authorized(created["id"], "user-b", write=False, is_admin=False) is None

    published = await store.request_publish(created["id"], "user-a", is_admin=False)
    assert published and published["publication_status"] == "published"
    assert [row["id"] for row in await store.list_visible("user-b", scope="public")] == [created["id"]]
    assert await store.get_authorized(created["id"], "user-b", write=False, is_admin=False) is not None
    assert await store.get_authorized(created["id"], "user-b", write=True, is_admin=False) is None


@pytest.mark.asyncio
async def test_conversation_deposit_mapping_is_system_owned_and_published(monkeypatch: pytest.MonkeyPatch) -> None:
    from types import SimpleNamespace

    from app.gateway import llmwiki_deposit

    class FakeClient:
        async def list_knowledge_bases(self) -> list[dict[str, object]]:
            return []

        async def create_knowledge_base(self, **_: object) -> dict[str, object]:
            return {
                "id": "remote-deposit",
                "name": "对话沉淀",
                "description": "global",
                "type": "document",
            }

    store = MemoryLlmWikiStore()
    app = SimpleNamespace(state=SimpleNamespace(llmwiki_store=store))
    monkeypatch.setattr(llmwiki_deposit, "_client", lambda _: FakeClient())

    row = await llmwiki_deposit.ensure_conversation_deposit_knowledge_base(app, owner_user_id="user-a")

    assert row is not None
    assert row["owner_user_id"] == "system"
    assert row["publication_status"] == "published"
    assert row["description"] == llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION
    assert row["id"] not in [item["id"] for item in await store.list_visible("user-a", scope="personal")]
    assert [item["id"] for item in await store.list_visible("user-a", scope="public")] == [row["id"]]

    markdown = llmwiki_deposit._format_markdown(
        llmwiki_deposit.ConversationTurnDeposit(
            thread_id="thread-one",
            question="question",
            answer="answer",
            human_message_id="human-one",
            assistant_message_id="assistant-one",
            assistant_id=None,
            knowledge_base_ids=[],
            created_by_user_id="user-a",
        ),
        "title",
    )
    assert "DeerFlow" not in markdown
    assert "source: cmzs" in markdown
    assert "#cmzs" in markdown


@pytest.mark.asyncio
async def test_existing_conversation_deposit_description_is_refreshed(monkeypatch: pytest.MonkeyPatch) -> None:
    from types import SimpleNamespace

    from app.gateway import llmwiki_deposit

    updates: list[tuple[str, dict[str, object]]] = []

    class FakeClient:
        async def update_knowledge_base(self, knowledge_base_id: str, changes: dict[str, object]) -> dict[str, object]:
            updates.append((knowledge_base_id, changes))
            return {"id": knowledge_base_id, **changes}

    store = MemoryLlmWikiStore()
    created = await store.create_mapping(
        weknora_id="remote-deposit",
        owner_user_id="system",
        name=llmwiki_deposit.CONVERSATION_DEPOSIT_KB_NAME,
        description="old description",
        kb_type="document",
    )
    await store.request_publish(created["id"], "system", is_admin=True)
    app = SimpleNamespace(state=SimpleNamespace(llmwiki_store=store))
    monkeypatch.setattr(llmwiki_deposit, "_client", lambda _: FakeClient())

    row = await llmwiki_deposit.ensure_conversation_deposit_knowledge_base(app, owner_user_id="user-a")

    assert row is not None
    assert row["description"] == llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION
    assert updates == [
        (
            "remote-deposit",
            {"description": llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION},
        )
    ]


@pytest.mark.asyncio
async def test_legacy_user_owned_conversation_deposit_is_adopted(monkeypatch: pytest.MonkeyPatch) -> None:
    from types import SimpleNamespace

    from app.gateway import llmwiki_deposit

    class FakeClient:
        async def list_knowledge_bases(self) -> list[dict[str, object]]:
            return [
                {
                    "id": "remote-deposit",
                    "name": llmwiki_deposit.CONVERSATION_DEPOSIT_KB_NAME,
                    "description": "DeerFlow legacy description",
                }
            ]

        async def update_knowledge_base(self, knowledge_base_id: str, changes: dict[str, object]) -> dict[str, object]:
            return {"id": knowledge_base_id, **changes}

    store = MemoryLlmWikiStore()
    legacy = await store.create_mapping(
        weknora_id="remote-deposit",
        owner_user_id="user-a",
        name=llmwiki_deposit.CONVERSATION_DEPOSIT_KB_NAME,
        description="DeerFlow legacy description",
        kb_type="document",
    )
    await store.request_publish(legacy["id"], "user-a", is_admin=False)
    app = SimpleNamespace(state=SimpleNamespace(llmwiki_store=store))
    monkeypatch.setattr(llmwiki_deposit, "_client", lambda _: FakeClient())

    row = await llmwiki_deposit.ensure_conversation_deposit_knowledge_base(app, owner_user_id="user-a")

    assert row is not None
    assert row["id"] == legacy["id"]
    assert row["owner_user_id"] == "system"
    assert row["publication_status"] == "published"
    assert row["description"] == llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION
    assert await store.list_visible("user-a", scope="personal") == []


@pytest.mark.asyncio
async def test_listing_conversation_deposit_refreshes_remote_description(monkeypatch: pytest.MonkeyPatch) -> None:
    from app.gateway import llmwiki_deposit
    from app.gateway.routers import llmwiki as llmwiki_router

    updates: list[tuple[str, dict[str, object]]] = []
    store = MemoryLlmWikiStore()
    created = await store.create_mapping(
        weknora_id="remote-deposit",
        owner_user_id="system",
        name=llmwiki_deposit.CONVERSATION_DEPOSIT_KB_NAME,
        description="DeerFlow legacy description",
        kb_type="document",
    )
    await store.request_publish(created["id"], "system", is_admin=True)

    class Client:
        async def list_knowledge_bases(self) -> list[dict[str, object]]:
            return [
                {
                    "id": "remote-deposit",
                    "name": llmwiki_deposit.CONVERSATION_DEPOSIT_KB_NAME,
                    "description": "DeerFlow legacy description",
                }
            ]

        async def create_knowledge_base(self, **_: object) -> dict[str, object]:
            raise AssertionError("The existing deposit should be reused")

        async def update_knowledge_base(self, knowledge_base_id: str, changes: dict[str, object]) -> dict[str, object]:
            updates.append((knowledge_base_id, changes))
            return {"id": knowledge_base_id, **changes}

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_router, "_actor", actor)
    monkeypatch.setattr(llmwiki_router, "_store", lambda _: store)
    monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client())

    payload = await llmwiki_router.list_knowledge_bases(object(), scope="public")

    assert payload["knowledge_bases"][0]["description"] == llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION
    assert "DeerFlow" not in json.dumps(payload, ensure_ascii=False)
    assert updates == [
        (
            "remote-deposit",
            {"description": llmwiki_deposit.CONVERSATION_DEPOSIT_KB_DESCRIPTION},
        )
    ]


def test_conversation_deposit_history_hides_legacy_brand() -> None:
    from app.gateway.routers import llmwiki as llmwiki_router

    value = {
        "content": "This page was generated by DeerFlow.",
        "metadata": {"source": "deer-flow", "tags": ["DEER_FLOW", "keep"]},
    }

    gateway_value = llmwiki_router._sanitize_conversation_deposit_value(value)
    tool_value = sanitize_tool_deposit_value(value)

    assert gateway_value == tool_value
    assert gateway_value == {
        "content": "This page was generated by cmzs.",
        "metadata": {"source": "cmzs", "tags": ["cmzs", "keep"]},
    }


@pytest.mark.asyncio
async def test_list_knowledge_bases_personal_scope_filters_to_actor_for_admin(monkeypatch: pytest.MonkeyPatch) -> None:
    from app.gateway.routers import llmwiki as llmwiki_router

    calls: list[dict[str, object]] = []

    class Store:
        async def list_visible(self, user_id: str, *, scope: str, is_admin: bool) -> list[dict[str, object]]:
            calls.append({"user_id": user_id, "scope": scope, "is_admin": is_admin})
            return [
                {
                    "id": "local-admin",
                    "weknora_id": "remote-admin",
                    "owner_user_id": user_id,
                    "name": "Admin private",
                    "description": "",
                    "kb_type": "document",
                    "publication_status": "private",
                }
            ]

    class Client:
        async def list_knowledge_bases(self) -> list[dict[str, object]]:
            return [{"id": "remote-admin", "name": "Admin private"}]

    async def actor(_: object) -> tuple[str, bool]:
        return "admin-id", True

    monkeypatch.setattr(llmwiki_router, "_actor", actor)
    monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store())
    monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client())

    payload = await llmwiki_router.list_knowledge_bases(object(), scope="personal")

    assert calls == [{"user_id": "admin-id", "scope": "personal", "is_admin": False}]
    assert payload["knowledge_bases"][0]["is_owner"] is True


def test_mapping_view_distinguishes_owner_from_admin_write_power() -> None:
    from app.gateway.routers import llmwiki as llmwiki_router

    view = llmwiki_router._mapping_view(
        {
            "id": "local-other",
            "weknora_id": "remote-other",
            "owner_user_id": "user-b",
            "name": "Other private",
            "description": "",
            "kb_type": "document",
            "publication_status": "private",
        },
        {"id": "remote-other", "name": "Other private"},
        actor_user_id="admin-id",
        is_admin=True,
    )

    assert view["is_owner"] is False
    assert view["can_write"] is True


def test_only_system_owned_named_base_is_treated_as_conversation_deposit() -> None:
    from app.gateway.routers import llmwiki as llmwiki_router

    personal = {
        "id": "personal-named-deposit",
        "owner_user_id": "user-a",
        "name": "对话沉淀",
        "publication_status": "private",
    }
    system = {
        "id": "system-deposit",
        "owner_user_id": "system",
        "name": "对话沉淀",
        "publication_status": "published",
    }

    assert not llmwiki_router._is_conversation_deposit_mapping(personal)
    assert llmwiki_router._is_conversation_deposit_mapping(system)
    system_view = llmwiki_router._mapping_view(
        system,
        {
            "id": "remote-system",
            "name": "对话沉淀",
            "description": "DeerFlow legacy description",
        },
        actor_user_id="admin-id",
        is_admin=True,
    )
    assert system_view["can_write"] is False
    assert system_view["description"] == "cmzs 全局问答沉淀知识库。所有用户可见，不默认参与问答检索。"
    assert "DeerFlow" not in system_view["description"]


@pytest.mark.asyncio
async def test_conversation_deposit_requires_thread_access(monkeypatch: pytest.MonkeyPatch) -> None:
    from types import SimpleNamespace

    from app.gateway.routers import llmwiki as llmwiki_router

    class ThreadStore:
        async def check_access(self, thread_id: str, user_id: str, *, require_existing: bool) -> bool:
            assert thread_id == "thread-other"
            assert user_id == "user-a"
            assert require_existing is True
            return False

    async def actor(_: object) -> tuple[str, bool]:
        return "user-a", False

    monkeypatch.setattr(llmwiki_router, "_actor", actor)
    monkeypatch.setattr(llmwiki_router, "get_thread_store", lambda _: ThreadStore())
    monkeypatch.setattr(
        llmwiki_router,
        "get_resolved_llmwiki_runtime",
        lambda _: SimpleNamespace(weknora_enabled=True, provider="weknora"),
    )
    request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(config=object())))
    body = llmwiki_router.ConversationDepositCreate(
        thread_id="thread-other",
        question="question",
        answer="answer",
        assistant_message_id="assistant-one",
    )

    with pytest.raises(HTTPException) as exc:
        await llmwiki_router.create_conversation_deposit(
            request,
            SimpleNamespace(add_task=lambda *_: None),
            body,
        )

    assert exc.value.status_code == 404


@pytest.mark.asyncio
async def test_source_context_prefers_wiki_page_over_raw_chunks(monkeypatch: pytest.MonkeyPatch) -> None:
    from app.gateway.routers import llmwiki as llmwiki_router

    class Client:
        async def get_chunk_context(self, chunk_id: str, *, expected_knowledge_base_id: str, radius: int) -> list[dict[str, object]]:
            assert chunk_id == "chunk-one"
            assert expected_knowledge_base_id == "remote-kb"
            return [
                {
                    "id": "chunk-one",
                    "chunk_index": 0,
                    "content": "raw split chunk",
                    "metadata": {"wiki_page_slug": "people/alice"},
                }
            ]

        async def get_wiki_page(self, knowledge_base_id: str, slug: str) -> dict[str, object]:
            assert knowledge_base_id == "remote-kb"
            assert slug == "people/alice"
            return {
                "id": "wiki-one",
                "slug": "people/alice",
                "title": "Alice",
                "summary": "clean summary",
                "content": "clean llmwiki page",
                "source_refs": ["chunk-one"],
            }

    async def authorized_mapping(*_: object, **__: object) -> tuple[dict[str, str], str, bool]:
        return {"id": "local-kb", "weknora_id": "remote-kb", "name": "Public LLMWiki"}, "user-a", False

    monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized_mapping)
    monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client())

    payload = await llmwiki_router.get_source_context(object(), knowledge_base_id="local-kb", chunk_id="chunk-one")

    assert payload["data"]["display_mode"] == "wiki"
    assert payload["data"]["wiki_page"]["content"] == "clean llmwiki page"
    assert payload["data"]["chunks"][0]["content"] == "raw split chunk"


@pytest.mark.asyncio
async def test_delete_cleans_owner_mapping_when_remote_is_already_missing(monkeypatch: pytest.MonkeyPatch) -> None:
    from app.gateway.routers import llmwiki as llmwiki_router

    deleted: list[str] = []

    class MissingClient:
        async def delete_knowledge_base(self, _: str) -> None:
            raise WeKnoraError("WeKnora resource was not found", status_code=404)

    class Store:
        async def delete_mapping(self, mapping_id: str) -> None:
            deleted.append(mapping_id)

    async def authorized_mapping(*_: object, **__: object) -> tuple[dict[str, str], str, bool]:
        return {"weknora_id": "remote-missing"}, "owner", False

    monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized_mapping)
    monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: MissingClient())
    monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store())

    await llmwiki_router.delete_knowledge_base(object(), "mapping-one")

    assert deleted == ["mapping-one"]


def test_weknora_iframe_cookie_is_signed_and_short_lived() -> None:
    from app.gateway.routers import llmwiki as llmwiki_router

    raw = llmwiki_router._sign_embed_payload(
        {
            "mapping_id": "local-kb",
            "weknora_id": "remote-kb",
            "user_id": "user-a",
            "is_admin": False,
            "can_write": True,
            "exp": 4_102_444_800,
        }
    )

    ctx = llmwiki_router._verify_embed_cookie(raw)

    assert ctx.mapping_id == "local-kb"
    assert ctx.weknora_id == "remote-kb"
    assert ctx.can_write is True

    with pytest.raises(HTTPException):
        llmwiki_router._verify_embed_cookie(raw + "tampered")


def test_weknora_iframe_proxy_is_fallback_after_deerflow_api_routes() -> None:
    assert _first_gateway_match("POST", "/api/v1/auth/login/username") == "/api/v1/auth/login/username"
    assert _first_gateway_match("POST", "/api/v1/knowledge-search") == "/api/v1/{proxied_path:path}"
    assert (
        _first_gateway_match("GET", "/api/llmwiki/weknora-embed/api/v1/auth/me")
        == "/api/llmwiki/weknora-embed/api/v1/{proxied_path:path}"
    )
    assert _first_gateway_match("GET", "/platform/knowledge-bases/remote-kb") == "/platform/{proxied_path:path}"


def test_weknora_iframe_filters_kb_list_to_current_mapping() -> None:
    from app.gateway.routers import llmwiki as llmwiki_router

    ctx = llmwiki_router.WeKnoraEmbedContext(
        mapping_id="local-kb",
        weknora_id="remote-kb",
        user_id="user-a",
        is_admin=False,
        can_write=False,
        exp=4_102_444_800,
    )
    response = httpx.Response(
        200,
        json={
            "success": True,
            "data": [
                {"id": "remote-kb", "name": "Allowed"},
                {"id": "other-kb", "name": "Blocked"},
            ],
        },
    )

    filtered = llmwiki_router._filter_knowledge_base_list_response(response, ctx)

    assert filtered is not None
    payload = json.loads(filtered)
    assert payload["data"] == [{"id": "remote-kb", "name": "Allowed"}]


@pytest.mark.asyncio
async def test_conversation_deposit_iframe_sanitizes_legacy_brand(monkeypatch: pytest.MonkeyPatch) -> None:
    from types import SimpleNamespace

    from app.gateway.routers import llmwiki as llmwiki_router

    async def fake_session(_: object, *, force_refresh: bool = False) -> dict[str, object]:
        return {"token": "token", "tenant": {}}

    async def authorize(*_: object) -> None:
        return None

    class FakeAsyncClient:
        def __init__(self, **_: object) -> None:
            pass

        async def __aenter__(self) -> "FakeAsyncClient":
            return self

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

        async def request(self, *_: object, **__: object) -> httpx.Response:
            return httpx.Response(
                200,
                json={"success": True, "data": {"summary": "Generated by DeerFlow", "source": "deer-flow"}},
            )

    class FakeRequest:
        method = "GET"
        headers = Headers({})
        query_params = QueryParams("")

        async def body(self) -> bytes:
            return b""

    ctx = llmwiki_router.WeKnoraEmbedContext(
        mapping_id="local-deposit",
        weknora_id="remote-deposit",
        user_id="user-a",
        is_admin=False,
        can_write=False,
        exp=4_102_444_800,
        is_conversation_deposit=True,
    )
    monkeypatch.setattr(llmwiki_router, "_runtime_for_iframe_proxy", lambda _: SimpleNamespace())
    monkeypatch.setattr(llmwiki_router, "_get_weknora_admin_session", fake_session)
    monkeypatch.setattr(llmwiki_router, "_authorize_weknora_api_proxy", authorize)
    monkeypatch.setattr(llmwiki_router.httpx, "AsyncClient", FakeAsyncClient)

    response = await llmwiki_router._proxy_weknora_request(
        FakeRequest(),
        target_base_url="http://weknora.local",
        upstream_path="/api/v1/knowledge-bases/remote-deposit",
        ctx=ctx,
        is_api=True,
    )

    payload = json.loads(response.body)
    assert payload["data"] == {"summary": "Generated by cmzs", "source": "cmzs"}


@pytest.mark.asyncio
async def test_weknora_iframe_proxy_blocks_other_knowledge_bases() -> None:
    from app.gateway.routers import llmwiki as llmwiki_router

    class FakeRequest:
        method = "GET"
        headers = Headers({})
        query_params = QueryParams("")

    ctx = llmwiki_router.WeKnoraEmbedContext(
        mapping_id="local-kb",
        weknora_id="remote-kb",
        user_id="user-a",
        is_admin=False,
        can_write=False,
        exp=4_102_444_800,
    )

    await llmwiki_router._authorize_weknora_api_proxy(
        FakeRequest(),
        ctx,
        object(),
        "/api/v1/knowledge-bases/remote-kb",
        b"",
        "token",
    )

    with pytest.raises(HTTPException) as exc_info:
        await llmwiki_router._authorize_weknora_api_proxy(
            FakeRequest(),
            ctx,
            object(),
            "/api/v1/knowledge-bases/other-kb",
            b"",
            "token",
        )

    assert exc_info.value.status_code == 403