DealDocumentScreening/tests/unit/test_telemetry.py
febux b279d6c61a Switch observability to passive collection: remove OTLP push, Vector replaces otel-collector
- core/telemetry.py is now a no-op (no opentelemetry imports); entrypoints
  no longer call setup/shutdown telemetry
- Remove all opentelemetry-* deps from pyproject groups; regenerate uv.lock
- Drop otel_exporter_otlp_endpoint/otel_service_name settings; sentry and
  auth use "contract-check" instead
- Stop publishing worker metrics ports; bind API metrics to 127.0.0.1 by
  default via API_METRICS_BIND_HOST
- Replace otel-collector with Vector in observer profile (docker_logs +
  prometheus_scrape -> OpenObserve); add deploy/observability/vector-config.yaml
- Update .env.example, docs (ARCHITECTURE, DEPLOY), README, Makefile
- Rewrite tests/unit/test_telemetry.py for the no-op implementation
2026-09-06 19:17:08 +03:00

87 lines
3 KiB
Python

"""Unit tests for telemetry after the passive-collection refactor.
The application no longer pushes OTLP. This module verifies that telemetry
setup/shutdown are safe no-ops and that structured logging still reaches the
configured root handler.
"""
from __future__ import annotations
import logging
import pytest
from contract_check.core.logging import configure_logging, get_logger
from contract_check.core.telemetry import get_meter, get_tracer, setup_telemetry, shutdown_telemetry
@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_setup_telemetry_is_safe_noop() -> None:
"""setup_telemetry() runs without error and does not add an OTLP handler."""
root = logging.getLogger()
before = list(root.handlers)
setup_telemetry("test")
try:
after = list(root.handlers)
# No new handlers are attached because OTLP is no longer initialized.
assert after == before
finally:
shutdown_telemetry()
def test_shutdown_telemetry_is_safe_noop() -> None:
"""shutdown_telemetry() can be called even when setup_telemetry() was skipped."""
shutdown_telemetry()
shutdown_telemetry() # idempotent
def test_otlp_endpoint_env_does_not_initialize_exporters(monkeypatch: pytest.MonkeyPatch) -> None:
"""A legacy OTLP endpoint env var must not cause the application to import OTel SDK."""
monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector:4318")
from contract_check.core.config import get_settings
# Settings explicitly ignores unknown env vars, but a stale env var must not
# be picked up by telemetry code. Refresh the cached settings instance.
get_settings.cache_clear()
root = logging.getLogger()
before = list(root.handlers)
setup_telemetry("test")
try:
after = list(root.handlers)
assert after == before
assert not any(type(h).__name__.startswith(("OTLP", "StructlogOTLP")) for h in after)
finally:
shutdown_telemetry()
get_settings.cache_clear()
def test_get_tracer_and_meter_are_safe_noop() -> None:
"""Tracer/meter accessors return objects that do not require OTel SDK."""
tracer = get_tracer("test")
meter = get_meter("test")
# They should be truthy and accept the expected creation calls without raising.
assert tracer
assert meter
span = tracer.start_span("ignored")
span.end()
counter = meter.create_counter("ignored")
counter.add(1)
def test_configure_logging_delivers_events_to_root_stream_handler() -> None:
"""After configure_logging, structlog events propagate to the root StreamHandler."""
root = logging.getLogger()
assert any(isinstance(h, logging.StreamHandler) for h in root.handlers)
log = get_logger("test")
log.info("smoke_test", answer=42)