deerflow-code/offline-backend-20260512/backend/tests/test_llmwiki_weknora.py

from __future__ import annotations

import asyncio
import json
from io import BytesIO
from types import SimpleNamespace

import httpx
import pytest
from fastapi import FastAPI, HTTPException, UploadFile
from fastapi.responses import Response
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.integrations.weknora.knowledge_profile import (
    are_test_questions_customized,
    default_test_questions,
    generated_test_questions_name,
    resolved_test_questions,
)
from deerflow.integrations.weknora.runtime import build_weknora_client
from deerflow.persistence.llmwiki import MemoryLlmWikiStore
from deerflow.tools.builtins.llmwiki_search_tool import (
    _sanitize_conversation_deposit_value as sanitize_tool_deposit_value,
)
from deerflow.tools.builtins.llmwiki_search_tool import (
    selected_knowledge_base_ids,
)


def _compatible_knowledge_package_bytes(*, bom: bool = False) -> bytes:
    records = [
        {
            "record_type": "source",
            "payload": {"package_version": 1},
        },
        {
            "record_type": "source",
            "payload": {
                "provider": "weknora",
                "export_format": "assistant-knowledge-jsonl",
            },
        },
        {
            "record_type": "wiki_page",
            "payload": {
                "slug": "wiki/cooling",
                "title": "散热",
                "content": "散热管理用于降低设备温度。",
            },
        },
    ]
    content = ("\n\n" + "\n".join(json.dumps(record, ensure_ascii=False) for record in records) + "\n").encode()
    return (b"\xef\xbb\xbf" if bom else b"") + content


