busya
54a0317e98
- executors: 13 tests, 100% coverage (init/shutdown/run_blocking/run_cpu_blocking) - fileio_utils: 33 tests, 45% coverage (sanitize_filename, get_filename_from_headers, calculate_crc32, create_temp_file, remove_empty_directories, create_dashboard_export, consolidate_archive_folders) - client_registry: 14 tests, 92% coverage (get_client, get_superset_client, get_semaphore, get_auth_lock, shutdown) - cot_logger: 24 tests, 100% coverage (seed/set/get trace_id, push/pop span, structured log with markers, MarkerLogger proxy) - encryption: 12 tests, 100% coverage (key validation, encrypt/decrypt cycle) - rate_limiter: 9 tests, 100% coverage (ban logic, window pruning, per-IP isolation) - auth/logger: 10 tests, 100% coverage (_mask_details, log_security_event)
200 lines
7.5 KiB
Python
200 lines
7.5 KiB
Python
# #region Test.CotLogger [C:3] [TYPE Module] [SEMANTICS test,logging,cot,trace,span,json]
|
|
# @BRIEF Tests for core/cot_logger.py — trace_id, span, log markers, MarkerLogger.
|
|
# @RELATION BINDS_TO -> [CotLoggerModule]
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
import logging
|
|
import re
|
|
from unittest.mock import patch, MagicMock, call
|
|
import pytest
|
|
|
|
|
|
class TestSeedTraceId:
|
|
"""seed_trace_id — generates UUID and sets ContextVar."""
|
|
|
|
def test_returns_uuid_string(self):
|
|
from src.core.cot_logger import seed_trace_id
|
|
tid = seed_trace_id()
|
|
assert isinstance(tid, str)
|
|
assert len(tid) == 36 # UUID4 with hyphens
|
|
assert tid.count("-") == 4
|
|
|
|
def test_sets_trace_id(self):
|
|
from src.core.cot_logger import seed_trace_id, get_trace_id
|
|
tid = seed_trace_id()
|
|
assert get_trace_id() == tid
|
|
|
|
|
|
class TestSetTraceId:
|
|
"""set_trace_id — sets explicit trace ID."""
|
|
|
|
def test_sets_provided_id(self):
|
|
from src.core.cot_logger import set_trace_id, get_trace_id
|
|
set_trace_id("custom-trace-123")
|
|
assert get_trace_id() == "custom-trace-123"
|
|
|
|
|
|
class TestGetTraceId:
|
|
"""get_trace_id — returns current trace ID."""
|
|
|
|
def test_default_empty(self):
|
|
from src.core.cot_logger import get_trace_id
|
|
# After seed, get_trace_id returns non-empty
|
|
# In a fresh test the ContextVar may be empty
|
|
pass # ContextVar has default ""
|
|
|
|
def test_after_seed_returns_value(self):
|
|
from src.core.cot_logger import seed_trace_id, get_trace_id
|
|
tid = seed_trace_id()
|
|
assert get_trace_id() == tid
|
|
|
|
|
|
class TestPushPopSpan:
|
|
"""push_span and pop_span — span stack operations."""
|
|
|
|
def test_push_returns_previous(self):
|
|
from src.core.cot_logger import push_span, pop_span
|
|
prev1 = push_span("span-1")
|
|
# Initially empty
|
|
prev2 = push_span("span-2")
|
|
assert prev2 == "span-1"
|
|
pop_span(prev2)
|
|
# After pop, current span is back to "span-1"
|
|
|
|
def test_push_pop_restores(self):
|
|
from src.core.cot_logger import push_span
|
|
prev = push_span("outer")
|
|
inner_prev = push_span("inner")
|
|
assert inner_prev == "outer"
|
|
# pop restores outer
|
|
from src.core.cot_logger import pop_span
|
|
pop_span(inner_prev)
|
|
|
|
|
|
class TestLog:
|
|
"""log — structured JSON logging with markers."""
|
|
|
|
def test_reason_marker_info_level(self):
|
|
from src.core.cot_logger import log
|
|
with patch("src.core.cot_logger.cot_logger.info") as mock_info:
|
|
log("TestModule", "REASON", "Testing reason")
|
|
mock_info.assert_called_once()
|
|
args, kwargs = mock_info.call_args
|
|
assert "Testing reason" in args[0]
|
|
extra = kwargs.get("extra", {})
|
|
assert extra["marker"] == "REASON"
|
|
assert extra["src"] == "TestModule"
|
|
assert extra["intent"] == "Testing reason"
|
|
|
|
def test_reflect_marker_info_level(self):
|
|
from src.core.cot_logger import log
|
|
with patch("src.core.cot_logger.cot_logger.info") as mock_info:
|
|
log("Test", "REFLECT", "Reflecting")
|
|
mock_info.assert_called_once()
|
|
|
|
def test_explore_marker_warning_level(self):
|
|
from src.core.cot_logger import log
|
|
with patch("src.core.cot_logger.cot_logger.warning") as mock_warn:
|
|
log("Test", "EXPLORE", "Exploring alternatives", error="Assumption violated")
|
|
mock_warn.assert_called_once()
|
|
|
|
def test_error_level_explicit(self):
|
|
from src.core.cot_logger import log
|
|
with patch("src.core.cot_logger.cot_logger.error") as mock_err:
|
|
log("Test", "REASON", "Error test", level="ERROR")
|
|
mock_err.assert_called_once()
|
|
|
|
def test_debug_level_explicit(self):
|
|
from src.core.cot_logger import log
|
|
with patch("src.core.cot_logger.cot_logger.debug") as mock_debug:
|
|
log("Test", "REASON", "Debug test", level="DEBUG")
|
|
mock_debug.assert_called_once()
|
|
|
|
def test_payload_included(self):
|
|
from src.core.cot_logger import log
|
|
with patch("src.core.cot_logger.cot_logger.info") as mock_info:
|
|
log("Test", "REASON", "With payload", payload={"key": "value"})
|
|
extra = mock_info.call_args[1]["extra"]
|
|
assert extra["payload"] == {"key": "value"}
|
|
|
|
def test_error_included(self):
|
|
from src.core.cot_logger import log
|
|
with patch("src.core.cot_logger.cot_logger.warning") as mock_warn:
|
|
log("Test", "EXPLORE", "Error case", error="Something broke")
|
|
extra = mock_warn.call_args[1]["extra"]
|
|
assert extra["error"] == "Something broke"
|
|
|
|
def test_payload_not_included_when_none(self):
|
|
from src.core.cot_logger import log
|
|
with patch("src.core.cot_logger.cot_logger.info") as mock_info:
|
|
log("Test", "REASON", "No payload")
|
|
extra = mock_info.call_args[1]["extra"]
|
|
assert "payload" not in extra
|
|
|
|
def test_error_not_included_when_none(self):
|
|
from src.core.cot_logger import log
|
|
with patch("src.core.cot_logger.cot_logger.info") as mock_info:
|
|
log("Test", "REASON", "No error")
|
|
extra = mock_info.call_args[1]["extra"]
|
|
assert "error" not in extra
|
|
|
|
def test_warning_level_inferred_from_explore(self):
|
|
from src.core.cot_logger import log
|
|
with patch("src.core.cot_logger.cot_logger.warning") as mock_warn:
|
|
log("Test", "EXPLORE", "Auto warning")
|
|
mock_warn.assert_called_once()
|
|
|
|
|
|
class TestMarkerLogger:
|
|
"""MarkerLogger — convenience proxy for .reason(), .reflect(), .explore()."""
|
|
|
|
@pytest.fixture
|
|
def logger(self):
|
|
from src.core.cot_logger import MarkerLogger
|
|
return MarkerLogger("MyComponent")
|
|
|
|
def test_reason(self, logger):
|
|
with patch("src.core.cot_logger.log") as mock_log:
|
|
logger.reason("Doing something", payload={"action": "test"})
|
|
mock_log.assert_called_once_with("MyComponent", "REASON", "Doing something", payload={"action": "test"})
|
|
|
|
def test_reflect(self, logger):
|
|
with patch("src.core.cot_logger.log") as mock_log:
|
|
logger.reflect("Done")
|
|
mock_log.assert_called_once_with("MyComponent", "REFLECT", "Done", payload=None)
|
|
|
|
def test_explore(self, logger):
|
|
with patch("src.core.cot_logger.log") as mock_log:
|
|
logger.explore("Failed", error="DB timeout")
|
|
mock_log.assert_called_once_with("MyComponent", "EXPLORE", "Failed", payload=None, error="DB timeout")
|
|
|
|
def test_explore_with_payload_and_error(self, logger):
|
|
with patch("src.core.cot_logger.log") as mock_log:
|
|
logger.explore("Retrying", payload={"attempt": 2}, error="Timeout")
|
|
mock_log.assert_called_once_with("MyComponent", "EXPLORE", "Retrying", payload={"attempt": 2}, error="Timeout")
|
|
|
|
|
|
class TestTraceIdPropagation:
|
|
"""Trace ID propagation across async contexts."""
|
|
|
|
def test_seed_then_get(self):
|
|
from src.core.cot_logger import seed_trace_id, get_trace_id
|
|
tid = seed_trace_id()
|
|
assert get_trace_id() == tid
|
|
|
|
def test_set_then_get(self):
|
|
from src.core.cot_logger import set_trace_id, get_trace_id
|
|
set_trace_id("abc-123")
|
|
assert get_trace_id() == "abc-123"
|
|
|
|
def test_set_overrides_seed(self):
|
|
from src.core.cot_logger import seed_trace_id, set_trace_id, get_trace_id
|
|
seed_trace_id()
|
|
set_trace_id("override")
|
|
assert get_trace_id() == "override"
|
|
# #endregion Test.CotLogger
|