-
Notifications
You must be signed in to change notification settings - Fork 1.2k
Expand file tree
/
Copy pathtest_telemetry.py
More file actions
117 lines (94 loc) · 4.64 KB
/
test_telemetry.py
File metadata and controls
117 lines (94 loc) · 4.64 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
"""Tests for OpenTelemetry telemetry helpers."""
from __future__ import annotations
from unittest.mock import patch
from copilot._telemetry import get_trace_context, trace_context
from copilot.client import TelemetryConfig
class TestGetTraceContext:
def test_returns_empty_dict_when_otel_not_installed(self):
"""get_trace_context() returns {} when opentelemetry is not importable."""
real_import = __import__
def _block_otel(name: str, *args, **kwargs):
if name.startswith("opentelemetry"):
raise ImportError("mocked")
return real_import(name, *args, **kwargs)
with patch("builtins.__import__", side_effect=_block_otel):
result = get_trace_context()
assert result == {}
def test_returns_dict_type(self):
"""get_trace_context() always returns a dict."""
result = get_trace_context()
assert isinstance(result, dict)
class TestTraceContext:
def test_yields_without_error_when_no_traceparent(self):
"""trace_context() with no traceparent should yield without error."""
with trace_context(None, None):
pass # should not raise
def test_yields_without_error_when_otel_not_installed(self):
"""trace_context() should gracefully yield even if opentelemetry is missing."""
real_import = __import__
def _block_otel(name: str, *args, **kwargs):
if name.startswith("opentelemetry"):
raise ImportError("mocked")
return real_import(name, *args, **kwargs)
with patch("builtins.__import__", side_effect=_block_otel):
with trace_context("00-abc-def-01", None):
pass # should not raise
def test_yields_without_error_with_traceparent(self):
"""trace_context() with a traceparent value should yield without error."""
tp = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"
with trace_context(tp, None):
pass # should not raise
def test_yields_without_error_with_tracestate(self):
"""trace_context() with both traceparent and tracestate should yield without error."""
tp = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"
with trace_context(tp, "congo=t61rcWkgMzE"):
pass # should not raise
class TestTelemetryConfig:
def test_telemetry_config_type(self):
"""TelemetryConfig can be constructed as a TypedDict."""
config: TelemetryConfig = {
"otlp_endpoint": "http://localhost:4318",
"exporter_type": "otlp-http",
"source_name": "my-app",
"capture_content": True,
}
assert config["otlp_endpoint"] == "http://localhost:4318"
assert config["capture_content"] is True
def test_telemetry_env_var_mapping(self):
"""TelemetryConfig fields map to expected environment variable names."""
config: TelemetryConfig = {
"otlp_endpoint": "http://localhost:4318",
"file_path": "/tmp/traces.jsonl",
"exporter_type": "file",
"source_name": "test-app",
"capture_content": True,
}
env: dict[str, str] = {}
env["COPILOT_OTEL_ENABLED"] = "true"
if "otlp_endpoint" in config:
env["OTEL_EXPORTER_OTLP_ENDPOINT"] = config["otlp_endpoint"]
if "file_path" in config:
env["COPILOT_OTEL_FILE_EXPORTER_PATH"] = config["file_path"]
if "exporter_type" in config:
env["COPILOT_OTEL_EXPORTER_TYPE"] = config["exporter_type"]
if "source_name" in config:
env["COPILOT_OTEL_SOURCE_NAME"] = config["source_name"]
if "capture_content" in config:
env["OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"] = str(
config["capture_content"]
).lower()
assert env["COPILOT_OTEL_ENABLED"] == "true"
assert env["OTEL_EXPORTER_OTLP_ENDPOINT"] == "http://localhost:4318"
assert env["COPILOT_OTEL_FILE_EXPORTER_PATH"] == "/tmp/traces.jsonl"
assert env["COPILOT_OTEL_EXPORTER_TYPE"] == "file"
assert env["COPILOT_OTEL_SOURCE_NAME"] == "test-app"
assert env["OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"] == "true"
def test_capture_content_false_maps_to_lowercase(self):
"""capture_content=False should map to 'false' string."""
config: TelemetryConfig = {"capture_content": False}
value = str(config["capture_content"]).lower()
assert value == "false"
def test_empty_telemetry_config(self):
"""An empty TelemetryConfig is valid since total=False."""
config: TelemetryConfig = {}
assert len(config) == 0