def _request(
    path: str,
    *,
    query: str = "",
    cookies: dict[str, str] | None = None,
    request_headers: dict[str, str] | None = None,
    root_path: str = "",
) -> Request:
    headers: list[tuple[bytes, bytes]] = [(key.lower().encode(), value.encode()) for key, value in (request_headers or {}).items()]
    if cookies:
        headers.append((b"cookie", "; ".join(f"{key}={value}" for key, value in cookies.items()).encode()))
    return Request(
        {
            "type": "http",
            "method": "GET",
            "path": path,
            "root_path": root_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 _knowledge_profile_mapping_row(*, name: str, questions: list[str] | None) -> dict[str, object]:
    return {
        "id": "mapping-one",
        "weknora_id": "remote-one",
        "owner_user_id": "user-one",
        "name": name,
        "description": "",
        "test_questions": questions,
        "retrieval_settings": None,
        "kb_type": "document",
        "publication_status": "private",
    }


def test_legacy_persisted_defaults_follow_renamed_knowledge_base() -> None:
    old_defaults = default_test_questions("旧名称")
    row = _knowledge_profile_mapping_row(name="新名称", questions=old_defaults)

    assert generated_test_questions_name(old_defaults) == "旧名称"
    assert resolved_test_questions(row) == default_test_questions("新名称")
    assert are_test_questions_customized(row) is False


def test_genuinely_custom_questions_are_not_rewritten_after_rename() -> None:
    questions = ["旧名称的自定义问题保留吗？"]
    row = _knowledge_profile_mapping_row(name="新名称", questions=questions)

    assert generated_test_questions_name(questions) is None
    assert resolved_test_questions(row) == questions
    assert are_test_questions_customized(row) is True


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

    row = _knowledge_profile_mapping_row(
        name="本地旧名称",
        questions=default_test_questions("更早的旧名称"),
    )

    view = llmwiki_router._mapping_view(
        row,
        {"id": "remote-one", "name": "远程新名称"},
        actor_user_id="user-one",
        is_admin=False,
    )

    assert view["name"] == "远程新名称"
    assert view["test_questions"] == default_test_questions("远程新名称")
    assert view["test_questions_customized"] is False


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"


@pytest.mark.asyncio
async def test_default_weknora_client_is_reused_on_the_same_event_loop() -> None:
    runtime = resolve_llmwiki_runtime(
        LlmWikiConfig.model_validate(
            {
                "weknora": {
                    "api_base_url": "http://shared-weknora.test:8080",
                    "admin_email": "admin@example.test",
                    "admin_password": "secret",
                }
            }
        )
    )

    first = build_weknora_client(runtime)
    second = build_weknora_client(runtime)

    assert second is first


def test_default_weknora_client_is_isolated_between_event_loops() -> None:
    runtime = resolve_llmwiki_runtime(
        LlmWikiConfig.model_validate(
            {
                "weknora": {
                    "api_base_url": "http://loop-local-weknora.test:8080",
                    "admin_email": "admin@example.test",
                    "admin_password": "secret",
                }
            }
        )
    )

    async def build_pair() -> tuple[WeKnoraClient, WeKnoraClient]:
        return build_weknora_client(runtime), build_weknora_client(runtime)

    first_loop_pair = asyncio.run(build_pair())
    second_loop_pair = asyncio.run(build_pair())

    assert first_loop_pair[0] is first_loop_pair[1]
    assert second_loop_pair[0] is second_loop_pair[1]
    assert first_loop_pair[0] is not second_loop_pair[0]


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_weknora_embed_follows_only_protected_binary_redirects() -> None:
    from app.gateway.routers import llmwiki as llmwiki_router

    assert llmwiki_router._should_follow_weknora_binary_redirect("/api/v1/knowledge/doc-one/preview")
    assert llmwiki_router._should_follow_weknora_binary_redirect("/api/v1/knowledge-bases/kb-one/files")
    assert not llmwiki_router._should_follow_weknora_binary_redirect("/api/v1/knowledge/doc-one")
    assert not llmwiki_router._should_follow_weknora_binary_redirect("/api/v1/knowledge-bases/kb-one")


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_llmwiki_rag_only_uses_explicit_conversation_selection() -> None:
    from app.gateway.llmwiki_rag import resolve_requested_llmwiki_knowledge_base_ids

    body = SimpleNamespace(
        assistant_id="agent-with-bound-kb",
        context={},
        metadata={"llmwiki_knowledge_base_ids": ["kb-one", "kb-one", "", "kb-two"]},
    )

    assert resolve_requested_llmwiki_knowledge_base_ids(body) == ["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("/deerflow/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("/deerflow/assets/index.js", cookies=cookies))
    for static_path in (
        "/deerflow/css/app.css",
        "/deerflow/fonts/app.woff2",
        "/deerflow/img/logo.png",
        "/deerflow/images/empty.svg",
        "/deerflow/js/app.js",
        "/deerflow/media/intro.mp4",
        "/deerflow/static/chunk.js",
    ):
        request = _request(static_path, cookies=cookies)
        assert auth_middleware._is_weknora_embed_proxy_request(request)
        assert csrf_middleware.is_weknora_embed_proxy_request(request)
    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("/deerflow/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 auth_middleware._is_weknora_embed_proxy_request(_request("/deerflow/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"}

        async def list_visible(self, *_: object, **__: object) -> list[dict[str, str]]:
            return [
                {"id": "local-kb", "weknora_id": "remote-kb"},
                {"id": "public-kb", "weknora_id": "other-public-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"] == "/deerflow/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
    set_cookie_headers = [value.decode("latin-1") for key, value in response.raw_headers if key.lower() == b"set-cookie"]
    assert any("Path=/" in value for value in set_cookie_headers)
    assert any("Path=/deerflow" in value for value in set_cookie_headers)
    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
    assert ctx.allowed_weknora_ids == ("remote-kb", "other-public-kb")


@pytest.mark.asyncio
async def test_weknora_session_endpoint_preserves_reverse_proxy_prefix(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"}

        async def list_visible(self, *_: object, **__: object) -> list[dict[str, str]]:
            return []

    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-Prefix": "/deerflow-api/"},
        ),
        "local-kb",
        tab="wiki",
    )

    payload = json.loads(response.body)
    assert payload["frame_path"] == "/deerflow-api/platform/knowledge-bases/remote-kb?tab=wiki"
    set_cookie_headers = [value.decode("latin-1") for key, value in response.raw_headers if key.lower() == b"set-cookie"]
    assert any("Path=/deerflow-api" in value for value in set_cookie_headers)


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

    request = _request(
        "/api/llmwiki/knowledge-bases/local-kb/weknora/session",
        request_headers={"X-Forwarded-Prefix": "//untrusted.example"},
    )

    assert llmwiki_router._request_public_prefix(request) == "/deerflow"


@pytest.mark.asyncio
async def test_weknora_session_endpoint_uses_backend_prefix_when_frontend_has_different_prefix(
    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"}

        async def list_visible(self, *_: object, **__: object) -> list[dict[str, str]]:
            return []

    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={"Referer": "https://example.test/magentweb/page/workspace/knowledge/local-kb"},
        ),
        "local-kb",
        tab="wiki",
    )

    payload = json.loads(response.body)
    assert payload["frame_path"] == "/deerflow/platform/knowledge-bases/remote-kb?tab=wiki"


@pytest.mark.asyncio
async def test_weknora_client_keeps_anonymous_mode_without_admin_credentials() -> 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_uses_admin_token_when_credentials_are_configured() -> None:
    requests: list[httpx.Request] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        if request.url.path == "/api/v1/auth/login":
            assert json.loads(request.content) == {"email": "admin@example.test", "password": "secret"}
            return httpx.Response(
                200,
                json={
                    "success": True,
                    "data": {
                        "token": "token-one",
                        "tenant": {"id": "tenant-one"},
                    },
                },
            )
        assert request.headers["Authorization"] == "Bearer token-one"
        assert request.headers["X-Tenant-ID"] == "tenant-one"
        return httpx.Response(200, json={"success": True, "data": []})

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

    assert results == []
    assert [request.url.path for request in requests] == [
        "/api/v1/auth/login",
        "/api/v1/knowledge-search",
    ]


@pytest.mark.asyncio
async def test_weknora_client_coalesces_concurrent_admin_logins() -> None:
    login_count = 0

    async def handler(request: httpx.Request) -> httpx.Response:
        nonlocal login_count
        if request.url.path == "/api/v1/auth/login":
            login_count += 1
            await asyncio.sleep(0.01)
            return httpx.Response(
                200,
                json={"success": True, "data": {"token": "shared-token"}},
            )
        assert request.headers["Authorization"] == "Bearer shared-token"
        return httpx.Response(200, json={"success": True, "data": []})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(
            base_url="http://weknora.local:8080",
            http_client=http_client,
            admin_email="admin@example.test",
            admin_password="secret",
        )
        await asyncio.gather(
            client.list_knowledge_bases(),
            client.list_knowledge_bases(),
        )

    assert login_count == 1


@pytest.mark.asyncio
async def test_weknora_client_caches_and_coalesces_knowledge_base_directory() -> None:
    list_count = 0

    async def handler(request: httpx.Request) -> httpx.Response:
        nonlocal list_count
        assert request.url.path == "/api/v1/knowledge-bases"
        list_count += 1
        await asyncio.sleep(0.01)
        return httpx.Response(200, json={"success": True, "data": [{"id": "kb-one", "name": "One"}]})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
        first, second = await asyncio.gather(
            client.list_knowledge_bases(),
            client.list_knowledge_bases(),
        )
        cached = await client.list_knowledge_bases()

    assert list_count == 1
    assert first == second == cached == [{"id": "kb-one", "name": "One"}]
    assert first is not second


@pytest.mark.asyncio
async def test_weknora_client_invalidates_knowledge_base_directory_after_mutation() -> None:
    list_count = 0

    async def handler(request: httpx.Request) -> httpx.Response:
        nonlocal list_count
        if request.method == "GET":
            list_count += 1
            return httpx.Response(200, json={"success": True, "data": [{"id": "kb-one", "name": f"One-{list_count}"}]})
        assert request.method == "PUT"
        return httpx.Response(200, json={"success": True, "data": {"id": "kb-one", "name": "Renamed"}})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
        assert (await client.list_knowledge_bases())[0]["name"] == "One-1"
        await client.update_knowledge_base("kb-one", {"name": "Renamed"})
        assert (await client.list_knowledge_bases())[0]["name"] == "One-2"

    assert list_count == 2


@pytest.mark.asyncio
async def test_weknora_client_invalidates_knowledge_base_directory_after_create_and_delete() -> None:
    list_count = 0

    async def handler(request: httpx.Request) -> httpx.Response:
        nonlocal list_count
        if request.method == "GET":
            list_count += 1
            return httpx.Response(200, json={"success": True, "data": [{"id": "kb-one", "name": f"One-{list_count}"}]})
        if request.method == "POST":
            return httpx.Response(201, json={"success": True, "data": {"id": "kb-created", "name": "Created"}})
        assert request.method == "DELETE"
        return httpx.Response(204)

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
        assert (await client.list_knowledge_bases())[0]["name"] == "One-1"
        await client.create_knowledge_base(name="Created", embedding_model_id="embed-one")
        assert (await client.list_knowledge_bases())[0]["name"] == "One-2"
        await client.delete_knowledge_base("kb-created")
        assert (await client.list_knowledge_bases())[0]["name"] == "One-3"

    assert list_count == 3


@pytest.mark.asyncio
async def test_weknora_client_invalidates_knowledge_base_directory_after_document_upload() -> None:
    list_count = 0

    async def handler(request: httpx.Request) -> httpx.Response:
        nonlocal list_count
        if request.method == "GET":
            list_count += 1
            return httpx.Response(
                200,
                json={
                    "success": True,
                    "data": [{"id": "kb-one", "processing_count": list_count}],
                },
            )
        assert request.url.path == "/api/v1/knowledge-bases/kb-one/knowledge/file"
        return httpx.Response(201, json={"success": True, "data": {"id": "doc-one"}})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
        assert (await client.list_knowledge_bases())[0]["processing_count"] == 1
        await client.upload_document("kb-one", filename="guide.md", content=b"# Guide")
        assert (await client.list_knowledge_bases())[0]["processing_count"] == 2

    assert list_count == 2


@pytest.mark.asyncio
async def test_weknora_client_refreshes_admin_token_after_rejection() -> None:
    login_count = 0
    search_count = 0

    async def handler(request: httpx.Request) -> httpx.Response:
        nonlocal login_count, search_count
        if request.url.path == "/api/v1/auth/login":
            login_count += 1
            return httpx.Response(
                200,
                json={"success": True, "data": {"token": f"token-{login_count}"}},
            )
        search_count += 1
        if search_count == 1:
            assert request.headers["Authorization"] == "Bearer token-1"
            return httpx.Response(401, json={"message": "expired"})
        assert request.headers["Authorization"] == "Bearer token-2"
        return httpx.Response(200, json={"success": True, "data": []})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(
            base_url="http://weknora.local:8080",
            http_client=http_client,
            admin_email="admin@example.test",
            admin_password="secret",
        )
        await client.search("deployment guide", ["kb-one"])

    assert login_count == 2
    assert search_count == 2


@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_merges_wiki_instructions_without_losing_runtime_config() -> 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": "Operational Wiki",
                        "wiki_config": {
                            "synthesis_model_id": "qa-one",
                            "extraction_granularity": "standard",
                            "content_instructions": "old content guidance",
                            "extraction_instructions": "old extraction guidance",
                        },
                    },
                },
            )
        assert request.method == "PUT"
        assert json.loads(request.content) == {
            "name": "Operational Wiki",
            "config": {
                "wiki_config": {
                    "synthesis_model_id": "qa-one",
                    "extraction_granularity": "standard",
                    "content_instructions": "先结论后依据",
                    "extraction_instructions": "old extraction guidance",
                }
            },
        }
        return httpx.Response(
            200,
            json={
                "success": True,
                "data": {
                    "id": "kb-one",
                    "name": "Operational Wiki",
                    "wiki_config": json.loads(request.content)["config"]["wiki_config"],
                },
            },
        )

    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",
            {
                "wiki_config": {
                    "content_instructions": " 先结论后依据 ",
                }
            },
        )

    assert updated["wiki_config"]["synthesis_model_id"] == "qa-one"
    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_wiki_create_reuses_catalog_models_without_debug_probe() -> None:
    requests: list[tuple[str, str, dict[str, object]]] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        payload = json.loads(request.content) if request.content else {}
        requests.append((request.method, request.url.path, payload))
        assert request.method == "POST"
        assert request.url.path == "/api/v1/knowledge-bases"
        return httpx.Response(201, json={"success": True, "data": {"id": "wiki-created", "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)
        await client.create_knowledge_base(
            name="Wiki",
            embedding_model_id="embedding-id",
            knowledge_qa_model_id="qa-id",
            wiki_enabled=True,
        )

    assert len(requests) == 1
    assert requests[0][2]["embedding_model_id"] == "embedding-id"
    assert requests[0][2]["summary_model_id"] == "qa-id"
    assert requests[0][2]["wiki_config"]["synthesis_model_id"] == "qa-id"


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

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

    class Client:
        async def list_models(self) -> list[dict[str, object]]:
            return [
                {
                    "id": "embed-local",
                    "type": "Embedding",
                    "name": "Local BGE",
                    "provider": "local",
                    "status": "active",
                    "is_default": True,
                    "api_key": "must-not-leak",
                    "config": {"secret": "must-not-leak"},
                },
                {
                    "id": "embed-openai",
                    "type": "embedding",
                    "display_name": "OpenAI text-embedding",
                    "source": "remote",
                    "parameters": {
                        "model": "text-embedding-3-large",
                        "provider": "openai",
                        "api_key": "must-not-leak",
                    },
                    "status": "active",
                },
                {
                    "id": "embed-disabled",
                    "type": "Embedding",
                    "name": "Disabled",
                    "status": "inactive",
                },
                {"id": "chat-one", "type": "LLM", "name": "chat"},
                {"id": "qa-default", "type": "KnowledgeQA", "name": "wiki-answer", "status": "active", "is_default": True},
            ]

        async def create_knowledge_base(self, **kwargs: object) -> dict[str, object]:
            create_calls.append(kwargs)
            return {
                "id": "remote-openai",
                "name": kwargs["name"],
                "description": kwargs["description"],
                "type": kwargs["kb_type"],
                "embedding_model_id": kwargs["embedding_model_id"],
            }

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

    store = MemoryLlmWikiStore()
    client = Client()
    monkeypatch.setattr(llmwiki_router, "_actor", actor)
    monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: client)
    monkeypatch.setattr(llmwiki_router, "_store", lambda _: store)
    monkeypatch.setattr(llmwiki_router, "_schedule_local_wiki_sync", lambda *_: None)

    listed = await llmwiki_router.list_embedding_models(object())

    assert listed["default_model_id"] == "embed-local"
    assert [item["id"] for item in listed["models"]] == [
        "embed-local",
        "embed-openai",
        "embed-disabled",
    ]
    assert listed["models"][1]["provider"] == "openai"
    assert listed["models"][1]["model_name"] == "text-embedding-3-large"
    assert listed["models"][2]["available"] is False
    assert "api_key" not in listed["models"][0]
    assert "config" not in listed["models"][0]
    assert "parameters" not in listed["models"][1]

    created = await llmwiki_router.create_knowledge_base(
        object(),
        llmwiki_router.KnowledgeBaseCreate(
            name="OpenAI docs",
            embedding_model_id="embed-openai",
            wiki_config=llmwiki_router.WikiGenerationInstructions(
                content_instructions="按事件时间线生成页面",
                extraction_instructions="重点提取人物和组织关系",
            ),
        ),
    )

    assert create_calls[0]["embedding_model_id"] == "embed-openai"
    assert create_calls[0]["knowledge_qa_model_id"] == "qa-default"
    assert create_calls[0]["wiki_enabled"] is True
    assert create_calls[0]["wiki_content_instructions"] == "按事件时间线生成页面"
    assert create_calls[0]["wiki_extraction_instructions"] == "重点提取人物和组织关系"
    assert created["embedding_model_id"] == "embed-openai"
    assert created["retrieval_settings"] == {
        "entity_expansion_enabled": True,
        "max_related_terms": 6,
        "statistics_enabled": True,
    }

    for rejected_model_id in ("embed-disabled", "chat-one", "missing"):
        with pytest.raises(HTTPException) as exc_info:
            await llmwiki_router.create_knowledge_base(
                object(),
                llmwiki_router.KnowledgeBaseCreate(
                    name="Rejected docs",
                    embedding_model_id=rejected_model_id,
                ),
            )
        assert exc_info.value.status_code == 422
    assert len(create_calls) == 1


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

    forwarded: list[tuple[str, dict[str, object]]] = []
    row = {
        "id": "local-wiki",
        "weknora_id": "remote-wiki",
        "owner_user_id": "user-a",
        "name": "Operations",
        "description": "",
        "kb_type": "document",
        "publication_status": "private",
    }

    class Client:
        async def get_knowledge_base(
            self,
            knowledge_base_id: str,
        ) -> dict[str, object]:
            return {
                "id": knowledge_base_id,
                "name": row["name"],
                "description": row["description"],
                "type": "document",
                "indexing_strategy": {"wiki_enabled": True},
                "wiki_config": {},
            }

        async def update_knowledge_base(
            self,
            knowledge_base_id: str,
            changes: dict[str, object],
        ) -> dict[str, object]:
            forwarded.append((knowledge_base_id, changes))
            return {
                "id": knowledge_base_id,
                "name": row["name"],
                "description": row["description"],
                "type": "document",
                "indexing_strategy": {"wiki_enabled": True},
                "wiki_config": changes["wiki_config"],
            }

    class Store:
        async def update_mapping(
            self,
            *_: object,
            **changes: object,
        ) -> dict[str, object]:
            return {**row, **changes}

    async def authorized_mapping(*_: object, **__: object) -> tuple[dict[str, object], str, bool]:
        return row, "user-a", False

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

    updated = await llmwiki_router.update_knowledge_base(
        object(),
        "local-wiki",
        llmwiki_router.KnowledgeBaseUpdate(
            wiki_config=llmwiki_router.WikiGenerationInstructions(
                content_instructions="正式、简洁，保留关键数字",
                extraction_instructions="优先人物、组织和时间线",
            )
        ),
    )

    assert forwarded == [
        (
            "remote-wiki",
            {
                "wiki_config": {
                    "content_instructions": "正式、简洁，保留关键数字",
                    "extraction_instructions": "优先人物、组织和时间线",
                }
            },
        )
    ]
    assert updated["wiki_config"]["content_instructions"] == "正式、简洁，保留关键数字"
    assert updated["wiki_config"]["extraction_instructions"] == "优先人物、组织和时间线"
    assert updated["wiki_config"]["wiki_config_schema_version"] == 3
    assert updated["wiki_config"]["wiki_config_revision"] == 1
    assert updated["wiki_config"]["apply_status"] == "active"


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

    class Client:
        async def list_models(self) -> list[dict[str, object]]:
            return [{"id": "chat-only", "type": "LLM"}]

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

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

    assert await llmwiki_router.list_embedding_models(object()) == {
        "models": [],
        "default_model_id": None,
    }


@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"},
                    ]
                },
            )
        if request.url.path == "/api/v1/models/qa-one/debug":
            assert request.method == "POST"
            assert request.headers["content-type"].startswith("multipart/form-data;")
            assert b"Reply with OK." in request.content
            return httpx.Response(200, json={"data": {"ok": True, "elapsed_ms": 12}})
        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"
        assert payload["wiki_config"]["content_instructions"] == "使用正式中文并保留关键数字"
        assert payload["wiki_config"]["extraction_instructions"] == "重点提取事件、人物与组织"
        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,
            wiki_content_instructions=" 使用正式中文并保留关键数字 ",
            wiki_extraction_instructions=" 重点提取事件、人物与组织 ",
        )

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


