Skip to content
Merged
Original file line number Diff line number Diff line change
Expand Up @@ -353,21 +353,20 @@ def _is_any_synthetic_source(properties: Optional[Any]) -> bool:

# pylint: disable=W0622
def _filter_custom_properties(properties: Attributes, filter=None) -> Dict[str, str]:
max_length = 64 * 1024
truncated_properties: Dict[str, str] = {}
filtered_properties: Dict[str, str] = {}
if not properties:
return truncated_properties
return filtered_properties
for key, val in properties.items():
# Apply filter function
if filter is not None:
if not filter(key, val):
continue
# Apply truncation rules
# Max key length is 150, value is 64 * 1024
# Apply filtering rules
# Max key length is 150
if not key or len(key) > 150 or val is None:
continue
truncated_properties[key] = str(val)[:max_length]
return truncated_properties
filtered_properties[key] = str(val)
return filtered_properties


def _get_auth_policy(credential, default_auth_policy, aad_audience=None):
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -35,10 +35,9 @@ class TestUtils(unittest.TestCase):
def setUp(self):
self._valid_instrumentation_key = "1234abcd-5678-4efa-8abc-1234567890ab"

def test_filter_custom_properties_truncates_and_drops_invalid_entries(self):
oversized_value = "v" * 9000
def test_filter_custom_properties_drops_invalid_entries(self):
properties = {
"valid_key": oversized_value,
"valid_key": "valid_value",
"k" * 151: "should_be_dropped",
"": "missing_key",
"short": "ok",
Expand All @@ -49,10 +48,23 @@ def test_filter_custom_properties_truncates_and_drops_invalid_entries(self):

self.assertEqual(len(filtered), 2)
self.assertIn("valid_key", filtered)
self.assertEqual(len(filtered["valid_key"]), 9000)
self.assertEqual(filtered["valid_key"], "valid_value")
self.assertEqual(filtered["short"], "ok")
self.assertNotIn("k" * 151, filtered)

def test_filter_custom_properties_preserves_large_values(self):
# Ensure values larger than 64KiB are not truncated
large_value = "x" * (64 * 1024 + 1000)
properties = {
"large_key": large_value,
}

filtered = _utils._filter_custom_properties(properties)

self.assertIn("large_key", filtered)
self.assertEqual(filtered["large_key"], large_value)
self.assertEqual(len(filtered["large_key"]), 64 * 1024 + 1000)

def test_nanoseconds_to_duration(self):
ns_to_duration = _utils.ns_to_duration
self.assertEqual(ns_to_duration(0), "0.00:00:00.000")
Expand Down
Loading