test_log_formatters.py
1 """ 2 Unit tests for src/solace_agent_mesh/common/utils/log_formatters.py 3 4 Tests the DatadogJsonFormatter implementation including: 5 - JSON log formatting 6 - Timestamp formatting 7 - Service name handling 8 - Datadog trace/span ID integration 9 - Exception handling and stack traces 10 - Thread information 11 - Code location information 12 """ 13 14 import json 15 import logging 16 import os 17 import sys 18 import traceback 19 from datetime import datetime, timezone 20 from unittest.mock import Mock, patch 21 import pytest 22 23 from src.solace_agent_mesh.common.utils.log_formatters import DatadogJsonFormatter 24 25 26 @pytest.fixture 27 def formatter(): 28 """Create a DatadogJsonFormatter instance for testing""" 29 return DatadogJsonFormatter() 30 31 32 @pytest.fixture 33 def sample_log_record(): 34 """Create a sample LogRecord for testing""" 35 record = logging.LogRecord( 36 name="test.logger", 37 level=logging.INFO, 38 pathname="/path/to/test_module.py", 39 lineno=42, 40 msg="Test message", 41 args=(), 42 exc_info=None 43 ) 44 record.created = 1640995200.0 # 2022-01-01 00:00:00 UTC 45 record.threadName = "MainThread" 46 record.module = "test_module" 47 record.funcName = "test_function" 48 return record 49 50 51 class TestDatadogJsonFormatterBasic: 52 """Tests for basic DatadogJsonFormatter functionality""" 53 54 def test_format_basic_message(self, formatter, sample_log_record): 55 """Test formatting a basic log message""" 56 formatted = formatter.format(sample_log_record) 57 log_data = json.loads(formatted) 58 59 # Verify basic fields 60 assert log_data["message"] == "Test message" 61 assert log_data["level"] == "INFO" 62 assert log_data["logger.name"] == "test.logger" 63 assert log_data["logger.thread_name"] == "MainThread" 64 assert log_data["code.filepath"] == "/path/to/test_module.py" 65 assert log_data["code.lineno"] == 42 66 assert log_data["code.module"] == "test_module" 67 assert log_data["code.funcName"] == "test_function" 68 69 def test_format_timestamp(self, formatter, sample_log_record): 70 """Test timestamp formatting""" 71 formatted = formatter.format(sample_log_record) 72 log_data = json.loads(formatted) 73 74 # Verify timestamp is in ISO format 75 timestamp = log_data["timestamp"] 76 assert timestamp == "2022-01-01T00:00:00+00:00" 77 78 # Verify it can be parsed back 79 parsed_dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00')) 80 assert parsed_dt.tzinfo == timezone.utc 81 82 def test_format_different_log_levels(self, formatter): 83 """Test formatting different log levels""" 84 levels = [ 85 (logging.DEBUG, "DEBUG"), 86 (logging.INFO, "INFO"), 87 (logging.WARNING, "WARNING"), 88 (logging.ERROR, "ERROR"), 89 (logging.CRITICAL, "CRITICAL") 90 ] 91 92 for level_num, level_name in levels: 93 record = logging.LogRecord( 94 name="test.logger", 95 level=level_num, 96 pathname="/path/to/test.py", 97 lineno=1, 98 msg="Test message", 99 args=(), 100 exc_info=None 101 ) 102 record.created = 1640995200.0 103 record.threadName = "MainThread" 104 record.module = "test" 105 record.funcName = "test_func" 106 107 formatted = formatter.format(record) 108 log_data = json.loads(formatted) 109 assert log_data["level"] == level_name 110 111 def test_format_message_with_args(self, formatter): 112 """Test formatting message with arguments""" 113 record = logging.LogRecord( 114 name="test.logger", 115 level=logging.INFO, 116 pathname="/path/to/test.py", 117 lineno=1, 118 msg="Hello %s, you have %d messages", 119 args=("Alice", 5), 120 exc_info=None 121 ) 122 record.created = 1640995200.0 123 record.threadName = "MainThread" 124 record.module = "test" 125 record.funcName = "test_func" 126 127 formatted = formatter.format(record) 128 log_data = json.loads(formatted) 129 assert log_data["message"] == "Hello Alice, you have 5 messages" 130 131 def test_format_returns_valid_json(self, formatter, sample_log_record): 132 """Test that format always returns valid JSON""" 133 formatted = formatter.format(sample_log_record) 134 135 # Should not raise exception 136 log_data = json.loads(formatted) 137 assert isinstance(log_data, dict) 138 139 140 class TestDatadogJsonFormatterServiceName: 141 """Tests for service name handling""" 142 143 def test_format_with_default_service_name(self, formatter, sample_log_record): 144 """Test formatting with default service name""" 145 with patch.dict(os.environ, {}, clear=True): 146 formatted = formatter.format(sample_log_record) 147 log_data = json.loads(formatted) 148 assert log_data["service"] == "solace_agent_mesh" 149 150 def test_format_with_custom_service_name(self, formatter, sample_log_record): 151 """Test formatting with custom service name from environment""" 152 with patch.dict(os.environ, {"SERVICE_NAME": "custom_service"}): 153 formatted = formatter.format(sample_log_record) 154 log_data = json.loads(formatted) 155 assert log_data["service"] == "custom_service" 156 157 def test_format_with_empty_service_name(self, formatter, sample_log_record): 158 """Test formatting with empty service name falls back to default""" 159 with patch.dict(os.environ, {"SERVICE_NAME": ""}): 160 formatted = formatter.format(sample_log_record) 161 log_data = json.loads(formatted) 162 assert log_data["service"] == "" # Should use the empty value 163 164 165 class TestDatadogJsonFormatterDatadogIntegration: 166 """Tests for Datadog trace and span ID integration""" 167 168 def test_format_with_datadog_trace_id(self, formatter, sample_log_record): 169 """Test formatting with Datadog trace ID""" 170 sample_log_record.dd = Mock() 171 sample_log_record.dd.trace_id = "123456789" 172 # Alternative attribute access pattern 173 setattr(sample_log_record, "dd.trace_id", "123456789") 174 175 formatted = formatter.format(sample_log_record) 176 log_data = json.loads(formatted) 177 assert log_data["dd.trace_id"] == "123456789" 178 179 def test_format_with_datadog_span_id(self, formatter, sample_log_record): 180 """Test formatting with Datadog span ID""" 181 setattr(sample_log_record, "dd.span_id", "987654321") 182 183 formatted = formatter.format(sample_log_record) 184 log_data = json.loads(formatted) 185 assert log_data["dd.span_id"] == "987654321" 186 187 def test_format_with_both_datadog_ids(self, formatter, sample_log_record): 188 """Test formatting with both Datadog trace and span IDs""" 189 setattr(sample_log_record, "dd.trace_id", "123456789") 190 setattr(sample_log_record, "dd.span_id", "987654321") 191 192 formatted = formatter.format(sample_log_record) 193 log_data = json.loads(formatted) 194 assert log_data["dd.trace_id"] == "123456789" 195 assert log_data["dd.span_id"] == "987654321" 196 197 def test_format_without_datadog_ids(self, formatter, sample_log_record): 198 """Test formatting without Datadog IDs (should not include them)""" 199 formatted = formatter.format(sample_log_record) 200 log_data = json.loads(formatted) 201 assert "dd.trace_id" not in log_data 202 assert "dd.span_id" not in log_data 203 204 def test_format_with_none_datadog_ids(self, formatter, sample_log_record): 205 """Test formatting with None Datadog IDs (should not include them)""" 206 setattr(sample_log_record, "dd.trace_id", None) 207 setattr(sample_log_record, "dd.span_id", None) 208 209 formatted = formatter.format(sample_log_record) 210 log_data = json.loads(formatted) 211 assert "dd.trace_id" not in log_data 212 assert "dd.span_id" not in log_data 213 214 215 class TestDatadogJsonFormatterExceptionHandling: 216 """Tests for exception handling and stack traces""" 217 218 def test_format_with_exception_info(self, formatter): 219 """Test formatting with exception information""" 220 try: 221 raise ValueError("Test exception") 222 except ValueError: 223 exc_info = sys.exc_info() 224 record = logging.LogRecord( 225 name="test.logger", 226 level=logging.ERROR, 227 pathname="/path/to/test.py", 228 lineno=1, 229 msg="An error occurred", 230 args=(), 231 exc_info=exc_info 232 ) 233 record.created = 1640995200.0 234 record.threadName = "MainThread" 235 record.module = "test" 236 record.funcName = "test_func" 237 238 formatted = formatter.format(record) 239 log_data = json.loads(formatted) 240 241 assert log_data["message"] == "An error occurred" 242 assert log_data["exception.type"] == "ValueError" 243 assert log_data["exception.message"] == "Test exception" 244 assert "exception.stacktrace" in log_data 245 assert "ValueError: Test exception" in log_data["exception.stacktrace"] 246 assert "Traceback" in log_data["exception.stacktrace"] 247 248 def test_format_with_nested_exception(self, formatter): 249 """Test formatting with nested exception""" 250 try: 251 try: 252 raise ValueError("Inner exception") 253 except ValueError as e: 254 raise RuntimeError("Outer exception") from e 255 except RuntimeError: 256 exc_info = sys.exc_info() 257 record = logging.LogRecord( 258 name="test.logger", 259 level=logging.ERROR, 260 pathname="/path/to/test.py", 261 lineno=1, 262 msg="Nested error occurred", 263 args=(), 264 exc_info=exc_info 265 ) 266 record.created = 1640995200.0 267 record.threadName = "MainThread" 268 record.module = "test" 269 record.funcName = "test_func" 270 271 formatted = formatter.format(record) 272 log_data = json.loads(formatted) 273 274 assert log_data["exception.type"] == "RuntimeError" 275 assert log_data["exception.message"] == "Outer exception" 276 assert "exception.stacktrace" in log_data 277 # Should contain both exceptions in the stack trace 278 assert "RuntimeError: Outer exception" in log_data["exception.stacktrace"] 279 assert "ValueError: Inner exception" in log_data["exception.stacktrace"] 280 281 def test_format_without_exception_info(self, formatter, sample_log_record): 282 """Test formatting without exception information""" 283 formatted = formatter.format(sample_log_record) 284 log_data = json.loads(formatted) 285 286 assert "exception.type" not in log_data 287 assert "exception.message" not in log_data 288 assert "exception.stacktrace" not in log_data 289 290 def test_format_with_custom_exception_class(self, formatter): 291 """Test formatting with custom exception class""" 292 class CustomError(Exception): 293 pass 294 295 try: 296 raise CustomError("Custom error message") 297 except CustomError: 298 exc_info = sys.exc_info() 299 record = logging.LogRecord( 300 name="test.logger", 301 level=logging.ERROR, 302 pathname="/path/to/test.py", 303 lineno=1, 304 msg="Custom error occurred", 305 args=(), 306 exc_info=exc_info 307 ) 308 record.created = 1640995200.0 309 record.threadName = "MainThread" 310 record.module = "test" 311 record.funcName = "test_func" 312 313 formatted = formatter.format(record) 314 log_data = json.loads(formatted) 315 316 assert log_data["exception.type"] == "CustomError" 317 assert log_data["exception.message"] == "Custom error message" 318 319 320 class TestDatadogJsonFormatterDeprecation: 321 """Tests for deprecation warning functionality""" 322 323 def test_deprecation_warning_emitted_once(self): 324 """Test that deprecation warning is emitted only once""" 325 import warnings 326 327 # Reset the class-level flag if it exists 328 if hasattr(DatadogJsonFormatter, '_deprecation_warned'): 329 delattr(DatadogJsonFormatter, '_deprecation_warned') 330 331 formatter = DatadogJsonFormatter() 332 record = logging.LogRecord( 333 name="test.logger", 334 level=logging.INFO, 335 pathname="/test.py", 336 lineno=1, 337 msg="Test", 338 args=(), 339 exc_info=None 340 ) 341 record.created = 1640995200.0 342 record.threadName = "MainThread" 343 record.module = "test" 344 record.funcName = "test_func" 345 346 with warnings.catch_warnings(record=True) as w: 347 warnings.simplefilter("always") 348 formatter.format(record) 349 formatter.format(record) # Second call should not warn 350 351 deprecation_warnings = [x for x in w if issubclass(x.category, DeprecationWarning)] 352 assert len(deprecation_warnings) == 1 353 assert "deprecated" in str(deprecation_warnings[0].message).lower() 354 355 def test_deprecation_warning_contains_alternative(self): 356 """Test that deprecation warning mentions the alternative""" 357 import warnings 358 359 # Reset the class-level flag if it exists 360 if hasattr(DatadogJsonFormatter, '_deprecation_warned'): 361 delattr(DatadogJsonFormatter, '_deprecation_warned') 362 363 formatter = DatadogJsonFormatter() 364 record = logging.LogRecord( 365 name="test.logger", 366 level=logging.INFO, 367 pathname="/test.py", 368 lineno=1, 369 msg="Test", 370 args=(), 371 exc_info=None 372 ) 373 record.created = 1640995200.0 374 record.threadName = "MainThread" 375 record.module = "test" 376 record.funcName = "test_func" 377 378 with warnings.catch_warnings(record=True) as w: 379 warnings.simplefilter("always") 380 formatter.format(record) 381 382 deprecation_warnings = [x for x in w if issubclass(x.category, DeprecationWarning)] 383 assert len(deprecation_warnings) == 1 384 warning_message = str(deprecation_warnings[0].message) 385 assert "pythonjsonlogger" in warning_message.lower() or "JsonFormatter" in warning_message 386 387 388 class TestDatadogJsonFormatterEdgeCases: 389 """Tests for edge cases and special scenarios""" 390 391 def test_format_with_unicode_message(self, formatter): 392 """Test formatting with Unicode characters in message""" 393 record = logging.LogRecord( 394 name="test.logger", 395 level=logging.INFO, 396 pathname="/path/to/test.py", 397 lineno=1, 398 msg="Unicode test: 你好世界 🌍", 399 args=(), 400 exc_info=None 401 ) 402 record.created = 1640995200.0 403 record.threadName = "MainThread" 404 record.module = "test" 405 record.funcName = "test_func" 406 407 formatted = formatter.format(record) 408 log_data = json.loads(formatted) 409 assert log_data["message"] == "Unicode test: 你好世界 🌍" 410 411 def test_format_with_special_characters_in_paths(self, formatter): 412 """Test formatting with special characters in file paths""" 413 record = logging.LogRecord( 414 name="test.logger", 415 level=logging.INFO, 416 pathname="/path/with spaces/and-dashes/test_file.py", 417 lineno=1, 418 msg="Test message", 419 args=(), 420 exc_info=None 421 ) 422 record.created = 1640995200.0 423 record.threadName = "Thread-1" 424 record.module = "test_file" 425 record.funcName = "test_function_with_underscores" 426 427 formatted = formatter.format(record) 428 log_data = json.loads(formatted) 429 assert log_data["code.filepath"] == "/path/with spaces/and-dashes/test_file.py" 430 assert log_data["logger.thread_name"] == "Thread-1" 431 assert log_data["code.funcName"] == "test_function_with_underscores" 432 433 def test_format_with_very_long_message(self, formatter): 434 """Test formatting with very long log message""" 435 long_message = "A" * 10000 # 10KB message 436 record = logging.LogRecord( 437 name="test.logger", 438 level=logging.INFO, 439 pathname="/path/to/test.py", 440 lineno=1, 441 msg=long_message, 442 args=(), 443 exc_info=None 444 ) 445 record.created = 1640995200.0 446 record.threadName = "MainThread" 447 record.module = "test" 448 record.funcName = "test_func" 449 450 formatted = formatter.format(record) 451 log_data = json.loads(formatted) 452 assert log_data["message"] == long_message 453 assert len(formatted) > 10000 454 455 def test_format_with_none_values(self, formatter): 456 """Test formatting with None values in record attributes""" 457 record = logging.LogRecord( 458 name="test.logger", 459 level=logging.INFO, 460 pathname="/path/to/test.py", 461 lineno=1, 462 msg="Test message", 463 args=(), 464 exc_info=None 465 ) 466 record.created = 1640995200.0 467 record.threadName = None 468 record.module = None 469 record.funcName = None 470 471 formatted = formatter.format(record) 472 log_data = json.loads(formatted) 473 474 # Should handle None values gracefully 475 assert log_data["logger.thread_name"] is None 476 assert log_data["code.module"] is None 477 assert log_data["code.funcName"] is None 478 479 480 def test_format_consistent_field_names(self, formatter, sample_log_record): 481 """Test that field names are consistent with Datadog conventions""" 482 formatted = formatter.format(sample_log_record) 483 log_data = json.loads(formatted) 484 485 # Verify expected field names 486 expected_fields = { 487 "timestamp", "level", "message", "logger.name", "logger.thread_name", 488 "service", "code.filepath", "code.lineno", "code.module", "code.funcName" 489 } 490 491 for field in expected_fields: 492 assert field in log_data, f"Expected field '{field}' not found in log data"