@pytest.mark.asyncio
async def test_weknora_client_prefers_a_healthy_explicit_knowledge_qa_default() -> None:
    attempts: list[str] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        if request.method == "GET" and request.url.path == "/api/v1/models":
            return httpx.Response(
                200,
                json={
                    "data": [
                        {
                            "id": "alphabetical-fallback",
                            "name": "Alpha",
                            "type": "KnowledgeQA",
                            "status": "active",
                            "is_default": False,
                        },
                        {
                            "id": "configured-default",
                            "name": "Zulu",
                            "type": "KnowledgeQA",
                            "status": "active",
                            "is_default": True,
                        },
                    ]
                },
            )
        attempts.append(request.url.path)
        return httpx.Response(200, json={"data": {"ok": True}})

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

    assert model_id == "configured-default"
    assert attempts == ["/api/v1/models/configured-default/debug"]


@pytest.mark.asyncio
async def test_weknora_client_uses_a_stable_healthy_knowledge_qa_fallback() -> None:
    attempts: list[str] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        if request.method == "GET" and request.url.path == "/api/v1/models":
            return httpx.Response(
                200,
                json={
                    "data": [
                        {
                            "id": "healthy-zulu",
                            "display_name": "Zulu",
                            "type": "KnowledgeQA",
                            "status": "active",
                            "is_default": False,
                        },
                        {
                            "id": "inactive-alpha",
                            "display_name": "Alpha",
                            "type": "KnowledgeQA",
                            "status": "inactive",
                            "is_default": False,
                        },
                        {
                            "id": "disabled-beta",
                            "display_name": "Beta",
                            "type": "KnowledgeQA",
                            "status": "active",
                            "is_default": False,
                        },
                    ]
                },
            )
        attempts.append(request.url.path)
        ok = request.url.path.endswith("/healthy-zulu/debug")
        return httpx.Response(
            200,
            json={"data": {"ok": ok, "error": "Model disabled." if not ok else ""}},
        )

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

    assert model_id == "healthy-zulu"
    assert attempts == [
        "/api/v1/models/disabled-beta/debug",
        "/api/v1/models/healthy-zulu/debug",
    ]


