knowledge-adapter-service/tests/test_resource_locator.py

import pytest

from knowledge_adapter.resource_locator import (
    InvalidResourceLocator,
    InvalidResourceReference,
    parse_oss_object_key,
    parse_resource_reference,
)

VALID_HANDLE = "AbCdEfGhIjKlMnOpQrSt_1"


def test_parse_resource_reference_accepts_only_canonical_handle() -> None:
    assert parse_resource_reference(f"resource://{VALID_HANDLE}") == VALID_HANDLE

    invalid_values = [
        "resource://short",
        "resource://AbCdEfGhIjKlMnOpQrSt/1",
        "resource://AbCdEfGhIjKlMnOpQrSt%1",
        "https://example.com/image.png",
        "http://example.com/image.png",
    ]
    for value in invalid_values:
        with pytest.raises(InvalidResourceReference):
            parse_resource_reference(value)


def test_parse_oss_object_key_accepts_matching_wrapped_locator() -> None:
    object_key = parse_oss_object_key(
        provider="oss",
        physical_path=(
            "storage://backend-test/"
            "oss://private-assets/weknora/10008/assets/image.png"
        ),
        storage_backend_id="backend-test",
        expected_storage_backend_id="backend-test",
        expected_bucket="private-assets",
    )

    assert object_key == "weknora/10008/assets/image.png"


@pytest.mark.parametrize(
    ("provider", "physical_path", "storage_backend_id"),
    [
        (
            "oss",
            "storage://backend-test/oss://private-assets/../secret.png",
            "backend-test",
        ),
        (
            "oss",
            "storage://backend-test/oss://private-assets//absolute.png",
            "backend-test",
        ),
        (
            "oss",
            "storage://backend-test/oss://private-assets/folder\\image.png",
            "backend-test",
        ),
        (
            "oss",
            "storage://backend-test/oss://private-assets/image.png\x00ignored",
            "backend-test",
        ),
        (
            "oss",
            "storage://backend-test/oss://private-assets/image.png?signature=bad",
            "backend-test",
        ),
        (
            "cos",
            "storage://backend-test/oss://private-assets/image.png",
            "backend-test",
        ),
        (
            "oss",
            "storage://other/oss://private-assets/image.png",
            "backend-test",
        ),
        (
            "oss",
            "storage://backend-test/oss://other-bucket/image.png",
            "backend-test",
        ),
        (
            "oss",
            "https://example.com/image.png",
            "backend-test",
        ),
        (
            "oss",
            "storage://backend-test/oss://private-assets/" + ("x" * 1024),
            "backend-test",
        ),
    ],
)
def test_parse_oss_object_key_rejects_locator_outside_fixed_boundary(
    provider: str,
    physical_path: str,
    storage_backend_id: str,
) -> None:
    with pytest.raises(InvalidResourceLocator):
        parse_oss_object_key(
            provider=provider,
            physical_path=physical_path,
            storage_backend_id=storage_backend_id,
            expected_storage_backend_id="backend-test",
            expected_bucket="private-assets",
        )