busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: test assertions for molecular CoT markers, separate propagate=False to ConfigManager, fix ruff in env.py

Browse Source
This commit is contained in:
busya
2026-05-27 11:26:14 +03:00
parent 143ab15744
commit 55c7e323c4
6 changed files with 72 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
backend/alembic/env.py
View File

@@ -1,3 +1,12 @@
# #region AlembicEnvModule [C:3] [TYPE Module] [SEMANTICS alembic, migration, env, logging]
# @BRIEF Alembic environment configuration — sets up DB connection, model metadata,
# and logging. Contains ADR [LOG-001] for suppress_existing_loggers=False.
# @LAYER Infrastructure
# @RELATION DEPENDS_ON -> [LoggerModule]
# @INVARIANT fileConfig() must be called with disable_existing_loggers=False
# to prevent disabling superset_tools_app logger (see ADR LOG-001).
# #endregion AlembicEnvModule
from logging.config import fileConfig
import os
from pathlib import Path
@@ -38,7 +47,7 @@ if config.config_file_name is not None:
# add your model's MetaData object here
# for 'autogenerate' support
# Import ALL model modules so their tables are registered in Base.metadata
from src.models import ( # noqa: F401
from src.models import ( # noqa: F401, E402
api_key,
assistant,
auth,
@@ -57,8 +66,8 @@ from src.models import ( # noqa: F401
)
# Import config models with explicit alias to avoid name collision with alembic config
import src.models.config as models_config # noqa: F401
from src.models.mapping import Base
import src.models.config as models_config # noqa: F401, E402
from src.models.mapping import Base # noqa: E402
target_metadata = Base.metadata

10
backend/alembic/versions/2df63b7ce038_add_missing_columns_policy_id_provider_.py
View File

@@ -13,17 +13,17 @@ Revises: 86c7b1d6a710
Create Date: 2026-05-27 11:17:26.789634
"""
from typing import Sequence, Union
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = '2df63b7ce038'
down_revision: Union[str, Sequence[str], None] = '86c7b1d6a710'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
down_revision: str | Sequence[str] | None = '86c7b1d6a710'
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:

8
backend/src/core/config_manager.py
View File

@@ -54,6 +54,14 @@ class ConfigManager:
self.raw_payload: dict[str, Any] = {}
self.config: AppConfig = self._load_config()
configure_logger(self.config.settings.logging)
# Disable propagation on app loggers to prevent duplicate output
# via uvicorn's root logger handler (see ADR LOG-002 in logger.py).
# We do this caller-side so tests calling configure_logger directly
# retain propagation for caplog capture.
from .cot_logger import cot_logger # noqa: F811
from .logger import logger as _app_logger
_app_logger.propagate = False # type: ignore[assignment]
cot_logger.propagate = False
if not isinstance(self.config, AppConfig):
logger.explore(
"Config loading resulted in invalid type",

24
backend/src/core/logger.py
View File

@@ -221,23 +221,13 @@ def configure_logger(config):
cot_handler.setFormatter(CotJsonFormatter())
cot_logger.addHandler(cot_handler)
# @ADR [LOG-002] Disable logger propagation to prevent duplicate output
# @RATIONALE Uvicorn's configure_logging() registers a StreamHandler on the root
# logger via logging.config.dictConfig(). Both 'superset_tools_app' and 'cot'
# loggers have their own dedicated handlers with CotJsonFormatter. With
# propagate=True (default), each log message is handled twice:
# 1. Our CotJsonFormatter → structured JSON (desired)
# 2. Propagated to root → uvicorn's DefaultFormatter → plain text (duplicate)
# Setting propagate=False prevents the second emission.
# @REJECTED Removing uvicorn's root logger handler via logging.getLogger().handlers = []
# was rejected because it couples SS-TOOLS to uvicorn's internal logging setup,
# which could change between uvicorn versions. Setting propagate on our own
# loggers is self-contained and version-safe.
# @REJECTED Overriding log level on root logger was rejected (root.setLevel(WARNING))
# because it would suppress legitimate warnings from third-party libraries
# (SQLAlchemy, Alembic, httpx, etc.).
logger.propagate = False
cot_logger.propagate = False
# ADR [LOG-002] propagation is disabled by the production caller
# (ConfigManager) after calling configure_logger(). We do NOT set it here
# because test fixtures (caplog) rely on propagation to capture records.
# See config_manager.py:56-58 for the production-side disable.
# @REJECTED Setting propagate=False inside configure_logger was tried but
# breaks all caplog-dependent tests. Caller-side disable keeps both
# testability and production correctness.
# #endregion configure_logger

63
backend/src/core/logger/__tests__/test_logger.py
View File

@@ -14,6 +14,11 @@ from src.core.config_models import LoggingConfig
from src.core.logger import belief_scope, configure_logger, get_task_log_level, logger, should_log_task_level
def _marker(record):
"""Get molecular CoT marker from a log record (set via extra dict)."""
return getattr(record, 'marker', None)
@pytest.fixture(autouse=True)
def reset_logger_state():
"""Reset logger state before each test to avoid cross-test contamination."""
@@ -52,10 +57,11 @@ def test_belief_scope_logs_reason_reflect_at_debug(caplog):
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]
assert any("[REASON]" in msg and "TestFunction" in msg for msg in log_messages), "REASON log not found"
log_records = list(caplog.records)
log_messages = [record.message for record in log_records]
assert any(_marker(r) == "REASON" and "TestFunction" in r.getMessage() for r in log_records), "REASON log not found"
assert any("Doing something important" in msg for msg in log_messages), "Inline info log not found"
assert any("[REFLECT]" in msg and "TestFunction" in msg for msg in log_messages), "REFLECT log not found"
assert any(_marker(r) == "REFLECT" for r in log_records), "REFLECT log not found"
# Reset to INFO
config = LoggingConfig(level="INFO", task_log_level="INFO", enable_belief_state=True)
@@ -79,10 +85,12 @@ def test_belief_scope_error_handling(caplog):
caplog.set_level("DEBUG")
with pytest.raises(ValueError), belief_scope("FailingFunction"):
raise ValueError("Something went wrong")
log_messages = [record.message for record in caplog.records]
assert any("[REASON]" in msg and "FailingFunction" in msg for msg in log_messages), "REASON log not found"
assert any("[EXPLORE]" in msg and "FailingFunction" in msg and "Something went wrong" in msg
for msg in log_messages), "EXPLORE log not found"
log_records = list(caplog.records)
assert any(_marker(r) == "REASON" and "FailingFunction" in r.getMessage() for r in log_records), "REASON log not found"
assert any(
_marker(r) == "EXPLORE" and "Something went wrong" in (getattr(r, 'error', '') or '')
for r in log_records
), "EXPLORE log not found"
# REFLECT should not be logged on failure
# Reset to INFO
@@ -107,8 +115,8 @@ def test_belief_scope_success_reflect(caplog):
caplog.set_level("DEBUG")
with belief_scope("SuccessFunction"):
pass
log_messages = [record.message for record in caplog.records]
assert any("[REFLECT]" in msg and "SuccessFunction" in msg for msg in log_messages), "REFLECT log not found"
log_records = list(caplog.records)
assert any(_marker(r) == "REFLECT" for r in log_records), "REFLECT log not found"
# #endregion test_belief_scope_success_reflect
# #region test_belief_scope_not_visible_at_info [TYPE Function]
@@ -117,17 +125,24 @@ def test_belief_scope_success_reflect(caplog):
# @PRE belief_scope is available. caplog fixture is used.
# @POST REASON is visible at INFO (uses info()); REFLECT is not (uses debug()).
def test_belief_scope_not_visible_at_info(caplog):
"""Test that belief_scope REFLECT logs are NOT visible at INFO level."""
"""Test that belief_scope REFLECT logs are NOT visible at INFO level.
NOTE: Molecular CoT protocol: belief_scope uses cot_log(level='DEBUG')
for both REASON (entry) and REFLECT (exit). At INFO level, neither
should appear in caplog. The inline logger.info() IS visible because
it goes through superset_tools_app at INFO.
"""
caplog.set_level("INFO")
with belief_scope("InfoLevelFunction"):
logger.info("Doing something important")
log_messages = [record.message for record in caplog.records]
log_records = list(caplog.records)
log_messages = [record.message for record in log_records]
# Inline INFO log should be visible
assert any("Doing something important" in msg for msg in log_messages), "INFO log not found"
# REASON uses info() → visible at INFO level
assert any("[REASON]" in msg and "InfoLevelFunction" in msg for msg in log_messages), "REASON log should be visible at INFO"
# REFLECT uses debug() → NOT visible at INFO level
assert not any("[REFLECT]" in msg for msg in log_messages), "REFLECT log should not be visible at INFO"
# REASON uses DEBUG level in molecular CoT → NOT visible at INFO
assert not any(_marker(r) == "REASON" for r in log_records), "REASON is DEBUG-level, should NOT be visible at INFO"
# REFLECT uses DEBUG level → NOT visible at INFO level
assert not any(_marker(r) == "REFLECT" for r in log_records), "REFLECT is DEBUG-level, should NOT be visible at INFO"
# #endregion test_belief_scope_not_visible_at_info
# #region test_task_log_level_default [TYPE Function]
# @RELATION BINDS_TO -> [test_logger]
@@ -189,12 +204,13 @@ def test_enable_belief_state_flag(caplog):
with belief_scope("DisabledFunction"):
logger.info("Doing something")
log_messages = [record.message for record in caplog.records]
log_records = list(caplog.records)
log_messages = [record.message for record in log_records]
# "Entering DisabledFunction" (explicit logger.reason()) should NOT be logged when disabled
assert not any("Entering" in msg for msg in log_messages), "REASON entry should not be logged when disabled"
# REFLECT should still be logged (internal tracking, not gated by _enable_belief_state)
assert any("[REFLECT]" in msg for msg in log_messages), "REFLECT should still be logged"
assert any(_marker(r) == "REFLECT" for r in log_records), "REFLECT should still be logged"
# #endregion test_enable_belief_state_flag
# #region test_belief_scope_missing_anchor [TYPE Function]
# @RELATION BINDS_TO -> [test_logger]
@@ -204,10 +220,8 @@ def test_belief_scope_missing_anchor():
import pytest
from src.core.logger import belief_scope
with pytest.raises(TypeError):
# Missing required positional argument 'anchor_id'
with belief_scope():
pass
with pytest.raises(TypeError), belief_scope():
pass
# #endregion test_belief_scope_missing_anchor
# #region test_configure_logger_post_conditions [TYPE Function]
# @RELATION BINDS_TO -> [test_logger]
@@ -219,7 +233,7 @@ def test_configure_logger_post_conditions(tmp_path):
from src.core.config_models import LoggingConfig
import src.core.logger as logger_module
from src.core.logger import BeliefFormatter, configure_logger, get_task_log_level, logger
from src.core.logger import configure_logger, get_task_log_level, logger
log_file = tmp_path / "test.log"
config = LoggingConfig(
level="WARNING",
@@ -239,9 +253,10 @@ def test_configure_logger_post_conditions(tmp_path):
import pathlib
assert pathlib.Path(file_handlers[0].baseFilename) == log_file.resolve()
# 3. Formatter is set to BeliefFormatter
# 3. Formatter is set to CotJsonFormatter (replaced BeliefFormatter)
from src.core.logger import CotJsonFormatter
for handler in logger.handlers:
assert isinstance(handler.formatter, BeliefFormatter)
assert isinstance(handler.formatter, CotJsonFormatter)
# 4. Global states
assert logger_module._enable_belief_state is False

2
build.sh
View File

@@ -16,7 +16,7 @@
# help Show this help
#
# Profiles: current (default), master, enterprise-clean
# [/DEF:build:Module]
# #endregion build
set -euo pipefail