@pytest.mark.asyncio
async def test_weknora_client_rejects_wiki_creation_when_no_knowledge_qa_model_is_usable() -> None:
    knowledge_base_creates = 0

    async def handler(request: httpx.Request) -> httpx.Response:
        nonlocal knowledge_base_creates
        if request.method == "GET" and request.url.path == "/api/v1/models":
            return httpx.Response(
                200,
                json={
                    "data": [
                        {"id": "embed-one", "type": "Embedding", "is_default": True},
                        {
                            "id": "disabled-alpha",
                            "display_name": "Alpha disabled",
                            "type": "KnowledgeQA",
                            "status": "active",
                            "is_default": False,
                        },
                        {
                            "id": "broken-zulu",
                            "display_name": "Zulu broken",
                            "type": "KnowledgeQA",
                            "status": "active",
                            "is_default": False,
                        },
                    ]
                },
            )
        if request.url.path == "/api/v1/models/disabled-alpha/debug":
            return httpx.Response(200, json={"data": {"ok": False, "error": "Model disabled."}})
        if request.url.path == "/api/v1/models/broken-zulu/debug":
            return httpx.Response(200, json={"data": {"ok": False, "error": "Provider rejected the model."}})
        knowledge_base_creates += 1
        return httpx.Response(201, json={"data": {"id": "must-not-be-created"}})

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

    assert raised.value.status_code == 400
    assert "Zulu broken, Alpha disabled" in str(raised.value)
    assert "working default model" in str(raised.value)
    assert knowledge_base_creates == 0


@pytest.mark.asyncio
async def test_weknora_client_preserves_model_debug_service_failures() -> None:
    attempts: list[str] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        if request.method == "GET" and request.url.path == "/api/v1/models":
            return httpx.Response(
                200,
                json={
                    "data": [
                        {
                            "id": "alpha",
                            "name": "Alpha",
                            "type": "KnowledgeQA",
                            "status": "active",
                        },
                        {
                            "id": "zulu",
                            "name": "Zulu",
                            "type": "KnowledgeQA",
                            "status": "active",
                        },
                    ]
                },
            )
        attempts.append(request.url.path)
        return httpx.Response(503, json={"message": "model debug service unavailable"})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
        with pytest.raises(WeKnoraError) as raised:
            await client.get_default_knowledge_qa_model_id()

    assert raised.value.status_code == 502
    assert str(raised.value) == "model debug service unavailable"
    assert attempts == ["/api/v1/models/alpha/debug"]


