log refactor
This commit is contained in:
@@ -1,29 +1,53 @@
|
||||
# [DEF:TestLogger:Module]
|
||||
# @COMPLEXITY: 3
|
||||
# @SEMANTICS: logging, tests, belief_state
|
||||
# @PURPOSE: Unit tests for the custom logger formatters and configuration context manager.
|
||||
# @SEMANTICS: logging, tests, belief_state, cot, json
|
||||
# @PURPOSE: Unit tests for the custom logger CoT JSON formatters and configuration context manager.
|
||||
# @LAYER: Logging (Tests)
|
||||
# @RELATION: VERIFIES -> src/core/logger.py
|
||||
# @INVARIANT: All required log statements must correctly check the threshold.
|
||||
|
||||
import pytest
|
||||
import logging
|
||||
|
||||
from src.core.logger import (
|
||||
belief_scope,
|
||||
logger,
|
||||
configure_logger,
|
||||
get_task_log_level,
|
||||
should_log_task_level
|
||||
should_log_task_level,
|
||||
CotJsonFormatter,
|
||||
)
|
||||
from src.core.config_models import LoggingConfig
|
||||
|
||||
|
||||
# [DEF:test_belief_scope_logs_entry_action_exit_at_debug:Function]
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_logger_state():
|
||||
"""Reset logger state before each test to avoid cross-test contamination."""
|
||||
config = LoggingConfig(
|
||||
level="INFO",
|
||||
task_log_level="INFO",
|
||||
enable_belief_state=True
|
||||
)
|
||||
configure_logger(config)
|
||||
# Also reset the logger level for caplog to work correctly
|
||||
logging.getLogger("superset_tools_app").setLevel(logging.DEBUG)
|
||||
yield
|
||||
# Reset after test too
|
||||
config = LoggingConfig(
|
||||
level="INFO",
|
||||
task_log_level="INFO",
|
||||
enable_belief_state=True
|
||||
)
|
||||
configure_logger(config)
|
||||
|
||||
|
||||
# [DEF:test_belief_scope_logs_reason_reflect_at_debug:Function]
|
||||
# @RELATION: BINDS_TO -> TestLogger
|
||||
# @PURPOSE: Test that belief_scope generates [ID][Entry], [ID][Action], and [ID][Exit] logs at DEBUG level.
|
||||
# @PURPOSE: Test that belief_scope generates REASON and REFLECT CoT markers at DEBUG level.
|
||||
# @PRE: belief_scope is available. caplog fixture is used. Logger configured to DEBUG.
|
||||
# @POST: Logs are verified to contain Entry, Action, and Exit tags at DEBUG level.
|
||||
def test_belief_scope_logs_entry_action_exit_at_debug(caplog):
|
||||
"""Test that belief_scope generates [ID][Entry], [ID][Action], and [ID][Exit] logs at DEBUG level."""
|
||||
# @POST: Logs are verified to contain REASON (entry) and REFLECT (coherence) markers.
|
||||
def test_belief_scope_logs_reason_reflect_at_debug(caplog):
|
||||
"""Test that belief_scope generates REASON and REFLECT CoT markers at DEBUG level."""
|
||||
# Configure logger to DEBUG level
|
||||
config = LoggingConfig(
|
||||
level="DEBUG",
|
||||
@@ -31,32 +55,43 @@ def test_belief_scope_logs_entry_action_exit_at_debug(caplog):
|
||||
enable_belief_state=True
|
||||
)
|
||||
configure_logger(config)
|
||||
|
||||
|
||||
caplog.set_level("DEBUG")
|
||||
|
||||
with belief_scope("TestFunction"):
|
||||
logger.info("Doing something important")
|
||||
|
||||
# Check that the logs contain the expected patterns
|
||||
log_messages = [record.message for record in caplog.records]
|
||||
# Check that the records contain the expected markers
|
||||
reason_records = [
|
||||
r for r in caplog.records
|
||||
if getattr(r, 'marker', None) == 'REASON' and getattr(r, 'src', None) == 'TestFunction'
|
||||
]
|
||||
reflect_records = [
|
||||
r for r in caplog.records
|
||||
if getattr(r, 'marker', None) == 'REFLECT' and getattr(r, 'src', None) == 'TestFunction'
|
||||
]
|
||||
info_records = [
|
||||
r for r in caplog.records
|
||||
if r.levelname == 'INFO' and 'Doing something important' in r.getMessage()
|
||||
]
|
||||
|
||||
assert len(reason_records) >= 1, "REASON marker not found for TestFunction"
|
||||
assert len(reflect_records) >= 1, "REFLECT marker not found for TestFunction"
|
||||
assert len(info_records) >= 1, "INFO log 'Doing something important' not found"
|
||||
|
||||
assert any("[TestFunction][Entry]" in msg for msg in log_messages), "Entry log not found"
|
||||
assert any("[TestFunction][Action] Doing something important" in msg for msg in log_messages), "Action log not found"
|
||||
assert any("[TestFunction][Exit]" in msg for msg in log_messages), "Exit log not found"
|
||||
|
||||
# Reset to INFO
|
||||
config = LoggingConfig(level="INFO", task_log_level="INFO", enable_belief_state=True)
|
||||
configure_logger(config)
|
||||
# [/DEF:test_belief_scope_logs_entry_action_exit_at_debug:Function]
|
||||
# [/DEF:test_belief_scope_logs_reason_reflect_at_debug:Function]
|
||||
|
||||
|
||||
# [DEF:test_belief_scope_error_handling:Function]
|
||||
# @RELATION: BINDS_TO -> TestLogger
|
||||
# @PURPOSE: Test that belief_scope logs Coherence:Failed on exception.
|
||||
# @PURPOSE: Test that belief_scope logs EXPLORE marker on exception.
|
||||
# @PRE: belief_scope is available. caplog fixture is used. Logger configured to DEBUG.
|
||||
# @POST: Logs are verified to contain Coherence:Failed tag.
|
||||
# @POST: Logs are verified to contain EXPLORE marker with error context.
|
||||
def test_belief_scope_error_handling(caplog):
|
||||
"""Test that belief_scope logs Coherence:Failed on exception."""
|
||||
"""Test that belief_scope logs EXPLORE marker on exception."""
|
||||
# Configure logger to DEBUG level
|
||||
config = LoggingConfig(
|
||||
level="DEBUG",
|
||||
@@ -64,19 +99,24 @@ def test_belief_scope_error_handling(caplog):
|
||||
enable_belief_state=True
|
||||
)
|
||||
configure_logger(config)
|
||||
|
||||
|
||||
caplog.set_level("DEBUG")
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
with belief_scope("FailingFunction"):
|
||||
raise ValueError("Something went wrong")
|
||||
|
||||
log_messages = [record.message for record in caplog.records]
|
||||
# Check that an EXPLORE marker was emitted
|
||||
explore_records = [
|
||||
r for r in caplog.records
|
||||
if getattr(r, 'marker', None) == 'EXPLORE'
|
||||
and getattr(r, 'src', None) == 'FailingFunction'
|
||||
]
|
||||
|
||||
assert len(explore_records) >= 1, f"EXPLORE marker not found for FailingFunction. Logs: {[(r.levelname, getattr(r, 'marker', ''), r.getMessage()) for r in caplog.records]}"
|
||||
assert 'Something went wrong' in explore_records[0].getMessage() or \
|
||||
'Something went wrong' in str(getattr(explore_records[0], 'error', ''))
|
||||
|
||||
assert any("[FailingFunction][Entry]" in msg for msg in log_messages), f"Entry log not found. Logs: {log_messages}"
|
||||
assert any("[FailingFunction][COHERENCE:FAILED]" in msg for msg in log_messages), f"Failed coherence log not found. Logs: {log_messages}"
|
||||
# Exit should not be logged on failure
|
||||
|
||||
# Reset to INFO
|
||||
config = LoggingConfig(level="INFO", task_log_level="INFO", enable_belief_state=True)
|
||||
configure_logger(config)
|
||||
@@ -85,11 +125,11 @@ def test_belief_scope_error_handling(caplog):
|
||||
|
||||
# [DEF:test_belief_scope_success_coherence:Function]
|
||||
# @RELATION: BINDS_TO -> TestLogger
|
||||
# @PURPOSE: Test that belief_scope logs Coherence:OK on success.
|
||||
# @PURPOSE: Test that belief_scope logs REFLECT marker on success.
|
||||
# @PRE: belief_scope is available. caplog fixture is used. Logger configured to DEBUG.
|
||||
# @POST: Logs are verified to contain Coherence:OK tag.
|
||||
# @POST: Logs are verified to contain REFLECT marker.
|
||||
def test_belief_scope_success_coherence(caplog):
|
||||
"""Test that belief_scope logs Coherence:OK on success."""
|
||||
"""Test that belief_scope logs REFLECT marker on success."""
|
||||
# Configure logger to DEBUG level
|
||||
config = LoggingConfig(
|
||||
level="DEBUG",
|
||||
@@ -97,43 +137,58 @@ def test_belief_scope_success_coherence(caplog):
|
||||
enable_belief_state=True
|
||||
)
|
||||
configure_logger(config)
|
||||
|
||||
|
||||
caplog.set_level("DEBUG")
|
||||
|
||||
with belief_scope("SuccessFunction"):
|
||||
pass
|
||||
|
||||
log_messages = [record.message for record in caplog.records]
|
||||
reflect_records = [
|
||||
r for r in caplog.records
|
||||
if getattr(r, 'marker', None) == 'REFLECT'
|
||||
and getattr(r, 'src', None) == 'SuccessFunction'
|
||||
]
|
||||
|
||||
assert len(reflect_records) >= 1, f"REFLECT marker not found for SuccessFunction. Logs: {[(r.levelname, getattr(r, 'marker', ''), r.getMessage()) for r in caplog.records]}"
|
||||
assert 'Coherence OK' in reflect_records[0].getMessage() or \
|
||||
getattr(reflect_records[0], 'intent', '') == 'Coherence OK'
|
||||
|
||||
|
||||
assert any("[SuccessFunction][COHERENCE:OK]" in msg for msg in log_messages), f"Success coherence log not found. Logs: {log_messages}"
|
||||
|
||||
# Reset to INFO
|
||||
config = LoggingConfig(level="INFO", task_log_level="INFO", enable_belief_state=True)
|
||||
configure_logger(config)
|
||||
# [/DEF:test_belief_scope_success_coherence:Function]
|
||||
|
||||
|
||||
# [DEF:test_belief_scope_not_visible_at_info:Function]
|
||||
# [DEF:test_belief_scope_reason_not_visible_at_info:Function]
|
||||
# @RELATION: BINDS_TO -> TestLogger
|
||||
# @PURPOSE: Test that belief_scope Entry/Exit/Coherence logs are NOT visible at INFO level.
|
||||
# @PURPOSE: Test that belief_scope REASON/REFLECT markers are NOT visible at INFO level.
|
||||
# @PRE: belief_scope is available. caplog fixture is used.
|
||||
# @POST: Entry/Exit/Coherence logs are not captured at INFO level.
|
||||
def test_belief_scope_not_visible_at_info(caplog):
|
||||
"""Test that belief_scope Entry/Exit/Coherence logs are NOT visible at INFO level."""
|
||||
# @POST: REASON/REFLECT markers are not captured at INFO level.
|
||||
def test_belief_scope_reason_not_visible_at_info(caplog):
|
||||
"""Test that belief_scope REASON/REFLECT markers are NOT visible at INFO level."""
|
||||
caplog.set_level("INFO")
|
||||
|
||||
with belief_scope("InfoLevelFunction"):
|
||||
logger.info("Doing something important")
|
||||
|
||||
log_messages = [record.message for record in caplog.records]
|
||||
# The REASON and REFLECT markers are debug-level, so they should NOT appear at INFO
|
||||
reason_records = [
|
||||
r for r in caplog.records
|
||||
if getattr(r, 'marker', None) == 'REASON' and getattr(r, 'src', None) == 'InfoLevelFunction'
|
||||
]
|
||||
reflect_records = [
|
||||
r for r in caplog.records
|
||||
if getattr(r, 'marker', None) == 'REFLECT' and getattr(r, 'src', None) == 'InfoLevelFunction'
|
||||
]
|
||||
|
||||
# Action log should be visible
|
||||
assert any("[InfoLevelFunction][Action] Doing something important" in msg for msg in log_messages), "Action log not found"
|
||||
# Entry/Exit/Coherence should NOT be visible at INFO level
|
||||
assert not any("[InfoLevelFunction][Entry]" in msg for msg in log_messages), "Entry log should not be visible at INFO"
|
||||
assert not any("[InfoLevelFunction][Exit]" in msg for msg in log_messages), "Exit log should not be visible at INFO"
|
||||
assert not any("[InfoLevelFunction][Coherence:OK]" in msg for msg in log_messages), "Coherence log should not be visible at INFO"
|
||||
# [/DEF:test_belief_scope_not_visible_at_info:Function]
|
||||
assert len(reason_records) == 0, "REASON marker should not be visible at INFO"
|
||||
assert len(reflect_records) == 0, "REFLECT marker should not be visible at INFO"
|
||||
|
||||
# But the INFO-level message should be visible
|
||||
info_records = [
|
||||
r for r in caplog.records
|
||||
if r.levelname == 'INFO' and 'Doing something important' in r.getMessage()
|
||||
]
|
||||
assert len(info_records) >= 1, "INFO log 'Doing something important' should be visible"
|
||||
# [/DEF:test_belief_scope_reason_not_visible_at_info:Function]
|
||||
|
||||
|
||||
# [DEF:test_task_log_level_default:Function]
|
||||
@@ -176,10 +231,10 @@ def test_configure_logger_task_log_level():
|
||||
enable_belief_state=True
|
||||
)
|
||||
configure_logger(config)
|
||||
|
||||
|
||||
assert get_task_log_level() == "DEBUG", "task_log_level should be DEBUG"
|
||||
assert should_log_task_level("DEBUG") is True, "DEBUG should be logged at DEBUG threshold"
|
||||
|
||||
|
||||
# Reset to INFO
|
||||
config = LoggingConfig(
|
||||
level="INFO",
|
||||
@@ -193,11 +248,11 @@ def test_configure_logger_task_log_level():
|
||||
|
||||
# [DEF:test_enable_belief_state_flag:Function]
|
||||
# @RELATION: BINDS_TO -> TestLogger
|
||||
# @PURPOSE: Test that enable_belief_state flag controls belief_scope logging.
|
||||
# @PURPOSE: Test that enable_belief_state flag controls belief_scope entry logging.
|
||||
# @PRE: LoggingConfig is available. caplog fixture is used.
|
||||
# @POST: belief_scope logs are controlled by the flag.
|
||||
# @POST: REASON entry marker suppressed when disabled; REFLECT coherence still logged.
|
||||
def test_enable_belief_state_flag(caplog):
|
||||
"""Test that enable_belief_state flag controls belief_scope logging."""
|
||||
"""Test that enable_belief_state flag controls belief_scope REASON entry logging."""
|
||||
# Disable belief state
|
||||
config = LoggingConfig(
|
||||
level="DEBUG",
|
||||
@@ -205,20 +260,26 @@ def test_enable_belief_state_flag(caplog):
|
||||
enable_belief_state=False
|
||||
)
|
||||
configure_logger(config)
|
||||
|
||||
|
||||
caplog.set_level("DEBUG")
|
||||
|
||||
|
||||
with belief_scope("DisabledFunction"):
|
||||
logger.info("Doing something")
|
||||
|
||||
log_messages = [record.message for record in caplog.records]
|
||||
|
||||
# Entry and Exit should NOT be logged when disabled
|
||||
assert not any("[DisabledFunction][Entry]" in msg for msg in log_messages), "Entry should not be logged when disabled"
|
||||
assert not any("[DisabledFunction][Exit]" in msg for msg in log_messages), "Exit should not be logged when disabled"
|
||||
# Coherence:OK should still be logged (internal tracking)
|
||||
assert any("[DisabledFunction][COHERENCE:OK]" in msg for msg in log_messages), "Coherence should still be logged"
|
||||
|
||||
|
||||
# REASON entry marker should NOT be logged when disabled
|
||||
reason_records = [
|
||||
r for r in caplog.records
|
||||
if getattr(r, 'marker', None) == 'REASON' and getattr(r, 'src', None) == 'DisabledFunction'
|
||||
]
|
||||
assert len(reason_records) == 0, "REASON entry should not be logged when disabled"
|
||||
|
||||
# REFLECT coherence marker should still be logged (internal tracking)
|
||||
reflect_records = [
|
||||
r for r in caplog.records
|
||||
if getattr(r, 'marker', None) == 'REFLECT' and getattr(r, 'src', None) == 'DisabledFunction'
|
||||
]
|
||||
assert len(reflect_records) >= 1, "REFLECT coherence should still be logged"
|
||||
|
||||
# Re-enable for other tests
|
||||
config = LoggingConfig(
|
||||
level="DEBUG",
|
||||
@@ -227,4 +288,73 @@ def test_enable_belief_state_flag(caplog):
|
||||
)
|
||||
configure_logger(config)
|
||||
# [/DEF:test_enable_belief_state_flag:Function]
|
||||
# [/DEF:TestLogger:Module]
|
||||
|
||||
# [DEF:test_cot_json_formatter_output:Function]
|
||||
# @RELATION: BINDS_TO -> TestLogger
|
||||
# @PURPOSE: Test that CotJsonFormatter produces valid JSON with expected fields.
|
||||
def test_cot_json_formatter_output():
|
||||
"""Test that CotJsonFormatter produces valid JSON with expected fields."""
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
|
||||
formatter = CotJsonFormatter()
|
||||
|
||||
# Create a LogRecord with structured extra data
|
||||
record = logging.LogRecord(
|
||||
name="test.module",
|
||||
level=logging.INFO,
|
||||
pathname="/fake/path.py",
|
||||
lineno=42,
|
||||
msg="Test action",
|
||||
args=(),
|
||||
exc_info=None,
|
||||
)
|
||||
record.marker = "REASON"
|
||||
record.intent = "Test action"
|
||||
record.src = "TestModule"
|
||||
record.payload = {"key": "value"}
|
||||
|
||||
output = formatter.format(record)
|
||||
parsed = json.loads(output)
|
||||
|
||||
assert parsed["level"] == "INFO"
|
||||
assert parsed["marker"] == "REASON"
|
||||
assert parsed["intent"] == "Test action"
|
||||
assert parsed["src"] == "TestModule"
|
||||
assert parsed["payload"] == {"key": "value"}
|
||||
assert "ts" in parsed
|
||||
assert "trace_id" in parsed
|
||||
# [/DEF:test_cot_json_formatter_output:Function]
|
||||
|
||||
# [DEF:test_cot_json_formatter_plain_message:Function]
|
||||
# @RELATION: BINDS_TO -> TestLogger
|
||||
# @PURPOSE: Test that CotJsonFormatter wraps plain messages (no extra) with default marker.
|
||||
def test_cot_json_formatter_plain_message():
|
||||
"""Test that CotJsonFormatter wraps plain messages with default marker."""
|
||||
import json
|
||||
|
||||
formatter = CotJsonFormatter()
|
||||
|
||||
# Create a LogRecord WITHOUT extra data (plain message)
|
||||
record = logging.LogRecord(
|
||||
name="test.module",
|
||||
level=logging.INFO,
|
||||
pathname="/fake/path.py",
|
||||
lineno=42,
|
||||
msg="Plain info message",
|
||||
args=(),
|
||||
exc_info=None,
|
||||
)
|
||||
|
||||
output = formatter.format(record)
|
||||
parsed = json.loads(output)
|
||||
|
||||
assert parsed["level"] == "INFO"
|
||||
assert parsed["marker"] == "REASON" # default for plain messages
|
||||
assert parsed["intent"] == "Plain info message"
|
||||
assert parsed["src"] == "test.module"
|
||||
assert "ts" in parsed
|
||||
assert "trace_id" in parsed
|
||||
# [/DEF:test_cot_json_formatter_plain_message:Function]
|
||||
|
||||
# [/DEF:TestLogger:Module]
|
||||
|
||||
Reference in New Issue
Block a user