Celes Renata
137 lines
4.0 KiB
Python
137 lines
4.0 KiB
Python
"""Tests for structured logging and distributed tracing."""
|
|
import json
|
|
import logging
|
|
|
|
from services.shared.logging import (
|
|
JSONFormatter,
|
|
Span,
|
|
extract_trace_context,
|
|
get_service_name,
|
|
get_span_id,
|
|
get_trace_id,
|
|
inject_trace_context,
|
|
new_span_id,
|
|
new_trace_id,
|
|
set_trace_context,
|
|
setup_logging,
|
|
)
|
|
|
|
|
|
def test_new_trace_id_format():
|
|
tid = new_trace_id()
|
|
assert len(tid) == 16
|
|
assert tid.isalnum()
|
|
|
|
|
|
def test_new_span_id_format():
|
|
sid = new_span_id()
|
|
assert len(sid) == 8
|
|
assert sid.isalnum()
|
|
|
|
|
|
def test_set_and_get_trace_context():
|
|
set_trace_context(trace_id="abc123", span_id="sp01", service="test_svc")
|
|
assert get_trace_id() == "abc123"
|
|
assert get_span_id() == "sp01"
|
|
assert get_service_name() == "test_svc"
|
|
|
|
|
|
def test_json_formatter_output():
|
|
set_trace_context(trace_id="trace42", span_id="span7", service="fmt_test")
|
|
formatter = JSONFormatter()
|
|
record = logging.LogRecord(
|
|
name="test_logger", level=logging.INFO, pathname="", lineno=0,
|
|
msg="hello world", args=(), exc_info=None,
|
|
)
|
|
output = formatter.format(record)
|
|
parsed = json.loads(output)
|
|
assert parsed["message"] == "hello world"
|
|
assert parsed["level"] == "INFO"
|
|
assert parsed["trace_id"] == "trace42"
|
|
assert parsed["span_id"] == "span7"
|
|
assert parsed["service"] == "fmt_test"
|
|
assert "timestamp" in parsed
|
|
|
|
|
|
def test_json_formatter_extra_fields():
|
|
set_trace_context(trace_id="t1", service="extra_test")
|
|
formatter = JSONFormatter()
|
|
record = logging.LogRecord(
|
|
name="test", level=logging.WARNING, pathname="", lineno=0,
|
|
msg="doc processed", args=(), exc_info=None,
|
|
)
|
|
record.ticker = "AAPL"
|
|
record.document_id = "doc-123"
|
|
output = formatter.format(record)
|
|
parsed = json.loads(output)
|
|
assert parsed["ticker"] == "AAPL"
|
|
assert parsed["document_id"] == "doc-123"
|
|
|
|
|
|
def test_span_sets_and_restores_context():
|
|
set_trace_context(trace_id="parent_trace", span_id="parent_span", service="span_test")
|
|
parent_span = get_span_id()
|
|
|
|
with Span("test_op", ticker="MSFT") as span:
|
|
assert get_trace_id() == "parent_trace"
|
|
assert get_span_id() == span.span_id
|
|
assert span.span_id != parent_span
|
|
|
|
# Context restored after span exits
|
|
assert get_span_id() == parent_span
|
|
|
|
|
|
def test_span_records_duration():
|
|
set_trace_context(service="dur_test")
|
|
with Span("slow_op") as span:
|
|
pass # instant
|
|
assert span.duration_ms >= 0
|
|
|
|
|
|
def test_span_generates_trace_id_if_missing():
|
|
set_trace_context(trace_id="", service="gen_test")
|
|
with Span("auto_trace") as span:
|
|
assert len(span.trace_id) == 16
|
|
|
|
|
|
def test_inject_trace_context():
|
|
set_trace_context(trace_id="inject_trace")
|
|
payload = inject_trace_context({"ticker": "GOOG"})
|
|
assert payload["_trace_id"] == "inject_trace"
|
|
assert payload["ticker"] == "GOOG"
|
|
|
|
|
|
def test_extract_trace_context():
|
|
payload = {"ticker": "TSLA", "_trace_id": "extracted_trace"}
|
|
extract_trace_context(payload)
|
|
assert get_trace_id() == "extracted_trace"
|
|
|
|
|
|
def test_extract_trace_context_generates_new_if_missing():
|
|
payload = {"ticker": "AMZN"}
|
|
extract_trace_context(payload)
|
|
assert len(get_trace_id()) == 16
|
|
|
|
|
|
def test_setup_logging_json_mode():
|
|
setup_logging("test_service", level="DEBUG", json_output=True)
|
|
root = logging.getLogger()
|
|
assert len(root.handlers) == 1
|
|
assert isinstance(root.handlers[0].formatter, JSONFormatter)
|
|
assert root.level == logging.DEBUG
|
|
assert get_service_name() == "test_service"
|
|
|
|
|
|
def test_setup_logging_text_mode():
|
|
setup_logging("text_service", level="WARNING", json_output=False)
|
|
root = logging.getLogger()
|
|
assert len(root.handlers) == 1
|
|
assert not isinstance(root.handlers[0].formatter, JSONFormatter)
|
|
assert root.level == logging.WARNING
|
|
|
|
|
|
def test_config_json_logs_field():
|
|
from services.shared.config import load_config
|
|
config = load_config()
|
|
assert isinstance(config.json_logs, bool)
|