@pytest.mark.asyncio
async def test_weknora_client_ignores_wiki_instructions_when_wiki_is_disabled() -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        if request.method == "GET":
            return httpx.Response(
                200,
                json={"data": [{"id": "embed-one", "type": "Embedding", "is_default": True}]},
            )
        payload = json.loads(request.content)
        assert payload["embedding_model_id"] == "embed-one"
        assert "wiki_config" not in payload
        assert "summary_model_id" not in payload
        return httpx.Response(201, json={"data": {"id": "docs-one", "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",
            wiki_enabled=False,
            wiki_content_instructions="must not be forwarded",
            wiki_extraction_instructions="must not be forwarded",
        )

    assert created["id"] == "docs-one"


@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_uses_native_wiki_regex_search_contract() -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        assert request.method == "GET"
        assert request.url.path == "/api/v1/knowledgebase/kb-one/wiki/search"
        assert request.url.params["q"] == "火炬|开幕|闭幕"
        # WeKnora v0.7.2 caps the endpoint at 50.
        assert request.url.params["limit"] == "50"
        return httpx.Response(
            200,
            json={
                "pages": [
                    {
                        "slug": "games/timeline",
                        "title": "赛事时间线",
                        "content": "火炬传递、开幕和闭幕安排。",
                    }
                ]
            },
        )

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
        pages = await client.search_wiki_pages(
            "kb-one",
            "火炬|开幕|闭幕",
            limit=500,
        )

    assert pages == [
        {
            "slug": "games/timeline",
            "title": "赛事时间线",
            "content": "火炬传递、开幕和闭幕安排。",
        }
    ]


@pytest.mark.asyncio
async def test_weknora_client_reads_ordered_wiki_index_groups() -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/api/v1/knowledgebase/kb-one/wiki/index"
        assert request.url.params["types"] == "entity,concept"
        assert request.url.params["limit"] == "50"
        assert request.url.params["cursor"] == "next-page"
        return httpx.Response(
            200,
            json={
                "data": {
                    "intro": "# Wiki Index",
                    "version": 2,
                    "groups": [
                        {
                            "type": "entity",
                            "total": 1,
                            "items": [{"slug": "entity/one", "title": "实体一"}],
                        }
                    ],
                }
            },
        )

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
        index = await client.get_wiki_index(
            "kb-one",
            types=["entity", "concept"],
            limit=50,
            cursor="next-page",
        )

    assert index["groups"][0]["items"][0]["slug"] == "entity/one"


@pytest.mark.asyncio
async def test_weknora_client_creates_updates_and_deletes_wiki_pages() -> None:
    seen: list[tuple[str, str]] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        seen.append((request.method, request.url.path))
        if request.method == "POST":
            assert request.url.path == "/api/v1/knowledgebase/kb-one/wiki/pages"
            assert json.loads(request.content) == {"slug": "folder/guide", "title": "Guide", "content": "# Guide"}
            return httpx.Response(200, json={"success": True, "data": {"slug": "folder/guide", "title": "Guide"}})
        assert request.url.path == "/api/v1/knowledgebase/kb-one/wiki/pages/folder/guide"
        if request.method == "PUT":
            assert json.loads(request.content) == {"title": "Guide", "content": "# Guide"}
            return httpx.Response(200, json={"success": True, "data": {"slug": "folder/guide", "title": "Guide"}})
        if request.method == "DELETE":
            return httpx.Response(204)
        return httpx.Response(405)

    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_wiki_page("kb-one", {"slug": "folder/guide", "title": "Guide", "content": "# Guide"})
        updated = await client.update_wiki_page("kb-one", "folder/guide", {"title": "Guide", "content": "# Guide"})
        await client.delete_wiki_page("kb-one", "folder/guide")

    assert created["slug"] == "folder/guide"
    assert updated["slug"] == "folder/guide"
    assert seen == [
        ("POST", "/api/v1/knowledgebase/kb-one/wiki/pages"),
        ("PUT", "/api/v1/knowledgebase/kb-one/wiki/pages/folder/guide"),
        ("DELETE", "/api/v1/knowledgebase/kb-one/wiki/pages/folder/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_weknora_client_fetches_protected_wiki_image() -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/api/v1/knowledge-bases/kb-one/files"
        assert request.url.params["file_path"] == "resource://wiki/figure.png"
        return httpx.Response(200, content=b"png-bytes", headers={"Content-Type": "image/png"})

    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_knowledge_base_file(
            "kb-one",
            "resource://wiki/figure.png",
        )

    assert content == b"png-bytes"
    assert media_type == "image/png"


@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, *, refresh: bool = False) -> 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, *, refresh: bool = False) -> 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, *, refresh: bool = False) -> 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",
        refresh=True,
    )

    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_agent_bound_knowledge_base_list_uses_local_mapping_only(monkeypatch: pytest.MonkeyPatch) -> None:
    from types import SimpleNamespace

    from app.gateway.routers import llmwiki as llmwiki_router
    from deerflow.persistence.agents import MemoryAgentStore
    from deerflow.persistence.llmwiki import MemoryLlmWikiStore

    store = MemoryLlmWikiStore()
    mapping = await store.create_mapping(
        weknora_id="remote-agent-kb",
        owner_user_id="user-a",
        name="本地目录缓存",
        description="",
        kb_type="document",
    )
    agent_store = MemoryAgentStore()
    await agent_store.ensure_builtin({"id": "agent-a", "name": "Agent A", "published": True})

    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, "get_agent_store", lambda _: agent_store)
    monkeypatch.setattr(
        llmwiki_router,
        "load_agent_config",
        lambda _agent_id: SimpleNamespace(llmwiki_knowledge_base_ids=[mapping["id"]]),
    )
    monkeypatch.setattr(
        llmwiki_router,
        "_client_or_503",
        lambda: (_ for _ in ()).throw(AssertionError("normal listing must not call WeKnora")),
    )

    payload = await llmwiki_router.list_agent_knowledge_bases(object(), "agent-a")
    assert [item["id"] for item in payload["knowledge_bases"]] == [mapping["id"]]
    assert payload["knowledge_bases"][0]["remote_status"] == "cached"


@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, *, refresh: bool = False) -> 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}"
    assert _first_gateway_match("GET", "/deerflow/platform/knowledge-bases/remote-kb") == "/deerflow/platform/{proxied_path:path}"
    assert _first_gateway_match("GET", "/deerflow/assets/index.js") == "/deerflow/assets/{proxied_path:path}"
    assert _first_gateway_match("GET", "/deerflow/js/index.js") == "/deerflow/{proxied_path:path}"
    assert _first_gateway_match("GET", "/deerflow/css/index.css") == "/deerflow/{proxied_path:path}"
    assert _first_gateway_match("GET", "/deerflow/fonts/index.woff2") == "/deerflow/{proxied_path:path}"
    assert _first_gateway_match("GET", "/deerflow/api/v1/knowledge-search") == "/deerflow/api/v1/{proxied_path:path}"
    assert _first_gateway_match("GET", "/js/index.js") == "/{proxied_path:path}"


@pytest.mark.parametrize(
    ("location", "target_base_url", "expected"),
    [
        ("/platform/knowledge-bases/kb?tab=wiki", "http://weknora-web:8080", "/deerflow/platform/knowledge-bases/kb?tab=wiki"),
        ("/deerflow/platform/knowledge-bases/kb", "http://weknora-web:8080", "/deerflow/platform/knowledge-bases/kb"),
        ("/deerflow-api/api/v1/auth/me", "http://gateway/deerflow-api", "/deerflow/api/v1/auth/me"),
        (
            "http://gateway/deerflow-api/api/v1/auth/me?next=1#top",
            "http://gateway/deerflow-api",
            "/deerflow/api/v1/auth/me?next=1#top",
        ),
        ("https://external.test/login", "http://gateway/deerflow-api", "https://external.test/login"),
    ],
)
def test_weknora_redirect_location_is_normalized_for_public_prefix(
    location: str,
    target_base_url: str,
    expected: str,
) -> None:
    from app.gateway.routers import llmwiki as llmwiki_router

    assert llmwiki_router._rewrite_weknora_location_header(location, "/deerflow", target_base_url) == expected


