busya
4205618ee6
Warnings fixed: - datetime.utcnow() → datetime.now(UTC) across 48+ files (src/ + tests/) - datetime.utcnow (callback ref) → lambda: datetime.now(UTC) in model fields (18 files) - Pydantic class Config → model_config = ConfigDict(...) (16 files) - Pydantic .dict() → .model_dump() (8 files) - ConfigDict(allow_population_by_field_name=True) → validate_by_name=True - SQLAlchemy declarative_base() import path updated - FastAPI on_event → lifespan context manager (app.py) - Import sorting (ruff I001) auto-fixed across all files - Fixed broken re-export chains that ruff F401 cleanup broke: _validate_bcp47: service.py now imports from dictionary_validation directly job_to_response: _job_routes.py and test imports from service_utils directly fetch_datasource_metadata: restored re-export in service.py - Added missing TranslateJobService import in _job_routes.py (was deleted by F401) - Added ConfigDict(protected_namespaces=()) for DashboardDatasetItem schema field - pytest.ini: replaced deprecated importmode with asyncio_mode All 440 tests pass with zero deprecation warnings.
358 lines
13 KiB
Python
358 lines
13 KiB
Python
# #region TestLogger [C:2] [TYPE Module]
|
|
# @SEMANTICS: logging, tests, belief_state, cot, json
|
|
# @PURPOSE: Unit tests for the custom logger CoT JSON formatters and configuration context manager.
|
|
# @LAYER Tests
|
|
# @RELATION BINDS_TO -> [LoggerModule]
|
|
# @INVARIANT: All required log statements must correctly check the threshold.
|
|
|
|
import logging
|
|
import pytest
|
|
|
|
from src.core.config_models import LoggingConfig
|
|
from src.core.logger import (
|
|
CotJsonFormatter,
|
|
belief_scope,
|
|
configure_logger,
|
|
get_task_log_level,
|
|
logger,
|
|
should_log_task_level,
|
|
)
|
|
|
|
|
|
@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)
|
|
|
|
|
|
# #region test_belief_scope_logs_reason_reflect_at_debug [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> TestLogger
|
|
# @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 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",
|
|
task_log_level="DEBUG",
|
|
enable_belief_state=True
|
|
)
|
|
configure_logger(config)
|
|
|
|
caplog.set_level("DEBUG")
|
|
|
|
with belief_scope("TestFunction"):
|
|
logger.info("Doing something important")
|
|
|
|
# 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"
|
|
|
|
# Reset to INFO
|
|
config = LoggingConfig(level="INFO", task_log_level="INFO", enable_belief_state=True)
|
|
configure_logger(config)
|
|
# #endregion test_belief_scope_logs_reason_reflect_at_debug
|
|
|
|
|
|
# #region test_belief_scope_error_handling [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> TestLogger
|
|
# @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 EXPLORE marker with error context.
|
|
def test_belief_scope_error_handling(caplog):
|
|
"""Test that belief_scope logs EXPLORE marker on exception."""
|
|
# Configure logger to DEBUG level
|
|
config = LoggingConfig(
|
|
level="DEBUG",
|
|
task_log_level="DEBUG",
|
|
enable_belief_state=True
|
|
)
|
|
configure_logger(config)
|
|
|
|
caplog.set_level("DEBUG")
|
|
|
|
with pytest.raises(ValueError), belief_scope("FailingFunction"):
|
|
raise ValueError("Something went wrong")
|
|
|
|
# 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', ''))
|
|
|
|
# Reset to INFO
|
|
config = LoggingConfig(level="INFO", task_log_level="INFO", enable_belief_state=True)
|
|
configure_logger(config)
|
|
# #endregion test_belief_scope_error_handling
|
|
|
|
|
|
# #region test_belief_scope_success_coherence [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> TestLogger
|
|
# @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 REFLECT marker.
|
|
def test_belief_scope_success_coherence(caplog):
|
|
"""Test that belief_scope logs REFLECT marker on success."""
|
|
# Configure logger to DEBUG level
|
|
config = LoggingConfig(
|
|
level="DEBUG",
|
|
task_log_level="DEBUG",
|
|
enable_belief_state=True
|
|
)
|
|
configure_logger(config)
|
|
|
|
caplog.set_level("DEBUG")
|
|
|
|
with belief_scope("SuccessFunction"):
|
|
pass
|
|
|
|
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'
|
|
|
|
|
|
# #endregion test_belief_scope_success_coherence
|
|
|
|
|
|
# #region test_belief_scope_reason_not_visible_at_info [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> TestLogger
|
|
# @PURPOSE: Test that belief_scope REASON/REFLECT markers are NOT visible at INFO level.
|
|
# @PRE: belief_scope is available. caplog fixture is used.
|
|
# @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")
|
|
|
|
# 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'
|
|
]
|
|
|
|
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"
|
|
# #endregion test_belief_scope_reason_not_visible_at_info
|
|
|
|
|
|
# #region test_task_log_level_default [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> TestLogger
|
|
# @PURPOSE: Test that default task log level is INFO.
|
|
# @PRE: None.
|
|
# @POST: Default level is INFO.
|
|
def test_task_log_level_default():
|
|
"""Test that default task log level is INFO."""
|
|
level = get_task_log_level()
|
|
assert level == "INFO"
|
|
# #endregion test_task_log_level_default
|
|
|
|
|
|
# #region test_should_log_task_level [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> TestLogger
|
|
# @PURPOSE: Test that should_log_task_level correctly filters log levels.
|
|
# @PRE: None.
|
|
# @POST: Filtering works correctly for all level combinations.
|
|
def test_should_log_task_level():
|
|
"""Test that should_log_task_level correctly filters log levels."""
|
|
# Default level is INFO
|
|
assert should_log_task_level("ERROR") is True, "ERROR should be logged at INFO threshold"
|
|
assert should_log_task_level("WARNING") is True, "WARNING should be logged at INFO threshold"
|
|
assert should_log_task_level("INFO") is True, "INFO should be logged at INFO threshold"
|
|
assert should_log_task_level("DEBUG") is False, "DEBUG should NOT be logged at INFO threshold"
|
|
# #endregion test_should_log_task_level
|
|
|
|
|
|
# #region test_configure_logger_task_log_level [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> TestLogger
|
|
# @PURPOSE: Test that configure_logger updates task_log_level.
|
|
# @PRE: LoggingConfig is available.
|
|
# @POST: task_log_level is updated correctly.
|
|
def test_configure_logger_task_log_level():
|
|
"""Test that configure_logger updates task_log_level."""
|
|
config = LoggingConfig(
|
|
level="DEBUG",
|
|
task_log_level="DEBUG",
|
|
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",
|
|
task_log_level="INFO",
|
|
enable_belief_state=True
|
|
)
|
|
configure_logger(config)
|
|
assert get_task_log_level() == "INFO", "task_log_level should be reset to INFO"
|
|
# #endregion test_configure_logger_task_log_level
|
|
|
|
|
|
# #region test_enable_belief_state_flag [C:2] [TYPE Function]
|
|
# @RELATION BINDS_TO -> TestLogger
|
|
# @PURPOSE: Test that enable_belief_state flag controls belief_scope entry logging.
|
|
# @PRE: LoggingConfig is available. caplog fixture is used.
|
|
# @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 REASON entry logging."""
|
|
# Disable belief state
|
|
config = LoggingConfig(
|
|
level="DEBUG",
|
|
task_log_level="DEBUG",
|
|
enable_belief_state=False
|
|
)
|
|
configure_logger(config)
|
|
|
|
caplog.set_level("DEBUG")
|
|
|
|
with belief_scope("DisabledFunction"):
|
|
logger.info("Doing something")
|
|
|
|
# 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",
|
|
task_log_level="DEBUG",
|
|
enable_belief_state=True
|
|
)
|
|
configure_logger(config)
|
|
# #endregion test_enable_belief_state_flag
|
|
|
|
# #region test_cot_json_formatter_output [C:2] [TYPE 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
|
|
|
|
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
|
|
# #endregion test_cot_json_formatter_output
|
|
|
|
# #region test_cot_json_formatter_plain_message [C:2] [TYPE 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
|
|
# #endregion test_cot_json_formatter_plain_message
|
|
|
|
# #endregion TestLogger
|