diff --git a/.changelog/5704.fixed b/.changelog/5704.fixed new file mode 100644 index 00000000000..aef00dd8750 --- /dev/null +++ b/.changelog/5704.fixed @@ -0,0 +1 @@ +`opentelemetry-exporter-zipkin-json`: warn and fall back to the default instead of raising `ValueError` when `OTEL_EXPORTER_ZIPKIN_TIMEOUT` holds a non-integer value; the same fix is applied to `opentelemetry-exporter-zipkin-proto-http`. diff --git a/.changelog/5879.added b/.changelog/5879.added new file mode 100644 index 00000000000..abd26b9a5ff --- /dev/null +++ b/.changelog/5879.added @@ -0,0 +1 @@ +`opentelemetry-test-utils`: document every hand-written module, class and callable diff --git a/.pylintrc b/.pylintrc index bda4148beb5..6ee96abff4d 100644 --- a/.pylintrc +++ b/.pylintrc @@ -36,10 +36,6 @@ persistent=yes # Specify a configuration file. #rcfile= -# When enabled, pylint would attempt to guess common misconfiguration and emit -# user-friendly hints instead of false-positive error messages. -suggestion-mode=yes - # Allow loading of arbitrary C extensions. Extensions are imported into the # active Python interpreter and may run arbitrary code. unsafe-load-any-extension=no diff --git a/dev-requirements.txt b/dev-requirements.txt index caa46ac0c4a..c96dbd33898 100644 --- a/dev-requirements.txt +++ b/dev-requirements.txt @@ -1,4 +1,4 @@ -pylint==3.3.4 +pylint==4.0.9 griffe==1.15.0 httpretty==1.1.4 inline-snapshot==0.34.2 diff --git a/exporter/opentelemetry-exporter-zipkin-json/src/opentelemetry/exporter/zipkin/json/__init__.py b/exporter/opentelemetry-exporter-zipkin-json/src/opentelemetry/exporter/zipkin/json/__init__.py index e976a07b387..426230615bc 100644 --- a/exporter/opentelemetry-exporter-zipkin-json/src/opentelemetry/exporter/zipkin/json/__init__.py +++ b/exporter/opentelemetry-exporter-zipkin-json/src/opentelemetry/exporter/zipkin/json/__init__.py @@ -81,10 +81,27 @@ DEFAULT_ENDPOINT = "http://localhost:9411/api/v2/spans" REQUESTS_SUCCESS_STATUS_CODES = (200, 202) +_DEFAULT_TIMEOUT = 10 logger = logging.getLogger(__name__) +def _timeout_from_env() -> int: + value = environ.get(OTEL_EXPORTER_ZIPKIN_TIMEOUT) + if value is None or not value.strip(): + return _DEFAULT_TIMEOUT + try: + return int(value) + except ValueError: + logger.warning( + "Invalid value %r for %s, using default of %s seconds", + value, + OTEL_EXPORTER_ZIPKIN_TIMEOUT, + _DEFAULT_TIMEOUT, + ) + return _DEFAULT_TIMEOUT + + class ZipkinExporter(SpanExporter): def __init__( self, @@ -128,7 +145,7 @@ def __init__( self.session = session or requests.Session() self.session.headers.update({"Content-Type": self.encoder.content_type()}) self._closed = False - self.timeout = timeout or int(environ.get(OTEL_EXPORTER_ZIPKIN_TIMEOUT, 10)) + self.timeout = timeout or _timeout_from_env() def export(self, spans: Sequence[Span]) -> SpanExportResult: # After the call to Shutdown subsequent calls to Export are diff --git a/exporter/opentelemetry-exporter-zipkin-json/tests/test_zipkin_exporter.py b/exporter/opentelemetry-exporter-zipkin-json/tests/test_zipkin_exporter.py index ced0bd27d81..0e49d1d81f5 100644 --- a/exporter/opentelemetry-exporter-zipkin-json/tests/test_zipkin_exporter.py +++ b/exporter/opentelemetry-exporter-zipkin-json/tests/test_zipkin_exporter.py @@ -63,6 +63,52 @@ def test_constructor_env_vars(self): self.assertEqual(exporter.local_node.port, None) self.assertEqual(exporter.timeout, 15) + def test_constructor_invalid_timeout_env_var_falls_back(self): + with patch.dict(os.environ, {OTEL_EXPORTER_ZIPKIN_TIMEOUT: "10s"}): + with self.assertLogs("opentelemetry.exporter.zipkin.json", level="WARNING") as cm: + exporter = ZipkinExporter() + + self.assertEqual(exporter.timeout, 10) + self.assertEqual(len(cm.records), 1) + self.assertTrue( + any("Invalid value" in message for message in cm.output), + cm.output, + ) + + def test_constructor_empty_timeout_env_var_falls_back(self): + for value in ("", " ", "\t\n"): + with self.subTest(value=value), patch.dict(os.environ, {OTEL_EXPORTER_ZIPKIN_TIMEOUT: value}): + with self.assertNoLogs("opentelemetry.exporter.zipkin.json", level="WARNING"): + exporter = ZipkinExporter() + self.assertEqual(exporter.timeout, 10) + + def test_constructor_unset_timeout_env_var_uses_default(self): + with patch.dict(os.environ): + os.environ.pop(OTEL_EXPORTER_ZIPKIN_TIMEOUT, None) + with self.assertNoLogs("opentelemetry.exporter.zipkin.json", level="WARNING"): + exporter = ZipkinExporter() + self.assertEqual(exporter.timeout, 10) + + def test_constructor_numeric_timeout_env_var_preserves_values(self): + for value, expected in (("15", 15), (" 15 ", 15), ("0", 0), ("-1", -1)): + with self.subTest(value=value), patch.dict(os.environ, {OTEL_EXPORTER_ZIPKIN_TIMEOUT: value}): + with self.assertNoLogs("opentelemetry.exporter.zipkin.json", level="WARNING"): + exporter = ZipkinExporter() + self.assertEqual(exporter.timeout, expected) + + def test_constructor_explicit_timeout_ignores_invalid_env_var(self): + for value in ("10s", "", "\t\n"): + with self.subTest(value=value), patch.dict(os.environ, {OTEL_EXPORTER_ZIPKIN_TIMEOUT: value}): + with self.assertNoLogs("opentelemetry.exporter.zipkin.json", level="WARNING"): + exporter = ZipkinExporter(timeout=7) + self.assertEqual(exporter.timeout, 7) + + def test_constructor_explicit_timeout_preserves_truthiness(self): + with patch.dict(os.environ, {OTEL_EXPORTER_ZIPKIN_TIMEOUT: "15"}): + with self.assertNoLogs("opentelemetry.exporter.zipkin.json", level="WARNING"): + self.assertEqual(ZipkinExporter(timeout=0).timeout, 15) + self.assertEqual(ZipkinExporter(timeout=-1).timeout, -1) + def test_constructor_protocol_endpoint(self): """Test the constructor for the common usage of providing the protocol and endpoint arguments.""" diff --git a/exporter/opentelemetry-exporter-zipkin-proto-http/src/opentelemetry/exporter/zipkin/proto/http/__init__.py b/exporter/opentelemetry-exporter-zipkin-proto-http/src/opentelemetry/exporter/zipkin/proto/http/__init__.py index e664f20624b..28dc8f2c85c 100644 --- a/exporter/opentelemetry-exporter-zipkin-proto-http/src/opentelemetry/exporter/zipkin/proto/http/__init__.py +++ b/exporter/opentelemetry-exporter-zipkin-proto-http/src/opentelemetry/exporter/zipkin/proto/http/__init__.py @@ -78,10 +78,27 @@ DEFAULT_ENDPOINT = "http://localhost:9411/api/v2/spans" REQUESTS_SUCCESS_STATUS_CODES = (200, 202) +_DEFAULT_TIMEOUT = 10 logger = logging.getLogger(__name__) +def _timeout_from_env() -> int: + value = environ.get(OTEL_EXPORTER_ZIPKIN_TIMEOUT) + if value is None or not value.strip(): + return _DEFAULT_TIMEOUT + try: + return int(value) + except ValueError: + logger.warning( + "Invalid value %r for %s, using default of %s seconds", + value, + OTEL_EXPORTER_ZIPKIN_TIMEOUT, + _DEFAULT_TIMEOUT, + ) + return _DEFAULT_TIMEOUT + + class ZipkinExporter(SpanExporter): def __init__( self, @@ -121,7 +138,7 @@ def __init__( self.session = session or requests.Session() self.session.headers.update({"Content-Type": self.encoder.content_type()}) self._closed = False - self.timeout = timeout or int(environ.get(OTEL_EXPORTER_ZIPKIN_TIMEOUT, 10)) + self.timeout = timeout or _timeout_from_env() def export(self, spans: Sequence[Span]) -> SpanExportResult: # After the call to Shutdown subsequent calls to Export are diff --git a/exporter/opentelemetry-exporter-zipkin-proto-http/tests/test_zipkin_exporter.py b/exporter/opentelemetry-exporter-zipkin-proto-http/tests/test_zipkin_exporter.py index 78e30b0b310..522d4fb43cb 100644 --- a/exporter/opentelemetry-exporter-zipkin-proto-http/tests/test_zipkin_exporter.py +++ b/exporter/opentelemetry-exporter-zipkin-proto-http/tests/test_zipkin_exporter.py @@ -65,6 +65,52 @@ def test_constructor_env_vars(self): self.assertEqual(exporter.local_node.port, None) self.assertEqual(exporter.timeout, 15) + def test_constructor_invalid_timeout_env_var_falls_back(self): + with patch.dict(os.environ, {OTEL_EXPORTER_ZIPKIN_TIMEOUT: "10s"}): + with self.assertLogs("opentelemetry.exporter.zipkin.proto.http", level="WARNING") as cm: + exporter = ZipkinExporter() + + self.assertEqual(exporter.timeout, 10) + self.assertEqual(len(cm.records), 1) + self.assertTrue( + any("Invalid value" in message for message in cm.output), + cm.output, + ) + + def test_constructor_empty_timeout_env_var_falls_back(self): + for value in ("", " ", "\t\n"): + with self.subTest(value=value), patch.dict(os.environ, {OTEL_EXPORTER_ZIPKIN_TIMEOUT: value}): + with self.assertNoLogs("opentelemetry.exporter.zipkin.proto.http", level="WARNING"): + exporter = ZipkinExporter() + self.assertEqual(exporter.timeout, 10) + + def test_constructor_unset_timeout_env_var_uses_default(self): + with patch.dict(os.environ): + os.environ.pop(OTEL_EXPORTER_ZIPKIN_TIMEOUT, None) + with self.assertNoLogs("opentelemetry.exporter.zipkin.proto.http", level="WARNING"): + exporter = ZipkinExporter() + self.assertEqual(exporter.timeout, 10) + + def test_constructor_numeric_timeout_env_var_preserves_values(self): + for value, expected in (("15", 15), (" 15 ", 15), ("0", 0), ("-1", -1)): + with self.subTest(value=value), patch.dict(os.environ, {OTEL_EXPORTER_ZIPKIN_TIMEOUT: value}): + with self.assertNoLogs("opentelemetry.exporter.zipkin.proto.http", level="WARNING"): + exporter = ZipkinExporter() + self.assertEqual(exporter.timeout, expected) + + def test_constructor_explicit_timeout_ignores_invalid_env_var(self): + for value in ("10s", "", "\t\n"): + with self.subTest(value=value), patch.dict(os.environ, {OTEL_EXPORTER_ZIPKIN_TIMEOUT: value}): + with self.assertNoLogs("opentelemetry.exporter.zipkin.proto.http", level="WARNING"): + exporter = ZipkinExporter(timeout=7) + self.assertEqual(exporter.timeout, 7) + + def test_constructor_explicit_timeout_preserves_truthiness(self): + with patch.dict(os.environ, {OTEL_EXPORTER_ZIPKIN_TIMEOUT: "15"}): + with self.assertNoLogs("opentelemetry.exporter.zipkin.proto.http", level="WARNING"): + self.assertEqual(ZipkinExporter(timeout=0).timeout, 15) + self.assertEqual(ZipkinExporter(timeout=-1).timeout, -1) + def test_constructor_protocol_endpoint(self): """Test the constructor for the common usage of providing the protocol and endpoint arguments.""" diff --git a/opentelemetry-configuration/src/opentelemetry/configuration/_conversion.py b/opentelemetry-configuration/src/opentelemetry/configuration/_conversion.py index e5ceb0f59f5..dca1e14c805 100644 --- a/opentelemetry-configuration/src/opentelemetry/configuration/_conversion.py +++ b/opentelemetry-configuration/src/opentelemetry/configuration/_conversion.py @@ -14,7 +14,7 @@ from collections.abc import Mapping from dataclasses import MISSING, fields, is_dataclass from enum import Enum -from types import UnionType +from types import NoneType, UnionType from typing import Any, TypeVar, Union, get_args, get_origin, get_type_hints _T = TypeVar("_T") @@ -27,7 +27,7 @@ def _unwrap_optional(type_hint: Any) -> Any: """ origin = get_origin(type_hint) if origin is UnionType or origin is Union: - non_none = [t for t in get_args(type_hint) if t is not type(None)] + non_none = [t for t in get_args(type_hint) if t is not NoneType] if len(non_none) == 1: return non_none[0] return type_hint diff --git a/opentelemetry-semantic-conventions/.pylintrc b/opentelemetry-semantic-conventions/.pylintrc index b034d8c75ad..621216e432e 100644 --- a/opentelemetry-semantic-conventions/.pylintrc +++ b/opentelemetry-semantic-conventions/.pylintrc @@ -36,10 +36,6 @@ persistent=yes # Specify a configuration file. #rcfile= -# When enabled, pylint would attempt to guess common misconfiguration and emit -# user-friendly hints instead of false-positive error messages. -suggestion-mode=yes - # Allow loading of arbitrary C extensions. Extensions are imported into the # active Python interpreter and may run arbitrary code. unsafe-load-any-extension=no diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/__init__.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/__init__.py index 070de9de7cb..1e7fab9490a 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/__init__.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/__init__.py @@ -3,34 +3,140 @@ # type: ignore +"""Shared test helpers for the OpenTelemetry Python packages. + +This package is published as ``opentelemetry-test-utils``. It is a test +dependency of the other distributions in this repository and of the +instrumentation repositories, never a runtime dependency of library or +application code. + +The module itself provides ``TestCase``, a ``unittest.TestCase`` extended with +an ``assertNotRaises`` assertion. The submodules provide the rest: base classes +that reset the API globals and capture spans and metrics in memory +(``opentelemetry.test.test_base``, ``opentelemetry.test.globals_test``), WSGI, +ASGI and HTTP server fixtures (``opentelemetry.test.wsgitestutil``, +``opentelemetry.test.asgitestutil``, ``opentelemetry.test.httptest``), builders +for spans, metrics and samplers (``opentelemetry.test.spantestutil``, +``opentelemetry.test.metrictestutil``, ``opentelemetry.test.samplertestutil``), +propagator doubles (``opentelemetry.test.mock_textmap``), an entry point double +(``opentelemetry.test.mock_test_classes``), helpers for provoking a thread race +(``opentelemetry.test.concurrency_test``), an in-process OTLP/HTTP receiver +(``opentelemetry.test._otlp_test_server``), a semantic convention check against +weaver (``opentelemetry.test.weaver_live_check``), and the version of the +distribution (``opentelemetry.test.version``). +""" + from traceback import format_tb from unittest import TestCase class _AssertNotRaisesMixin: + """Mixin that adds an ``assertNotRaises`` assertion to a test case. + + ``unittest`` has ``assertRaises`` but no inverse, because an unexpected + exception already fails a test by propagating out of it. ``assertNotRaises`` + exists so that a test can state the expectation explicitly and fail with the + traceback of the unwanted exception formatted into the failure message, + rather than with the raw exception. + + Mix it in before ``unittest.TestCase`` in the base class list, so that its + ``__init__`` runs first and installs the assertion. The ``TestCase`` defined + in this module already does that. + """ + class _AssertNotRaises: + """Callable context manager that fails its test case on given exceptions. + + An instance is bound to ``self.assertNotRaises`` on the test case. It is + both the callable and the context manager: calling it records the + exception types to watch for and returns the same instance, which is + then entered. Entering it without calling it first succeeds, because + ``__enter__`` does not read the exception types; the omission surfaces + as an ``AttributeError`` out of ``__exit__``, and only when the block + raised something. + + The instance is shared across every use within one test case, so nesting + two ``assertNotRaises`` blocks does not work: the inner call overwrites + the exception types of the outer one. + + Args: + test_case: The test case to call ``fail`` on. Stored, not copied, so + the failure is reported against the running test. + """ + def __init__(self, test_case): + """Parameters are documented on the class docstring.""" self._test_case = test_case def __enter__(self): + """Return this instance so the ``with`` body can run. + + Returns: + The context manager itself. The value is of no use to the body + and is returned only because entering a context manager binds a + value. + """ return self def __exit__(self, type_, value, tb): # pylint: disable=invalid-name + """Fail the test case when one of the watched exceptions was raised. + + The exception type is matched by identity against the types passed + to ``__call__``, not with ``issubclass``, so a subclass of a watched + type does not match. + + Args: + type_: Class of the exception raised in the ``with`` body, or + ``None`` when the body completed normally. + value: The exception instance, or ``None``. + tb: The traceback of the exception, or ``None``. It is formatted + into the failure message. + + Returns: + Always ``True``, which suppresses whatever exception the ``with`` + body raised, including an exception of a type that was not + watched. + """ if value is not None and type_ in self._exception_types: self._test_case.fail("Unexpected exception was raised:\n{}".format("\n".join(format_tb(tb)))) return True def __call__(self, exception, *exceptions): + """Record the exception types to watch for and return this instance. + + Args: + exception: An exception class that must not be raised. At least + one is required, so that an empty watch list cannot be set + by accident. + *exceptions: Further exception classes that must not be raised. + + Returns: + The context manager itself, ready to be entered. The types + recorded here replace any recorded by an earlier call. + """ # pylint: disable=attribute-defined-outside-init self._exception_types = (exception, *exceptions) return self def __init__(self, *args, **kwargs): + """Install ``assertNotRaises`` on this test case. + + Args: + *args: Forwarded unchanged to the next class in the method + resolution order, which is the ``unittest.TestCase`` this mixin + is combined with. + **kwargs: Forwarded unchanged for the same reason. + """ super().__init__(*args, **kwargs) # pylint: disable=invalid-name self.assertNotRaises = self._AssertNotRaises(self) class TestCase(_AssertNotRaisesMixin, TestCase): # pylint: disable=function-redefined - pass + """A ``unittest.TestCase`` that also provides ``assertNotRaises``. + + This class deliberately shadows the ``unittest.TestCase`` imported in this + module, so that a test module can switch to it by changing only its import + and keep writing ``class MyTest(TestCase)``. + """ diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/_otlp_test_server.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/_otlp_test_server.py index f7c3b053ed3..76419e1a9cb 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/_otlp_test_server.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/_otlp_test_server.py @@ -1,6 +1,24 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +"""An in-process OTLP/HTTP receiver for tests of the OTLP exporters. + +:class:`OtlpProtoTestServer` listens on a loopback port, accepts the protobuf +over HTTP that the OTLP exporters send, and puts every span, metric and log +record it decodes on a queue a test can read. A request on one of the three +OTLP paths whose body the protobuf parser accepts is answered with the empty +success response of the corresponding service, so the exporter under test sees +an ordinary successful export. ``do_POST`` describes the two cases answered +otherwise. + +The module is private, because the queue-per-signal shape and the ``Recorded*`` +dataclasses are chosen to suit the exporter tests in this repository. + +``opentelemetry-proto`` is imported lazily, so importing this module costs +nothing when the server is not used. Constructing the server without that +distribution installed raises ``ImportError`` with an explanation. +""" + from __future__ import annotations import gzip @@ -23,6 +41,22 @@ @dataclass class RecordedSpan: + """One span received by the server, with the context it arrived in. + + An OTLP request nests spans under a scope and a resource, so a span taken + out of it on its own has lost both. This keeps the three together. + + Attributes: + span: The protobuf span, as received. It is the object from the parsed + request, not a copy. + resource: The protobuf resource of the ``ResourceSpans`` the span + arrived in. + scope: The protobuf instrumentation scope of the ``ScopeSpans`` the + span arrived in. + resource_schema_url: Schema URL declared on the ``ResourceSpans``, which + is the empty string when the exporter sent none. + """ + span: Span resource: Resource scope: InstrumentationScope @@ -31,6 +65,19 @@ class RecordedSpan: @dataclass class RecordedMetric: + """One metric received by the server, with the context it arrived in. + + Attributes: + metric: The protobuf metric, as received, carrying all of its data + points. + resource: The protobuf resource of the ``ResourceMetrics`` the metric + arrived in. + scope: The protobuf instrumentation scope of the ``ScopeMetrics`` the + metric arrived in. + resource_schema_url: Schema URL declared on the ``ResourceMetrics``, + which is the empty string when the exporter sent none. + """ + metric: Metric resource: Resource scope: InstrumentationScope @@ -39,6 +86,18 @@ class RecordedMetric: @dataclass class RecordedLogRecord: + """One log record received by the server, with the context it arrived in. + + Attributes: + log_record: The protobuf log record, as received. + resource: The protobuf resource of the ``ResourceLogs`` the record + arrived in. + scope: The protobuf instrumentation scope of the ``ScopeLogs`` the + record arrived in. + resource_schema_url: Schema URL declared on the ``ResourceLogs``, which + is the empty string when the exporter sent none. + """ + log_record: LogRecord resource: Resource scope: InstrumentationScope @@ -53,6 +112,33 @@ def _make_handler( metrics_path: str, logs_path: str, ) -> type[BaseHTTPRequestHandler]: + """Build a request handler class bound to given queues and paths. + + The handler is built here rather than defined at module level for two + reasons: the generated protobuf modules are imported inside this function, + so importing this module does not require ``opentelemetry-proto``, and the + queues and paths are closed over, so the handler needs no per-instance + configuration that ``HTTPServer`` has no way to supply. + + Args: + spans_queue: Queue that each decoded span is put on, one + :class:`RecordedSpan` per span. + metrics_queue: Queue that each decoded metric is put on. + logs_queue: Queue that each decoded log record is put on. + traces_path: Request path that carries ``ExportTraceServiceRequest`` + bodies, compared for equality after the query string is stripped. + metrics_path: Request path that carries + ``ExportMetricsServiceRequest`` bodies. + logs_path: Request path that carries ``ExportLogsServiceRequest`` + bodies. + + Returns: + A ``BaseHTTPRequestHandler`` subclass. Each call returns a new class + closing over the arguments of that call. + + Raises: + ImportError: When ``opentelemetry-proto`` is not installed. + """ # pylint: disable=import-outside-toplevel,no-name-in-module from opentelemetry.proto.collector.logs.v1.logs_service_pb2 import ( # noqa: PLC0415 ExportLogsServiceRequest, @@ -68,7 +154,20 @@ def _make_handler( ) class _Handler(BaseHTTPRequestHandler): + """Handler that decodes an OTLP export request onto the right queue.""" + def do_POST(self): # pylint: disable=invalid-name + """Decode one export request and answer with the success response. + + The path decides which signal the body holds. A path matching none + of the three is answered with ``404`` and nothing is queued. A + recognised path whose body the protobuf parser accepts is answered + with ``200`` and the empty success response of that OTLP service, + so partial success is never reported. A body the parser rejects + raises out of this method instead, and the connection is closed + with no status line, so the exporter sees a transport failure and + takes the retry path. + """ content_length = int(self.headers.get("Content-Length", 0)) body = self._decompress(self.rfile.read(content_length)) path = urlparse(self.path).path @@ -90,6 +189,18 @@ def do_POST(self): # pylint: disable=invalid-name self.wfile.write(response_body) def _decompress(self, body: bytes) -> bytes: + """Undo the compression named by the ``Content-Encoding`` header. + + Args: + body: The raw request body. + + Returns: + The decompressed body for ``gzip`` and ``deflate``, and the + body unchanged for any other encoding including an absent + header. An encoding the exporter applied that is not one of + those two therefore reaches the protobuf parser compressed and + fails there. + """ encoding = self.headers.get("Content-Encoding", "") if encoding == "gzip": return gzip.decompress(body) @@ -99,6 +210,15 @@ def _decompress(self, body: bytes) -> bytes: @staticmethod def _handle_traces(body: bytes) -> bytes: + """Queue every span in a trace export request. + + Args: + body: The decompressed ``ExportTraceServiceRequest`` protobuf. + + Returns: + A serialized empty ``ExportTraceServiceResponse``, which tells + the exporter that every span was accepted. + """ request = ExportTraceServiceRequest() request.ParseFromString(body) for rs in request.resource_spans: @@ -116,6 +236,15 @@ def _handle_traces(body: bytes) -> bytes: @staticmethod def _handle_metrics(body: bytes) -> bytes: + """Queue every metric in a metrics export request. + + Args: + body: The decompressed ``ExportMetricsServiceRequest`` + protobuf. + + Returns: + A serialized empty ``ExportMetricsServiceResponse``. + """ request = ExportMetricsServiceRequest() request.ParseFromString(body) for rm in request.resource_metrics: @@ -133,6 +262,14 @@ def _handle_metrics(body: bytes) -> bytes: @staticmethod def _handle_logs(body: bytes) -> bytes: + """Queue every log record in a logs export request. + + Args: + body: The decompressed ``ExportLogsServiceRequest`` protobuf. + + Returns: + A serialized empty ``ExportLogsServiceResponse``. + """ request = ExportLogsServiceRequest() request.ParseFromString(body) for rl in request.resource_logs: @@ -149,13 +286,62 @@ def _handle_logs(body: bytes) -> bytes: return ExportLogsServiceResponse().SerializeToString() def log_message(self, format, *args): # pylint: disable=redefined-builtin - pass + """Discard the access log line ``BaseHTTPRequestHandler`` writes. + + The default implementation writes to stderr, which would put a line + per export into the output of every passing test. + + Args: + format: Printf-style format string of the line to discard. + *args: Values the format string would have been given. + """ return _Handler class OtlpProtoTestServer: + """An OTLP/HTTP receiver that keeps what it receives in memory. + + Use it as a context manager, which starts it on entry and stops it on exit:: + + with OtlpProtoTestServer() as server: + exporter = OTLPSpanExporter(endpoint=server.traces_endpoint) + exporter.export([span]) + recorded = server.wait_for_span(name="my-span") + + The three endpoint properties give the URLs to configure an exporter with. + Under the default ``port`` of ``0`` they are meaningful only after + ``start``, because the port is assigned on bind; when an explicit port was + requested they already hold it before ``start``. + Received telemetry is read either by waiting for a specific item, with + ``get_span``, ``wait_for_span`` and their metric and log counterparts, or by + taking whatever has arrived, with ``drain_spans``, ``drain_metrics`` and + ``drain_log_records``. Every read removes what it returns, so an item is + seen once. A waiting read also removes and discards what it passes over: a + ``wait_for_*`` call drops every non-matching item it takes off the queue, + and a ``get_*`` call that gives up drops the items it had already collected + along with raising ``TimeoutError``. + + Requests are served from a single daemon thread, so exports are decoded one + at a time in the order they arrive. + + Args: + host: Address to bind to. The default keeps the server off the network. + port: Port to bind to. The default of ``0`` lets the operating system + choose a free one, which is what lets several tests run at once; + read the chosen port from the ``port`` property after ``start``. + base_path: Prefix the three OTLP paths are served under, for example + ``/otlp`` to serve ``/otlp/v1/traces``. A trailing slash is + stripped. The default serves ``/v1/traces``, ``/v1/metrics`` and + ``/v1/logs``, which is where an exporter looks by default. + + Raises: + ImportError: When ``opentelemetry-proto`` is not installed, since the + server cannot decode a request without it. + """ + def __init__(self, host: str = "127.0.0.1", port: int = 0, base_path: str = "") -> None: + """Parameters are documented on the class docstring.""" try: # pylint: disable-next=import-outside-toplevel,unused-import import opentelemetry.proto # noqa: PLC0415, F401 @@ -174,6 +360,16 @@ def __init__(self, host: str = "127.0.0.1", port: int = 0, base_path: str = "") self._thread: Thread | None = None def start(self) -> OtlpProtoTestServer: + """Bind the socket and serve requests from a background thread. + + Returns: + This server, so the call can be chained. After it returns, ``port`` + holds the bound port and the three endpoint properties are usable. + + Raises: + OSError: When the address and port cannot be bound, which is what + happens when an explicit ``port`` is already in use. + """ handler = _make_handler( self._spans_queue, self._metrics_queue, @@ -193,6 +389,14 @@ def start(self) -> OtlpProtoTestServer: return self def stop(self) -> None: + """Stop serving, join the thread and close the socket. + + Does nothing when the server was never started. Queued telemetry is + kept, so a test may read it after stopping. Calling ``start`` again + builds a new ``HTTPServer`` and works, and it rebinds the port assigned + to the previous run rather than asking for a new one, because that port + was written over the ``0`` requested at construction. + """ if self._server is not None: self._server.shutdown() if self._thread is not None: @@ -200,34 +404,103 @@ def stop(self) -> None: self._server.server_close() def __enter__(self) -> OtlpProtoTestServer: + """Start the server for the duration of the block. + + Returns: + This server, already serving. + """ return self.start() def __exit__(self, *exc: object) -> None: + """Stop the server, whether the block completed or raised. + + Args: + *exc: The exception type, value and traceback that ``with`` + supplies. They are ignored, so an exception raised in the block + propagates. + """ self.stop() @property def port(self) -> int: + """Return the TCP port the server listens on. + + Returns: + The port assigned by the operating system once ``start`` has bound + the socket, or the port requested at construction before that. It + is ``0`` before ``start`` when no explicit port was requested. + """ return self._port @property def traces_endpoint(self) -> str: + """Return the URL to configure a span exporter with. + + Returns: + The full ``http://`` URL of the traces path, including the base + path. Under the default ``port`` of ``0`` it is meaningful only + after ``start``, since the port is part of it; an explicit port is + already in the URL before that. The scheme is plain HTTP, so an + exporter has to be configured not to use TLS. + """ return f"http://{self._host}:{self._port}{self._base_path}/v1/traces" @property def metrics_endpoint(self) -> str: + """Return the URL to configure a metric exporter with. + + Returns: + The full ``http://`` URL of the metrics path, including the base + path, meaningful under the same condition as ``traces_endpoint``. + """ return f"http://{self._host}:{self._port}{self._base_path}/v1/metrics" @property def logs_endpoint(self) -> str: + """Return the URL to configure a log exporter with. + + Returns: + The full ``http://`` URL of the logs path, including the base path, + meaningful under the same condition as ``traces_endpoint``. + """ return f"http://{self._host}:{self._port}{self._base_path}/v1/logs" def get_span(self, timeout: float = 5.0) -> RecordedSpan: + """Take the next span off the queue, waiting for one to arrive. + + Args: + timeout: Seconds to wait before giving up. + + Returns: + The oldest span not yet taken, removed from the queue. + + Raises: + TimeoutError: When no span arrives within ``timeout``. + """ try: return self._spans_queue.get(timeout=timeout) except Empty: raise TimeoutError(f"No span received within {timeout}s") from None def get_spans(self, count: int = 1, timeout: float = 5.0) -> list[RecordedSpan]: + """Take a given number of spans off the queue. + + Args: + count: How many spans to collect. + timeout: Seconds to wait for all of them together, not for each + one, so the call returns within roughly this long however the + spans are spread out. + + Returns: + A list of ``count`` spans, oldest first, removed from the queue. + + Raises: + TimeoutError: When fewer than ``count`` spans arrive in time. The + spans already taken off the queue are discarded with the + exception. The message names how many were collected only when + the deadline passes between two successful reads; the usual + case is the timeout of the read itself, which names no count. + """ deadline = time.monotonic() + timeout spans = [] for _ in range(count): @@ -238,6 +511,22 @@ def get_spans(self, count: int = 1, timeout: float = 5.0) -> list[RecordedSpan]: return spans def wait_for_span(self, *, name: str | None = None, timeout: float = 5.0) -> RecordedSpan: + """Wait for a span with a given name, discarding the ones before it. + + Args: + name: Exact span name to wait for. ``None`` accepts the next span + whatever its name, which makes the call equivalent to + ``get_span``. + timeout: Seconds to wait before giving up. + + Returns: + The first span whose name matches, removed from the queue. + + Raises: + TimeoutError: When no matching span arrives within ``timeout``. + Every non-matching span taken off the queue while waiting is + discarded, so a test that needs those has to read them first. + """ deadline = time.monotonic() + timeout while True: remaining = deadline - time.monotonic() @@ -251,6 +540,14 @@ def wait_for_span(self, *, name: str | None = None, timeout: float = 5.0) -> Rec return recorded def drain_spans(self) -> list[RecordedSpan]: + """Take every span queued so far, without waiting. + + Returns: + A list of the queued spans, oldest first, leaving the queue empty. + Nothing queued gives an empty list. Because it does not wait, an + export still in flight is missed; flush the exporter under test + first. + """ result = [] while True: try: @@ -259,12 +556,39 @@ def drain_spans(self) -> list[RecordedSpan]: return result def get_metric(self, timeout: float = 5.0) -> RecordedMetric: + """Take the next metric off the queue, waiting for one to arrive. + + Args: + timeout: Seconds to wait before giving up. + + Returns: + The oldest metric not yet taken, removed from the queue. One + exported metric is one item however many data points it carries. + + Raises: + TimeoutError: When no metric arrives within ``timeout``. + """ try: return self._metrics_queue.get(timeout=timeout) except Empty: raise TimeoutError(f"No metric received within {timeout}s") from None def get_metrics(self, count: int = 1, timeout: float = 5.0) -> list[RecordedMetric]: + """Take a given number of metrics off the queue. + + Args: + count: How many metrics to collect. + timeout: Seconds to wait for all of them together. + + Returns: + A list of ``count`` metrics, oldest first, removed from the queue. + + Raises: + TimeoutError: When fewer than ``count`` metrics arrive in time. The + metrics already taken off the queue are discarded with the + exception. The message names how many were collected only when + the deadline passes between two successful reads. + """ deadline = time.monotonic() + timeout metrics = [] for _ in range(count): @@ -275,6 +599,22 @@ def get_metrics(self, count: int = 1, timeout: float = 5.0) -> list[RecordedMetr return metrics def wait_for_metric(self, *, name: str | None = None, timeout: float = 5.0) -> RecordedMetric: + """Wait for a metric with a given name, discarding the ones before it. + + Args: + name: Exact metric name to wait for. ``None`` accepts the next + metric whatever its name. + timeout: Seconds to wait before giving up. + + Returns: + The first metric whose name matches, removed from the queue. + + Raises: + TimeoutError: When no matching metric arrives within ``timeout``. + Every non-matching metric taken off the queue while waiting is + discarded, which matters more here than for spans because one + export carries every instrument of a meter. + """ deadline = time.monotonic() + timeout while True: remaining = deadline - time.monotonic() @@ -288,6 +628,12 @@ def wait_for_metric(self, *, name: str | None = None, timeout: float = 5.0) -> R return recorded def drain_metrics(self) -> list[RecordedMetric]: + """Take every metric queued so far, without waiting. + + Returns: + A list of the queued metrics, oldest first, leaving the queue + empty. Nothing queued gives an empty list. + """ result = [] while True: try: @@ -296,12 +642,39 @@ def drain_metrics(self) -> list[RecordedMetric]: return result def get_log_record(self, timeout: float = 5.0) -> RecordedLogRecord: + """Take the next log record off the queue, waiting for one to arrive. + + Args: + timeout: Seconds to wait before giving up. + + Returns: + The oldest log record not yet taken, removed from the queue. + + Raises: + TimeoutError: When no log record arrives within ``timeout``. + """ try: return self._logs_queue.get(timeout=timeout) except Empty: raise TimeoutError(f"No log record received within {timeout}s") from None def get_log_records(self, count: int = 1, timeout: float = 5.0) -> list[RecordedLogRecord]: + """Take a given number of log records off the queue. + + Args: + count: How many log records to collect. + timeout: Seconds to wait for all of them together. + + Returns: + A list of ``count`` log records, oldest first, removed from the + queue. + + Raises: + TimeoutError: When fewer than ``count`` log records arrive in time. + The records already taken off the queue are discarded with the + exception. The message names how many were collected only when + the deadline passes between two successful reads. + """ deadline = time.monotonic() + timeout log_records = [] for _ in range(count): @@ -312,12 +685,33 @@ def get_log_records(self, count: int = 1, timeout: float = 5.0) -> list[Recorded return log_records def wait_for_log_record(self, *, timeout: float = 5.0) -> RecordedLogRecord: + """Wait for the next log record to arrive. + + There is no name to filter on, so unlike ``wait_for_span`` and + ``wait_for_metric`` this discards nothing and is the same as + ``get_log_record`` with a keyword-only timeout. + + Args: + timeout: Seconds to wait before giving up. + + Returns: + The oldest log record not yet taken, removed from the queue. + + Raises: + TimeoutError: When no log record arrives within ``timeout``. + """ try: return self._logs_queue.get(timeout=timeout) except Empty: raise TimeoutError(f"No log record received within {timeout}s") from None def drain_log_records(self) -> list[RecordedLogRecord]: + """Take every log record queued so far, without waiting. + + Returns: + A list of the queued log records, oldest first, leaving the queue + empty. Nothing queued gives an empty list. + """ result = [] while True: try: @@ -326,6 +720,12 @@ def drain_log_records(self) -> list[RecordedLogRecord]: return result def clear(self) -> None: + """Discard everything queued on all three signals. + + Call it between phases of a test so that telemetry from the setup phase + is not read as telemetry from the phase under test. The server keeps + serving, so anything exported after this call is queued normally. + """ for queue in ( self._spans_queue, self._metrics_queue, diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/asgitestutil.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/asgitestutil.py index c831a33721d..56a25715d11 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/asgitestutil.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/asgitestutil.py @@ -1,6 +1,20 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +"""``TestBase`` subclasses that drive an ASGI application through ``asgiref``. + +An ASGI application is a coroutine taking a scope dict and a pair of ``receive`` +and ``send`` callables, so a test has to build the scope, feed messages in and +collect messages out. ``asgiref.testing.ApplicationCommunicator`` does that +bookkeeping; the classes here hold one, supply a default HTTP scope and expose +the send and receive steps as methods. + +Pick :class:`AsyncAsgiTestBase` for new tests: it inherits +``unittest.IsolatedAsyncioTestCase``, so its test methods are coroutines that +the runner awaits directly. :class:`AsgiTestBase` is the older form, whose test +methods are ordinary functions that drive the event loop themselves. +""" + import asyncio from unittest import IsolatedAsyncioTestCase @@ -10,6 +24,21 @@ def setup_testing_defaults(scope): + """Fill an ASGI scope with the fields of a plain ``GET /`` over HTTP. + + Named after ``wsgiref.util.setup_testing_defaults``, and like it, this + updates the mapping in place and returns nothing. Keys the mapping already + carries are overwritten, so set the fields a test cares about after calling + this rather than before. + + Args: + scope: The connection scope to fill. It gains ``client``, ``headers``, + ``http_version``, ``method``, ``path``, ``query_string``, + ``scheme``, ``server`` and ``type``. The defaults describe an + HTTP/1.0 ``GET /`` from ``127.0.0.1`` with no headers and no query + string. Fields an ASGI server would also set, ``asgi`` and + ``root_path`` among them, are not added. + """ scope.update( { "client": ("127.0.0.1", 32767), @@ -26,7 +55,32 @@ def setup_testing_defaults(scope): class AsgiTestBase(TestBase): + """Base class for a synchronous test that drives an ASGI application. + + Test methods are ordinary functions, and each call into the application + runs the event loop to completion through + ``asyncio.get_event_loop().run_until_complete``. Prefer + :class:`AsyncAsgiTestBase` in new code, whose test methods are coroutines. + + Call ``seed_app`` with the application first, then ``send_input`` or + ``send_default_request``, then ``get_output`` or ``get_all_output``. + ``self.scope`` is the scope the application will see. The communicator + keeps a reference to it and does not start the application, which begins on + the first ``send_input`` or ``get_output``, so an edit made any time before + that first call still reaches the application. + + ``tearDown`` waits up to one second for the application to finish and + cancels it after that, and it does not call ``TestBase.tearDown``, so the + trace and metrics globals this class installed stay installed until the next + ``setUp`` replaces them. + """ + def setUp(self): + """Install the in-memory pipelines and build a default HTTP scope. + + Leaves ``self.communicator`` as ``None``: there is no application until + ``seed_app`` is called. + """ super().setUp() self.scope = {} @@ -34,23 +88,79 @@ def setUp(self): self.communicator = None def tearDown(self): + """Wait up to one second for the seeded application to finish. + + Waiting surfaces an exception the application raised after the last + message a test read, which would otherwise be swallowed. One second is + the default of ``ApplicationCommunicator.wait`` and this call does not + override it, so an application still running when that second expires + is cancelled rather than awaited, and a test that leaves one running + pays the full second. This override does not call + ``super().tearDown()``, so the trace and metrics globals are not reset + here. + """ if self.communicator: asyncio.get_event_loop().run_until_complete(self.communicator.wait()) def seed_app(self, app): + """Attach an ASGI application to the scope built by ``setUp``. + + Call this once per test. Calling it again replaces the communicator and + abandons the previous application without waiting for it. + + Args: + app: The ASGI application to drive. The communicator is given a + reference to ``self.scope`` and does not instantiate the + application here: that happens on the first ``send_input`` or + ``get_output``, so an edit to ``self.scope`` made before that + first call does reach the application. + """ self.communicator = ApplicationCommunicator(app, self.scope) def send_input(self, message): + """Send one ASGI message to the application and run the loop. + + Args: + message: The message to deliver, for example + ``{"type": "http.request", "body": b""}``. Its ``type`` has to + be one the application expects for the scope type. + """ asyncio.get_event_loop().run_until_complete(self.communicator.send_input(message)) def send_default_request(self): + """Send an ``http.request`` message with an empty body. + + This is the one message a plain ``GET`` needs, since the absence of + ``more_body`` tells the application the body is complete. + """ self.send_input({"type": "http.request", "body": b""}) def get_output(self): + """Take the next message the application sent. + + Returns: + The next ASGI message, for example an ``http.response.start``. + + Raises: + asyncio.TimeoutError: When no message is already queued. The wait + is zero seconds, so this reads what the application has + produced rather than waiting for it to produce more. A read + that times out also cancels the application when it has not + finished, so this ends the conversation rather than pausing it. + """ output = asyncio.get_event_loop().run_until_complete(self.communicator.receive_output(0)) return output def get_all_output(self): + """Take every message the application has already sent. + + Returns: + A list of ASGI messages, in the order the application sent them, + emptied out until nothing is queued. An application that has sent + nothing yields an empty list rather than an error. Draining ends + with a read that times out, which cancels an application that has + not finished, so do not drain one that still has input to receive. + """ outputs = [] while True: try: @@ -61,7 +171,22 @@ def get_all_output(self): class AsyncAsgiTestBase(TestBase, IsolatedAsyncioTestCase): + """Base class for a test whose methods are coroutines. + + Same shape as :class:`AsgiTestBase`, but ``send_input``, + ``send_default_request``, ``get_output`` and ``get_all_output`` are + coroutines that a test method awaits, and ``IsolatedAsyncioTestCase`` runs + each test method on its own event loop. Prefer this class in new code. + + ``tearDown`` has the same limit as in :class:`AsgiTestBase`: it does not + call ``TestBase.tearDown``, so the trace and metrics globals are not reset. + It is also synchronous, and drives the loop with ``run_until_complete`` + rather than awaiting, so a subclass that needs cleanup it can await has to + override ``asyncTearDown`` instead. + """ + def setUp(self): + """Install the in-memory pipelines and build a default HTTP scope.""" super().setUp() self.scope = {} @@ -69,22 +194,72 @@ def setUp(self): self.communicator = None def tearDown(self): + """Wait up to one second for the seeded application to finish. + + One second is the default of ``ApplicationCommunicator.wait``, which + this call does not override, so an application still running when that + second expires is cancelled rather than awaited. This override does not + call ``super().tearDown()``, so the trace and metrics globals are not + reset here. + """ if self.communicator: asyncio.get_event_loop().run_until_complete(self.communicator.wait()) def seed_app(self, app): + """Attach an ASGI application to the scope built by ``setUp``. + + Args: + app: The ASGI application to drive. The communicator is given a + reference to ``self.scope`` and does not instantiate the + application here: that happens on the first ``send_input`` or + ``get_output``, so an edit to ``self.scope`` made before that + first call does reach the application. + """ self.communicator = ApplicationCommunicator(app, self.scope) async def send_input(self, message): + """Send one ASGI message to the application. + + Args: + message: The message to deliver, for example + ``{"type": "http.request", "body": b""}``. + """ await self.communicator.send_input(message) async def send_default_request(self): + """Send an ``http.request`` message with an empty body.""" await self.send_input({"type": "http.request", "body": b""}) async def get_output(self, timeout=0.01): + """Wait for the next message the application sends. + + Args: + timeout: Seconds to wait for a message before giving up. + + Returns: + The next ASGI message. + + Raises: + asyncio.TimeoutError: When no message arrives within ``timeout``. + The timed-out read also cancels the application when it has not + finished, so this ends the conversation rather than pausing it. + """ return await self.communicator.receive_output(timeout) async def get_all_output(self, timeout=0.01): + """Collect every message the application sends, then cancel it. + + Args: + timeout: Seconds to wait for each further message. The call takes + at least this long, because it ends by waiting for a message + that never arrives. + + Returns: + A list of ASGI messages in the order the application sent them. + An application that sends nothing yields an empty list. The read + that ends the loop cancels an application that has not finished, so + do not drain one that still has input to receive. + """ outputs = [] while True: try: diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/concurrency_test.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/concurrency_test.py index f275abe7024..2a5b7f0da20 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/concurrency_test.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/concurrency_test.py @@ -1,6 +1,21 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +"""Helpers for tests that have to provoke a race rather than wait for one. + +A data race in code guarded by a lock shows up only when two threads reach the +guarded region at the same moment, which a test cannot arrange by starting +threads and hoping. The two pieces here arrange it: ``run_with_many_threads`` +holds every thread at a barrier and releases them together, and +``ConcurrencyTestBase`` shortens the interpreter's thread switch interval for +the duration of the test class so that a switch is likely to land inside the +guarded region. + +Neither makes a race certain. A test built on them fails intermittently when +the code under test is broken, which is enough to catch a regression over many +runs and is not enough to prove a single run correct. +""" + import sys import threading import unittest @@ -20,11 +35,32 @@ class MockFunc: """ def __init__(self) -> None: + """Build the lock, the call counter and the mock to hand back. + + ``call_count`` starts at zero and is the number this class exists to + make reliable. ``mock`` is the object ``__call__`` returns and is never + called here. ``lock`` guards ``call_count`` and nothing else. + """ self.lock = threading.Lock() self.call_count = 0 self.mock = Mock() def __call__(self, *args, **kwargs): + """Count this call and return the shared ``Mock``. + + ``self.mock`` is returned but never called here, so its own + ``call_count`` counts what callers did with the returned object and not + calls to this ``MockFunc``. Assert on ``call_count``, which is the only + count of the latter and the only one taken under the lock. + + Args: + *args: Accepted and discarded, so any call signature works. + **kwargs: Accepted and discarded for the same reason. + + Returns: + The same ``Mock`` instance on every call, so a test can assert on + what the caller did with the returned value. + """ with self.lock: self.call_count += 1 return self.mock @@ -43,12 +79,24 @@ class ConcurrencyTestBase(unittest.TestCase): @classmethod def setUpClass(cls) -> None: + """Shorten the interpreter thread switch interval for this class. + + The interval is process-wide, so every test running in the same + interpreter is slowed while this class runs. ``tearDownClass`` puts the + original value back. + """ super().setUpClass() # switch threads more often to increase chance of contention sys.setswitchinterval(1e-12) @classmethod def tearDownClass(cls) -> None: + """Restore the thread switch interval read when this module was imported. + + ``orig_switch_interval`` is a class attribute evaluated at import time, + so the value restored is the one in force before any test ran, not the + one in force when ``setUpClass`` was called. + """ super().tearDownClass() sys.setswitchinterval(cls.orig_switch_interval) @@ -57,12 +105,41 @@ def run_with_many_threads( func_to_test: Callable[[], ReturnT], num_threads: int = 100, ) -> list[ReturnT]: - """Util to run ``func_to_test`` in ``num_threads`` concurrently""" + """Util to run ``func_to_test`` in ``num_threads`` concurrently + + Every thread is started, then held at a barrier until all of them have + arrived, then released at once. Holding them is what makes the calls + overlap: without it the first threads would finish before the last ones + started. + + Args: + func_to_test: Called once per thread, with no arguments. It has to + be safe to call concurrently with itself, since testing that is + the point. An exception it raises propagates out of the thread + and is printed by ``threading.excepthook`` rather than failing + the test, and that thread then leaves ``None`` in its slot of + the result list. + num_threads: How many threads to run, which is also the number of + calls and the size of the returned list. The barrier is sized to + this number, so every thread has to arrive before any proceeds. + + Returns: + A list of the return values, where position ``i`` holds the value + returned by the ``i``-th thread. The order is the order the threads + were created in, not the order they finished in. + """ barrier = threading.Barrier(num_threads) results: list[ReturnT | None] = [None] * num_threads def thread_start(idx: int) -> None: + """Wait for every other thread, then call the function under test. + + Args: + idx: Position in the result list to store the return value at. + Each thread gets its own, so the threads write to disjoint + slots and the list needs no lock. + """ nonlocal results # Get all threads here before releasing them to create contention barrier.wait() diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/globals_test.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/globals_test.py index fc0bec3a1e7..3e814743a7f 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/globals_test.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/globals_test.py @@ -1,6 +1,24 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +"""Resetting the API-level provider globals between tests. + +The trace, metrics and logs APIs each hold the configured provider in a module +level global, guarded by an ``opentelemetry.util._once.Once`` so that the +corresponding ``set_*_provider`` function takes effect only the first time it is +called. That is the behaviour an application wants and the opposite of what a +test suite wants, because the second test to install a provider would silently +keep the one installed by the first. + +The ``reset_*_globals`` functions here reach into those private globals and put +them back to their unset state, and the ``*GlobalsTest`` classes call them from +``setUp`` and ``tearDown`` so a test does not have to. Reaching into private +names is why each of the three functions that assign to those globals carries a +warning against use outside a test suite: nothing here is part of the public API +of the packages it mutates, and a release may move or rename the globals it +assigns to. +""" + import unittest from opentelemetry import trace as trace_api @@ -12,7 +30,21 @@ # pylint: disable=protected-access def reset_trace_globals() -> None: - """WARNING: only use this for tests.""" + """Put the trace API globals back to their unset state. + + Clears the configured tracer provider, replaces the proxy tracer provider + with a fresh one so that a tracer requested after this call is not bound to + the provider the previous test installed, and resets the ``Once`` guard that + makes ``opentelemetry.trace.set_tracer_provider`` effective only once. + After this call, ``set_tracer_provider`` installs a provider again. + + A tracer obtained before this call is not detached by it: a proxy tracer + caches the real tracer the first time it resolves one, so it keeps sending + spans to the earlier provider. Request the tracer again after the reset. + + WARNING: only use this for tests. It assigns to private module globals of + ``opentelemetry.trace``. + """ trace_api._TRACER_PROVIDER_SET_ONCE = Once() trace_api._TRACER_PROVIDER = None trace_api._PROXY_TRACER_PROVIDER = trace_api.ProxyTracerProvider() @@ -20,7 +52,20 @@ def reset_trace_globals() -> None: # pylint: disable=protected-access def reset_metrics_globals() -> None: - """WARNING: only use this for tests.""" + """Put the metrics API globals back to their unset state. + + Clears the configured meter provider, replaces the proxy meter provider with + a fresh one so that a meter requested after this call is not bound to the + provider the previous test installed, and resets the ``Once`` guard that + makes ``opentelemetry.metrics.set_meter_provider`` effective only once. + + A meter or instrument obtained before this call is not detached by it: the + proxy caches the real one the first time it resolves it, so measurements + keep reaching the earlier provider. Create them again after the reset. + + WARNING: only use this for tests. It assigns to private module globals of + ``opentelemetry.metrics._internal``. + """ metrics_api._METER_PROVIDER_SET_ONCE = Once() # type: ignore[attr-defined] metrics_api._METER_PROVIDER = None # type: ignore[attr-defined] metrics_api._PROXY_METER_PROVIDER = _ProxyMeterProvider() # type: ignore[attr-defined] @@ -28,7 +73,20 @@ def reset_metrics_globals() -> None: # pylint: disable=protected-access def reset_logging_globals() -> None: - """WARNING: only use this for tests.""" + """Put the logs API globals back to their unset state. + + Clears the configured logger provider, replaces the proxy logger provider + with a fresh one so that a logger requested after this call is not bound to + the provider the previous test installed, and resets the ``Once`` guard that + makes ``opentelemetry._logs.set_logger_provider`` effective only once. + + A logger obtained before this call is not detached by it: a proxy logger + caches the real logger the first time it resolves one. Request the logger + again after the reset. + + WARNING: only use this for tests. It assigns to private module globals of + ``opentelemetry._logs._internal``. + """ logging_api._LOGGER_PROVIDER_SET_ONCE = Once() # type: ignore[attr-defined] logging_api._LOGGER_PROVIDER = None # type: ignore[attr-defined] logging_api._PROXY_LOGGER_PROVIDER = logging_api.ProxyLoggerProvider() # type: ignore[attr-defined] @@ -38,7 +96,10 @@ def reset_logging_globals() -> None: def reset_event_globals() -> None: """Deprecated no-op kept for backward compatibility. - The Events API has been removed; this function does nothing. + The Events API has been removed, so there are no event globals left to + reset and this function does nothing. It is still exported so that a test + suite calling it against an older release of this package keeps importing + and running. Delete the call rather than keeping it. """ @@ -49,10 +110,16 @@ class TraceGlobalsTest(unittest.TestCase): """ def setUp(self) -> None: + """Reset the trace API globals before the test method runs.""" super().setUp() reset_trace_globals() def tearDown(self) -> None: + """Reset the trace API globals again, so the next test starts clean. + + Resetting on both sides means a test that fails part way through still + leaves the globals unset for whatever runs next. + """ super().tearDown() reset_trace_globals() @@ -64,10 +131,12 @@ class MetricsGlobalsTest(unittest.TestCase): """ def setUp(self) -> None: + """Reset the metrics API globals before the test method runs.""" super().setUp() reset_metrics_globals() def tearDown(self) -> None: + """Reset the metrics API globals again, so the next test starts clean.""" super().tearDown() reset_metrics_globals() @@ -79,10 +148,12 @@ class LoggingGlobalsTest(unittest.TestCase): """ def setUp(self) -> None: + """Reset the logs API globals before the test method runs.""" super().setUp() reset_logging_globals() def tearDown(self) -> None: + """Reset the logs API globals again, so the next test starts clean.""" super().tearDown() reset_logging_globals() @@ -90,11 +161,17 @@ def tearDown(self) -> None: class EventsGlobalsTest(unittest.TestCase): """Deprecated no-op kept for backward compatibility. - The Events API has been removed; this class does nothing. + The Events API has been removed, so there are no event globals left to + reset and inheriting from this class is the same as inheriting from + ``unittest.TestCase``. It is still exported so that a test suite naming it + as a base class keeps importing and running. Replace it with + ``unittest.TestCase`` rather than keeping it. """ def setUp(self) -> None: # pylint: disable=useless-parent-delegation + """Delegate to ``unittest.TestCase``, resetting nothing.""" super().setUp() def tearDown(self) -> None: # pylint: disable=useless-parent-delegation + """Delegate to ``unittest.TestCase``, resetting nothing.""" super().tearDown() diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/httptest.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/httptest.py index 3e62c51f318..0669268bfe0 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/httptest.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/httptest.py @@ -1,6 +1,15 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +"""A real HTTP server for tests of HTTP client instrumentation. + +Instrumentation of an HTTP client is hard to test against a mock, because the +behaviour under test includes what the client library does with a real socket, +with keep-alive, and with an error status. :class:`HttpTestBase` starts an +actual ``http.server`` on a loopback port for the duration of the test class, so +the client under test makes a real request. +""" + import re import unittest from http import HTTPStatus @@ -9,15 +18,46 @@ class HttpTestBase(unittest.TestCase): + """Base class that runs an HTTP server for the whole test class. + + ``setUpClass`` binds a server to an ephemeral port on ``127.0.0.1`` and + serves it from a daemon thread; ``tearDownClass`` shuts it down and joins + that thread. The port is chosen by the operating system, so read it from + ``cls.server.server_address[1]`` rather than assuming one. + + The server answers ``GET /`` with ``DEFAULT_RESPONSE`` and ``200``, and + ``GET /status/`` with ````, which is how a test drives the + client under test into an error path. Every other method gets the ``501`` + that ``BaseHTTPRequestHandler`` returns for an unimplemented verb. + + Override ``Handler`` in a subclass to answer differently. ``create_server`` + reads ``cls.Handler``, so the override takes effect without touching the + other methods. Overriding ``DEFAULT_RESPONSE`` instead does nothing: + ``do_GET`` reads ``HttpTestBase.DEFAULT_RESPONSE`` off this class, so a + subclass attribute of that name is never consulted and the body stays + ``b"Hello!"``. + """ + DEFAULT_RESPONSE = b"Hello!" class Handler(BaseHTTPRequestHandler): + """Request handler answering ``/`` and ``/status/``.""" + protocol_version = "HTTP/1.1" # Support keep-alive. timeout = 3 # Seconds STATUS_RE = re.compile(r"/status/(\d+)") def do_GET(self): # pylint:disable=invalid-name + """Answer a GET request, with the status the path asks for. + + A path of the form ``/status/`` is answered with that status + and, unless it is ``200``, with the error body that + ``send_error`` generates. Any other path is answered with ``200`` + and ``HttpTestBase.DEFAULT_RESPONSE``. The path is matched in full, + so ``/status/404/extra`` is treated as an ordinary path and gets + ``200``. + """ status_match = self.STATUS_RE.fullmatch(self.path) status = 200 if status_match: @@ -33,11 +73,27 @@ def do_GET(self): # pylint:disable=invalid-name @classmethod def create_server(cls): + """Build a server bound to an ephemeral loopback port, without serving. + + Returns: + An ``HTTPServer`` bound to ``127.0.0.1`` on a port the operating + system picked, using ``cls.Handler``. It is bound but not yet + serving, so a caller that wants requests answered has to drive + ``serve_forever`` itself, which is what ``run_server`` does. + """ server_address = ("127.0.0.1", 0) # Only bind to localhost. return HTTPServer(server_address, cls.Handler) @classmethod def run_server(cls): + """Build a server and serve it from a background thread. + + Returns: + A tuple of the worker thread and the ``HTTPServer`` it serves. The + thread is a daemon, so it does not keep the interpreter alive, but + shutting the server down and joining the thread is still the + caller's job, which is what ``tearDownClass`` does. + """ httpd = cls.create_server() worker = Thread(target=httpd.serve_forever, daemon=True, name="Test server worker") worker.start() @@ -45,11 +101,23 @@ def run_server(cls): @classmethod def setUpClass(cls): + """Start the shared server and record it on the class. + + One server serves every test method of the class, so a test that wants + the server to have seen no earlier request needs its own class. + """ super().setUpClass() cls.server_thread, cls.server = cls.run_server() @classmethod def tearDownClass(cls): + """Stop the shared server and wait for its thread to finish. + + ``shutdown`` already blocks until the serve loop has returned, so no + request is still being answered once it comes back and the ``join`` + only reaps a thread that has already finished. The listening socket is + not closed here, since ``server_close`` is never called. + """ cls.server.shutdown() cls.server_thread.join() super().tearDownClass() diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/metrictestutil.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/metrictestutil.py index f51bfcd5256..4478ebe3cd3 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/metrictestutil.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/metrictestutil.py @@ -1,6 +1,21 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +"""Builders for the SDK metric objects an exporter test compares against. + +A metrics exporter is tested by handing it a ``Metric`` and checking what it +produces, which means a test needs a ``Metric`` with a known shape. Building one +by hand takes a nested ``Metric``, data and data point construction every time, +so these functions fill in the parts a test does not care about: the timestamps +are fixed constants rather than the current time, except the gauge start +timestamp, which is ``None``; the temporality is cumulative wherever the data +type carries one, which a ``Gauge`` does not; and the description and unit +default to ``"foo"`` and ``"s"``. + +Every name here is private, because the exact values are chosen to suit the +exporter tests in this repository and are changed whenever those tests need +different ones. +""" from opentelemetry.attributes import BoundedAttributes from opentelemetry.sdk.metrics.export import ( @@ -16,6 +31,26 @@ def _generate_metric(name, data, attributes=None, description=None, unit=None) -> Metric: + """Wrap already-built metric data in a ``Metric``. + + Args: + name: The metric name. + data: The ``Sum``, ``Gauge``, ``Histogram`` or other data object the + metric carries, already holding its data points. Passing ``None`` + produces a metric with no data, which is what + ``_generate_unsupported_metric`` relies on. + attributes: Ignored. Attributes belong to the individual data points + inside ``data``, so there is nothing for this function to do with + them. The parameter only keeps the signature uniform with the other + builders in this module, none of which pass their own ``attributes`` + on to this function. + description: The metric description. ``None`` means ``"foo"``, so a + test that does not care about the description gets a non-empty one. + unit: The metric unit. ``None`` means ``"s"``. + + Returns: + A ``Metric`` carrying the given name, data, description and unit. + """ if description is None: description = "foo" if unit is None: @@ -36,6 +71,24 @@ def _generate_sum( unit=None, is_monotonic=True, ) -> Metric: + """Build a ``Metric`` holding one cumulative ``Sum`` data point. + + Args: + name: The metric name. + value: The value of the single data point. + attributes: Attributes of the data point. ``None`` means + ``{"a": 1, "b": True}`` wrapped in ``BoundedAttributes``, which is + what an attribute set that has been through the SDK looks like. + description: The metric description. ``None`` means ``"foo"``. + unit: The metric unit. ``None`` means ``"s"``. + is_monotonic: Whether the sum only ever increases, which is what + distinguishes a counter from an up-down counter. + + Returns: + A ``Metric`` whose data is a ``Sum`` with cumulative temporality and + exactly one ``NumberDataPoint``. Its start and end timestamps are fixed + constants one second apart, so a test can assert on them. + """ if attributes is None: attributes = BoundedAttributes(attributes={"a": 1, "b": True}) return _generate_metric( @@ -58,6 +111,22 @@ def _generate_sum( def _generate_gauge(name, value, attributes=None, description=None, unit=None) -> Metric: + """Build a ``Metric`` holding one ``Gauge`` data point. + + Args: + name: The metric name. + value: The value of the single data point. + attributes: Attributes of the data point. ``None`` means + ``{"a": 1, "b": True}`` wrapped in ``BoundedAttributes``. + description: The metric description. ``None`` means ``"foo"``. + unit: The metric unit. ``None`` means ``"s"``. + + Returns: + A ``Metric`` whose data is a ``Gauge`` with exactly one + ``NumberDataPoint``. Its start timestamp is ``None``, because a gauge + reports an instantaneous value and has no interval, which is a case an + exporter has to encode correctly. + """ if attributes is None: attributes = BoundedAttributes(attributes={"a": 1, "b": True}) return _generate_metric( @@ -78,6 +147,21 @@ def _generate_gauge(name, value, attributes=None, description=None, unit=None) - def _generate_unsupported_metric(name, attributes=None, description=None, unit=None) -> Metric: + """Build a ``Metric`` whose data is ``None``. + + No instrument produces such a metric. It exists so a test can check that an + exporter handed a data type it does not recognise drops the metric and + reports it, rather than raising. + + Args: + name: The metric name. + attributes: Ignored. There are no data points to attach attributes to. + description: The metric description. ``None`` means ``"foo"``. + unit: The metric unit. ``None`` means ``"s"``. + + Returns: + A ``Metric`` whose ``data`` attribute is ``None``. + """ return _generate_metric( name, None, @@ -92,6 +176,26 @@ def _generate_histogram( description: str | None = None, unit: str | None = None, ) -> Metric: + """Build a ``Metric`` holding one cumulative ``Histogram`` data point. + + Unlike the other builders here, this one takes no value: the counts, sum, + bounds and extremes of the data point are fixed. It records 6 measurements + summing to 579.0, spread over the buckets ``[1, 3, 2]`` with boundaries + ``[123.0, 456.0]``, a minimum of 1 and a maximum of 457. + + Args: + name: The metric name. + attributes: Attributes of the data point. ``None`` means + ``{"a": 1, "b": True}`` wrapped in ``BoundedAttributes``. + description: The metric description. ``None`` means ``"foo"``. + unit: The metric unit. ``None`` means ``"s"``. + + Returns: + A ``Metric`` whose data is a ``Histogram`` with cumulative temporality + and exactly one ``HistogramDataPoint``. Its start and end timestamps are + the same fixed constant, so the data point covers a zero-length + interval. + """ if attributes is None: attributes = BoundedAttributes(attributes={"a": 1, "b": True}) return _generate_metric( diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/mock_test_classes.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/mock_test_classes.py index 680c223ab6c..a2bc53020c2 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/mock_test_classes.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/mock_test_classes.py @@ -1,11 +1,41 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +"""A stand-in for an ``importlib.metadata`` entry point. + +Several packages pick an implementation by name from an entry point group, for +example ``opentelemetry_tracer_provider``. Testing that selection against real +entry points would mean installing a distribution, so a test patches +``entry_points`` to return instances of :class:`IterEntryPoint` instead. +""" + class IterEntryPoint: + """An entry point that loads a class given to it rather than importing one. + + Only the ``name`` attribute and the ``load`` method are provided, which is + the part of the ``importlib.metadata.EntryPoint`` interface the selection + code uses. ``value``, ``group`` and ``dist`` are absent, so code that reads + those raises ``AttributeError``. + + Args: + name: The entry point name, which is what selection code matches a + configured value against. + class_type: The object ``load`` returns. Named for the usual case of a + class, but nothing checks it, so any object works. + """ + def __init__(self, name, class_type): + """Parameters are documented on the class docstring.""" self.name = name self.class_type = class_type def load(self): + """Return the object this entry point stands for. + + Returns: + The object passed as ``class_type``, returned as it is. No import + happens and no module is executed, so an entry point that would fail + to import in reality cannot be simulated with this class. + """ return self.class_type diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/mock_textmap.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/mock_textmap.py index 4e37fc81858..af140de4325 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/mock_textmap.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/mock_textmap.py @@ -1,6 +1,18 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +"""Two text map propagators that stand in for a real one in tests. + +:class:`NOOPTextMapPropagator` carries nothing, so it exercises the path where +extraction yields no span context. :class:`MockTextMapPropagator` carries the +trace id and span id as decimal strings under two keys of its own, so a test can +assert on the exact carrier contents without having to parse the W3C +``traceparent`` format. + +Neither is a valid propagator to use outside a test: the format of +:class:`MockTextMapPropagator` is not interoperable with anything, and +:class:`NOOPTextMapPropagator` loses context silently. +""" from opentelemetry import trace from opentelemetry.context import Context @@ -27,6 +39,18 @@ def extract( context: Context | None = None, getter: Getter = default_getter, ) -> Context: + """Return an empty context, reading nothing from the carrier. + + Args: + carrier: Ignored. Nothing is read from it. + context: Ignored. The context passed in is discarded rather than + returned, so any baggage or span it carried is lost. + getter: Ignored, since the carrier is never read. + + Returns: + A new empty ``Context``. A span started from it has no parent, which + is the case this propagator exists to produce. + """ return Context() def inject( @@ -35,15 +59,33 @@ def inject( context: Context | None = None, setter: Setter = default_setter, ) -> None: - return None + """Write nothing to the carrier. + + Args: + carrier: Left unchanged. + context: Ignored. + setter: Ignored, since the carrier is never written. + """ @property def fields(self): + """Return the carrier keys this propagator writes. + + Returns: + An empty set, since ``inject`` writes nothing. A framework that + clears these keys before injecting therefore clears nothing. + """ return set() class MockTextMapPropagator(TextMapPropagator): - """Mock propagator for testing purposes.""" + """Mock propagator for testing purposes. + + Carries the trace id and the span id as decimal strings under the + ``mock-traceid`` and ``mock-spanid`` keys. Trace flags, trace state and + baggage are not carried, so a span extracted from a carrier this propagator + wrote is never sampled through the propagated flags. + """ TRACE_ID_KEY = "mock-traceid" SPAN_ID_KEY = "mock-spanid" @@ -54,6 +96,28 @@ def extract( context: Context | None = None, getter: Getter = default_getter, ) -> Context: + """Read the trace id and span id from the carrier into a span context. + + Args: + carrier: The carrier to read ``mock-traceid`` and ``mock-spanid`` + from. Only the first value of each is used when the getter + returns several. + context: Context to extract into. ``None`` means a new empty + ``Context``. + getter: Reads the two keys out of the carrier. + + Returns: + A context whose current span is a ``NonRecordingSpan`` with the + trace id and span id from the carrier, marked remote. When either + key is missing or empty, the context is returned unchanged, so the + caller cannot distinguish that from a carrier that was never + written to. + + Raises: + ValueError: When either value is present but is not a decimal + integer. A real propagator reports a malformed carrier by + returning the context unchanged instead. + """ if context is None: context = Context() trace_id_list = getter.get(carrier, self.TRACE_ID_KEY) @@ -79,10 +143,29 @@ def inject( context: Context | None = None, setter: Setter = default_setter, ) -> None: + """Write the trace id and span id of the current span to the carrier. + + Both are written as decimal strings. No check is made that the span + context is valid, so an invalid context is written out as two zeros + rather than skipped, which is how a test can tell that injection + happened with no span in context. + + Args: + carrier: The carrier to write the two keys into. + context: Context to read the current span from. ``None`` means the + current context. + setter: Writes the two keys into the carrier. + """ span = trace.get_current_span(context) setter.set(carrier, self.TRACE_ID_KEY, str(span.get_span_context().trace_id)) setter.set(carrier, self.SPAN_ID_KEY, str(span.get_span_context().span_id)) @property def fields(self): + """Return the carrier keys this propagator writes. + + Returns: + A set holding ``mock-traceid`` and ``mock-spanid``. A framework + reusing a carrier clears these before injecting. + """ return {self.TRACE_ID_KEY, self.SPAN_ID_KEY} diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/samplertestutil.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/samplertestutil.py index a196514fceb..2a180753bcc 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/samplertestutil.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/samplertestutil.py @@ -1,6 +1,15 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +"""A sampler that records the arguments the SDK passes to it. + +A sampling decision is taken before a span exists, from the name, kind and +attributes handed to ``start_span``. Instrumentation that sets an attribute +after the span has started leaves a sampler unable to see it, which is a +regression a test cannot detect by inspecting the finished span. +:class:`CapturingSampler` makes it detectable by keeping what it was given. +""" + from __future__ import annotations from collections.abc import Sequence @@ -21,6 +30,16 @@ class CapturingSampler(Sampler): Every span is recorded and sampled unless ``required_attributes`` is given, in which case a span is dropped when any of these attributes is missing or has a different value. + + After a span has started, ``name``, ``kind`` and ``attributes`` hold what + the SDK passed for the most recent sampling decision, and are ``None``, + ``None`` and an empty dict before the first one. + + Args: + required_attributes: Attributes a span has to carry at start time to be + sampled, compared by equality. ``None`` or an empty mapping samples + every span. The mapping is copied, so later edits to the argument do + not change the sampling decisions. """ name: str | None @@ -28,6 +47,7 @@ class CapturingSampler(Sampler): attributes: dict[str, AttributeValue] def __init__(self, required_attributes: Attributes = None) -> None: + """Parameters are documented on the class docstring.""" self.required_attributes = dict(required_attributes or {}) self.name = None self.kind = None @@ -43,6 +63,35 @@ def should_sample( links: Sequence[Link] | None = None, trace_state: TraceState | None = None, ) -> SamplingResult: + """Record the sampling arguments and decide whether to sample. + + Only the last call is kept, so a test that starts several spans sees + the arguments of the last one. + + Args: + parent_context: Context the new span is starting in. Its current + span supplies the trace state carried into the result. ``None`` + means the current context. + trace_id: Trace id of the new span. Ignored by this sampler, so the + decision does not vary across traces. + name: Name the span is starting with, recorded as ``self.name``. + kind: Kind the span is starting with, recorded as ``self.kind``. + attributes: Attributes passed at span start, copied into + ``self.attributes``. Copied rather than stored, so attributes + set after the span starts cannot make a missing one look + present. ``None`` is recorded as an empty dict. + links: Links of the new span. Ignored by this sampler. + trace_state: Trace state proposed for the new span. Ignored: the + result carries the trace state of the parent instead. + + Returns: + A ``SamplingResult`` with ``RECORD_AND_SAMPLE`` when every entry of + ``required_attributes`` is present in ``attributes`` with the same + value, and ``DROP`` otherwise. A dropped span carries no attributes + in the result. The trace state of the parent span is carried over + when the parent context is valid, and is ``None`` when there is no + valid parent. + """ self.name = name self.kind = kind # Snapshot now so attributes set after span creation cannot hide a regression. @@ -64,4 +113,16 @@ def should_sample( ) def get_description(self) -> str: + """Return the human-readable description of this sampler. + + The SDK uses a sampler description only when a composite sampler such + as ``ParentBased`` builds its own description out of the samplers it + wraps. Nothing attaches it to a span, so a test cannot assert on it + through exported telemetry. + + Returns: + The constant ``"CapturingSampler"``. It does not vary with + ``required_attributes``, so two instances configured differently + describe themselves identically. + """ return "CapturingSampler" diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/spantestutil.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/spantestutil.py index 2b9e6c06dc9..2d7c733f13c 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/spantestutil.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/spantestutil.py @@ -1,6 +1,13 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +"""Builders for spans that exporter tests need. + +Both functions here build SDK objects directly instead of going through the API +globals, so they can be used from a test that is not based on +:class:`opentelemetry.test.test_base.TestBase`. +""" + from functools import partial from opentelemetry import trace as trace_api @@ -9,6 +16,26 @@ def new_tracer(span_limits=None, resource=None) -> trace_api.Tracer: + """Build a tracer from a brand new SDK tracer provider. + + The provider has no span processor, so a span this tracer starts is + recorded in memory and exported nowhere. Use it when the test is about what + a span contains rather than about what an exporter does with it. + + Args: + span_limits: Limits on attribute, event and link counts and on + attribute value length, applied to spans this tracer starts. When + ``None``, the provider reads the limits from the environment, which + means a test that sets one of the ``OTEL_SPAN_*`` variables + influences the result. + resource: Resource to attach to spans this tracer starts. When + ``None``, the provider builds the default resource, which detects + the service name and the SDK attributes from the environment. + + Returns: + A tracer named after this module, ``opentelemetry.test.spantestutil``, + from a provider that no other call shares. + """ provider_factory = trace_sdk.TracerProvider if resource is not None: provider_factory = partial(provider_factory, resource=resource) @@ -16,6 +43,23 @@ def new_tracer(span_limits=None, resource=None) -> trace_api.Tracer: def get_span_with_dropped_attributes_events_links(): + """Build one ended span that has dropped attributes, events and links. + + Exporters have to report what the SDK discarded, and the counts are only + non-zero once a limit is exceeded, so this builds a span that exceeds every + one of them. The default ``SpanLimits`` cap attributes, events and links at + 128 each; this span is given 130 attributes, 131 events and 129 links, and + its resource and every one of its events and links are given the same 130 + attributes. Only the resource keeps all 130, because no limit applies to it; + the span, each event and each link keep the 128 the limits allow. The span + name is ``"span"``. + + Returns: + The span, already ended, because it is returned from inside the + ``start_as_current_span`` block and the context manager ends it on the + way out. Reading ``dropped_attributes``, ``dropped_events`` and + ``dropped_links`` on it gives non-zero counts. + """ attributes = {} for index in range(130): attributes[f"key{index}"] = [f"value{index}"] diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/test_base.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/test_base.py index 06f819ea9c0..be174ecd310 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/test_base.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/test_base.py @@ -1,6 +1,19 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +"""A ``unittest`` base class that captures spans and metrics in memory. + +:class:`TestBase` wires a tracer provider and a meter provider that keep +everything they are given in process, installs them as the API globals, and +exposes the captured telemetry to assertions. No exporter reaches the network, +so a test that uses it needs no collector and no port. + +The module also provides :class:`FinishedTestSpans`, the list subclass that +:meth:`TestBase.get_finished_spans` returns, which can look a span up by name or +by attribute and fails the test instead of returning ``None`` when nothing +matches. +""" + import logging import unittest from collections.abc import Sequence @@ -31,9 +44,40 @@ class TestBase(unittest.TestCase): + """Base class for a test that needs to assert on emitted telemetry. + + ``setUp`` installs a fresh tracer provider and meter provider as the API + globals, each feeding an in-memory sink, and ``tearDown`` removes them + again, so instrumentation under test reaches this class through the ordinary + ``opentelemetry.trace.get_tracer`` and ``opentelemetry.metrics.get_meter`` + calls and needs no test-specific wiring of its own. + + The captured telemetry is reachable in three ways. ``self.memory_exporter`` + is the span exporter, ``self.memory_metrics_reader`` is the metric reader, + and the ``get_finished_spans`` and ``get_sorted_metrics`` methods read from + them. Every test method gets its own providers, so nothing recorded by one + test is visible to another. + + Override the ``create_tracer_provider`` or ``create_meter_provider`` + staticmethod to pass arguments to a provider, forwarding ``**kwargs`` to + the ``TestBase`` implementation. ``setUp`` calls the override and installs + what it returns as the API global. Reassigning ``self.tracer_provider`` + after ``super().setUp()`` does not work: the provider built by ``setUp`` is + already the installed global, and ``set_tracer_provider`` will not replace + it, so instrumentation keeps reporting to it while the reassigned exporter + stays empty. + """ + # pylint: disable=C0103 def setUp(self): + """Install in-memory trace and metrics pipelines as the API globals. + + Resets the trace and metrics globals first, because + ``set_tracer_provider`` and ``set_meter_provider`` ignore the call when + a provider is already configured, which would otherwise leave this test + reading the telemetry of whichever test ran first. + """ super().setUp() result = self.create_tracer_provider() self.tracer_provider, self.memory_exporter = result @@ -53,18 +97,60 @@ def setUp(self): metrics_api.set_meter_provider(self.meter_provider) def tearDown(self): + """Unset the trace and metrics globals installed by ``setUp``. + + The providers and their in-memory sinks are not shut down, so a span or + a metric reached for after the test method has returned is still + readable through the attributes ``setUp`` assigned. + """ super().tearDown() reset_trace_globals() reset_metrics_globals() def get_finished_spans(self): + """Return the spans ended so far, in the order they were exported. + + Returns: + A :class:`FinishedTestSpans` over the spans the in-memory exporter + holds, which supports ``by_name`` and ``by_attr`` lookups on top of + the usual list operations. The list is a snapshot: a span ended + after this call does not appear in it. + """ return FinishedTestSpans(self, self.memory_exporter.get_finished_spans()) def assertEqualSpanInstrumentationScope(self, span, module): + """Assert that a span was produced by the tracer of a given module. + + Instrumentation is expected to name its tracer after its own module and + version, so this checks both at once rather than leaving a test to spell + out two ``assertEqual`` calls. + + Args: + span: The span to inspect. Its ``instrumentation_scope`` is read, so + it has to be an SDK span and not a non-recording one. + module: The module object the span is expected to come from. Its + ``__name__`` is compared against the scope name first, and its + ``__version__`` against the scope version only once that + comparison has passed. A module without ``__version__`` + therefore raises ``AttributeError`` rather than failing the + assertion, but only when its ``__name__`` already matches; a + name mismatch fails the first assertion and never reads + ``__version__``. + """ self.assertEqual(span.instrumentation_scope.name, module.__name__) self.assertEqual(span.instrumentation_scope.version, module.__version__) def assertSpanHasAttributes(self, span, attributes): + """Assert that a span carries at least the given attributes. + + Each key has to be present and to compare equal. Attributes the span + carries beyond the given ones are ignored, so this does not detect an + attribute that should have been dropped. + + Args: + span: The span whose ``attributes`` mapping is checked. + attributes: Mapping of attribute key to expected value. + """ for key, val in attributes.items(): self.assertIn(key, span.attributes) self.assertEqual(val, span.attributes[key]) @@ -75,6 +161,14 @@ def sorted_spans(self, spans): # pylint: disable=R0201 Note: This method should not be used to sort spans in a deterministic way as the order depends on timing precision provided by the platform. + + Args: + spans: The spans to sort. Their private ``_start_time`` is read, so + they have to be SDK spans. + + Returns: + A new list holding the same spans, newest first. The argument is not + modified. """ return sorted( spans, @@ -91,8 +185,12 @@ def create_tracer_provider(**kwargs): All the parameters passed are forwarded to the TracerProvider constructor. + Args: + **kwargs: Forwarded unchanged to the ``TracerProvider`` + constructor, for example ``resource`` or ``sampler``. + Returns: - A list with the tracer provider in the first element and the + A tuple with the tracer provider in the first element and the in-memory span exporter in the second. """ tracer_provider = TracerProvider(**kwargs) @@ -104,11 +202,20 @@ def create_tracer_provider(**kwargs): @staticmethod def create_meter_provider(**kwargs) -> tuple[MeterProvider, MetricReader]: - """Helper to create a configured meter provider + """Helper to create a configured meter provider. + Creates a `MeterProvider` and an `InMemoryMetricReader`. + + Args: + **kwargs: Forwarded unchanged to the ``MeterProvider`` + constructor. A ``metric_readers`` list given here is appended + to rather than replaced, and the list object passed in is + mutated, so the in-memory reader is added alongside any reader + the caller supplies. + Returns: A tuple with the meter provider in the first element and the - in-memory metrics exporter in the second + in-memory metric reader in the second. """ memory_reader = InMemoryMetricReader() metric_readers = kwargs.get("metric_readers", []) @@ -120,6 +227,21 @@ def create_meter_provider(**kwargs) -> tuple[MeterProvider, MetricReader]: @staticmethod @contextmanager def disable_logging(highest_level=logging.CRITICAL): + """Silence the ``logging`` module for the duration of the block. + + Use it around code that is expected to log an error, so the failure + under test does not fill the output of a passing run. + + Args: + highest_level: The highest level to suppress. Everything at this + level and below is dropped, so the default of + ``logging.CRITICAL`` suppresses every record. + + Yields: + ``None``. The block runs with logging suppressed, and suppression is + lifted on exit whether the block completed or raised. Suppression is + process-wide, so a thread logging concurrently is silenced too. + """ logging.disable(highest_level) try: @@ -132,7 +254,16 @@ def get_sorted_metrics(self, scope: str | None = None): Args: scope: Optional scope name to filter metrics by. If unset, - all metrics are returned. + all metrics are returned. A name that matches no scope + yields an empty list rather than an error. + + Returns: + The metrics the reader holds after a fresh collection, sorted by + name. Reading them performs that collection. ``create_meter_provider`` + builds the reader with the default cumulative temporality, so a + measurement already reported is reported again on every later call + and the values accumulate rather than reset. Calling this twice with + nothing recorded in between returns the same values twice. """ metrics_data = self.memory_metrics_reader.get_metrics_data() resource_metrics = metrics_data.resource_metrics if metrics_data else [] @@ -150,6 +281,13 @@ def get_sorted_metrics(self, scope: str | None = None): def sorted_metrics(metrics): """ Sorts metrics by metric name. + + Args: + metrics: The metrics to sort. + + Returns: + A new list holding the same metrics in ascending name order. The + argument is not modified. """ return sorted( metrics, @@ -162,6 +300,21 @@ def assert_metric_expected( expected_data_points: Sequence[DataPointT], est_value_delta: float | None = 0, ): + """Assert that a metric carries exactly the expected data points. + + The counts have to match, and every expected data point has to have a + match somewhere in the metric. Order is not compared, so a data point + may appear anywhere in the metric. + + Args: + metric: The metric to check. + expected_data_points: The data points the metric is expected to + carry. + est_value_delta: Tolerance on the numeric value, for a measurement + that cannot be predicted exactly, such as a duration. Passing + ``0`` demands equality and additionally demands that the + ``min`` and ``max`` of a histogram data point match. + """ self.assertEqual(len(expected_data_points), len(metric.data.data_points)) for expected_data_point in expected_data_points: self.assert_data_point_expected(expected_data_point, metric.data.data_points, est_value_delta) @@ -173,6 +326,25 @@ def is_data_points_equal( data_point: DataPointT, est_value_delta: float | None = 0, ): + """Report whether two data points describe the same measurement. + + Args: + expected_data_point: The data point a test is looking for. + data_point: The data point that was recorded. + est_value_delta: Largest difference in value that still counts as + equal. When it is ``0``, the ``min`` and ``max`` of a histogram + data point also have to match exactly. + + Returns: + ``True`` when the two are of the same concrete type, that type is + ``NumberDataPoint`` or ``HistogramDataPoint``, their values agree + within ``est_value_delta``, and their attributes are equal. Two + ``HistogramDataPoint`` objects additionally have to carry the same + ``count`` and the same ``explicit_bounds``, and the same ``min`` and + ``max`` when ``est_value_delta`` is ``0``. A data point of any other + type returns ``False`` rather than raising, so an unsupported type + is reported as a mismatch. + """ if type(expected_data_point) != type(data_point) or not isinstance( # noqa: E721 expected_data_point, (HistogramDataPoint, NumberDataPoint) ): @@ -200,6 +372,15 @@ def assert_data_point_expected( data_points: Sequence[DataPointT], est_value_delta: float | None = 0, ): + """Assert that one of the recorded data points matches the expected one. + + Fails the test naming the expected data point when none of them does. + + Args: + expected_data_point: The data point a test is looking for. + data_points: The recorded data points to search, in any order. + est_value_delta: Tolerance passed on to ``is_data_points_equal``. + """ is_data_point_exist = False for data_point in data_points: if self.is_data_points_equal(expected_data_point, data_point, est_value_delta): @@ -213,6 +394,18 @@ def assert_data_point_expected( @staticmethod def create_number_data_point(value, attributes): + """Build a ``NumberDataPoint`` to compare a recorded one against. + + Both timestamps are set to ``0``, because ``is_data_points_equal`` + ignores them and a test cannot predict the real ones. + + Args: + value: The expected value of the counter or gauge. + attributes: The expected attributes of the data point. + + Returns: + A ``NumberDataPoint`` carrying the given value and attributes. + """ return NumberDataPoint( value=value, attributes=attributes, @@ -229,6 +422,26 @@ def create_histogram_data_point( attributes, explicit_bounds=None, ): + """Build a ``HistogramDataPoint`` to compare a recorded one against. + + Both timestamps are set to ``0`` and ``bucket_counts`` to an empty list, + because ``is_data_points_equal`` compares neither. + + Args: + sum_data_point: The expected sum of the recorded measurements. + count: The expected number of recorded measurements. + max_data_point: The expected largest recorded measurement. It is + compared only when ``est_value_delta`` is ``0``. + min_data_point: The expected smallest recorded measurement, compared + under the same condition. + attributes: The expected attributes of the data point. + explicit_bounds: The expected bucket boundaries. When ``None``, the + SDK default boundaries are used, which is what an instrument + configured with no explicit view reports. + + Returns: + A ``HistogramDataPoint`` carrying the given values. + """ return HistogramDataPoint( count=count, sum=sum_data_point, @@ -245,11 +458,36 @@ def create_histogram_data_point( class FinishedTestSpans(list): + """A list of ended spans that can look one up and fail the test if absent. + + Returned by :meth:`TestBase.get_finished_spans`. Every ``list`` operation + works on it. The ``by_name`` and ``by_attr`` lookups exist so that a missing + span is reported as a named test failure rather than as an + ``AttributeError`` on ``None`` several lines later. + + Args: + test: The test case to call ``fail`` on when a lookup finds nothing. + spans: The ended spans, copied into this list in the order given. + """ + def __init__(self, test, spans): + """Parameters are documented on the class docstring.""" super().__init__(spans) self.test = test def by_name(self, name): + """Return the first span with a given name, failing the test if absent. + + Args: + name: The exact span name to look for. Matching is by equality, not + by substring or pattern. + + Returns: + The first span in this list whose name matches. When no span + matches, the test case is failed, so the ``None`` in the return + statement is never the value a caller sees under the default + ``unittest`` runner. + """ for span in self: if span.name == name: return span @@ -257,6 +495,20 @@ def by_name(self, name): return None def by_attr(self, key, value): + """Return the first span carrying a given attribute value. + + Args: + key: The attribute key to read. + value: The value that key is expected to have. A span that does not + carry the key at all is treated as a span whose value is + ``None``, so passing ``None`` here matches such a span. + + Returns: + The first span in this list whose attribute matches. When no span + matches, the test case is failed, so the ``None`` in the return + statement is never the value a caller sees under the default + ``unittest`` runner. + """ for span in self: if span.attributes.get(key) == value: return span diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/version/__init__.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/version/__init__.py index b83363231d3..7b81b303fbc 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/version/__init__.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/version/__init__.py @@ -1,4 +1,16 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +"""Version of the ``opentelemetry-test-utils`` distribution. + +``[tool.hatch.version]`` in ``tests/opentelemetry-test-utils/pyproject.toml`` +points at this module, so ``__version__`` is the version that the built sdist +and wheel carry. ``scripts/eachdist.py`` rewrites the line during a release, +resolving the file through that same key. Nothing in this repository imports +``__version__`` from here. A test asserting on an instrumentation scope version +wants the version module of the package under test instead, which for +``opentelemetry-api`` and ``opentelemetry-sdk`` is on the stable version line +rather than the prerelease one this distribution shares with twelve others. +""" + __version__ = "0.67b0.dev" diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/weaver_live_check.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/weaver_live_check.py index 6219315e1c3..9c58fa1c89f 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/weaver_live_check.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/weaver_live_check.py @@ -1,6 +1,31 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +"""Checking emitted telemetry against the OpenTelemetry semantic conventions. + +Weaver is the OpenTelemetry tool that reads the semantic convention registry and +checks telemetry against it. Its ``registry live-check`` subcommand runs as a +server: it accepts OTLP over gRPC, accumulates what it is sent, and reports on +request which attributes and signals departed from the registry. + +:class:`WeaverLiveCheck` runs that subcommand as a subprocess and exposes it to +a test. A test points an OTLP exporter at ``otlp_endpoint``, emits telemetry, +then calls :meth:`WeaverLiveCheck.end_and_check` to assert that weaver found +nothing, or :meth:`WeaverLiveCheck.end` to get the report and assert on it +directly. :class:`LiveCheckReport` wraps the JSON report and +:class:`LiveCheckError` is what a failed check raises. + +The ``weaver`` binary has to be on ``PATH``, and constructing +:class:`WeaverLiveCheck` raises ``RuntimeError`` when no binary of that name is +found. Version 0.27 or later is required, and nothing checks the version: an +older weaver is accepted by the constructor and rejects one of the managed flags +later, which surfaces as a ``RuntimeError`` out of +:meth:`WeaverLiveCheck.start` naming only the exit status. +Downloads are at https://github.com/open-telemetry/weaver/releases + +This module is experimental and its API may change without notice. +""" + import functools import json import logging @@ -25,6 +50,14 @@ def _find_free_port() -> int: + """Ask the operating system for a port that is free right now. + + Returns: + A TCP port number. Binding to port ``0`` and reading back the assigned + port is the only portable way to get one, and the socket is closed + before the number is handed over, so another process can take the port + between this call and weaver binding to it. + """ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.bind(("", 0)) return sock.getsockname()[1] @@ -40,10 +73,36 @@ def _extract_violations(report: dict) -> list: | group_by(.id, .message, .context, .signal_name, .signal_type) | map({ ..., count: length }) | sort_by(-.count, .message) + + The same violation is reported by weaver once per sample that triggered it, + so a single wrong attribute on a thousand spans produces a thousand + entries. Grouping them turns that into one entry with a count. + + Args: + report: The full report as weaver returned it, already parsed from + JSON. It is searched recursively, so the shape below the top level + does not have to be known. + + Returns: + One dict per distinct violation, with the keys ``id``, ``message``, + ``context``, ``signal_name``, ``signal_type`` and ``count``, sorted by + descending count and then by message. ``context`` is the original dict + from weaver, not the JSON string used to group by it. A report with no + violation gives an empty list. """ raw: list[dict] = [] def _collect(obj: Any) -> list[dict]: + """Walk a parsed JSON value and gather every violation under it. + + Args: + obj: Any value from the parsed report: a dict, a list, or a scalar. + + Returns: + Every advice entry at level ``violation`` found anywhere below + ``obj``, in traversal order and with duplicates kept. A scalar + gives an empty list, which is what ends the recursion. + """ if isinstance(obj, dict): result: list[dict] = [] lcr = obj.get("live_check_result") @@ -88,7 +147,17 @@ def _collect(obj: Any) -> list[dict]: def _format_violations(violations: list) -> str: - """Format violations list as human-readable text.""" + """Format violations list as human-readable text. + + Args: + violations: The grouped violations produced by ``_extract_violations``. + + Returns: + One line per violation, newline separated, each naming the violation + id, the message, the occurrence count and the signal it was seen on. + The signal part is omitted when weaver reported neither a signal type + nor a signal name. An empty list gives an empty string. + """ lines = [] for violation in violations: signal = "" @@ -109,10 +178,10 @@ def _format_violations(violations: list) -> str: class LiveCheckError(AssertionError): """Raised by :meth:`WeaverLiveCheck.end_and_check` when semconv violations are found. - Captured process output is available as :attr:`stdout` and :attr:`stderr`, + Captured process output is available as ``stdout`` and ``stderr``, including after the live-check context has closed. - The full :class:`LiveCheckReport` is attached as :attr:`report` for + The full :class:`LiveCheckReport` is attached as ``report`` for structured inspection beyond the human-readable message:: with pytest.raises(LiveCheckError) as exc_info: @@ -123,9 +192,19 @@ class LiveCheckError(AssertionError): v["id"] == "my_policy_check" and v["context"]["attribute_name"] == "my.attribute" for v in err.report.violations ) + + Args: + message: The failure message, which :meth:`WeaverLiveCheck.end_and_check` + builds from the formatted violation list. + report: The full report the violations came from, kept as ``report``. + stdout: Everything weaver wrote to stdout up to the failure, kept as + ``stdout``. + stderr: Everything weaver wrote to stderr up to the failure, kept as + ``stderr``. """ def __init__(self, message: str, report: "LiveCheckReport", stdout: str = "", stderr: str = "") -> None: + """Parameters are documented on the class docstring.""" super().__init__(message) self.report = report self.stdout = stdout @@ -152,23 +231,54 @@ class LiveCheckReport: assert any( v["id"] == "my_policy_check" and v["context"]["attribute_name"] == "my.attribute" for v in report.violations ) + + Args: + report: The report as weaver returned it, already parsed from JSON. It + is stored, not copied, so the caller should not keep editing it. """ def __init__(self, report: dict[str, Any]) -> None: + """Parameters are documented on the class docstring.""" self._report = report def to_dict(self) -> dict[str, Any]: - """Return a deep copy of the full report for inspection or JSON serialization.""" + """Return a deep copy of the full report for inspection or JSON serialization. + + Returns: + A deep copy, so editing it does not affect this report or any later + call. Use ``__getitem__`` or ``get`` to read one key without + copying the whole report. + """ return deepcopy(self._report) @property def samples(self) -> list[dict[str, Any]]: - """Return a deep copy of the samples, or an empty list if absent.""" + """Return a deep copy of the samples, or an empty list if absent. + + A sample is one span, metric or log record as weaver received it, + annotated with the advice weaver attached to it. + + Returns: + A deep copy of the ``samples`` entry of the report, or an empty + list when weaver recorded none. A fresh copy is built on every + access, so bind it to a name rather than indexing the property + repeatedly. + """ return deepcopy(self._report.get("samples", [])) @property def statistics(self) -> dict[str, Any]: - """Return a deep copy of the statistics, or an empty dict if absent.""" + """Return a deep copy of the statistics, or an empty dict if absent. + + The statistics count what weaver saw, for example which registry + metrics appeared and how often, which is how a test asserts that an + instrument fired without inspecting individual samples. + + Returns: + A deep copy of the ``statistics`` entry of the report, or an empty + dict when weaver recorded none. A fresh copy is built on every + access. + """ return deepcopy(self._report.get("statistics", {})) @functools.cached_property @@ -178,19 +288,69 @@ def violations(self) -> list[dict[str, Any]]: Each item is a dict with keys: ``id``, ``message``, ``context`` (the raw context dict from weaver, e.g. ``{"attribute_name": "foo"}``), ``signal_name``, ``signal_type``, ``count``. + + Returns: + The violations, most frequent first and then by message. The list + is computed once and cached on the instance, and it is the cached + list itself rather than a copy, so editing it changes what later + reads return. A report with no violation gives an empty list, which + is not the same as :meth:`WeaverLiveCheck.end_and_check` passing: + that method + decides on weaver's exit status, so weaver can exit non-zero for a + reason that produces no violation entry and raise + :class:`LiveCheckError` with an empty list in the message. """ return _extract_violations(self._report) def __getitem__(self, key: str) -> Any: + """Read a top-level key of the raw report. + + Args: + key: Top-level key of the weaver report, for example + ``"statistics"``. + + Returns: + The value as weaver returned it, not copied, so editing a dict or + list obtained this way changes the report this object holds. + + Raises: + KeyError: When the report has no such key, which is what happens + for a section weaver omits when it is empty. + """ return self._report[key] def get(self, key: str, default: Any = None) -> Any: + """Read a top-level key of the raw report, with a fallback. + + Args: + key: Top-level key of the weaver report. + default: Returned when the key is absent. + + Returns: + The value as weaver returned it, not copied, or ``default`` when + the key is absent. + """ return self._report.get(key, default) def __contains__(self, key: object) -> bool: + """Report whether the raw report has a given top-level key. + + Args: + key: Key to look for. Only top-level keys are considered. + + Returns: + ``True`` when the report has that key. + """ return key in self._report def __repr__(self) -> str: + """Return a short description naming the number of violations. + + Returns: + A string such as ``LiveCheckReport(3 violations)``. Computing it + extracts the violations, so the first ``repr`` of a report does the + grouping work that :attr:`violations` caches. + """ num_violations = len(self.violations) return f"LiveCheckReport({num_violations} violation{'s' if num_violations != 1 else ''})" @@ -215,8 +375,9 @@ def test_my_telemetry(self): # ... configure provider, emit telemetry ... provider.force_flush() - # Signals weaver to stop, raises LiveCheckError listing violations - # if any, or returns a LiveCheckReport on success. + # Signals weaver to stop, returns a LiveCheckReport when weaver + # exits zero, and raises LiveCheckError with the violations + # listed in the message when it does not. report = weaver.end_and_check() # __exit__ calls close(), which is idempotent if end_and_check() was already called @@ -235,9 +396,11 @@ def test_my_telemetry(self): Lifecycle: - :meth:`start` — launches weaver and waits for it to become ready. - :attr:`otlp_endpoint` — gRPC OTLP endpoint to point exporters at. - - :meth:`end` — signals weaver to stop and always returns a - :class:`LiveCheckReport`. Never raises for semconv violations; use - this when you want to write your own assertions. + - :meth:`end` — signals weaver to stop and returns the + :class:`LiveCheckReport` whether or not it holds violations. Never + raises for semconv violations, which is why you write your own + assertions against it; see the method itself for the three ways it + raises instead of returning. - :meth:`end_and_check` — signals weaver to stop and raises :class:`LiveCheckError` with a human-readable violation list and the full report attached if weaver exits non-zero. Returns a @@ -245,6 +408,46 @@ def test_my_telemetry(self): - :meth:`close` — stops weaver if not already stopped and terminates the process. Never raises for semconv violations. Idempotent; safe to call even if :meth:`end_and_check` or :meth:`end` was already called. + + Args: + registry: Semantic convention registry to check against, as a + directory, an archive URL or any other value weaver's + ``--registry`` accepts. An existing local directory is made + absolute. ``None`` downloads the published registry archive for + ``schema_version`` from GitHub, which means the constructor + configures a network download that ``start`` then performs. + schema_version: Version of the published registry to download, for + example ``"1.37.0"``. ``None`` takes the newest version known to + ``opentelemetry.semconv.schemas.Schemas``, so the check follows the + semantic conventions the installed SDK was built against. Ignored + when ``registry`` is given. + policies_dir: Directory of extra advice policies, passed to weaver's + ``--advice-policies`` and made absolute. ``None`` runs only the + policies built into weaver. + inactivity_timeout: Seconds weaver waits without receiving telemetry + before stopping on its own. A test slower than this between + ``start`` and the first export finds weaver already gone. + otlp_port: Port for weaver's OTLP gRPC listener. ``0`` picks a free one, + which is what lets several of these run at once. + admin_port: Port for weaver's admin HTTP listener, which serves the + health, stop, report and shutdown endpoints this class calls. + ``0`` picks a free one. + extra_args: Further weaver arguments, appended verbatim after the + managed flags, for example ``["--quiet"]``. ``None`` adds none. + startup_timeout: Seconds to wait for weaver's health endpoint to answer + before giving up. + config: Weaver TOML configuration file, made absolute and passed to + ``--config``. The managed flags above take precedence over values + in it. + advice_data: Value for weaver's ``--advice-data``, which loads JSON or + YAML data for advice policies. An existing local path is made + absolute; a URL or a glob is passed unchanged. + + Raises: + ValueError: When ``startup_timeout`` is not positive and finite. + RuntimeError: When no ``weaver`` binary is on ``PATH``. Nothing is + started by the constructor, so this is raised before any + subprocess exists. """ def __init__( @@ -262,18 +465,11 @@ def __init__( ): """Build the ``weaver registry live-check`` command. - ``extra_args`` is appended verbatim to the weaver command after the - managed flags (``--registry``, ``--otlp-grpc-port``, etc.) and lets - callers pass additional weaver options — for example ``["--quiet"]`` - or ``["--skip-policies"]`` — without subclassing. - - ``config`` selects a Weaver TOML configuration file. Managed command - flags take precedence over values in that file. - ``advice_data`` is passed to Weaver's ``--advice-data`` option to load - JSON/YAML data for advice policies. Existing local paths are made - absolute; other values (URLs, globs) are passed unchanged. - ``startup_timeout`` controls how long to wait for the health endpoint, - in seconds. It defaults to 30 seconds. + Parameters are documented on the class docstring. No subprocess is + started here: the command is assembled and the port numbers are chosen, + and :meth:`start` runs it. Choosing a port does not hold it, as + ``_find_free_port`` explains, so another process can take it before + weaver binds. """ if not isfinite(startup_timeout) or startup_timeout <= 0: raise ValueError("startup_timeout must be positive and finite") @@ -332,14 +528,45 @@ def __init__( logger.debug("Weaver command: %s", command) def __enter__(self) -> "WeaverLiveCheck": + """Start weaver for the duration of the block. + + Returns: + This instance, with weaver running and ready to receive telemetry. + """ return self.start() def __exit__(self, exc_type: Any, *_: object) -> None: + """Close weaver, whether the block completed or raised. + + When the block raised, the instance is marked stopped before closing, + so no report is requested. That keeps the original exception visible + instead of replacing it with whatever the report request would fail + with. + + Args: + exc_type: Class of the exception raised in the block, or ``None``. + *_: The exception value and traceback, which are not needed. + """ if exc_type is not None: self._stopped = True self.close() def start(self) -> "WeaverLiveCheck": + """Launch weaver and wait for its health endpoint to answer. + + Weaver's stdout and stderr are redirected to temporary files, which + :attr:`stdout` and :attr:`stderr` read and :meth:`close` deletes. + + Returns: + This instance, so the call can be chained, as ``__enter__`` does. + + Raises: + RuntimeError: When weaver exits before becoming ready. The + collected output is logged at error level first, and the + process is killed while collecting it. + TimeoutError: When weaver is still running but has not answered the + health endpoint within ``startup_timeout``. + """ logger.debug("Starting WeaverLiveCheck process...") # Redirect weaver's stdout/stderr to tempfiles stdout_fd, self._stdout_path = tempfile.mkstemp(prefix="weaver-stdout-", suffix=".log") @@ -363,6 +590,19 @@ def start(self) -> "WeaverLiveCheck": return self def _wait_for_ready(self) -> None: + """Poll weaver's health endpoint until it answers or the deadline passes. + + Polls roughly four times a second. A connection error is expected while + weaver is still binding its listener, so it is kept as the cause of + whichever exception is eventually raised rather than reported at once. + + Raises: + RuntimeError: When the weaver process exited, which is checked + before each poll so a crash is reported straight away instead + of after the full timeout. + TimeoutError: When the deadline passes with weaver still running + and still not answering. + """ deadline = time.monotonic() + self._startup_timeout last_error: RequestException | None = None while True: @@ -389,6 +629,14 @@ def _wait_for_ready(self) -> None: @property def otlp_endpoint(self) -> str: + """Return the endpoint to configure an OTLP exporter with. + + Returns: + The endpoint of weaver's OTLP gRPC listener. The port is fixed at + construction, so this is readable before :meth:`start`, but nothing + listens there until weaver is running. It is plain gRPC with no + TLS, so an exporter pointed at it has to be configured insecure. + """ return f"http://localhost:{self._otlp_port}" def _do_stop(self, timeout: int) -> tuple["LiveCheckReport", int]: @@ -396,6 +644,23 @@ def _do_stop(self, timeout: int) -> tuple["LiveCheckReport", int]: Raises for infrastructure errors (HTTP failure, process communication). Never raises for semconv violations. + + Args: + timeout: Seconds allowed for each admin HTTP request and for the + process to exit, not for the sequence as a whole. + + Returns: + A tuple of the parsed report and the exit status of the weaver + process. The status is the one ``_shutdown`` returns and means what + that method documents: non-zero is any failure weaver reports, + usually violations. It is returned rather than raised on. + + Raises: + RuntimeError: When weaver never became ready, so there is nothing + to stop. + requests.RequestException: When an admin request fails or answers + with an error status. Shutdown is still attempted in that case, + and the collected weaver output is logged at error level. """ if not self._ready: raise RuntimeError("WeaverLiveCheck process did not start successfully") @@ -420,12 +685,44 @@ def _do_stop(self, timeout: int) -> tuple["LiveCheckReport", int]: return report, exit_code def _shutdown(self, timeout: int) -> int: + """Ask weaver to exit and wait for the process to do so. + + Args: + timeout: Seconds allowed for the shutdown request and, separately, + for the process to exit. + + Returns: + The exit status of the weaver process. Non-zero is how weaver + reports a failed check, which is usually violations but covers any + failure it reports, and is negative when a signal killed it. + + Raises: + requests.RequestException: When the shutdown request fails or + answers with an error status. + subprocess.TimeoutExpired: When the process has not exited within + ``timeout`` after being asked to. + """ response = post(f"http://localhost:{self._admin_port}/shutdown", timeout=timeout) response.raise_for_status() assert self._process is not None return self._process.wait(timeout=timeout) def _end(self, timeout: int) -> tuple[LiveCheckReport, int]: + """Stop weaver once and return the cached outcome on every later call. + + Args: + timeout: Seconds passed on to ``_do_stop``, used only on the call + that actually stops weaver. + + Returns: + The tuple of report and exit status produced by the first call. + + Raises: + RuntimeError: When weaver was already stopped without a report + having been collected, which is the state :meth:`close` leaves + behind and the state ``__exit__`` forces after the block + raised. + """ if not self._stopped: self._stopped = True self._result = self._do_stop(timeout) @@ -444,11 +741,21 @@ def end(self, timeout: int = 30) -> "LiveCheckReport": Raises :exc:`RuntimeError` if closed without requesting a report or if an earlier report request or shutdown failed. - Raises :exc:`RuntimeError` for infrastructure problems (weaver failed - to start, HTTP communication error, etc.). + Raises :exc:`RuntimeError` when weaver never became ready, so there is + nothing to stop. An admin request that fails or answers with an error + status is not turned into a :exc:`RuntimeError`: the + ``requests.RequestException`` raised by ``_do_stop`` propagates. See https://github.com/open-telemetry/weaver/tree/main/crates/weaver_live_check#output for the report structure. + + Args: + timeout: Seconds allowed for each step of stopping weaver: the stop + request, the report request, the shutdown request and the + process exit. + + Returns: + The report, whether or not it holds violations. """ report, _ = self._end(timeout) return report @@ -462,15 +769,23 @@ def end_and_check(self, timeout: int = 30) -> "LiveCheckReport": Does **not** return if weaver exits with a non-zero status — raises :exc:`LiveCheckError` (a subclass of :exc:`AssertionError`) with a human-readable list of violations and the full :class:`LiveCheckReport` - attached as :attr:`LiveCheckError.report`. + attached as ``LiveCheckError.report``. Use :meth:`end` if you need the report regardless of violations. Repeated calls check the cached result and raise again for violations. Raises :exc:`RuntimeError` if closed without requesting a report or if an earlier report request or shutdown failed. - Raises :exc:`RuntimeError` for infrastructure problems (weaver failed - to start, HTTP communication error, etc.). + Raises :exc:`RuntimeError` when weaver never became ready, so there is + nothing to stop. An admin request that fails or answers with an error + status is not turned into a :exc:`RuntimeError`: the + ``requests.RequestException`` raised by ``_do_stop`` propagates. + + Args: + timeout: Seconds allowed for each step of stopping weaver. + + Returns: + The report, when weaver exited with status zero. """ report, exit_code = self._end(timeout) if exit_code == 0: @@ -485,6 +800,20 @@ def end_and_check(self, timeout: int = 30) -> "LiveCheckReport": @staticmethod def _read_output(path: str | None, cached: str) -> str: + """Read one of weaver's output files, falling back to what was cached. + + Args: + path: The temporary file to read, or ``None`` once :meth:`close` + has deleted it. + cached: The text captured before the file was deleted, returned + when the file cannot be read. + + Returns: + The file contents decoded as UTF-8, with undecodable bytes replaced + rather than raising, since weaver output is read for a human. When + ``path`` is ``None`` or the file cannot be opened, ``cached`` is + returned and the failure is logged at debug level. + """ if path is None: return cached try: @@ -496,17 +825,39 @@ def _read_output(path: str | None, cached: str) -> str: @property def stdout(self) -> str: - """Captured stdout so far, retained after close. Reading does not stop Weaver.""" + """Captured stdout so far, retained after close. Reading does not stop Weaver. + + Returns: + Everything weaver has written to stdout. Reading re-reads the + temporary file while weaver is running, so the value grows between + calls, and returns the text captured at close afterwards. + """ self._stdout = self._read_output(self._stdout_path, self._stdout) return self._stdout @property def stderr(self) -> str: - """Captured stderr so far, retained after close. Reading does not stop Weaver.""" + """Captured stderr so far, retained after close. Reading does not stop Weaver. + + Returns: + Everything weaver has written to stderr, under the same rules as + :attr:`stdout`. + """ self._stderr = self._read_output(self._stderr_path, self._stderr) return self._stderr def _read_weaver_logs(self) -> str | None: + """Kill weaver and return everything it wrote, for a failure message. + + This is called on the error paths of :meth:`start` and ``_do_stop``, + where weaver is not going to be asked politely again. It kills the + process first so that the output files are complete. + + Returns: + The stdout and stderr captured so far, joined by a newline. + ``None`` when weaver was never started, and also ``None`` when + reading failed, in which case the failure is logged at error level. + """ if self._process is None: return None try: diff --git a/tests/opentelemetry-test-utils/src/opentelemetry/test/wsgitestutil.py b/tests/opentelemetry-test-utils/src/opentelemetry/test/wsgitestutil.py index b12482bad93..bf610e35f86 100644 --- a/tests/opentelemetry-test-utils/src/opentelemetry/test/wsgitestutil.py +++ b/tests/opentelemetry-test-utils/src/opentelemetry/test/wsgitestutil.py @@ -1,6 +1,16 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +"""A ``TestBase`` that supplies the WSGI environment and the ``start_response``. + +A WSGI application is a callable taking an environment dict and a +``start_response`` callable, so a test of WSGI instrumentation has to supply +both and then read back what the application passed to ``start_response``. +:class:`WsgiTestBase` builds a default environment and acts as the +``start_response`` itself, recording the status, the headers and the exception +information on the test case. +""" + import io import wsgiref.util as wsgiref_util @@ -9,7 +19,31 @@ class WsgiTestBase(TestBase): + """Base class for a test that calls a WSGI application directly. + + Inherits the in-memory trace and metrics pipelines of + :class:`opentelemetry.test.test_base.TestBase`, so a test can assert on the + spans the application produced as well as on the response it returned. + + ``setUp`` leaves four attributes to pass into the application and three to + read back out of it. ``self.environ`` is the WSGI environment, filled with + the defaults of ``wsgiref.util.setup_testing_defaults``, which describe a + ``GET /`` over HTTP/1.0 whose only request header is ``HTTP_HOST``, set to + ``127.0.0.1``; edit it before the call to change the request. + ``self.start_response`` is the callable to hand to the application, + ``self.write`` is the write callable it returns, and ``self.write_buffer`` + is the ``BytesIO`` that write callable appends to. + After the call, ``self.status``, ``self.response_headers`` and + ``self.exc_info`` hold what the application passed to ``start_response``, + and stay ``None`` if it never called it. + """ + def setUp(self): + """Build a fresh WSGI environment and clear the recorded response. + + Every attribute is replaced rather than cleared in place, so a + reference one test method kept does not leak into the next. + """ super().setUp() self.write_buffer = io.BytesIO() @@ -23,12 +57,46 @@ def setUp(self): self.exc_info = None def start_response(self, status, response_headers, exc_info=None): + """Record the response the application is starting, as WSGI requires. + + This is the ``start_response`` callable to pass to the application + under test. It records its arguments rather than writing anything, and + it does not implement the WSGI rule that a second call with + ``exc_info`` has to re-raise, so a test of that rule needs its own + callable. + + Args: + status: The status line the application is sending, for example + ``"200 OK"``. + response_headers: The list of header name and value pairs. + exc_info: The ``sys.exc_info()`` triple when the application is + replacing an already-started response after an error, and + ``None`` otherwise. + + Returns: + The ``write`` callable of ``self.write_buffer``, which is the + legacy WSGI write callable. An application that uses it appends to + that buffer. + """ self.status = status self.response_headers = response_headers self.exc_info = exc_info return self.write def assertTraceResponseHeaderMatchesSpan(self, headers, span): # pylint: disable=invalid-name + """Assert that the ``traceresponse`` header identifies a given span. + + Checks that the header is present, that + ``access-control-expose-headers`` names it so a browser can read it + cross-origin, and that its value is the W3C Trace Context version 0 + encoding of the trace id, span id and trace flags of the span. + + Args: + headers: The response headers, looked up by lowercase name, so a + plain dict has to use lowercase keys. + span: The span the header is expected to identify. Its span context + is read, so a non-recording span with a valid context works. + """ self.assertIn("traceresponse", headers) self.assertEqual( headers["access-control-expose-headers"],