X-Financial/server/tests/test_settings_service.py

from __future__ import annotations

from app.schemas.settings import ModelConnectivityTestRequest
from app.services.model_connectivity import ConnectivityCheckError, probe_model_connectivity


def test_probe_openai_compatible_chat_model(monkeypatch) -> None:
    captured: dict[str, object] = {}

    def fake_send_json_request(method, url, *, headers, payload):
        captured["method"] = method
        captured["url"] = url
        captured["headers"] = headers
        captured["payload"] = payload
        return 200, {"id": "ok"}

    monkeypatch.setattr("app.services.model_connectivity._send_json_request", fake_send_json_request)

    result = probe_model_connectivity(
        ModelConnectivityTestRequest(
            provider="OpenAI Compatible",
            endpoint="https://api.example.com/v1",
            model="gpt-test",
            api_key="secret",
            capability="chat",
        )
    )

    assert result.ok is True
    assert result.status_code == 200
    assert captured["method"] == "POST"
    assert captured["url"] == "https://api.example.com/v1/chat/completions"
    assert captured["headers"]["Authorization"] == "Bearer secret"
    assert captured["payload"]["model"] == "gpt-test"


def test_probe_azure_embedding_model(monkeypatch) -> None:
    captured: dict[str, object] = {}

    def fake_send_json_request(method, url, *, headers, payload):
        captured["url"] = url
        captured["headers"] = headers
        captured["payload"] = payload
        return 200, {"data": []}

    monkeypatch.setattr("app.services.model_connectivity._send_json_request", fake_send_json_request)

    result = probe_model_connectivity(
        ModelConnectivityTestRequest(
            provider="Azure OpenAI",
            endpoint="https://resource.openai.azure.com",
            model="embedding-deployment",
            api_key="azure-key",
            capability="embedding",
        )
    )

    assert result.ok is True
    assert (
        captured["url"]
        == "https://resource.openai.azure.com/openai/deployments/embedding-deployment/embeddings?api-version=2024-10-21"
    )
    assert captured["headers"]["api-key"] == "azure-key"
    assert captured["payload"]["input"] == "connectivity test"


def test_probe_openai_compatible_reranker_model(monkeypatch) -> None:
    captured: dict[str, object] = {}

    def fake_send_json_request(method, url, *, headers, payload):
        captured["method"] = method
        captured["url"] = url
        captured["headers"] = headers
        captured["payload"] = payload
        return 200, {"results": []}

    monkeypatch.setattr("app.services.model_connectivity._send_json_request", fake_send_json_request)

    result = probe_model_connectivity(
        ModelConnectivityTestRequest(
            provider="OpenAI Compatible",
            endpoint="https://api.example.com/v1",
            model="reranker-v1",
            api_key="secret",
            capability="reranker",
        )
    )

    assert result.ok is True
    assert captured["method"] == "POST"
    assert captured["url"] == "https://api.example.com/v1/rerank"
    assert captured["headers"]["Authorization"] == "Bearer secret"
    assert captured["payload"] == {
        "model": "reranker-v1",
        "query": "connectivity test",
        "documents": ["sample document"],
    }


def test_probe_ollama_failure_returns_error_payload(monkeypatch) -> None:
    def fake_send_json_request(method, url, *, headers, payload):
        raise ConnectivityCheckError("模型不存在或尚未拉取。", status_code=404)

    monkeypatch.setattr("app.services.model_connectivity._send_json_request", fake_send_json_request)

    result = probe_model_connectivity(
        ModelConnectivityTestRequest(
            provider="Ollama",
            endpoint="http://127.0.0.1:11434",
            model="llama3.1",
            capability="chat",
        )
    )

    assert result.ok is False
    assert result.status_code == 404
    assert "模型不存在" in result.detail