mirror of
https://github.com/hoshikawa2/compass_backoffice.git
synced 2026-07-09 13:54:20 +00:00
421 lines
11 KiB
Python
421 lines
11 KiB
Python
"""
|
|
Testes unitários para o módulo de logging.
|
|
"""
|
|
|
|
import logging
|
|
import json
|
|
from io import StringIO
|
|
import pytest
|
|
|
|
import src.core.logging as _logging_module
|
|
from src.core.logging import (
|
|
setup_logging,
|
|
get_logger,
|
|
set_request_id,
|
|
set_session_id,
|
|
set_user_id,
|
|
clear_context,
|
|
MetricsLogger,
|
|
ContextFilter,
|
|
_StructuredFormatter,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def reset_logging():
|
|
"""Reset logging configuration and singleton state after each test."""
|
|
yield
|
|
root = logging.getLogger()
|
|
for handler in root.handlers[:]:
|
|
root.removeHandler(handler)
|
|
_logging_module._configured = False
|
|
clear_context()
|
|
|
|
|
|
@pytest.fixture
|
|
def capture_logs():
|
|
"""Capture log output to a string buffer"""
|
|
log_capture = StringIO()
|
|
handler = logging.StreamHandler(log_capture)
|
|
handler.setLevel(logging.DEBUG)
|
|
|
|
# Use simple formatter for easier testing
|
|
formatter = logging.Formatter('%(levelname)s - %(name)s - %(message)s')
|
|
handler.setFormatter(formatter)
|
|
|
|
logger = logging.getLogger()
|
|
logger.addHandler(handler)
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
yield log_capture
|
|
|
|
logger.removeHandler(handler)
|
|
|
|
|
|
def test_setup_logging_creates_logger(reset_logging):
|
|
"""Test that setup_logging creates and configures a logger"""
|
|
logger = setup_logging()
|
|
|
|
assert logger is not None
|
|
assert isinstance(logger, logging.Logger)
|
|
assert len(logger.handlers) > 0
|
|
|
|
|
|
def test_get_logger_returns_logger():
|
|
"""Test that get_logger returns a logger with the specified name"""
|
|
logger = get_logger("test_module")
|
|
|
|
assert logger is not None
|
|
assert logger.name == "test_module"
|
|
|
|
|
|
def test_context_filter_adds_request_id():
|
|
"""Test that ContextFilter adds request_id to log records"""
|
|
set_request_id("test-request-123")
|
|
|
|
record = logging.LogRecord(
|
|
name="test",
|
|
level=logging.INFO,
|
|
pathname="",
|
|
lineno=0,
|
|
msg="test message",
|
|
args=(),
|
|
exc_info=None
|
|
)
|
|
|
|
filter = ContextFilter()
|
|
filter.filter(record)
|
|
|
|
assert hasattr(record, 'request_id')
|
|
assert record.request_id == "test-request-123"
|
|
|
|
clear_context()
|
|
|
|
|
|
def test_context_filter_adds_session_id():
|
|
"""Test that ContextFilter adds session_id to log records"""
|
|
set_session_id("session-456")
|
|
|
|
record = logging.LogRecord(
|
|
name="test",
|
|
level=logging.INFO,
|
|
pathname="",
|
|
lineno=0,
|
|
msg="test message",
|
|
args=(),
|
|
exc_info=None
|
|
)
|
|
|
|
filter = ContextFilter()
|
|
filter.filter(record)
|
|
|
|
assert hasattr(record, 'session_id')
|
|
assert record.session_id == "session-456"
|
|
|
|
clear_context()
|
|
|
|
|
|
def test_context_filter_adds_user_id():
|
|
"""Test that ContextFilter adds user_id to log records"""
|
|
set_user_id("user-789")
|
|
|
|
record = logging.LogRecord(
|
|
name="test",
|
|
level=logging.INFO,
|
|
pathname="",
|
|
lineno=0,
|
|
msg="test message",
|
|
args=(),
|
|
exc_info=None
|
|
)
|
|
|
|
filter = ContextFilter()
|
|
filter.filter(record)
|
|
|
|
assert hasattr(record, 'user_id')
|
|
assert record.user_id == "user-789"
|
|
|
|
clear_context()
|
|
|
|
|
|
def test_set_and_clear_context():
|
|
"""Test setting and clearing context variables"""
|
|
set_request_id("req-1")
|
|
set_session_id("sess-1")
|
|
set_user_id("user-1")
|
|
|
|
# Create a record and apply filter
|
|
record = logging.LogRecord(
|
|
name="test",
|
|
level=logging.INFO,
|
|
pathname="",
|
|
lineno=0,
|
|
msg="test",
|
|
args=(),
|
|
exc_info=None
|
|
)
|
|
|
|
filter = ContextFilter()
|
|
filter.filter(record)
|
|
|
|
assert record.request_id == "req-1"
|
|
assert record.session_id == "sess-1"
|
|
assert record.user_id == "user-1"
|
|
|
|
# Clear context
|
|
clear_context()
|
|
|
|
# Create new record
|
|
record2 = logging.LogRecord(
|
|
name="test",
|
|
level=logging.INFO,
|
|
pathname="",
|
|
lineno=0,
|
|
msg="test",
|
|
args=(),
|
|
exc_info=None
|
|
)
|
|
|
|
filter.filter(record2)
|
|
|
|
assert record2.request_id is None
|
|
assert record2.session_id is None
|
|
assert record2.user_id is None
|
|
|
|
|
|
def test_metrics_logger_measures_time():
|
|
"""Test that MetricsLogger correctly measures elapsed time"""
|
|
import time
|
|
|
|
metrics = MetricsLogger("test_operation")
|
|
metrics.start()
|
|
|
|
time.sleep(0.1) # Sleep for 100ms
|
|
|
|
elapsed = metrics.stop()
|
|
|
|
assert elapsed >= 0.1
|
|
assert elapsed < 0.2 # Should be close to 0.1
|
|
|
|
|
|
def test_metrics_logger_logs_basic_metrics(capture_logs):
|
|
"""Test that MetricsLogger logs basic metrics"""
|
|
logger = get_logger("test")
|
|
metrics = MetricsLogger("test_op", logger=logger)
|
|
|
|
metrics.start()
|
|
metrics.stop()
|
|
metrics.log()
|
|
|
|
log_output = capture_logs.getvalue()
|
|
assert "test_op" in log_output or "Metrics" in log_output
|
|
|
|
|
|
def test_metrics_logger_logs_tokens_and_cost(capture_logs):
|
|
"""Test that MetricsLogger logs tokens and cost"""
|
|
logger = get_logger("test")
|
|
metrics = MetricsLogger("llm_call", logger=logger)
|
|
|
|
metrics.start()
|
|
metrics.stop()
|
|
metrics.log(tokens=150, cost=0.003)
|
|
|
|
log_output = capture_logs.getvalue()
|
|
assert "llm_call" in log_output or "Metrics" in log_output
|
|
|
|
|
|
def test_metrics_logger_logs_extra_metrics(capture_logs):
|
|
"""Test that MetricsLogger logs extra metrics"""
|
|
logger = get_logger("test")
|
|
metrics = MetricsLogger("custom_op", logger=logger)
|
|
|
|
metrics.start()
|
|
metrics.stop()
|
|
metrics.log(custom_metric=42, another_metric="value")
|
|
|
|
log_output = capture_logs.getvalue()
|
|
assert "custom_op" in log_output or "Metrics" in log_output
|
|
|
|
|
|
def test_metrics_logger_auto_stops_on_log():
|
|
"""Test that MetricsLogger automatically stops when logging"""
|
|
import time
|
|
|
|
metrics = MetricsLogger("test_op")
|
|
metrics.start()
|
|
|
|
# Sleep a tiny bit to ensure some time passes
|
|
time.sleep(0.001)
|
|
|
|
# Don't call stop() explicitly
|
|
assert metrics.end_time is None
|
|
|
|
metrics.log()
|
|
|
|
# Should have stopped automatically
|
|
assert metrics.end_time is not None
|
|
assert metrics.elapsed_time >= 0 # Changed from > 0 to >= 0
|
|
|
|
|
|
def test_logging_different_levels(capture_logs):
|
|
"""Test logging at different levels"""
|
|
logger = get_logger("test")
|
|
|
|
logger.debug("Debug message")
|
|
logger.info("Info message")
|
|
logger.warning("Warning message")
|
|
logger.error("Error message")
|
|
|
|
log_output = capture_logs.getvalue()
|
|
|
|
# Depending on log level, some messages might not appear
|
|
# But at least one should be there
|
|
assert len(log_output) > 0
|
|
|
|
|
|
def test_context_persists_across_multiple_logs(capture_logs):
|
|
"""Test that context persists across multiple log calls"""
|
|
set_request_id("persistent-req")
|
|
set_session_id("persistent-sess")
|
|
|
|
logger = get_logger("test")
|
|
logger.info("First log")
|
|
logger.info("Second log")
|
|
logger.info("Third log")
|
|
|
|
# Context should persist for all logs
|
|
# This is verified by the ContextFilter being applied to all records
|
|
|
|
clear_context()
|
|
|
|
|
|
@pytest.mark.parametrize("level", [
|
|
logging.DEBUG,
|
|
logging.INFO,
|
|
logging.WARNING,
|
|
logging.ERROR,
|
|
])
|
|
def test_metrics_logger_respects_log_level(level):
|
|
"""Test that MetricsLogger respects different log levels"""
|
|
logger = get_logger("test")
|
|
metrics = MetricsLogger("test_op", logger=logger)
|
|
|
|
metrics.start()
|
|
metrics.stop()
|
|
|
|
# Should not raise an error
|
|
metrics.log(level=level)
|
|
|
|
|
|
def test_metrics_logger_elapsed_time_before_stop():
|
|
"""Test that elapsed_time works even before stop() is called"""
|
|
import time
|
|
|
|
metrics = MetricsLogger("test_op")
|
|
metrics.start()
|
|
|
|
time.sleep(0.05)
|
|
|
|
# Should return current elapsed time even without stop()
|
|
elapsed = metrics.elapsed_time
|
|
assert elapsed >= 0.05
|
|
assert elapsed < 0.1
|
|
|
|
|
|
def test_metrics_logger_without_start():
|
|
"""Test that MetricsLogger handles case where start() wasn't called"""
|
|
metrics = MetricsLogger("test_op")
|
|
|
|
# Don't call start()
|
|
elapsed = metrics.elapsed_time
|
|
|
|
assert elapsed == 0.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Correlation fields — session_id, user_id, request_id must always be present
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_record(msg: str = "test") -> logging.LogRecord:
|
|
record = logging.LogRecord(
|
|
name="test", level=logging.INFO, pathname="",
|
|
lineno=0, msg=msg, args=(), exc_info=None,
|
|
)
|
|
ContextFilter().filter(record)
|
|
return record
|
|
|
|
|
|
def test_json_always_has_correlation_fields_when_context_empty():
|
|
"""JSON output must contain request_id/session_id/user_id even when None."""
|
|
clear_context()
|
|
formatter = _StructuredFormatter(use_json=True)
|
|
output = formatter.format(_make_record())
|
|
obj = json.loads(output)
|
|
|
|
assert "request_id" in obj
|
|
assert "session_id" in obj
|
|
assert "user_id" in obj
|
|
assert obj["request_id"] is None
|
|
assert obj["session_id"] is None
|
|
assert obj["user_id"] is None
|
|
|
|
|
|
def test_json_correlation_fields_carry_set_values():
|
|
"""JSON output must carry the values set via set_* helpers."""
|
|
set_request_id("req-abc")
|
|
set_session_id("sess-xyz")
|
|
set_user_id("user-123")
|
|
|
|
formatter = _StructuredFormatter(use_json=True)
|
|
output = formatter.format(_make_record())
|
|
obj = json.loads(output)
|
|
|
|
assert obj["request_id"] == "req-abc"
|
|
assert obj["session_id"] == "sess-xyz"
|
|
assert obj["user_id"] == "user-123"
|
|
|
|
clear_context()
|
|
|
|
|
|
def test_text_always_has_correlation_fields_when_context_empty():
|
|
"""Text output must contain the three correlation keys even when None."""
|
|
clear_context()
|
|
formatter = _StructuredFormatter(use_json=False)
|
|
output = formatter.format(_make_record())
|
|
|
|
assert "request_id=None" in output
|
|
assert "session_id=None" in output
|
|
assert "user_id=None" in output
|
|
|
|
|
|
def test_text_correlation_fields_carry_set_values():
|
|
"""Text output must carry the values set via set_* helpers."""
|
|
set_request_id("req-t1")
|
|
set_session_id("sess-t1")
|
|
set_user_id("user-t1")
|
|
|
|
formatter = _StructuredFormatter(use_json=False)
|
|
output = formatter.format(_make_record())
|
|
|
|
assert "request_id=req-t1" in output
|
|
assert "session_id=sess-t1" in output
|
|
assert "user_id=user-t1" in output
|
|
|
|
clear_context()
|
|
|
|
|
|
def test_json_correlation_fields_reset_after_clear_context():
|
|
"""After clear_context, correlation fields revert to None in JSON."""
|
|
set_request_id("req-temp")
|
|
set_session_id("sess-temp")
|
|
set_user_id("user-temp")
|
|
clear_context()
|
|
|
|
formatter = _StructuredFormatter(use_json=True)
|
|
output = formatter.format(_make_record())
|
|
obj = json.loads(output)
|
|
|
|
assert obj["request_id"] is None
|
|
assert obj["session_id"] is None
|
|
assert obj["user_id"] is None
|