@pytest.mark.asyncio
async def test_prefixed_weknora_proxy_serves_page_static_assets_and_api_without_auth_errors(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway.auth_middleware import AuthMiddleware
    from app.gateway.csrf_middleware import CSRFMiddleware
    from app.gateway.routers import llmwiki as llmwiki_router
    from app.gateway.weknora_embed import WEKNORA_EMBED_COOKIE, sign_weknora_embed_payload

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

    runtime = SimpleNamespace(
        web_base_url="http://weknora-web.test",
        api_base_url="http://weknora-api.test",
    )
    upstream_requests: list[tuple[str, str, dict[str, str]]] = []

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

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

        async def __aenter__(self) -> FakeUpstreamClient:
            return self

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

        async def request(
            self,
            method: str,
            url: str,
            *,
            headers: dict[str, str],
            **__: object,
        ) -> httpx.Response:
            upstream_requests.append((method, url, headers))
            path = httpx.URL(url).path
            if url.startswith(runtime.api_base_url):
                return httpx.Response(200, json={"success": True, "data": {"id": "admin"}})
            if path == "/platform/knowledge-bases/remote-kb":
                return httpx.Response(
                    200,
                    text=('<html><head><script src="/assets/main.js"></script><link rel="stylesheet" href="css/app.css"><script src="/config.js"></script><script src="runtime-config.js"></script></head><body></body></html>'),
                    headers={"content-type": "text/html; charset=utf-8"},
                )
            if path == "/assets/main.js":
                return httpx.Response(
                    200,
                    text=('const __vite__mapDeps=(i,m=__vite__mapDeps,d=(m.f||(m.f=["assets/lazy.css","assets/lazy.js"])))=>i.map(i=>d[i]);const chunk="/static/chunk.js",matcher=/platform\\//;fetch("/api/v1/auth/me")'),
                    headers={"content-type": "application/javascript"},
                )
            if path == "/css/app.css":
                return httpx.Response(
                    200,
                    text='@font-face{src:url("/fonts/app.woff2")}',
                    headers={"content-type": "text/css"},
                )
            if path in {"/config.js", "/runtime-config.js", "/static/chunk.js", "/assets/deep.js"}:
                return httpx.Response(200, text=f'window.loaded="{path}"', headers={"content-type": "application/javascript"})
            if path == "/fonts/app.woff2":
                return httpx.Response(200, content=b"font", headers={"content-type": "font/woff2"})
            return httpx.Response(404, text=f"missing upstream path: {path}")

    monkeypatch.setattr(llmwiki_router, "_runtime_for_iframe_proxy", lambda _: runtime)
    monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store())
    monkeypatch.setattr(llmwiki_router, "_get_weknora_admin_session", fake_session)

    real_async_client = httpx.AsyncClient
    monkeypatch.setattr(llmwiki_router.httpx, "AsyncClient", FakeUpstreamClient)

    test_app = FastAPI()
    test_app.add_middleware(AuthMiddleware)
    test_app.add_middleware(CSRFMiddleware)
    test_app.include_router(llmwiki_router.router)
    test_app.include_router(llmwiki_router.proxy_router, prefix="/deerflow")
    test_app.include_router(llmwiki_router.proxy_router)

    cookie = sign_weknora_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,
        }
    )
    transport = httpx.ASGITransport(app=test_app)
    async with real_async_client(transport=transport, base_url="http://deerflow.test") as client:
        client.cookies.set(WEKNORA_EMBED_COOKIE, cookie)

        page = await client.get("/deerflow/platform/knowledge-bases/remote-kb?tab=wiki")
        assert page.status_code == 200
        assert 'src="/deerflow/assets/main.js"' in page.text
        assert 'href="/deerflow/css/app.css"' in page.text
        assert 'src="/deerflow/config.js"' in page.text
        assert 'src="/deerflow/runtime-config.js"' in page.text

        # The release Nginx strips /deerflow-api before forwarding and tells
        # the Gateway which public prefix to put back into iframe asset URLs.
        release_page = await client.get(
            "/platform/knowledge-bases/remote-kb?tab=wiki",
            headers={"X-Forwarded-Prefix": "/deerflow-api"},
        )
        assert release_page.status_code == 200
        assert 'src="/deerflow-api/assets/main.js"' in release_page.text
        assert 'href="/deerflow-api/css/app.css"' in release_page.text
        assert '"apiPrefix":"/deerflow-api/api/llmwiki/weknora-embed"' in release_page.text

        script = await client.get("/deerflow/assets/main.js")
        assert script.status_code == 200
        assert '"/deerflow/static/chunk.js"' in script.text
        assert '"deerflow/assets/lazy.css"' in script.text
        assert '"deerflow/assets/lazy.js"' in script.text
        assert '"/deerflow/assets/lazy.css"' not in script.text
        assert "matcher=/platform\\//" in script.text
        assert 'fetch("/deerflow/api/v1/auth/me")' in script.text

        stylesheet = await client.get("/deerflow/css/app.css")
        assert stylesheet.status_code == 200
        assert 'url("/deerflow/fonts/app.woff2")' in stylesheet.text

        for path in (
            "/deerflow/config.js",
            "/deerflow/runtime-config.js",
            "/deerflow/platform/knowledge-bases/runtime-config.js",
            "/deerflow/static/chunk.js",
            "/deerflow/fonts/app.woff2",
            "/deerflow/platform/knowledge-bases/remote-kb/assets/deep.js",
        ):
            response = await client.get(path)
            assert response.status_code == 200, f"{path}: {response.status_code} {response.text}"

        api = await client.get("/deerflow/api/llmwiki/weknora-embed/api/v1/auth/me")
        assert api.status_code == 200

    api_request = next(request for request in upstream_requests if request[1].startswith(runtime.api_base_url))
    assert api_request[2]["Authorization"] == "Bearer admin-token"
    assert api_request[2]["X-Tenant-ID"] == "tenant-1"


def test_weknora_iframe_hides_current_outer_sidebar_shell() -> 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,
        allowed_weknora_ids=("remote-kb", "other-visible-kb"),
    )
    injection = llmwiki_router._weknora_embed_injection(
        allowed_path="/platform/knowledge-bases/remote-kb",
        tab="wiki",
        session={},
        ctx=ctx,
    )

    assert "#app > .main > .aside_box" in injection
    assert '".main > .aside_box"' in injection


def test_weknora_iframe_html_rewrites_public_prefix_paths() -> 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,
        allowed_weknora_ids=("remote-kb", "other-visible-kb"),
    )
    html = b'<html><head><script src="/assets/index.js"></script></head><body>fetch("/api/v1/auth/me")</body></html>'

    injected = llmwiki_router._inject_weknora_embed_html(
        html,
        allowed_path="/deerflow/platform/knowledge-bases/remote-kb",
        tab="wiki",
        session={},
        ctx=ctx,
        public_prefix="/deerflow",
    ).decode()

    assert 'src="/deerflow/assets/index.js"' in injected
    assert 'fetch("/deerflow/api/v1/auth/me")' in injected
    assert '"apiPrefix":"/deerflow/api/llmwiki/weknora-embed"' in injected
    assert '"allowedPath":"/deerflow/platform/knowledge-bases/remote-kb"' in injected
    assert '"allowedWeKnoraIds":["remote-kb","other-visible-kb"]' in injected
    assert ".breadcrumb-link.dropdown" not in injected
    assert "reloadForKnowledgeBaseSwitch" in injected
    assert "function protectedImageApiUrl" in injected
    assert "img[data-protected-src]" in injected
    assert "encodeURIComponent(cfg.weknoraId)" in injected
    assert "image.src = proxyUrl;" in injected
    assert "URL.createObjectURL(blob)" not in injected
    assert 'attributeFilter: ["data-protected-src", "src"]' in injected


