opentelemetry: cleanup linter errors

Author: callmesi

tracing.py

@@ -34,6 +34,9 @@
 """
 from __future__ import annotations
 
+# opentelemetry packages are available at runtime but not in the linter env
+# pylint: disable=import-error
+
 import builtins
 import os
 from typing import Optional, Set
@@ -62,7 +65,8 @@ def force_flush(timeout_millis: int = 5000) -> bool:
 # pytest plugin entry point
 # ---------------------------------------------------------------------------
 
-def pytest_configure(config) -> None:
+
+def pytest_configure(config) -> None: # pylint: disable=unused-argument
     """Called once per worker process before test collection starts."""
     if not _TRACEPARENT:
         return
@@ -73,6 +77,7 @@ def pytest_configure(config) -> None:
 # Custom headers propagator
 # ---------------------------------------------------------------------------
 
+
 class _TraceIdSpanIdPropagator:
     """
     Minimal propagator that injects 'trace-id' and 'span-id' headers
@@ -84,9 +89,37 @@ class _TraceIdSpanIdPropagator:
     _FIELDS: Set[str] = {"trace-id", "span-id"}
 
     def inject(self, carrier, context: Optional[object] = None, setter=None) -> None:
+        """
+        Minimal propagator for injecting custom 'trace-id' and 'span-id' headers.
+
+        This class implements a propagator conforming to the OpenTelemetry propagators API,
+        but tailored for systems that expect lowercase, hyphenated 'trace-id' and 'span-id'
+        headers instead of standard OTel formats.
+
+        The propagator only performs injection (outbound context propagation) and does not
+        extract or parse incoming headers (extract() is a no-op). This is useful for test
+        environments or systems with custom trace context expectations.
+
+        Attributes:
+            _FIELDS (Set[str]): The set of field names injected by this propagator.
+
+        Methods:
+            inject(carrier, context=None, setter=None):
+                Injects 'trace-id' and 'span-id' headers into the provided carrier
+                using the current span context.
+
+            extract(carrier, context=None, getter=None):
+                No-op. Returns the context unmodified.
+
+            fields:
+                Returns the set of header names injected by this propagator.
+        """
+        # pylint: disable=import-outside-toplevel
         from opentelemetry import trace as otel_trace
         from opentelemetry.propagators.textmap import default_setter
 
+        # pylint: enable=import-outside-toplevel
+
         if setter is None:
             setter = default_setter
 
@@ -97,27 +130,44 @@ def inject(self, carrier, context: Optional[object] = None, setter=None) -> None
         setter.set(carrier, "trace-id", format(span_ctx.trace_id, "032x"))
         setter.set(carrier, "span-id", format(span_ctx.span_id, "016x"))
 
-    def extract(self, carrier, context=None, getter=None):
+    def extract(self, _carrier, context=None, _getter=None):
+        """
+        Returns the docstring for the _TraceIdSpanIdPropagator class.
+
+        :return: The class docstring.
+        :rtype: str
+        """
         return context
 
     @property
     def fields(self) -> Set[str]:
+        """
+        Returns the docstring for the _TraceIdSpanIdPropagator class.
+
+        :return: The class docstring as a string.
+        :rtype: str
+        """
         return self._FIELDS
 
 
 # ---------------------------------------------------------------------------
 # Internal helpers
 # ---------------------------------------------------------------------------
 
+
 def _attach_root_context() -> None:
     """
     Reconstruct the root SpanContext from the bash-generated TRACEPARENT env
     var and attach it to the running context so every span created during the
     test session inherits the same trace ID.
     """
-    from opentelemetry import trace as otel_trace, context as otel_context
+    # pylint: disable=import-outside-toplevel
+    from opentelemetry import trace as otel_trace
+    from opentelemetry import context as otel_context
     from opentelemetry.trace import NonRecordingSpan, SpanContext, TraceFlags
 
+    # pylint: enable=import-outside-toplevel
+
     parts = _TRACEPARENT.split("-")
     if len(parts) != 4:
         return
@@ -134,51 +184,86 @@ def _attach_root_context() -> None:
         pass
 
 
