113 lines
4.1 KiB
Python
113 lines
4.1 KiB
Python
"""OpenTelemetry wiring tests.
|
|
|
|
The actual OpenTelemetry packages are an optional extra (``kca-app[otel]``),
|
|
so the kaya-otel mixin is stubbed here. Provider-dependent tests live in
|
|
``test_telemetry.py``.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import types
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from kca.app import otel_mixin_from_settings
|
|
from kca.config import Settings
|
|
from kca.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():
|
|
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": "kca-test",
|
|
"OTEL_EXPORTER_OTLP_ENDPOINT": "http://collector:4318",
|
|
"OTEL_EXPORTER_OTLP_HEADERS": "Authorization=Bearer t, X-Tenant=one",
|
|
"OTEL_METRIC_EXPORT_INTERVAL_MILLIS": "15000",
|
|
}
|
|
)
|
|
)
|
|
assert isinstance(mixin, _StubOTelMixin)
|
|
self.assertEqual("kca-test", mixin.kwargs["service_name"])
|
|
self.assertEqual("http://collector:4318", mixin.kwargs["endpoint"])
|
|
self.assertEqual(
|
|
{"Authorization": "Bearer t", "X-Tenant": "one"},
|
|
mixin.kwargs["headers"],
|
|
)
|
|
self.assertEqual(("/api/health",), mixin.kwargs["excluded_paths"])
|
|
self.assertEqual(15000, mixin.kwargs["metric_export_interval_millis"])
|
|
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("kca", mixin.kwargs["service_name"])
|
|
self.assertIsNone(mixin.kwargs["endpoint"])
|
|
self.assertIsNone(mixin.kwargs["headers"])
|
|
self.assertEqual(("/api/health",), mixin.kwargs["excluded_paths"])
|
|
|
|
def test_providers_are_passed_through(self):
|
|
with _stub_kaya_otel():
|
|
meter = object()
|
|
tracer = object()
|
|
mixin = otel_mixin_from_settings(
|
|
_settings({"OTEL_ENABLED": "1"}),
|
|
meter_provider=meter,
|
|
tracer_provider=tracer,
|
|
)
|
|
assert isinstance(mixin, _StubOTelMixin)
|
|
self.assertIs(meter, mixin.kwargs["meter_provider"])
|
|
self.assertIs(tracer, 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,/api/docs",
|
|
}
|
|
)
|
|
)
|
|
assert isinstance(mixin, _StubOTelMixin)
|
|
self.assertEqual(("/api/health", "/api/docs"), 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()
|