knowledge-adapter-service/tests/test_weknora_client.py

from uuid import UUID

import httpx
import pytest
from pydantic import SecretStr

from knowledge_adapter.weknora_client import (
    WeKnoraBadResponseError,
    WeKnoraClient,
    WeKnoraUnavailableError,
)

KB_ID = UUID("3904b40e-278b-4273-a6fa-db6cbc975587")
WEKNORA_KEY = "test-weknora-key"


async def test_search_uses_only_readonly_endpoint_and_expected_contract() -> None:
    requests: list[httpx.Request] = []

    def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        return httpx.Response(200, json={"success": True, "data": []})

    client = WeKnoraClient(
        base_url="http://weknora.internal:8080",
        api_key=SecretStr(WEKNORA_KEY),
        timeout_seconds=15,
        transport=httpx.MockTransport(handler),
    )
    async with client:
        results = await client.search(query="normalized query", knowledge_base_id=KB_ID)

    assert results == []
    assert len(requests) == 1
    request = requests[0]
    assert request.method == "POST"
    assert request.url.path == "/api/v1/knowledge-search"
    assert request.headers["X-API-Key"] == WEKNORA_KEY
    assert request.headers.get("Authorization") is None
    assert request.read().decode() == (
        '{"query":"normalized query",'
        '"knowledge_base_ids":["3904b40e-278b-4273-a6fa-db6cbc975587"]}'
    )


async def test_timeout_becomes_sanitized_unavailable_error() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ReadTimeout("internal timeout details", request=request)

    client = WeKnoraClient(
        base_url="http://weknora.internal:8080",
        api_key=SecretStr(WEKNORA_KEY),
        timeout_seconds=15,
        transport=httpx.MockTransport(handler),
    )

    with pytest.raises(WeKnoraUnavailableError) as captured:
        async with client:
            await client.search(query="sensitive query", knowledge_base_id=KB_ID)

    assert str(captured.value) == "WeKnora unavailable"


async def test_500_becomes_sanitized_bad_response_error() -> None:
    def handler(_request: httpx.Request) -> httpx.Response:
        return httpx.Response(500, text="sensitive upstream failure")

    client = WeKnoraClient(
        base_url="http://weknora.internal:8080",
        api_key=SecretStr(WEKNORA_KEY),
        timeout_seconds=15,
        transport=httpx.MockTransport(handler),
    )

    with pytest.raises(WeKnoraBadResponseError) as captured:
        async with client:
            await client.search(query="sensitive query", knowledge_base_id=KB_ID)

    assert str(captured.value) == "invalid WeKnora response"


async def test_non_json_200_becomes_sanitized_bad_response_error() -> None:
    def handler(_request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, text="not-json sensitive body")

    client = WeKnoraClient(
        base_url="http://weknora.internal:8080",
        api_key=SecretStr(WEKNORA_KEY),
        timeout_seconds=15,
        transport=httpx.MockTransport(handler),
    )

    with pytest.raises(WeKnoraBadResponseError) as captured:
        async with client:
            await client.search(query="sensitive query", knowledge_base_id=KB_ID)

    assert str(captured.value) == "invalid WeKnora response"


async def test_success_false_becomes_bad_response_error() -> None:
    def handler(_request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"success": False, "data": []})

    client = WeKnoraClient(
        base_url="http://weknora.internal:8080",
        api_key=SecretStr(WEKNORA_KEY),
        timeout_seconds=15,
        transport=httpx.MockTransport(handler),
    )

    with pytest.raises(WeKnoraBadResponseError):
        async with client:
            await client.search(query="query", knowledge_base_id=KB_ID)