-def _bootstrap_otel() -> None:
-    global _provider
-    from opentelemetry import trace as otel_trace, propagate
+def _create_provider():
+    """Build a TracerProvider with a service-name resource and return it."""
+    # pylint: disable=import-outside-toplevel
+    from opentelemetry.sdk.resources import SERVICE_NAME, Resource
     from opentelemetry.sdk.trace import TracerProvider
-    from opentelemetry.sdk.resources import Resource, SERVICE_NAME
-    from opentelemetry.propagators.composite import CompositePropagator
-    from opentelemetry.propagators.b3 import B3MultiFormat
-    from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
+
+    # pylint: enable=import-outside-toplevel
 
     service_name = os.environ.get("OTEL_SERVICE_NAME", "atp3-python-runner")
     resource = Resource({SERVICE_NAME: service_name})
-    provider = TracerProvider(resource=resource)
-    _provider = provider
+    return TracerProvider(resource=resource)
+
 
+def _add_otlp_exporter(provider) -> None:
+    """Attach a BatchSpanProcessor with OTLPSpanExporter if an endpoint is configured."""
     otlp_endpoint = os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT")
-    if otlp_endpoint:
-        from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
-        from opentelemetry.sdk.trace.export import BatchSpanProcessor
-        provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
+    if not otlp_endpoint:
+        return
+    # pylint: disable=import-outside-toplevel
+    from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+    from opentelemetry.sdk.trace.export import BatchSpanProcessor
 
-    otel_trace.set_tracer_provider(provider)
+    # pylint: enable=import-outside-toplevel
+
+    provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
+
+
+def _setup_propagators(provider) -> None:
+    """Register the tracer provider and composite propagator globally."""
+    # pylint: disable=import-outside-toplevel
+    from opentelemetry import propagate
+    from opentelemetry import trace as otel_trace
+    from opentelemetry.propagators.b3 import B3MultiFormat
+    from opentelemetry.propagators.composite import CompositePropagator
+    from opentelemetry.trace.propagation.tracecontext import (
+        TraceContextTextMapPropagator,
+    )
+
+    # pylint: enable=import-outside-toplevel
 
-    composite = CompositePropagator([
-        TraceContextTextMapPropagator(),
-        B3MultiFormat(),
-        _TraceIdSpanIdPropagator(),
-    ])
+    otel_trace.set_tracer_provider(provider)
+    composite = CompositePropagator(
+        [
+            TraceContextTextMapPropagator(),
+            B3MultiFormat(),
+            _TraceIdSpanIdPropagator(),
+        ]
+    )
     propagate.set_global_textmap(composite)
 
-    _attach_root_context()
 
+def _instrument_http() -> None:
+    """Auto-instrument outgoing HTTP calls via requests and urllib3 if available."""
     try:
+        # pylint: disable=import-outside-toplevel
         from opentelemetry.instrumentation.requests import RequestsInstrumentor
+
+        # pylint: enable=import-outside-toplevel
         RequestsInstrumentor().instrument()
     except ImportError as e:
         print("RequestsInstrumentation import error, skipping tracing.", e.name, e.path)
-        pass
 
     try:
+        # pylint: disable=import-outside-toplevel
         from opentelemetry.instrumentation.urllib3 import URLLib3Instrumentor
+
+        # pylint: enable=import-outside-toplevel
         URLLib3Instrumentor().instrument()
     except ImportError as e:
         print("URLLib3Instrumentation import error, skipping tracing.", e.name, e.path)
-        pass
 
+
+def _bootstrap_otel() -> None:
+    """Orchestrate OTel initialisation: provider, exporter, propagators, instrumentation."""
+    global _provider  # pylint: disable=global-statement
+    _provider = _create_provider()
+    _add_otlp_exporter(_provider)
+    _setup_propagators(_provider)
+    _attach_root_context()
+    _instrument_http()
     _patch_print()
 
 
@@ -190,7 +275,11 @@ def _patch_print() -> None:
     _orig = builtins.print
 
     def _traced(*args, **kwargs):
+        # pylint: disable=import-outside-toplevel
         from opentelemetry import trace as otel_trace
+
+        # pylint: enable=import-outside-toplevel
+
         span_ctx = otel_trace.get_current_span().get_span_context()
         if span_ctx and span_ctx.is_valid:
             _orig(f"[traceId={format(span_ctx.trace_id, '032x')}]", *args, **kwargs)