knowledge-adapter-service/tests/test_asset_route.py

from uuid import UUID

import httpx

from knowledge_adapter.app import create_app
from knowledge_adapter.asset_redirect import (
    AssetNotFoundError,
    AssetRedirectUnavailableError,
)

ASSET_ID = UUID("c7f61f89-19ce-45b1-ac60-dbeed5e2a491")
SIGNED_URL = (
    "https://private-assets.oss-cn-shenzhen.aliyuncs.com/"
    "weknora/10008/assets/image.png?x-oss-signature=redacted"
)


class RecordingRedirectService:
    def __init__(
        self,
        url: str = SIGNED_URL,
        *,
        error: Exception | None = None,
    ) -> None:
        self.url = url
        self.error = error
        self.calls: list[UUID] = []

    async def get_redirect_url(self, asset_id: UUID) -> str:
        self.calls.append(asset_id)
        if self.error is not None:
            raise self.error
        return self.url


async def test_asset_route_returns_anonymous_302() -> None:
    service = RecordingRedirectService()
    app = create_app(asset_redirect_service=service)
    transport = httpx.ASGITransport(app=app)

    async with httpx.AsyncClient(
        transport=transport,
        base_url="http://test",
        follow_redirects=False,
    ) as client:
        response = await client.get(f"/i/{ASSET_ID}")

    assert response.status_code == 302
    assert response.headers["location"] == SIGNED_URL
    assert response.headers["cache-control"] == "no-store"
    assert response.headers["referrer-policy"] == "no-referrer"
    assert response.content == b""
    assert service.calls == [ASSET_ID]


async def test_asset_route_returns_empty_404_for_unknown_asset() -> None:
    service = RecordingRedirectService(error=AssetNotFoundError("asset not found"))
    app = create_app(asset_redirect_service=service)
    transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)

    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get(f"/i/{ASSET_ID}")

    assert response.status_code == 404
    assert response.content == b""
    assert "asset" not in response.text.lower()
    assert str(ASSET_ID) not in response.text


async def test_asset_route_rejects_malformed_uuid_before_service() -> None:
    service = RecordingRedirectService()
    app = create_app(asset_redirect_service=service)
    transport = httpx.ASGITransport(app=app)

    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/i/not-a-uuid")

    assert response.status_code == 404
    assert response.content == b""
    assert service.calls == []


async def test_asset_route_returns_empty_503_when_redirect_is_unavailable() -> None:
    service = RecordingRedirectService(
        error=AssetRedirectUnavailableError(
            "metadata http://100.100.100.200?token=secret object_key=private/image.png"
        )
    )
    app = create_app(asset_redirect_service=service)
    transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)

    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get(f"/i/{ASSET_ID}")

    assert response.status_code == 503
    assert response.content == b""
    assert "100.100.100.200" not in response.text
    assert "object_key" not in response.text
    assert "secret" not in response.text


async def test_asset_route_does_not_accept_bucket_object_key_or_url_overrides() -> None:
    service = RecordingRedirectService()
    app = create_app(asset_redirect_service=service)
    transport = httpx.ASGITransport(app=app)

    async with httpx.AsyncClient(
        transport=transport,
        base_url="http://test",
        follow_redirects=False,
    ) as client:
        response = await client.get(
            f"/i/{ASSET_ID}",
            params={
                "bucket": "attacker-bucket",
                "object_key": "other/secret.png",
                "url": "https://attacker.example/file",
            },
            headers={
                "X-OSS-Bucket": "attacker-bucket",
                "X-Object-Key": "other/secret.png",
            },
        )

    assert response.status_code == 302
    assert response.headers["location"] == SIGNED_URL
    assert service.calls == [ASSET_ID]