FilesExpand file tree

main

/

test_telemetry.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

128 lines (104 loc) · 5.08 KB

main

/

test_telemetry.py

Copy path

Top

File metadata and controls

• Code
• Blame

128 lines (104 loc) · 5.08 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

118

119

120

121

122

123

124

125

126

127

128

"""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.types import SubprocessConfig, 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_config_in_subprocess_config(self):

"""TelemetryConfig can be used in SubprocessConfig."""

config = SubprocessConfig(

telemetry={

"otlp_endpoint": "http://localhost:4318",

"exporter_type": "otlp-http",

}

)

assert config.telemetry is not None

assert config.telemetry["otlp_endpoint"] == "http://localhost:4318"

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