def test_weknora_iframe_filters_kb_list_to_signed_visible_mappings() -> 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,
        allowed_weknora_ids=("remote-kb", "other-visible-kb"),
    )
    response = httpx.Response(
        200,
        json={
            "success": True,
            "data": [
                {"id": "remote-kb", "name": "Allowed"},
                {"id": "other-visible-kb", "name": "Also 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"},
        {"id": "other-visible-kb", "name": "Also allowed"},
    ]


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

    class Store:
        async def get_by_weknora_id(self, weknora_id: str) -> dict[str, str] | None:
            if weknora_id == "other-visible-kb":
                return {"id": "public-kb", "weknora_id": weknora_id, "owner_user_id": "owner-b"}
            return None

        async def get_authorized(self, mapping_id: str, _: str, *, write: bool, **__: object) -> dict[str, str] | None:
            if mapping_id != "public-kb":
                return None
            if write:
                return None
            return {"id": "public-kb", "weknora_id": "other-visible-kb", "owner_user_id": "owner-b"}

    monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store())
    original = llmwiki_router.WeKnoraEmbedContext(
        mapping_id="local-kb",
        weknora_id="remote-kb",
        user_id="user-a",
        is_admin=False,
        can_write=True,
        exp=4_102_444_800,
        allowed_weknora_ids=("remote-kb", "other-visible-kb"),
    )

    selected = await llmwiki_router._switch_embed_context(
        _request("/platform/knowledge-bases/other-visible-kb"),
        original,
        "other-visible-kb",
    )

    assert selected.mapping_id == "public-kb"
    assert selected.weknora_id == "other-visible-kb"
    assert selected.can_write is False


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

    current = 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,
        allowed_weknora_ids=("remote-kb", "other-visible-kb"),
    )
    selected = llmwiki_router.WeKnoraEmbedContext(
        mapping_id="public-kb",
        weknora_id="other-visible-kb",
        user_id="user-a",
        is_admin=False,
        can_write=False,
        exp=4_102_444_800,
        allowed_weknora_ids=current.allowed_weknora_ids,
    )
    seen: dict[str, object] = {}

    async def read_context(_: Request) -> object:
        return current

    async def switch_context(_: Request, ctx: object, weknora_id: str) -> object:
        seen["current"] = ctx
        seen["target"] = weknora_id
        return selected

    async def proxy_request(*_: object, **kwargs: object) -> Response:
        seen.update(kwargs)
        return Response("ok")

    monkeypatch.setattr(llmwiki_router, "_read_embed_context", read_context)
    monkeypatch.setattr(llmwiki_router, "_switch_embed_context", switch_context)
    monkeypatch.setattr(
        llmwiki_router,
        "_runtime_for_iframe_proxy",
        lambda _: SimpleNamespace(web_base_url="http://weknora-web"),
    )
    monkeypatch.setattr(llmwiki_router, "_proxy_weknora_request", proxy_request)

    response = await llmwiki_router.proxy_weknora_platform_page(
        _request("/platform/knowledge-bases/other-visible-kb", request_headers={"X-Forwarded-Proto": "https"}),
        "knowledge-bases/other-visible-kb",
    )

    assert seen["current"] == current
    assert seen["target"] == "other-visible-kb"
    assert seen["upstream_path"] == "/platform/knowledge-bases/other-visible-kb"
    raw_cookie = response.headers["set-cookie"].split("deerflow_weknora_embed=", 1)[1].split(";", 1)[0]
    switched_context = llmwiki_router._verify_embed_cookie(raw_cookie)
    assert switched_context.mapping_id == "public-kb"
    assert switched_context.weknora_id == "other-visible-kb"


