Spaces:
Paused
Paused
File size: 3,476 Bytes
8c1b9fe | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | """Tests for the optional Langfuse trace exporter (ADR-0019).
All offline: the `langfuse` SDK is stubbed so we exercise enable-detection and the
export path without the real dependency, and confirm export is a safe no-op when
disabled and never raises.
"""
from __future__ import annotations
import importlib.machinery
import importlib.util
import sys
import types
from auralynq.telemetry.langfuse_export import export_trace, langfuse_enabled
from auralynq.telemetry.tracing import Trace
def _trace() -> Trace:
t = Trace(trace_id="t-test")
with t.span("planner", q="x"):
pass
with t.span("synthesizer", provider="extractive"):
pass
return t
def test_disabled_without_keys(monkeypatch):
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "")
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "")
from auralynq.config import reload_settings
reload_settings()
assert langfuse_enabled() is False
# export is a no-op (returns False), never raises
assert export_trace(_trace(), question="q", answer="a") is False
def _install_fake_langfuse(monkeypatch, recorder):
fake = types.ModuleType("langfuse")
fake.__spec__ = importlib.machinery.ModuleSpec("langfuse", loader=None)
class _Span:
def span(self, **kw):
recorder["spans"].append(kw)
return self
class _Trace(_Span):
pass
class _Langfuse:
def __init__(self, **kw):
recorder["init"] = kw
def trace(self, **kw):
recorder["trace"] = kw
return _Trace()
def flush(self):
recorder["flushed"] = True
fake.Langfuse = _Langfuse
monkeypatch.setitem(sys.modules, "langfuse", fake)
def test_enabled_and_exports_with_keys_and_sdk(monkeypatch):
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-test")
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-test")
from auralynq.config import reload_settings
reload_settings()
rec = {"spans": []}
_install_fake_langfuse(monkeypatch, rec)
# reset the cached client so it picks up the stub + keys
from auralynq.telemetry import langfuse_export as lx
lx._client.cache_clear()
assert langfuse_enabled() is True
ok = export_trace(
_trace(),
question="What is the capital of France?",
answer="Paris [1]",
metadata={"route": "fast"},
)
assert ok is True
assert rec["trace"]["input"] == {"question": "What is the capital of France?"}
assert rec["trace"]["output"] == {"answer": "Paris [1]"}
assert [s["name"] for s in rec["spans"]] == ["planner", "synthesizer"]
assert rec.get("flushed") is True
lx._client.cache_clear()
def test_export_never_raises_on_sdk_error(monkeypatch):
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-test")
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-test")
from auralynq.config import reload_settings
reload_settings()
fake = types.ModuleType("langfuse")
fake.__spec__ = importlib.machinery.ModuleSpec("langfuse", loader=None)
class _Boom:
def __init__(self, **kw):
raise RuntimeError("network down")
fake.Langfuse = _Boom
monkeypatch.setitem(sys.modules, "langfuse", fake)
from auralynq.telemetry import langfuse_export as lx
lx._client.cache_clear()
# must swallow the error and return False, not propagate
assert export_trace(_trace(), question="q", answer="a") is False
lx._client.cache_clear()
|