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