diff --git a/.changelog/5739.added b/.changelog/5739.added new file mode 100644 index 0000000000..6c3364c2d4 --- /dev/null +++ b/.changelog/5739.added @@ -0,0 +1,2 @@ +`opentelemetry-test-utils`: expose live-check reports and retained stdout/stderr, +and add named Weaver configuration and advice-data options. 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 cef8e40cad..6219315e1c 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 @@ -12,6 +12,7 @@ import time from collections import defaultdict from collections.abc import Sequence +from copy import deepcopy from itertools import chain from math import isfinite from typing import Any @@ -108,6 +109,9 @@ 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`, + including after the live-check context has closed. + The full :class:`LiveCheckReport` is attached as :attr:`report` for structured inspection beyond the human-readable message:: @@ -121,9 +125,11 @@ class LiveCheckError(AssertionError): ) """ - def __init__(self, message: str, report: "LiveCheckReport") -> None: + def __init__(self, message: str, report: "LiveCheckReport", stdout: str = "", stderr: str = "") -> None: super().__init__(message) self.report = report + self.stdout = stdout + self.stderr = stderr class LiveCheckReport: @@ -151,6 +157,20 @@ class LiveCheckReport: def __init__(self, report: dict[str, Any]) -> None: self._report = report + def to_dict(self) -> dict[str, Any]: + """Return a deep copy of the full report for inspection or JSON serialization.""" + 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 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 deepcopy(self._report.get("statistics", {})) + @functools.cached_property def violations(self) -> list[dict[str, Any]]: """Deduplicated list of semconv violations found in the report. @@ -237,6 +257,8 @@ def __init__( admin_port: int = 0, extra_args: Sequence[str] | None = None, startup_timeout: float = 30, + config: str | None = None, + advice_data: str | None = None, ): """Build the ``weaver registry live-check`` command. @@ -245,6 +267,11 @@ def __init__( 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. """ @@ -265,6 +292,8 @@ def __init__( self._process: subprocess.Popen[bytes] | None = None self._stdout_path: str | None = None self._stderr_path: str | None = None + self._stdout = "" + self._stderr = "" command = [ weaver_bin, @@ -280,6 +309,13 @@ def __init__( if policies_dir: command += ["--advice-policies", os.path.abspath(policies_dir)] + if config is not None: + command += ["--config", os.path.abspath(config)] + if advice_data is not None: + if os.path.exists(advice_data): + advice_data = os.path.abspath(advice_data) + command += ["--advice-data", advice_data] + if registry is None: if schema_version is None: schema_version = list(Schemas)[-1].value.rsplit("/", 1)[-1] @@ -443,8 +479,33 @@ def end_and_check(self, timeout: int = 30) -> "LiveCheckReport": raise LiveCheckError( f"Semconv violations found:\n{_format_violations(report.violations)}", report, + stdout=self.stdout, + stderr=self.stderr, ) + @staticmethod + def _read_output(path: str | None, cached: str) -> str: + if path is None: + return cached + try: + with open(path, "rb") as fp: + return fp.read().decode(errors="replace") + except OSError as exc: + logger.debug("Could not read weaver output from %s: %s", path, exc) + return cached + + @property + def stdout(self) -> str: + """Captured stdout so far, retained after close. Reading does not stop Weaver.""" + 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.""" + self._stderr = self._read_output(self._stderr_path, self._stderr) + return self._stderr + def _read_weaver_logs(self) -> str | None: if self._process is None: return None @@ -456,13 +517,7 @@ def _read_weaver_logs(self) -> str | None: except subprocess.TimeoutExpired: pass - def _read(path: str | None) -> str: - if path is None or not os.path.exists(path): - return "" - with open(path, "rb") as fp: - return fp.read().decode(errors="replace") - - return f"{_read(self._stdout_path)}\n{_read(self._stderr_path)}" + return f"{self.stdout}\n{self.stderr}" except Exception as exc: # pylint: disable=broad-except logger.error("Could not get weaver logs: %s", exc) return None @@ -488,6 +543,12 @@ def close(self) -> None: self._process.wait(timeout=5) except subprocess.TimeoutExpired: self._process.kill() + try: + self._process.wait(timeout=5) + except subprocess.TimeoutExpired: + logger.debug("Weaver did not exit after kill; continuing output cleanup") + self._stdout = self.stdout + self._stderr = self.stderr for path in (self._stdout_path, self._stderr_path): if path is not None: try: diff --git a/tests/opentelemetry-test-utils/tests/test_weaver_live_check.py b/tests/opentelemetry-test-utils/tests/test_weaver_live_check.py index 2219ded203..9da0d179ea 100644 --- a/tests/opentelemetry-test-utils/tests/test_weaver_live_check.py +++ b/tests/opentelemetry-test-utils/tests/test_weaver_live_check.py @@ -7,9 +7,12 @@ https://github.com/open-telemetry/weaver/releases """ +import json import os import shutil +import tempfile import unittest +from pathlib import Path from opentelemetry.sdk.resources import SERVICE_NAME, Resource from opentelemetry.sdk.trace import TracerProvider @@ -50,6 +53,41 @@ def _make_provider(otlp_endpoint: str) -> TracerProvider: "weaver binary not found on PATH — install from https://github.com/open-telemetry/weaver/releases", ) class TestSDKInitLiveCheck(unittest.TestCase): + def test_config_advice_data_and_retained_diagnostics(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + policies = root / "policies" + policies.mkdir() + (policies / "test.rego").write_text( + "package live_check_advice\nimport rego.v1\n" + "deny contains result if {\n" + ' input.sample.attribute.name == "never.use.this.attribute"\n' + ' result := {"type": "advice", "advice_type": "data_check",\n' + ' "advice_level": "violation", "message": data.settings.message}\n' + "}\n", + encoding="utf-8", + ) + data = root / "data" + data.mkdir() + (data / "settings.json").write_text('{"message": "message from advice data"}', encoding="utf-8") + config = root / "weaver.toml" + config.write_text(f"[live-check]\nadvice_policies = {json.dumps(str(policies))}\n", encoding="utf-8") + with WeaverLiveCheck(registry=_REGISTRY_DIR, config=str(config), advice_data=str(data)) as weaver: + provider = _make_provider(weaver.otlp_endpoint) + try: + with provider.get_tracer("test-tracer").start_as_current_span("test-span") as span: + span.set_attribute("never.use.this.attribute", "bad value") + provider.force_flush() + with self.assertRaisesRegex(LiveCheckError, "message from advice data") as caught: + weaver.end_and_check() + finally: + provider.shutdown() + self.assertTrue(weaver.stderr) + self.assertEqual(caught.exception.stderr, weaver.stderr) + self.assertEqual(caught.exception.stdout, weaver.stdout) + self.assertTrue(caught.exception.report.samples) + self.assertTrue(caught.exception.report.statistics) + def test_end_and_check_no_violations(self): """end_and_check() returns a LiveCheckReport with no violations on conformant telemetry.""" with WeaverLiveCheck(registry=_REGISTRY_DIR) as weaver: diff --git a/tests/opentelemetry-test-utils/tests/test_weaver_live_check_access.py b/tests/opentelemetry-test-utils/tests/test_weaver_live_check_access.py new file mode 100644 index 0000000000..5c496e6181 --- /dev/null +++ b/tests/opentelemetry-test-utils/tests/test_weaver_live_check_access.py @@ -0,0 +1,155 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +# pylint: disable=protected-access + +import json +import os +import subprocess +import tempfile +import unittest +from contextlib import ExitStack +from pathlib import Path +from unittest.mock import Mock, patch + +from opentelemetry.test.weaver_live_check import LiveCheckError, LiveCheckReport, WeaverLiveCheck + + +class TestReportAccess(unittest.TestCase): + def test_report_can_be_serialized_without_losing_fields(self): + raw = {"samples": [{"span": {"name": "test"}}], "statistics": {"count": 1}, "extra": [1]} + report = LiveCheckReport(raw) + self.assertEqual(json.loads(json.dumps(report.to_dict())), raw) + self.assertEqual(report.samples, raw["samples"]) + self.assertEqual(report.statistics, raw["statistics"]) + report.to_dict()["extra"].append(2) + report.samples[0]["span"]["name"] = "changed" + report.statistics["count"] = 2 + self.assertEqual(report["extra"], [1]) + self.assertEqual(report.samples[0]["span"]["name"], "test") + self.assertEqual(report.statistics["count"], 1) + + def test_missing_report_sections(self): + report = LiveCheckReport({}) + self.assertEqual(report.samples, []) + self.assertEqual(report.statistics, {}) + + +class TestWeaverOptions(unittest.TestCase): + @patch("opentelemetry.test.weaver_live_check.shutil.which", return_value="weaver") + def test_named_options(self, _which): + for data in ("data/*.json", "data with spaces", "https://example.com/data.tar.gz[policies]"): + with self.subTest(advice_data=data): + weaver = WeaverLiveCheck( + registry="registry", + otlp_port=4317, + admin_port=4320, + config="config with spaces.toml", + advice_data=data, + extra_args=["--quiet"], + ) + command = weaver._command + self.assertEqual(command[command.index("--config") + 1], os.path.abspath("config with spaces.toml")) + self.assertEqual(command[command.index("--advice-data") + 1], data) + self.assertEqual(command[-1], "--quiet") + + @patch("opentelemetry.test.weaver_live_check.shutil.which", return_value="weaver") + def test_local_advice_data_is_absolute(self, _which): + with tempfile.TemporaryDirectory() as directory: + Path(directory, "data.json").write_text("{}", encoding="utf-8") + cwd = os.getcwd() + os.chdir(directory) + try: + weaver = WeaverLiveCheck(registry="registry", otlp_port=4317, admin_port=4320, advice_data="data.json") + expected = os.path.abspath("data.json") + finally: + os.chdir(cwd) + command = weaver._command + self.assertEqual(command[command.index("--advice-data") + 1], expected) + + @patch("opentelemetry.test.weaver_live_check.shutil.which", return_value="weaver") + def test_optional_flags_are_omitted(self, _which): + weaver = WeaverLiveCheck(registry="registry", otlp_port=4317, admin_port=4320) + self.assertNotIn("--config", weaver._command) + self.assertNotIn("--advice-data", weaver._command) + + +class TestWeaverOutput(unittest.TestCase): + def setUp(self): + stack = ExitStack() + self.addCleanup(stack.close) + stack.enter_context(patch("opentelemetry.test.weaver_live_check.shutil.which", return_value="weaver")) + directory = Path(stack.enter_context(tempfile.TemporaryDirectory())) + self.stdout_path = directory / "stdout.log" + self.stderr_path = directory / "stderr.log" + self.stdout_path.write_bytes(b"output\xff\n") + self.stderr_path.write_bytes(b"diagnostic\n") + self.weaver = WeaverLiveCheck(registry="registry", otlp_port=4317, admin_port=4320) + self.weaver._stdout_path = str(self.stdout_path) + self.weaver._stderr_path = str(self.stderr_path) + self.process = Mock() + self.process.poll.return_value = None + self.weaver._process = self.process + self.addCleanup(self.weaver.close) + + def test_reading_output_does_not_stop_process(self): + self.assertEqual(self.weaver.stdout, "output\ufffd\n") + self.assertEqual(self.weaver.stderr, "diagnostic\n") + self.process.assert_not_called() + self.assertEqual(self.process.mock_calls, []) + self.stderr_path.write_bytes(b"more diagnostics\n") + self.assertEqual(self.weaver.stderr, "more diagnostics\n") + + def test_close_retains_output_and_deletes_files(self): + self.weaver.close() + self.weaver.close() + self.assertFalse(self.stdout_path.exists()) + self.assertFalse(self.stderr_path.exists()) + self.assertEqual(self.weaver.stdout, "output\ufffd\n") + self.assertEqual(self.weaver.stderr, "diagnostic\n") + + def test_forced_exit_retains_output(self): + self.process.wait.side_effect = [subprocess.TimeoutExpired("weaver", 5), 0] + self.process.poll.side_effect = [None, 0] + self.weaver.close() + self.process.kill.assert_called_once_with() + self.assertEqual(self.process.wait.call_count, 2) + self.assertEqual(self.weaver.stderr, "diagnostic\n") + + def test_output_preserves_windows_line_endings(self): + self.stderr_path.write_bytes(b"diagnostic\r\n") + self.weaver.close() + self.assertEqual(self.weaver.stderr, "diagnostic\r\n") + + def test_post_kill_timeout_preserves_original_error_and_output(self): + self.process.wait.side_effect = subprocess.TimeoutExpired("weaver", 5) + self.process.poll.side_effect = [None, 0] + with patch.object(self.weaver, "start", return_value=self.weaver): + with self.assertRaisesRegex(RuntimeError, "original failure"): + with self.weaver: + raise RuntimeError("original failure") + self.process.kill.assert_called_once_with() + self.assertEqual(self.process.wait.call_count, 2) + self.assertFalse(self.stdout_path.exists()) + self.assertFalse(self.stderr_path.exists()) + self.assertEqual(self.weaver.stdout, "output\ufffd\n") + self.assertEqual(self.weaver.stderr, "diagnostic\n") + + def test_error_keeps_output_after_close(self): + report = LiveCheckReport({}) + with patch.object(self.weaver, "_do_stop", return_value=(report, 1)): + with self.assertRaises(LiveCheckError) as caught: + self.weaver.end_and_check() + self.weaver.close() + self.assertEqual(caught.exception.stdout, "output\ufffd\n") + self.assertEqual(caught.exception.stderr, "diagnostic\n") + + def test_missing_output_file_preserves_last_read(self): + self.assertEqual(self.weaver.stderr, "diagnostic\n") + self.stderr_path.unlink() + self.assertEqual(self.weaver.stderr, "diagnostic\n") + + def test_error_constructor_remains_compatible(self): + error = LiveCheckError("violation", LiveCheckReport({})) + self.assertEqual(error.stdout, "") + self.assertEqual(error.stderr, "")