117 lines
4.4 KiB
Python
117 lines
4.4 KiB
Python
"""Unit tests for the OpenTelemetry wiring in :mod:`tavolo.app`.
|
|
|
|
The mixin under test is kaya-otel's :class:`~kaya.otel.OTelMixin`, an
|
|
optional dependency (the ``otel`` extra); these tests only verify that
|
|
:func:`tavolo.app.otel_mixin_from_settings` maps the ``OTEL_*`` settings
|
|
onto mixin construction. The ``kaya.otel`` module is stubbed in
|
|
``sys.modules`` so the suite does not need the extra installed.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import types
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from tavolo.app import otel_mixin_from_settings
|
|
from tavolo.config import Settings
|
|
from tavolo.telemetry import session_tag_hook
|
|
|
|
|
|
def _settings(env: dict) -> Settings:
|
|
with patch.dict(os.environ, env, clear=True):
|
|
return Settings.from_env()
|
|
|
|
|
|
class _StubOTelMixin:
|
|
def __init__(self, **kwargs):
|
|
self.kwargs = kwargs
|
|
|
|
|
|
def _stub_kaya_otel():
|
|
"""Install a fake ``kaya.otel`` module and return it."""
|
|
module = types.ModuleType("kaya.otel")
|
|
module.OTelMixin = _StubOTelMixin # type: ignore[attr-defined]
|
|
return patch.dict(sys.modules, {"kaya.otel": module})
|
|
|
|
|
|
class OTelMixinFromSettingsTests(unittest.TestCase):
|
|
def test_disabled_by_default(self):
|
|
self.assertIsNone(otel_mixin_from_settings(_settings({})))
|
|
|
|
def test_enabled_by_otel_enabled(self):
|
|
with _stub_kaya_otel():
|
|
mixin = otel_mixin_from_settings(_settings({"OTEL_ENABLED": "1"}))
|
|
self.assertIsNotNone(mixin)
|
|
|
|
def test_settings_are_passed_through(self):
|
|
with _stub_kaya_otel():
|
|
mixin = otel_mixin_from_settings(_settings({
|
|
"OTEL_ENABLED": "true",
|
|
"OTEL_SERVICE_NAME": "cards",
|
|
"OTEL_EXPORTER_OTLP_ENDPOINT": "http://collector:4318",
|
|
"OTEL_EXPORTER_OTLP_HEADERS": "Authorization=Bearer t, X-Tenant=one",
|
|
}))
|
|
assert isinstance(mixin, _StubOTelMixin)
|
|
self.assertEqual({
|
|
"service_name": "cards",
|
|
"endpoint": "http://collector:4318",
|
|
"headers": {"Authorization": "Bearer t", "X-Tenant": "one"},
|
|
"excluded_paths": ("/api/health",),
|
|
"server_response_hook": session_tag_hook,
|
|
"websocket_close_hook": session_tag_hook,
|
|
}, mixin.kwargs)
|
|
|
|
def test_session_tag_hooks_are_wired(self):
|
|
with _stub_kaya_otel():
|
|
mixin = otel_mixin_from_settings(_settings({"OTEL_ENABLED": "1"}))
|
|
assert isinstance(mixin, _StubOTelMixin)
|
|
self.assertIs(session_tag_hook, mixin.kwargs["server_response_hook"])
|
|
self.assertIs(session_tag_hook, mixin.kwargs["websocket_close_hook"])
|
|
|
|
def test_defaults_when_only_enabled(self):
|
|
with _stub_kaya_otel():
|
|
mixin = otel_mixin_from_settings(_settings({"OTEL_ENABLED": "on"}))
|
|
assert isinstance(mixin, _StubOTelMixin)
|
|
self.assertEqual("tavolo", mixin.kwargs["service_name"])
|
|
self.assertIsNone(mixin.kwargs["endpoint"])
|
|
self.assertIsNone(mixin.kwargs["headers"])
|
|
self.assertEqual(("/api/health",), mixin.kwargs["excluded_paths"])
|
|
|
|
def test_meter_provider_is_passed_through(self):
|
|
with _stub_kaya_otel():
|
|
provider = object()
|
|
mixin = otel_mixin_from_settings(
|
|
_settings({"OTEL_ENABLED": "1"}), meter_provider=provider
|
|
)
|
|
assert isinstance(mixin, _StubOTelMixin)
|
|
self.assertIs(provider, mixin.kwargs["meter_provider"])
|
|
|
|
def test_tracer_provider_is_passed_through(self):
|
|
with _stub_kaya_otel():
|
|
provider = object()
|
|
mixin = otel_mixin_from_settings(
|
|
_settings({"OTEL_ENABLED": "1"}), tracer_provider=provider
|
|
)
|
|
assert isinstance(mixin, _StubOTelMixin)
|
|
self.assertIs(provider, mixin.kwargs["tracer_provider"])
|
|
|
|
def test_excluded_paths_are_passed_through(self):
|
|
with _stub_kaya_otel():
|
|
mixin = otel_mixin_from_settings(_settings({
|
|
"OTEL_ENABLED": "1",
|
|
"OTEL_EXCLUDED_PATHS": "/api/health,/metrics",
|
|
}))
|
|
assert isinstance(mixin, _StubOTelMixin)
|
|
self.assertEqual(("/api/health", "/metrics"), mixin.kwargs["excluded_paths"])
|
|
|
|
def test_missing_extra_raises_runtime_error(self):
|
|
with patch.dict(sys.modules, {"kaya.otel": None}):
|
|
with self.assertRaises(RuntimeError):
|
|
otel_mixin_from_settings(_settings({"OTEL_ENABLED": "1"}))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|