from adapters.prometheus import PrometheusMetrics from config import Settings from rag.instrumentation import InstrumentedGenerator from rag.metrics import ( CLARIFY_ASKED, PROVIDER_FAILURE, REQUEST_DURATION, REQUESTS, STAGE_DURATION, TRACE_WRITE_FAILED, InMemoryMetrics, ) from rag.telemetry import ( configure_telemetry, correlation_context, request_span, stage, ) def test_prometheus_registers_domain_failures_histograms_and_bounds_labels(): metrics = PrometheusMetrics() metrics.increment(TRACE_WRITE_FAILED) metrics.increment(CLARIFY_ASKED, reason="raw-user-controlled-reason") metrics.increment( REQUESTS, method="POST", route="/v1/rag/query", status="2xx" ) metrics.observe( REQUEST_DURATION, 1.25, method="POST", route="/v1/rag/query", status="2xx", ) metrics.observe(STAGE_DURATION, 0.2, stage="retrieval", outcome="ok") body, content_type = metrics.render() rendered = body.decode("utf-8") assert "application/openmetrics-text" in content_type assert "duocthu_trace_write_failed_total 1.0" in rendered assert 'duocthu_clarify_asked_total{reason="other"} 1.0' in rendered assert "duocthu_request_duration_seconds_bucket" in rendered assert "duocthu_stage_duration_seconds_bucket" in rendered def test_stage_records_duration_without_requiring_an_otel_collector(): metrics = InMemoryMetrics() configure_telemetry(Settings(otel_enabled=False), metrics) with stage("grounding"): pass values = metrics.observed(STAGE_DURATION, stage="grounding", outcome="ok") assert len(values) == 1 assert values[0] >= 0 class _FailingGenerator: model_id = "fake" def generate(self, system: str, user: str, schema: dict) -> str: raise ConnectionError("provider offline") def test_provider_failure_uses_bounded_classification_not_exception_text(): metrics = InMemoryMetrics() generator = InstrumentedGenerator(_FailingGenerator(), metrics) try: generator.generate("system", "user", {}) except ConnectionError: pass else: # pragma: no cover - assertion guard raise AssertionError("expected provider failure") assert metrics.total( PROVIDER_FAILURE, provider="other", operation="generate", reason="provider_unavailable", ) == 1 def test_otel_server_and_stage_spans_share_trace_and_emit_metric_exemplar(): otel_trace = pytest.importorskip("opentelemetry.trace") sdk_trace = pytest.importorskip("opentelemetry.sdk.trace") exporter_module = pytest.importorskip( "opentelemetry.sdk.trace.export.in_memory_span_exporter" ) export_module = pytest.importorskip("opentelemetry.sdk.trace.export") exporter = exporter_module.InMemorySpanExporter() provider = sdk_trace.TracerProvider() provider.add_span_processor(export_module.SimpleSpanProcessor(exporter)) otel_trace.set_tracer_provider(provider) metrics = PrometheusMetrics() configure_telemetry( Settings( otel_enabled=True, otel_exporter_otlp_endpoint="http://127.0.0.1:4318/v1/traces", ), metrics, ) with correlation_context("req-span-test"): with request_span("POST", "/v1/rag/query", {}): with stage("retrieval"): pass provider.force_flush() spans = exporter.get_finished_spans() assert {item.name for item in spans} == { "POST /v1/rag/query", "rag.stage.retrieval", } assert len({item.context.trace_id for item in spans}) == 1 assert spans[-1].attributes["duocthu.correlation_id"] == "req-span-test" rendered = metrics.render()[0].decode("utf-8") assert "duocthu_stage_duration_seconds_bucket" in rendered assert 'trace_id="' in rendered import pytest def test_langfuse_exporter_added_only_when_fully_configured(): """Langfuse is opt-in: all three settings, or no extra exporter at all. A half-configured Langfuse (say a base URL pasted in but no keys yet) must not produce an exporter that would fail auth on every batch. """ from rag.telemetry import _langfuse_exporters captured = [] def fake_exporter(**kwargs): captured.append(kwargs) return object() # Explicit empty values, not Settings() defaults: Settings reads the # developer's real .env, which on a configured machine already carries # live Langfuse keys and would make this assertion pass or fail by # accident depending on whose machine runs it. unset = Settings( langfuse_base_url="", langfuse_public_key="", langfuse_secret_key="" ) assert _langfuse_exporters(unset, fake_exporter) == [] assert ( _langfuse_exporters( Settings( langfuse_base_url="https://langfuse.example", langfuse_public_key="", langfuse_secret_key="", ), fake_exporter, ) == [] ) assert captured == [] result = _langfuse_exporters( Settings( langfuse_base_url="https://langfuse.example/", langfuse_public_key="pk-lf-public", langfuse_secret_key="sk-lf-secret", ), fake_exporter, ) assert len(result) == 1 assert len(captured) == 1 # Trailing slash stripped, so the path is not doubled. assert captured[0]["endpoint"] == "https://langfuse.example/api/public/otel/v1/traces" # Basic auth is public-key-as-username, secret-key-as-password. import base64 expected = base64.b64encode(b"pk-lf-public:sk-lf-secret").decode() assert captured[0]["headers"] == {"Authorization": f"Basic {expected}"} def test_langfuse_score_client_is_all_or_nothing(): """Same partial-config rule as the span exporter, for the same reason.""" from adapters.langfuse_scores import LangfuseScoreClient unset = Settings( langfuse_base_url="", langfuse_public_key="", langfuse_secret_key="" ) assert LangfuseScoreClient.from_settings(unset) is None assert ( LangfuseScoreClient.from_settings( Settings( langfuse_base_url="https://langfuse.example", langfuse_public_key="pk-lf-x", langfuse_secret_key="", ) ) is None ) client = LangfuseScoreClient.from_settings( Settings( langfuse_base_url="https://langfuse.example/", langfuse_public_key="pk-lf-public", langfuse_secret_key="sk-lf-secret", ) ) assert client is not None # Trailing slash stripped, so the path is not doubled. assert client.endpoint == "https://langfuse.example/api/public/scores" def test_langfuse_score_post_never_raises(monkeypatch): """A Langfuse outage must not propagate into the feedback request path.""" from adapters import langfuse_scores client = langfuse_scores.LangfuseScoreClient( "https://langfuse.example", "pk", "sk" ) def boom(*args, **kwargs): raise OSError("connection refused") monkeypatch.setattr(langfuse_scores.urllib.request, "urlopen", boom) assert client.post_score(otel_trace_id="a" * 32, name="user_feedback", value=1.0) is False