zhuangxiu/services/api/tests/test_settings.py

from pathlib import Path

import pytest

from app.config import Settings
from app.runtime_settings import (
    EncryptedSettingsStore,
    RuntimeSettingsTestResult,
    generate_secret,
    test_model_pool_item as run_model_pool_test,
)


def create_store(path: Path) -> EncryptedSettingsStore:
    settings = Settings(
        _env_file=None,
        runtime_settings_dir=str(path),
        database_url="postgresql://user:secret@postgres:5432/app",
        redis_url="redis://:secret@redis:6379/0",
    )
    return EncryptedSettingsStore(settings)


def test_runtime_secrets_are_encrypted_and_never_returned(tmp_path: Path) -> None:
    store = create_store(tmp_path)
    store.update(
        {
            "s3_endpoint": "http://minio:9000",
            "s3_public_endpoint": "http://localhost:9000",
            "s3_access_key_id": "service-account",
            "s3_secret_access_key": "very-private-secret",
        }
    )

    response = store.public_response()
    assert "s3_secret_access_key" not in response.values
    assert response.configured["s3_secret_access_key"] is True
    assert b"very-private-secret" not in store.data_path.read_bytes()


def test_readiness_requires_configuration_and_successful_tests(tmp_path: Path) -> None:
    store = create_store(tmp_path)
    store.update(
        {
            "s3_endpoint": "http://minio:9000",
            "s3_public_endpoint": "http://localhost:9000",
            "s3_access_key_id": "access",
            "s3_secret_access_key": "secret",
            "model_pool": [
                {
                    "id": "planner",
                    "name": "总调度",
                    "model_id": "planner-model",
                    "category": "language",
                    "provider": "lingke",
                    "base_url": "https://example.com/v1",
                    "api_key": "planner-secret",
                    "capabilities": ["orchestration"],
                },
                {
                    "id": "designer",
                    "name": "空间与生图",
                    "model_id": "designer-model",
                    "category": "multimodal",
                    "provider": "lingke",
                    "base_url": "https://example.com/v1",
                    "api_key": "designer-secret",
                    "capabilities": ["spatial_understanding", "image_generation"],
                },
            ],
            "orchestrator_model_id": "planner",
            "spatial_routing_mode": "auto",
            "image_routing_mode": "manual",
            "image_model_id": "designer",
        }
    )
    assert store.public_response().readiness.ready is False

    for target in ("infrastructure", "storage", "model:planner", "model:designer", "generation:designer"):
        store.record_test(RuntimeSettingsTestResult(target=target, ok=True, message="ok"))

    assert store.public_response().readiness.ready is True

    pool = store.merged_values()["model_pool"]
    pool[1]["model_id"] = "another-image-model"
    store.update({"model_pool": pool})
    assert store.public_response().readiness.ready is False
    assert store.public_response().tests["model:designer"]["ok"] is False
    assert "重新测试" in store.public_response().tests["model:designer"]["message"]


def test_model_pool_keys_are_encrypted_and_redacted(tmp_path: Path) -> None:
    store = create_store(tmp_path)
    store.update(
        {
            "model_pool": [
                {
                    "id": "planner",
                    "name": "总调度",
                    "model_id": "planner-model",
                    "category": "language",
                    "provider": "custom",
                    "base_url": "https://example.com/v1",
                    "api_key": "nested-model-secret",
                    "capabilities": ["orchestration"],
                }
            ]
        }
    )

    public_model = store.public_response().values["model_pool"][0]
    assert "api_key" not in public_model
    assert public_model["api_key_configured"] is True
    assert b"nested-model-secret" not in store.data_path.read_bytes()


def test_secret_generators_use_expected_lengths() -> None:
    assert len(generate_secret("hex24")) == 48
    assert len(generate_secret("hex32")) == 64
    assert len(generate_secret("base64_32")) == 44


@pytest.mark.asyncio
async def test_individual_model_error_names_the_failing_model(tmp_path: Path, monkeypatch) -> None:
    store = create_store(tmp_path)
    store.update(
        {
            "model_pool": [
                {
                    "id": "image-model",
                    "name": "客厅效果图模型",
                    "model_id": "wrong-image-id",
                    "category": "multimodal",
                    "provider": "custom",
                    "base_url": "https://example.com/v1",
                    "api_key": "secret",
                    "capabilities": ["image_generation"],
                }
            ]
        }
    )

    class FakeResponse:
        status_code = 200

        def raise_for_status(self) -> None:
            return None

        def json(self) -> dict:
            return {"data": [{"id": "available-image-id"}]}

    class FakeClient:
        async def __aenter__(self):
            return self

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

        async def get(self, *args, **kwargs) -> FakeResponse:
            return FakeResponse()

    monkeypatch.setattr("app.runtime_settings.httpx.AsyncClient", lambda **kwargs: FakeClient())
    result = await run_model_pool_test(store.merged_values(), "image-model")

    assert result.ok is False
    assert "客厅效果图模型" in result.message
    assert "wrong-image-id" in result.message