Chris CoutinhoandClaude Opus 4.8
2e49e442f4
Reviewer findings: - Fix double-count of exhausted-retry failures: the inner final-retry branch and the outer except both recorded a processing error. Consolidate to the outer handler (single call site); inner branch keeps only the Qdrant-upsert error metric. Regression test added. - Deletes are no longer counted as indexing events: the delete success path drops doc_type so astrolabe_documents_indexed_total is not inflated. Regression test added. - Reuse the already-resolved `settings` in _index_document instead of a second get_settings() call. - Use explicit `> 0` guards in record_document_parse / record_embedding instead of truthiness checks. SonarCloud: - S1244 (BUG): replace float `==` equality in metric tests with pytest.approx. - S5332 (hotspot): use https in the gateway-URL test fixture. - S1192: extract the repeated "vector_sync.chunk_count" span-attribute literal into a module constant. Review nit: move the duplicated `_sample` test helper into a shared `metric_sample` fixture in tests/unit/conftest.py. Refs Deck #175, PR #831. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
123 lines
4.2 KiB
Python
123 lines
4.2 KiB
Python
"""Unit tests for embedding observability.
|
|
|
|
Covers:
|
|
1. ``Settings.get_embedding_provider_family()`` — the single source of truth for
|
|
the ``provider`` metric label / span attribute — across provider configs.
|
|
2. The ``record_embedding`` helper — that it increments the right
|
|
``astrolabe_embedding_*`` series and skips the throughput counters on error.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from nextcloud_mcp_server.config import Settings
|
|
from nextcloud_mcp_server.observability.metrics import record_embedding
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
# ``metric_sample`` is provided as a shared fixture in tests/unit/conftest.py.
|
|
|
|
|
|
class TestProviderFamily:
|
|
"""Provider-family detection mirrors ProviderRegistry priority."""
|
|
|
|
def test_bedrock(self):
|
|
assert (
|
|
Settings(aws_region="us-east-1").get_embedding_provider_family()
|
|
== "bedrock"
|
|
)
|
|
|
|
def test_openai(self):
|
|
settings = Settings(
|
|
openai_api_key="sk-test",
|
|
aws_region=None,
|
|
bedrock_embedding_model=None,
|
|
bedrock_generation_model=None,
|
|
)
|
|
assert settings.get_embedding_provider_family() == "openai"
|
|
|
|
def test_mistral(self):
|
|
settings = Settings(
|
|
mistral_api_key="m-test",
|
|
aws_region=None,
|
|
bedrock_embedding_model=None,
|
|
bedrock_generation_model=None,
|
|
openai_api_key=None,
|
|
)
|
|
assert settings.get_embedding_provider_family() == "mistral"
|
|
|
|
def test_ollama(self):
|
|
settings = Settings(
|
|
ollama_base_url="http://localhost:11434",
|
|
aws_region=None,
|
|
bedrock_embedding_model=None,
|
|
bedrock_generation_model=None,
|
|
openai_api_key=None,
|
|
mistral_api_key=None,
|
|
)
|
|
assert settings.get_embedding_provider_family() == "ollama"
|
|
|
|
def test_simple_fallback(self):
|
|
settings = Settings(
|
|
aws_region=None,
|
|
bedrock_embedding_model=None,
|
|
bedrock_generation_model=None,
|
|
openai_api_key=None,
|
|
mistral_api_key=None,
|
|
ollama_base_url=None,
|
|
)
|
|
assert settings.get_embedding_provider_family() == "simple"
|
|
|
|
def test_gateway_uses_model_prefix(self):
|
|
settings = Settings(
|
|
embedding_provider="gateway",
|
|
embedding_gateway_url="https://gateway:8080",
|
|
embedding_gateway_model="mistral/mistral-embed",
|
|
)
|
|
assert settings.get_embedding_provider_family() == "mistral"
|
|
|
|
|
|
class TestRecordEmbedding:
|
|
def test_dense_success_increments_throughput(self, metric_sample):
|
|
labels = {"kind": "dense", "provider": "uttest-prov"}
|
|
before_chunks = metric_sample("astrolabe_embedding_chunks_total", labels)
|
|
before_chars = metric_sample("astrolabe_embedding_chars_total", labels)
|
|
before_req = metric_sample(
|
|
"astrolabe_embedding_requests_total", {**labels, "status": "success"}
|
|
)
|
|
|
|
record_embedding("dense", "uttest-prov", 0.42, chunks=12, chars=3400)
|
|
|
|
assert metric_sample(
|
|
"astrolabe_embedding_chunks_total", labels
|
|
) == pytest.approx(before_chunks + 12)
|
|
assert metric_sample(
|
|
"astrolabe_embedding_chars_total", labels
|
|
) == pytest.approx(before_chars + 3400)
|
|
assert metric_sample(
|
|
"astrolabe_embedding_requests_total", {**labels, "status": "success"}
|
|
) == pytest.approx(before_req + 1)
|
|
assert (
|
|
metric_sample(
|
|
"astrolabe_embedding_duration_seconds_count",
|
|
{**labels, "status": "success"},
|
|
)
|
|
>= 1
|
|
)
|
|
|
|
def test_sparse_error_skips_throughput(self, metric_sample):
|
|
labels = {"kind": "sparse", "provider": "bm25-uttest"}
|
|
record_embedding(
|
|
"sparse", "bm25-uttest", 0.1, chunks=5, chars=100, status="error"
|
|
)
|
|
assert metric_sample(
|
|
"astrolabe_embedding_chunks_total", labels
|
|
) == pytest.approx(0.0)
|
|
assert metric_sample(
|
|
"astrolabe_embedding_chars_total", labels
|
|
) == pytest.approx(0.0)
|
|
assert metric_sample(
|
|
"astrolabe_embedding_requests_total", {**labels, "status": "error"}
|
|
) == pytest.approx(1.0)
|