80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
|
|
from config_manager.v2.trace import TraceManager
|
|
|
|
|
|
@dataclass
|
|
class MemoryTraceTransport:
|
|
contexts: list = field(default_factory=list)
|
|
messages: list = field(default_factory=list)
|
|
|
|
def write_context(self, record) -> None:
|
|
self.contexts.append(record)
|
|
|
|
def write_message(self, record) -> None:
|
|
self.messages.append(record)
|
|
|
|
|
|
def test_bind_context_writes_context_and_returns_trace_id():
|
|
transport = MemoryTraceTransport()
|
|
trace = TraceManager(transport=transport)
|
|
|
|
trace_id = trace.bind_context(alias="email-1", type="email", attrs={"message_id": "abc"})
|
|
|
|
assert trace_id
|
|
assert transport.contexts[0].trace_id == trace_id
|
|
assert transport.contexts[0].alias == "email-1"
|
|
assert transport.contexts[0].type == "email"
|
|
assert transport.contexts[0].attrs == {"message_id": "abc"}
|
|
|
|
|
|
def test_open_context_restores_parent_after_exit():
|
|
transport = MemoryTraceTransport()
|
|
trace = TraceManager(transport=transport)
|
|
|
|
parent_id = trace.bind_context(alias="email-1", type="email")
|
|
with trace.open_context(alias="order.xlsx", type="attachment") as child_id:
|
|
assert trace.current_trace_id() == child_id
|
|
|
|
assert trace.current_trace_id() == parent_id
|
|
|
|
|
|
def test_messages_use_current_step_and_attrs():
|
|
transport = MemoryTraceTransport()
|
|
trace = TraceManager(transport=transport)
|
|
|
|
trace.bind_context(alias="email-1", type="email")
|
|
trace.step("parse_email")
|
|
trace.info("Письмо распарсено", status="completed", attrs={"attachments_count": 2})
|
|
|
|
message = transport.messages[0]
|
|
assert message.step == "parse_email"
|
|
assert message.status == "completed"
|
|
assert message.level == "INFO"
|
|
assert message.message == "Письмо распарсено"
|
|
assert message.attrs == {"attachments_count": 2}
|
|
|
|
|
|
def test_bind_context_keeps_parent_empty_by_default():
|
|
transport = MemoryTraceTransport()
|
|
trace = TraceManager(transport=transport)
|
|
|
|
parent_id = trace.bind_context(alias="email-1", type="email")
|
|
child_id = trace.bind_context(alias="order.xlsx", type="attachment")
|
|
|
|
assert child_id != parent_id
|
|
assert transport.contexts[-1].parent_id is None
|
|
|
|
|
|
def test_bind_context_uses_explicit_parent_id():
|
|
transport = MemoryTraceTransport()
|
|
trace = TraceManager(transport=transport)
|
|
|
|
parent_id = trace.bind_context(alias="email-1", type="email")
|
|
child_id = trace.bind_context(alias="order.xlsx", type="attachment", parent_id=parent_id)
|
|
|
|
assert child_id != parent_id
|
|
assert transport.contexts[-1].parent_id == parent_id
|