local-inference-service / tests /test_embeddings_api.py
fashxp's picture lukmzig's picture
BPT adjustments: batched embedding API, normalized vectors, resource limits, lint/type/test gate (#3)
ed962ac
Raw History Blame Contribute Delete
2.76 kB
# -------------------------------------------------------------------
# This source file is available under the terms of the
# Pimcore Open Core License (POCL)
# Full copyright and license information is available in
# LICENSE.md which is distributed with this source code.
# -------------------------------------------------------------------
"""API-layer tests: the /text-embedding and /image-embedding endpoints share one request/response
shape. The model-backed services are mocked so no weights are loaded."""
from fastapi.testclient import TestClient
from src import main
from src.embeddings import EmbeddingData, EmbeddingsRequest, EmbeddingsResponse
client = TestClient(main.app)
def test_text_embedding_returns_list_envelope(monkeypatch):
async def fake(request: EmbeddingsRequest) -> EmbeddingsResponse:
return EmbeddingsResponse(
data=[EmbeddingData(index=0, embedding=[0.1, 0.2, 0.3])],
model=request.model,
)
monkeypatch.setattr(main.text_embedding_service, "generate_embeddings", fake)
response = client.post("/text-embedding", json={"model": "nomic-text", "input": "hello"})
assert response.status_code == 200
payload = response.json()
assert payload["object"] == "list"
assert payload["model"] == "nomic-text"
assert payload["data"][0]["embedding"] == [0.1, 0.2, 0.3]
def test_image_embedding_returns_list_envelope(monkeypatch):
async def fake(request: EmbeddingsRequest) -> EmbeddingsResponse:
return EmbeddingsResponse(
data=[EmbeddingData(index=0, embedding=[1.0, 2.0])],
model=request.model,
)
monkeypatch.setattr(main.image_embedding_service, "generate_embeddings", fake)
response = client.post("/image-embedding", json={"model": "nomic-vision", "input": "BASE64IMG"})
assert response.status_code == 200
assert response.json()["data"][0]["embedding"] == [1.0, 2.0]
def test_text_embedding_accepts_batch_input(monkeypatch):
async def fake(request: EmbeddingsRequest) -> EmbeddingsResponse:
items = request.input if isinstance(request.input, list) else [request.input]
return EmbeddingsResponse(
data=[EmbeddingData(index=i, embedding=[float(i)]) for i in range(len(items))],
model=request.model,
)
monkeypatch.setattr(main.text_embedding_service, "generate_embeddings", fake)
response = client.post("/text-embedding", json={"model": "m", "input": ["a", "b"]})
assert response.status_code == 200
assert [d["index"] for d in response.json()["data"]] == [0, 1]
def test_text_embedding_rejects_missing_model():
response = client.post("/text-embedding", json={"input": "hello"})
assert response.status_code == 422