237 lines
8.5 KiB
Python
237 lines
8.5 KiB
Python
from agentops.common.otel import otel_attributes_to_nested
|
|
|
|
|
|
class TestOtelAttributesToNested:
|
|
"""Tests for the otel_attributes_to_nested function."""
|
|
|
|
def test_simple_nested_attributes(self):
|
|
"""Test basic nested attribute conversion."""
|
|
attributes = {"foo.bar": "value1", "foo.baz": "value2", "top": "level"}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
assert result == {"foo": {"bar": "value1", "baz": "value2"}, "top": "level"}
|
|
|
|
def test_array_attributes(self):
|
|
"""Test conversion of attributes with numeric indices to arrays."""
|
|
attributes = {
|
|
"items.0.name": "first",
|
|
"items.0.value": "100",
|
|
"items.1.name": "second",
|
|
"items.1.value": "200",
|
|
}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
assert result == {"items": [{"name": "first", "value": "100"}, {"name": "second", "value": "200"}]}
|
|
|
|
def test_mixed_nested_and_arrays(self):
|
|
"""Test complex structure with both nested objects and arrays."""
|
|
attributes = {
|
|
"user.name": "John",
|
|
"user.tags.0": "admin",
|
|
"user.tags.1": "developer",
|
|
"user.metadata.age": "30",
|
|
"user.metadata.city": "NYC",
|
|
}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
assert result == {
|
|
"user": {"name": "John", "tags": ["admin", "developer"], "metadata": {"age": "30", "city": "NYC"}}
|
|
}
|
|
|
|
def test_sparse_array(self):
|
|
"""Test that sparse arrays are filled with None values."""
|
|
attributes = {"array.0": "first", "array.5": "sixth"}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
assert result == {"array": ["first", None, None, None, None, "sixth"]}
|
|
|
|
def test_string_terminal_value_skip(self):
|
|
"""Test that we skip attributes when trying to set a key on a string value."""
|
|
attributes = {
|
|
"field": "string_value",
|
|
"field.subfield": "should_be_skipped", # Can't add subfield to a string
|
|
"field.another.deep": "also_skipped",
|
|
}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
# Only the first string value should be set
|
|
assert result == {"field": "string_value"}
|
|
|
|
def test_type_mismatch_string_key_on_list(self):
|
|
"""Test that string keys on lists are skipped."""
|
|
attributes = {
|
|
"field.0": "array_item",
|
|
"field.invalid": "should_be_skipped", # String key on what should be a list
|
|
}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
# Only the valid array item should be set
|
|
assert result == {"field": ["array_item"]}
|
|
|
|
def test_type_mismatch_int_key_on_dict(self):
|
|
"""Test that integer keys on dicts are skipped."""
|
|
attributes = {
|
|
"field.name": "dict_value",
|
|
"field.0": "should_be_skipped", # Integer key on what should be a dict
|
|
}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
# Only the valid dict entry should be set
|
|
assert result == {"field": {"name": "dict_value"}}
|
|
|
|
def test_deep_nesting(self):
|
|
"""Test deeply nested structures."""
|
|
attributes = {"a.b.c.d.e.f": "deep_value", "a.b.c.d.e.g": "another_deep", "a.b.x": "sibling"}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
assert result == {
|
|
"a": {"b": {"c": {"d": {"e": {"f": "deep_value", "g": "another_deep"}}}, "x": "sibling"}}
|
|
}
|
|
|
|
def test_legacy_gen_ai_prompt_conversion(self):
|
|
"""Test that legacy gen_ai.prompt string format is converted to indexed format."""
|
|
attributes = {"gen_ai.prompt": "What is the weather today?", "gen_ai.model": "gpt-4"}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
assert result == {
|
|
"gen_ai": {
|
|
"prompt": [{"content": "What is the weather today?", "role": "user"}],
|
|
"model": "gpt-4",
|
|
}
|
|
}
|
|
|
|
def test_indexed_gen_ai_prompt_not_converted(self):
|
|
"""Test that properly indexed gen_ai.prompt is not modified."""
|
|
attributes = {
|
|
"gen_ai.prompt.0.content": "Hello",
|
|
"gen_ai.prompt.0.role": "user",
|
|
"gen_ai.prompt.1.content": "Hi there!",
|
|
"gen_ai.prompt.1.role": "assistant",
|
|
}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
assert result == {
|
|
"gen_ai": {
|
|
"prompt": [
|
|
{"content": "Hello", "role": "user"},
|
|
{"content": "Hi there!", "role": "assistant"},
|
|
]
|
|
}
|
|
}
|
|
|
|
def test_gen_ai_non_string_prompt_not_converted(self):
|
|
"""Test that gen_ai.prompt that's already an object/array is not converted."""
|
|
# First, let's build a structure where gen_ai.prompt is already an array
|
|
attributes = {"gen_ai.prompt.0": "already_indexed"}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
# Should not apply legacy conversion since it's already indexed
|
|
assert result == {"gen_ai": {"prompt": ["already_indexed"]}}
|
|
|
|
def test_empty_attributes(self):
|
|
"""Test that empty attributes dict returns empty result."""
|
|
result = otel_attributes_to_nested({})
|
|
assert result == {}
|
|
|
|
def test_single_level_attributes(self):
|
|
"""Test attributes with no nesting."""
|
|
attributes = {"field1": "value1", "field2": "value2", "field3": "value3"}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
assert result == attributes
|
|
|
|
def test_numeric_string_keys_not_converted(self):
|
|
"""Test that numeric strings that aren't at array positions stay as strings."""
|
|
attributes = {
|
|
"map.123": "value", # This should create a dict with string key "123"
|
|
"array.0": "item", # This should create an array
|
|
}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
assert result == {"map": {"123": "value"}, "array": ["item"]}
|
|
|
|
def test_conflicting_paths_first_wins(self):
|
|
"""Test that when paths conflict, the first one processed wins."""
|
|
attributes = {
|
|
"field.sub": "string_value",
|
|
"field.sub.nested": "should_be_skipped", # Can't nest under string
|
|
}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
assert result == {"field": {"sub": "string_value"}}
|
|
|
|
def test_complex_real_world_example(self):
|
|
"""Test a complex real-world example with OpenTelemetry span attributes."""
|
|
attributes = {
|
|
"gen_ai.prompt.0.content": "Write a function",
|
|
"gen_ai.prompt.0.role": "user",
|
|
"gen_ai.completion.0.content": "def hello(): pass",
|
|
"gen_ai.completion.0.role": "assistant",
|
|
"gen_ai.model": "gpt-4",
|
|
"gen_ai.usage.prompt_tokens": "10",
|
|
"gen_ai.usage.completion_tokens": "5",
|
|
"llm.is_streaming": "false",
|
|
"span.kind": "CLIENT",
|
|
"error.message": "none",
|
|
"tags.0": "production",
|
|
"tags.1": "v2",
|
|
}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
assert result == {
|
|
"gen_ai": {
|
|
"prompt": [{"content": "Write a function", "role": "user"}],
|
|
"completion": [{"content": "def hello(): pass", "role": "assistant"}],
|
|
"model": "gpt-4",
|
|
"usage": {"prompt_tokens": "10", "completion_tokens": "5"},
|
|
},
|
|
"llm": {"is_streaming": "false"},
|
|
"span": {"kind": "CLIENT"},
|
|
"error": {"message": "none"},
|
|
"tags": ["production", "v2"],
|
|
}
|
|
|
|
def test_edge_case_existing_string_in_path(self):
|
|
"""Test handling when trying to traverse through an existing string value."""
|
|
attributes = {
|
|
"a.b": "string",
|
|
"a.b.c.d": "should_skip", # Can't traverse through string at a.b
|
|
}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
assert result == {"a": {"b": "string"}}
|
|
|
|
def test_multiple_legacy_prompts(self):
|
|
"""Test multiple gen_ai attributes with legacy format."""
|
|
attributes = {
|
|
"gen_ai.prompt": "User question here",
|
|
"gen_ai.completion": "AI response here", # This won't be converted, only prompt
|
|
"other.field": "value",
|
|
}
|
|
|
|
result = otel_attributes_to_nested(attributes)
|
|
|
|
assert result == {
|
|
"gen_ai": {
|
|
"prompt": [{"content": "User question here", "role": "user"}],
|
|
"completion": "AI response here", # Not converted
|
|
},
|
|
"other": {"field": "value"},
|
|
}
|