103 lines
3.6 KiB
Python
103 lines
3.6 KiB
Python
"""Unit tests for OpenTelemetry integration.
|
|
|
|
Covers the structlog-aware OTEL log handler that unwraps event dicts so
|
|
OpenObserve receives structured attributes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
import pytest
|
|
|
|
from contract_check.core.logging import configure_logging, get_logger
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_logging():
|
|
"""Configure logging before each test and clean handlers after."""
|
|
configure_logging("INFO", json_output=True, service="test-service", env="test")
|
|
yield
|
|
root = logging.getLogger()
|
|
root.handlers.clear()
|
|
|
|
|
|
def test_structlog_otlp_handler_unwraps_event_dict(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""A structlog-wrapped record is converted to body + attributes."""
|
|
from contract_check.core.config import get_settings
|
|
from contract_check.core.telemetry import setup_telemetry, shutdown_telemetry
|
|
|
|
monkeypatch.setattr(
|
|
get_settings(), "otel_exporter_otlp_endpoint", "http://localhost:9999/v1/logs"
|
|
)
|
|
|
|
# setup_telemetry defines the handler class when OTEL is imported.
|
|
setup_telemetry("test")
|
|
root = logging.getLogger()
|
|
try:
|
|
otel_handler = next(
|
|
h for h in root.handlers if type(h).__name__ == "StructlogOTLPLogHandler"
|
|
)
|
|
|
|
record = logging.LogRecord(
|
|
name="test.logger",
|
|
level=logging.INFO,
|
|
pathname="test.py",
|
|
lineno=5,
|
|
msg=({"event": "hello_openobserve", "foo": "bar", "count": 42},),
|
|
args=(),
|
|
exc_info=None,
|
|
)
|
|
record._logger = logging.getLogger("test.logger") # type: ignore[attr-defined]
|
|
record._name = "info" # type: ignore[attr-defined]
|
|
|
|
otel_record = otel_handler._translate(record)
|
|
|
|
assert otel_record.body == "hello_openobserve"
|
|
attrs = {k: v for k, v in otel_record.attributes.items() if not k.startswith("code.")}
|
|
assert attrs["foo"] == "bar"
|
|
assert attrs["count"] == 42
|
|
assert "_logger" not in attrs
|
|
assert "_name" not in attrs
|
|
finally:
|
|
shutdown_telemetry()
|
|
root.handlers = [h for h in root.handlers if type(h).__name__ != "StructlogOTLPLogHandler"]
|
|
|
|
|
|
def test_structlog_logs_reach_root_handlers() -> None:
|
|
"""After configure_logging, structlog events propagate to stdlib handlers."""
|
|
root = logging.getLogger()
|
|
assert (
|
|
any(type(h).__name__ in ("StreamHandler", "ProcessorFormatter") for h in root.handlers)
|
|
or root.handlers
|
|
)
|
|
|
|
log = get_logger("test")
|
|
log.info("smoke_test", answer=42)
|
|
|
|
|
|
@pytest.mark.parametrize("level", ["DEBUG", "INFO", "WARNING"])
|
|
def test_otel_handler_respects_configured_log_level(
|
|
monkeypatch: pytest.MonkeyPatch, level: str
|
|
) -> None:
|
|
"""The OTLP handler level matches settings.log_level so DEBUG logs can reach OpenObserve."""
|
|
from contract_check.core.config import get_settings
|
|
from contract_check.core.logging import configure_logging
|
|
from contract_check.core.telemetry import setup_telemetry, shutdown_telemetry
|
|
|
|
configure_logging(level, json_output=True, service="test-service", env="test")
|
|
monkeypatch.setattr(
|
|
get_settings(), "otel_exporter_otlp_endpoint", "http://localhost:9999/v1/logs"
|
|
)
|
|
monkeypatch.setattr(get_settings(), "log_level", level)
|
|
|
|
setup_telemetry("test")
|
|
root = logging.getLogger()
|
|
try:
|
|
otel_handler = next(
|
|
h for h in root.handlers if type(h).__name__ == "StructlogOTLPLogHandler"
|
|
)
|
|
assert otel_handler.level == getattr(logging, level)
|
|
finally:
|
|
shutdown_telemetry()
|
|
root.handlers = [h for h in root.handlers if type(h).__name__ != "StructlogOTLPLogHandler"]
|