opentelemetry-test-utils: expose Weaver reports and diagnostics (#5739)

* opentelemetry-test-utils: expose Weaver reports and diagnostics

Assisted-by: ChatGPT

* Name changelog fragment for PR 5739

* Preserve Weaver diagnostics after kill timeout and fix test portability

Assisted-by: ChatGPT

* Make local --advice-data paths absolute

Assisted-by: Claude Opus 4.6

* Keep startup_timeout position and fix Windows test

Assisted-by: Claude Opus 4.6

* Specify encoding in advice data test

Assisted-by: Claude Opus 4.6

Author: lmolkova

.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.

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:

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:

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, "")