@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
@pytest.mark.parametrize("filename", ["wiki-vector.jsonl", "renamed-package.txt"])
async def test_compatible_package_is_not_uploaded_as_an_ordinary_document(
    monkeypatch: pytest.MonkeyPatch,
    filename: str,
) -> None:
    from app.gateway.routers import llmwiki as llmwiki_router

    forwarded: list[dict[str, object]] = []
    syncs: list[dict[str, object]] = []

    class Client:
        async def upload_document(self, knowledge_base_id: str, **kwargs: object) -> dict[str, object]:
            forwarded.append({"knowledge_base_id": knowledge_base_id, **kwargs})
            return {"id": "unexpected-document"}

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

    app = FastAPI()
    request = Request(
        {
            "type": "http",
            "method": "POST",
            "path": "/api/llmwiki/knowledge-bases/local-kb/documents",
            "headers": [],
            "app": app,
        }
    )
    upload = UploadFile(
        filename=filename,
        file=BytesIO(_compatible_knowledge_package_bytes(bom=True)),
        headers=Headers({"content-type": "text/plain"}),
    )
    monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized_mapping)
    monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client())
    monkeypatch.setattr(llmwiki_router, "_schedule_local_wiki_sync", lambda _request, row: syncs.append(row))

    with pytest.raises(HTTPException) as exc_info:
        await llmwiki_router.upload_document(request, "local-kb", upload)

    assert exc_info.value.status_code == 422
    assert "本次文件未送入模型" in str(exc_info.value.detail)
    assert forwarded == []
    assert syncs == []


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

    content = '{"question":"如何散热？","answer":"保持风道畅通。"}\n'.encode()
    forwarded: list[dict[str, object]] = []
    syncs: list[dict[str, object]] = []

    class Client:
        async def upload_document(self, knowledge_base_id: str, **kwargs: object) -> dict[str, object]:
            forwarded.append({"knowledge_base_id": knowledge_base_id, **kwargs})
            return {"id": "document-one"}

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

    app = FastAPI()
    request = Request(
        {
            "type": "http",
            "method": "POST",
            "path": "/api/llmwiki/knowledge-bases/local-kb/documents",
            "headers": [],
            "app": app,
        }
    )
    upload = UploadFile(
        filename="faq.jsonl",
        file=BytesIO(content),
        headers=Headers({"content-type": "application/x-ndjson"}),
    )
    monkeypatch.setattr(llmwiki_router, "_authorized_mapping", authorized_mapping)
    monkeypatch.setattr(llmwiki_router, "_client_or_503", lambda: Client())
    monkeypatch.setattr(llmwiki_router, "_schedule_local_wiki_sync", lambda _request, row: syncs.append(row))

    result = await llmwiki_router.upload_document(request, "local-kb", upload)

    assert result == {"id": "document-one"}
    assert forwarded == [
        {
            "knowledge_base_id": "remote-kb",
            "filename": "faq.jsonl",
            "content": content,
            "content_type": "application/x-ndjson",
        }
    ]
    assert syncs == [{"id": "local-kb", "weknora_id": "remote-kb"}]


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

    boundary = "deerflow-test-boundary"
    body = (
        (f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="wiki-vector.jsonl"\r\nContent-Type: application/x-ndjson\r\n\r\n').encode()
        + _compatible_knowledge_package_bytes(bom=True)
        + f"\r\n--{boundary}--\r\n".encode()
    )

    class FakeRequest:
        method = "POST"
        headers = Headers({"content-type": f"multipart/form-data; boundary={boundary}"})
        query_params = QueryParams("")

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

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

    assert exc_info.value.status_code == 422
    assert "本次文件未送入模型" in str(exc_info.value.detail)


@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",
    )

    await llmwiki_router._authorize_weknora_api_proxy(
        FakeRequest(),
        ctx,
        object(),
        "/api/v1/me/invitations/pending-count",
        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


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

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

        def __init__(self, method: str) -> None:
            self.method = method

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

    await llmwiki_router._authorize_weknora_api_proxy(
        FakeRequest("GET"),
        ctx,
        object(),
        "/api/v1/tenants/kv/retrieval-config",
        b"",
        "token",
    )

    for method, path in (
        ("GET", "/api/v1/tenants/kv/other-setting"),
        ("PUT", "/api/v1/tenants/kv/retrieval-config"),
    ):
        with pytest.raises(HTTPException) as exc_info:
            await llmwiki_router._authorize_weknora_api_proxy(
                FakeRequest(method),
                ctx,
                object(),
                path,
                b"{}",
                "token",
            )

        assert exc_info.value.status_code == 403


@pytest.mark.asyncio
async def test_incremental_sync_endpoint_queues_owner_authorized_non_force_job(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway.routers import llmwiki_index as index_router

    mapping = {
        "id": "local-kb",
        "weknora_id": "remote-kb",
        "name": "KB",
        "wiki_index_enabled": True,
    }
    authorization_calls: list[dict[str, object]] = []
    sync_calls: list[dict[str, object]] = []
    queued_coroutines: list[object] = []

    class Store:
        async def get_authorized(
            self,
            mapping_id: str,
            user_id: str,
            *,
            write: bool,
            is_admin: bool,
        ) -> dict[str, object] | None:
            authorization_calls.append(
                {
                    "mapping_id": mapping_id,
                    "user_id": user_id,
                    "write": write,
                    "is_admin": is_admin,
                }
            )
            return mapping

    class Service:
        embedding = SimpleNamespace(dimensions=3)

        async def sync_mapping(
            self,
            selected_mapping: dict[str, object],
            *,
            force: bool,
            sync_id: str | None = None,
        ) -> None:
            sync_calls.append(
                {
                    "mapping": selected_mapping,
                    "force": force,
                    "sync_id": sync_id,
                }
            )

    async def runtime_config(_: object) -> object:
        return SimpleNamespace(enabled=True)

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

    def queue(_: FastAPI, coroutine: object) -> None:
        queued_coroutines.append(coroutine)

    app = FastAPI()
    app.state.llmwiki_store = Store()
    app.state.llmwiki_sync_service = Service()
    request = _request("/api/llmwiki/knowledge-bases/local-kb/sync")
    request.scope["app"] = app
    monkeypatch.setattr(index_router, "rebuild_local_wiki_index_runtime", runtime_config)
    monkeypatch.setattr(index_router, "_actor", actor)
    monkeypatch.setattr(index_router, "spawn_local_wiki_index_job", queue)

    response = await index_router.sync_knowledge_base_incrementally(request, "local-kb")

    assert response["status"] == "accepted"
    assert response["mode"] == "incremental"
    assert response["knowledge_base_id"] == "local-kb"
    assert response["status_url"].endswith("/local-kb/wiki-index-status")
    assert authorization_calls == [
        {
            "mapping_id": "local-kb",
            "user_id": "owner-a",
            "write": True,
            "is_admin": False,
        }
    ]
    assert len(queued_coroutines) == 1
    await queued_coroutines.pop()
    assert sync_calls == [
        {
            "mapping": mapping,
            "force": False,
            "sync_id": response["job_id"],
        }
    ]
    route = next(
        route
        for route in index_router.router.routes
        if route.path == "/api/llmwiki/knowledge-bases/{mapping_id}/sync"
    )
    assert route.status_code == 202


@pytest.mark.asyncio
async def test_incremental_sync_endpoint_rejects_before_encoder_probe(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from app.gateway.routers import llmwiki_index as index_router

    class Store:
        async def get_authorized(self, *_: object, **__: object) -> None:
            return None

    class Embedding:
        dimensions = None

        async def embed_query(self, _: str) -> list[float]:
            raise AssertionError("unauthorized requests must not probe the encoder")

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

    async def unexpected_runtime(_: object) -> object:
        raise AssertionError("authorization must precede runtime initialization")

    app = FastAPI()
    app.state.llmwiki_store = Store()
    app.state.llmwiki_sync_service = SimpleNamespace(embedding=Embedding())
    request = _request("/api/llmwiki/knowledge-bases/private-kb/sync")
    request.scope["app"] = app
    monkeypatch.setattr(index_router, "_actor", actor)
    monkeypatch.setattr(
        index_router,
        "rebuild_local_wiki_index_runtime",
        unexpected_runtime,
    )

    with pytest.raises(HTTPException) as exc_info:
        await index_router.sync_knowledge_base_incrementally(request, "private-kb")

    assert exc_info.value.status_code == 404


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

    mapping = {"id": "local-kb", "weknora_id": "remote-kb", "name": "KB"}
    scheduled: list[dict[str, object]] = []

    class Store:
        async def get_authorized(
            self,
            mapping_id: str,
            user_id: str,
            *,
            write: bool,
            is_admin: bool,
        ) -> dict[str, object] | None:
            assert (mapping_id, user_id, write, is_admin) == (
                "local-kb",
                "owner-a",
                False,
                False,
            )
            return mapping

    monkeypatch.setattr(llmwiki_router, "_store", lambda _: Store())
    monkeypatch.setattr(
        llmwiki_router,
        "_schedule_local_wiki_sync",
        lambda _request_value, value: scheduled.append(value),
    )
    context = llmwiki_router.WeKnoraEmbedContext(
        mapping_id="local-kb",
        weknora_id="remote-kb",
        user_id="owner-a",
        is_admin=False,
        can_write=True,
        exp=4_102_444_800,
    )

    await llmwiki_router._schedule_proxy_local_wiki_sync(
        _request("/api/v1/knowledge-bases/remote-kb"),
        context,
    )

    assert scheduled == [mapping]


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("method", "upstream_status", "expected_schedules"),
    [("POST", 201, 1), ("POST", 422, 0), ("GET", 200, 0)],
)
async def test_proxy_only_schedules_after_successful_content_mutation(
    monkeypatch: pytest.MonkeyPatch,
    method: str,
    upstream_status: int,
    expected_schedules: int,
) -> None:
    from app.gateway.routers import llmwiki as llmwiki_router

    scheduled: list[object] = []

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

        def __init__(self) -> None:
            self.method = method

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

    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(upstream_status, json={"ok": upstream_status < 400})

    async def session(*_: object, **__: object) -> dict[str, object]:
        return {"token": "token", "tenant": {}}

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

    async def schedule(*args: object) -> None:
        scheduled.append(args)

    context = llmwiki_router.WeKnoraEmbedContext(
        mapping_id="local-kb",
        weknora_id="remote-kb",
        user_id="owner-a",
        is_admin=False,
        can_write=True,
        exp=4_102_444_800,
    )
    monkeypatch.setattr(llmwiki_router, "_runtime_for_iframe_proxy", lambda _: object())
    monkeypatch.setattr(llmwiki_router, "_get_weknora_admin_session", session)
    monkeypatch.setattr(llmwiki_router, "_authorize_weknora_api_proxy", authorize)
    monkeypatch.setattr(llmwiki_router, "_schedule_proxy_local_wiki_sync", schedule)
    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-kb/wiki-pages",
        ctx=context,
        is_api=True,
    )

    assert response.status_code == upstream_status
    assert len(scheduled) == expected_schedules