knowledge-adapter-service/tests/test_settings.py

from pathlib import Path
from typing import Any
from uuid import UUID

import pytest
from pydantic import SecretStr, ValidationError

from knowledge_adapter.content_budget import TRUNCATION_MARKER
from knowledge_adapter.settings import Settings

BASE_SETTINGS: dict[str, Any] = {
    "DIFY_KEY_HASH_CURRENT": "a" * 64,
    "DIFY_KEY_HASH_NEXT": "",
    "WEKNORA_BASE_URL": "http://weknora.internal:8080",
    "WEKNORA_ALLOWED_KB_IDS": (
        "3904b40e-278b-4273-a6fa-db6cbc975587,"
        "7fa6e8d2-05d8-4e74-b98c-d8591b680db9"
    ),
    "WEKNORA_TENANT_ID": 10008,
    "WEKNORA_OSS_STORAGE_BACKEND_ID": "backend-test",
    "OSS_REGION": "cn-shenzhen",
    "OSS_BUCKET": "private-assets",
    "OSS_RAM_ROLE_NAME": "knowledge-adapter-role",
    "ASSET_PUBLIC_BASE_URL": "https://assets.example.com",
}


def test_settings_parse_allowlist_secrets_and_required_defaults() -> None:
    settings = Settings(**BASE_SETTINGS)

    assert settings.dify_key_hash_current == "a" * 64
    assert settings.dify_key_hash_next is None
    assert settings.weknora_api_key_file == Path("/run/secrets/weknora_api_key")
    assert settings.weknora_allowed_kb_ids == frozenset(
        {
            UUID("3904b40e-278b-4273-a6fa-db6cbc975587"),
            UUID("7fa6e8d2-05d8-4e74-b98c-d8591b680db9"),
        }
    )
    assert settings.weknora_tenant_id == 10008
    assert settings.weknora_oss_storage_backend_id == "backend-test"
    assert settings.max_query_chars == 2000
    assert settings.max_top_k == 5
    assert settings.max_chunk_chars == 4000
    assert settings.max_total_chars == 12000
    assert settings.weknora_timeout_seconds == 15
    assert settings.asset_database_url_file == Path("/run/secrets/asset_database_url")
    assert settings.weknora_readonly_database_url_file == Path(
        "/run/secrets/weknora_readonly_database_url"
    )


def test_settings_reject_malformed_current_key_hash() -> None:
    values = {**BASE_SETTINGS, "DIFY_KEY_HASH_CURRENT": "not-a-sha256"}

    with pytest.raises(ValidationError):
        Settings(**values)


def test_settings_uses_weknora_api_key_secret_file(tmp_path: Path) -> None:
    secret_file = tmp_path / "weknora-api-key"
    values = {
        key: value
        for key, value in BASE_SETTINGS.items()
        if key != "WEKNORA_API_KEY"
    }
    values["WEKNORA_API_KEY_FILE"] = secret_file

    settings = Settings(**values)

    assert settings.weknora_api_key_file == secret_file
    assert not hasattr(settings, "weknora_api_key")


def test_settings_reads_weknora_api_key_at_runtime(tmp_path: Path) -> None:
    secret_file = tmp_path / "weknora-api-key"
    secret_file.write_text("weknora-test-secret\n", encoding="utf-8")
    values = {
        key: value
        for key, value in BASE_SETTINGS.items()
        if key != "WEKNORA_API_KEY"
    }
    values["WEKNORA_API_KEY_FILE"] = secret_file
    settings = Settings(**values)

    api_key = settings.read_weknora_api_key()

    assert isinstance(api_key, SecretStr)
    assert api_key.get_secret_value() == "weknora-test-secret"
    assert "weknora-test-secret" not in repr(settings)


def test_static_ak_mode_reads_secret_files_without_ram_role(tmp_path: Path) -> None:
    access_key_id_file = tmp_path / "oss-access-key-id"
    access_key_secret_file = tmp_path / "oss-access-key-secret"
    access_key_id_file.write_text("test-access-key-id\n", encoding="utf-8")
    access_key_secret_file.write_text("test-access-key-secret\n", encoding="utf-8")
    values = {
        **BASE_SETTINGS,
        "OSS_CREDENTIAL_MODE": "static_ak",
        "OSS_RAM_ROLE_NAME": None,
        "OSS_ACCESS_KEY_ID_FILE": access_key_id_file,
        "OSS_ACCESS_KEY_SECRET_FILE": access_key_secret_file,
    }

    settings = Settings(**values)

    assert settings.oss_credential_mode == "static_ak"
    assert settings.oss_ram_role_name is None
    assert settings.read_oss_access_key_id().get_secret_value() == "test-access-key-id"
    assert settings.read_oss_access_key_secret().get_secret_value() == "test-access-key-secret"
    assert "test-access-key-id" not in repr(settings)
    assert "test-access-key-secret" not in repr(settings)


@pytest.mark.parametrize("content", [None, "  \n"])
def test_settings_rejects_unavailable_weknora_api_key_secret(
    tmp_path: Path,
    content: str | None,
) -> None:
    secret_file = tmp_path / "weknora-api-key"
    if content is not None:
        secret_file.write_text(content, encoding="utf-8")
    values = {
        key: value
        for key, value in BASE_SETTINGS.items()
        if key != "WEKNORA_API_KEY"
    }
    values["WEKNORA_API_KEY_FILE"] = secret_file
    settings = Settings(**values)

    with pytest.raises(RuntimeError, match=r"^required secret unavailable$"):
        settings.read_weknora_api_key()


def test_settings_reject_chunk_limit_smaller_than_truncation_marker() -> None:
    values = {**BASE_SETTINGS, "MAX_CHUNK_CHARS": len(TRUNCATION_MARKER) - 1}

    with pytest.raises(ValidationError):
        Settings(**values)


@pytest.mark.parametrize(
    ("name", "value"),
    [
        ("OSS_REGION", "https://attacker.example"),
        ("OSS_REGION", "cn-shenzhen/internal"),
        ("OSS_BUCKET", "Private_Assets"),
        ("OSS_BUCKET", "https://attacker.example"),
        ("OSS_RAM_ROLE_NAME", ""),
        ("OSS_RAM_ROLE_NAME", "role/name"),
    ],
)
def test_settings_reject_unsafe_oss_identifiers(name: str, value: str) -> None:
    values = {**BASE_SETTINGS, name: value}

    with pytest.raises(ValidationError):
        Settings(**values)