"""Tests for the embedding generation service (T91, Phase 4). Phase 4's first cut ships a deterministic local pseudo-embedding so the vector retrieval pipeline can land without an external embeddings API or a heavy local model dependency. These tests pin the contract: * the result has the right shape (vector length, ``dim`` metadata), * the default ``model`` string is reported back unchanged, * output is byte-identical for the same input (deterministic), * distinct inputs produce distinct vectors (so cosine actually discriminates), * empty / whitespace-only input collapses to the ``"fallback"`` sentinel with a zero vector — callers detect this and skip indexing, * the vector is unit-normalized so cosine similarity behaves. The pseudo path doesn't touch the LLMClient, so we pass an empty ``MockLLMClient`` — any accidental call into it would raise ``IndexError`` and surface as a regression. """ from __future__ import annotations import logging import math import pytest from chat.llm.mock import MockLLMClient from chat.services.embeddings import ( DEFAULT_EMBEDDING_DIM, DEFAULT_EMBEDDING_MODEL, FALLBACK_EMBEDDING_MODEL, EmbeddingResult, generate_embedding, ) def _client() -> MockLLMClient: # Pseudo path never calls the client — empty canned list ensures any # accidental call raises and surfaces the regression loudly. return MockLLMClient(canned=[]) @pytest.mark.asyncio async def test_generate_embedding_returns_vector_of_correct_dim(): result = await generate_embedding(_client(), text="hello") assert isinstance(result, EmbeddingResult) assert isinstance(result.vector, list) assert len(result.vector) == DEFAULT_EMBEDDING_DIM == 384 assert result.dim == 384 assert all(isinstance(x, float) for x in result.vector) @pytest.mark.asyncio async def test_generate_embedding_returns_correct_model_metadata(): result = await generate_embedding(_client(), text="hello") assert result.model == DEFAULT_EMBEDDING_MODEL == "pseudo-sha256-384" @pytest.mark.asyncio async def test_generate_embedding_is_deterministic(): a = await generate_embedding(_client(), text="hello world") b = await generate_embedding(_client(), text="hello world") assert a.vector == b.vector @pytest.mark.asyncio async def test_generate_embedding_distinct_text_produces_distinct_vectors(): a = await generate_embedding(_client(), text="hello world") b = await generate_embedding(_client(), text="totally different content") assert a.vector != b.vector # Sanity-check cosine similarity — both vectors are unit-normalized, # so this reduces to a plain dot product. cosine = sum(x * y for x, y in zip(a.vector, b.vector)) assert cosine < 0.99 @pytest.mark.asyncio async def test_generate_embedding_empty_text_returns_fallback(): for empty in ("", " ", "\n\t"): result = await generate_embedding(_client(), text=empty) assert result.model == FALLBACK_EMBEDDING_MODEL == "fallback" assert result.dim == DEFAULT_EMBEDDING_DIM assert len(result.vector) == DEFAULT_EMBEDDING_DIM assert all(x == 0.0 for x in result.vector) @pytest.mark.asyncio async def test_generate_embedding_unit_normalized(): result = await generate_embedding(_client(), text="some non-empty text") norm_sq = sum(x * x for x in result.vector) assert math.isclose(norm_sq, 1.0, abs_tol=1e-6) @pytest.mark.asyncio async def test_generate_embedding_non_default_model_logs_warning(caplog): """T107: non-default model falls through to fallback and must warn. A Phase 4.5+ caller pointing at a real model that isn't yet wired up would otherwise silently degrade (zero vector → useless cosine). The warning surfaces the misconfiguration in logs. """ caplog.set_level(logging.WARNING, logger="chat.services.embeddings") result = await generate_embedding(_client(), text="hello", model="real-model") # Behavior unchanged: still returns the fallback sentinel. assert result.model == FALLBACK_EMBEDDING_MODEL == "fallback" assert all(x == 0.0 for x in result.vector) # Warning fired and names the offending model. warnings = [r for r in caplog.records if r.levelno == logging.WARNING] assert any("non-default model" in r.getMessage() for r in warnings) assert any("real-model" in r.getMessage() for r in warnings) @pytest.mark.asyncio async def test_generate_embedding_default_model_does_not_warn(caplog): """T107: the silent default path must stay silent.""" caplog.set_level(logging.WARNING, logger="chat.services.embeddings") await generate_embedding(_client(), text="hello") warnings = [r for r in caplog.records if r.levelno == logging.WARNING] assert warnings == []