Skip to content

App

app

logger

Application-wide logging system for Mango Tango CLI.

Provides structured JSON logging with: - Console output (ERROR and CRITICAL levels only) to stderr - File output (INFO and above) with automatic rotation - Configurable log levels via CLI flag

ContextEnrichmentFilter

Bases: Filter

Filter that enriches log records with contextual information.

Adds: - process_id: Current process ID - thread_id: Current thread ID - app_version: Application version (if available)

Source code in app/logger.py
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class ContextEnrichmentFilter(logging.Filter):
    """
    Filter that enriches log records with contextual information.

    Adds:
    - process_id: Current process ID
    - thread_id: Current thread ID
    - app_version: Application version (if available)
    """

    def __init__(self, app_version: str = "unknown"):
        super().__init__()
        self.app_version = app_version
        self.process_id = os.getpid()

    def filter(self, record: logging.LogRecord) -> bool:
        # Add contextual information to the log record
        record.process_id = self.process_id
        record.thread_id = threading.get_ident()
        record.app_version = self.app_version
        return True

get_logger(name)

Get a logger instance for the specified module.

Parameters:

Name Type Description Default
name
str

Logger name (typically name)

required

Returns:

Type Description
Logger

Configured logger instance

Source code in app/logger.py
137
138
139
140
141
142
143
144
145
146
147
def get_logger(name: str) -> logging.Logger:
    """
    Get a logger instance for the specified module.

    Args:
        name: Logger name (typically __name__)

    Returns:
        Configured logger instance
    """
    return logging.getLogger(name)

setup_logging(log_file_path, level=logging.INFO, app_version='unknown')

Configure application-wide logging with structured JSON output.

Parameters:

Name Type Description Default
log_file_path
Path

Path to the log file

required
level
int

Minimum logging level (default: logging.INFO)

INFO
app_version
str

Application version to include in logs

'unknown'
Source code in app/logger.py
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def setup_logging(
    log_file_path: Path, level: int = logging.INFO, app_version: str = "unknown"
) -> None:
    """
    Configure application-wide logging with structured JSON output.

    Args:
        log_file_path: Path to the log file
        level: Minimum logging level (default: logging.INFO)
        app_version: Application version to include in logs
    """
    # Ensure the log directory exists
    log_file_path.parent.mkdir(parents=True, exist_ok=True)

    # Logging configuration dictionary
    config: Dict[str, Any] = {
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "json": {
                "()": "pythonjsonlogger.jsonlogger.JsonFormatter",
                "format": "%(asctime)s %(name)s %(levelname)s %(message)s %(process_id)s %(thread_id)s %(app_version)s",
                "rename_fields": {"levelname": "level", "asctime": "timestamp"},
            }
        },
        "filters": {
            "context_enrichment": {
                "()": ContextEnrichmentFilter,
                "app_version": app_version,
            }
        },
        "handlers": {
            "console": {
                "class": "logging.StreamHandler",
                "level": "ERROR",
                "formatter": "json",
                "filters": ["context_enrichment"],
                "stream": sys.stderr,
            },
            "file": {
                "class": "logging.handlers.RotatingFileHandler",
                "level": level,
                "formatter": "json",
                "filters": ["context_enrichment"],
                "filename": str(log_file_path),
                "maxBytes": 10485760,  # 10MB
                "backupCount": 5,
                "encoding": "utf-8",
            },
        },
        "root": {"level": level, "handlers": ["console", "file"]},
        "loggers": {
            # Third-party library loggers - keep them quieter by default
            "urllib3": {"level": "WARNING", "propagate": True},
            "requests": {"level": "WARNING", "propagate": True},
            "dash": {"level": "WARNING", "propagate": True},
            "plotly": {"level": "WARNING", "propagate": True},
            "shiny": {"level": "WARNING", "propagate": True},
            "uvicorn": {"level": "WARNING", "propagate": True},
            "starlette": {"level": "WARNING", "propagate": True},
            # Application loggers - inherit from root level
            "mangotango": {"level": level, "propagate": True},
            "app": {"level": level, "propagate": True},
            "analyzers": {"level": level, "propagate": True},
            "components": {"level": level, "propagate": True},
            "storage": {"level": level, "propagate": True},
            "importing": {"level": level, "propagate": True},
        },
    }

    # Apply the configuration
    logging.config.dictConfig(config)

    # Set up global exception handler
    def handle_exception(exc_type, exc_value, exc_traceback):
        """Handle uncaught exceptions by logging them."""
        if issubclass(exc_type, KeyboardInterrupt):
            # Let KeyboardInterrupt be handled normally
            sys.__excepthook__(exc_type, exc_value, exc_traceback)
            return

        logger = logging.getLogger("uncaught_exception")
        logger.critical(
            "Uncaught exception",
            exc_info=(exc_type, exc_value, exc_traceback),
            extra={
                "exception_type": exc_type.__name__,
                "exception_message": str(exc_value),
            },
        )

    # Install the global exception handler
    sys.excepthook = handle_exception

test_logger

Unit tests for the logging configuration module.

TestContextEnrichment

Test cases for context enrichment features.

Source code in app/test_logger.py
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
class TestContextEnrichment:
    """Test cases for context enrichment features."""

    def test_context_filter_adds_metadata(self):
        """Test that context filter adds process_id, thread_id, and app_version."""
        with tempfile.TemporaryDirectory() as temp_dir:
            log_file_path = Path(temp_dir) / "test.log"

            setup_logging(log_file_path, logging.INFO, "context_test_version")
            logger = get_logger("context_test")
            logger.info("Test context enrichment")

            # Force flush
            for handler in logging.getLogger().handlers:
                handler.flush()

            # Read and verify log contains enriched context
            if log_file_path.exists():
                log_content = log_file_path.read_text().strip()
                if log_content:
                    log_entry = json.loads(log_content)
                    assert "process_id" in log_entry
                    assert "thread_id" in log_entry
                    assert log_entry["app_version"] == "context_test_version"
                    assert isinstance(log_entry["process_id"], int)
                    assert isinstance(log_entry["thread_id"], int)

    def test_third_party_logger_levels(self):
        """Test that third-party loggers are set to WARNING level."""
        with tempfile.TemporaryDirectory() as temp_dir:
            log_file_path = Path(temp_dir) / "test.log"

            setup_logging(log_file_path, logging.DEBUG, "hierarchy_test")

            # Test third-party loggers
            urllib3_logger = logging.getLogger("urllib3")
            requests_logger = logging.getLogger("requests")
            dash_logger = logging.getLogger("dash")

            # They should be set to WARNING level
            assert urllib3_logger.level == logging.WARNING
            assert requests_logger.level == logging.WARNING
            assert dash_logger.level == logging.WARNING

            # Application loggers should inherit root level (DEBUG)
            app_logger = logging.getLogger("app")
            assert app_logger.level == logging.DEBUG

    def test_cli_level_controls_file_handler(self):
        """Test that CLI log level properly controls file handler level."""
        with tempfile.TemporaryDirectory() as temp_dir:
            log_file_path = Path(temp_dir) / "test.log"

            # Set up with DEBUG level
            setup_logging(log_file_path, logging.DEBUG, "cli_test")

            root_logger = logging.getLogger()
            file_handler = None

            # Find the file handler
            for handler in root_logger.handlers:
                if hasattr(handler, "baseFilename"):
                    file_handler = handler
                    break

            assert file_handler is not None
            # File handler level should match the CLI level
            assert file_handler.level == logging.DEBUG

    def test_global_exception_handler_setup(self):
        """Test that global exception handler is installed."""
        with tempfile.TemporaryDirectory() as temp_dir:
            log_file_path = Path(temp_dir) / "test.log"

            # Store original exception handler
            original_excepthook = sys.excepthook

            try:
                setup_logging(log_file_path, logging.INFO, "exception_test")

                # Exception handler should be modified
                assert sys.excepthook != original_excepthook

            finally:
                # Restore original handler
                sys.excepthook = original_excepthook
test_cli_level_controls_file_handler()

Test that CLI log level properly controls file handler level.

Source code in app/test_logger.py
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
def test_cli_level_controls_file_handler(self):
    """Test that CLI log level properly controls file handler level."""
    with tempfile.TemporaryDirectory() as temp_dir:
        log_file_path = Path(temp_dir) / "test.log"

        # Set up with DEBUG level
        setup_logging(log_file_path, logging.DEBUG, "cli_test")

        root_logger = logging.getLogger()
        file_handler = None

        # Find the file handler
        for handler in root_logger.handlers:
            if hasattr(handler, "baseFilename"):
                file_handler = handler
                break

        assert file_handler is not None
        # File handler level should match the CLI level
        assert file_handler.level == logging.DEBUG
test_context_filter_adds_metadata()

Test that context filter adds process_id, thread_id, and app_version.

Source code in app/test_logger.py
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def test_context_filter_adds_metadata(self):
    """Test that context filter adds process_id, thread_id, and app_version."""
    with tempfile.TemporaryDirectory() as temp_dir:
        log_file_path = Path(temp_dir) / "test.log"

        setup_logging(log_file_path, logging.INFO, "context_test_version")
        logger = get_logger("context_test")
        logger.info("Test context enrichment")

        # Force flush
        for handler in logging.getLogger().handlers:
            handler.flush()

        # Read and verify log contains enriched context
        if log_file_path.exists():
            log_content = log_file_path.read_text().strip()
            if log_content:
                log_entry = json.loads(log_content)
                assert "process_id" in log_entry
                assert "thread_id" in log_entry
                assert log_entry["app_version"] == "context_test_version"
                assert isinstance(log_entry["process_id"], int)
                assert isinstance(log_entry["thread_id"], int)
test_global_exception_handler_setup()

Test that global exception handler is installed.

Source code in app/test_logger.py
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
def test_global_exception_handler_setup(self):
    """Test that global exception handler is installed."""
    with tempfile.TemporaryDirectory() as temp_dir:
        log_file_path = Path(temp_dir) / "test.log"

        # Store original exception handler
        original_excepthook = sys.excepthook

        try:
            setup_logging(log_file_path, logging.INFO, "exception_test")

            # Exception handler should be modified
            assert sys.excepthook != original_excepthook

        finally:
            # Restore original handler
            sys.excepthook = original_excepthook
test_third_party_logger_levels()

Test that third-party loggers are set to WARNING level.

Source code in app/test_logger.py
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def test_third_party_logger_levels(self):
    """Test that third-party loggers are set to WARNING level."""
    with tempfile.TemporaryDirectory() as temp_dir:
        log_file_path = Path(temp_dir) / "test.log"

        setup_logging(log_file_path, logging.DEBUG, "hierarchy_test")

        # Test third-party loggers
        urllib3_logger = logging.getLogger("urllib3")
        requests_logger = logging.getLogger("requests")
        dash_logger = logging.getLogger("dash")

        # They should be set to WARNING level
        assert urllib3_logger.level == logging.WARNING
        assert requests_logger.level == logging.WARNING
        assert dash_logger.level == logging.WARNING

        # Application loggers should inherit root level (DEBUG)
        app_logger = logging.getLogger("app")
        assert app_logger.level == logging.DEBUG

TestGetLogger

Test cases for the get_logger function.

Source code in app/test_logger.py
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
class TestGetLogger:
    """Test cases for the get_logger function."""

    def test_get_logger_returns_logger_instance(self):
        """Test that get_logger returns a logging.Logger instance."""
        logger = get_logger("test")
        assert isinstance(logger, logging.Logger)

    def test_get_logger_with_different_names(self):
        """Test that get_logger returns different loggers for different names."""
        logger1 = get_logger("test1")
        logger2 = get_logger("test2")

        assert logger1.name == "test1"
        assert logger2.name == "test2"
        assert logger1 is not logger2

    def test_get_logger_with_same_name_returns_same_instance(self):
        """Test that get_logger returns the same instance for the same name."""
        logger1 = get_logger("test")
        logger2 = get_logger("test")

        assert logger1 is logger2
test_get_logger_returns_logger_instance()

Test that get_logger returns a logging.Logger instance.

Source code in app/test_logger.py
188
189
190
191
def test_get_logger_returns_logger_instance(self):
    """Test that get_logger returns a logging.Logger instance."""
    logger = get_logger("test")
    assert isinstance(logger, logging.Logger)
test_get_logger_with_different_names()

Test that get_logger returns different loggers for different names.

Source code in app/test_logger.py
193
194
195
196
197
198
199
200
def test_get_logger_with_different_names(self):
    """Test that get_logger returns different loggers for different names."""
    logger1 = get_logger("test1")
    logger2 = get_logger("test2")

    assert logger1.name == "test1"
    assert logger2.name == "test2"
    assert logger1 is not logger2
test_get_logger_with_same_name_returns_same_instance()

Test that get_logger returns the same instance for the same name.

Source code in app/test_logger.py
202
203
204
205
206
207
def test_get_logger_with_same_name_returns_same_instance(self):
    """Test that get_logger returns the same instance for the same name."""
    logger1 = get_logger("test")
    logger2 = get_logger("test")

    assert logger1 is logger2

TestIntegration

Integration tests for the logging system.

Source code in app/test_logger.py
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
class TestIntegration:
    """Integration tests for the logging system."""

    def test_full_logging_workflow(self):
        """Test the complete logging workflow."""
        with tempfile.TemporaryDirectory() as temp_dir:
            log_file_path = Path(temp_dir) / "integration_test.log"

            # Setup logging
            setup_logging(log_file_path, logging.INFO, "integration_test_version")

            # Get logger and log messages
            logger = get_logger("integration_test")
            logger.info("Integration test started")
            logger.warning("This is a warning")
            logger.error("This is an error")

            # Force flush
            for handler in logging.getLogger().handlers:
                handler.flush()

            # Verify log file exists and contains expected content
            assert log_file_path.exists()
            log_content = log_file_path.read_text()
            assert "Integration test started" in log_content
            assert "This is a warning" in log_content
            assert "This is an error" in log_content

            # Verify JSON format
            for line in log_content.strip().split("\n"):
                if line.strip():
                    log_entry = json.loads(line)
                    assert log_entry["name"] == "integration_test"
                    assert log_entry["level"] in [
                        "INFO",
                        "WARNING",
                        "ERROR",
                    ]  # renamed field
                    assert log_entry["app_version"] == "integration_test_version"
test_full_logging_workflow()

Test the complete logging workflow.

Source code in app/test_logger.py
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
def test_full_logging_workflow(self):
    """Test the complete logging workflow."""
    with tempfile.TemporaryDirectory() as temp_dir:
        log_file_path = Path(temp_dir) / "integration_test.log"

        # Setup logging
        setup_logging(log_file_path, logging.INFO, "integration_test_version")

        # Get logger and log messages
        logger = get_logger("integration_test")
        logger.info("Integration test started")
        logger.warning("This is a warning")
        logger.error("This is an error")

        # Force flush
        for handler in logging.getLogger().handlers:
            handler.flush()

        # Verify log file exists and contains expected content
        assert log_file_path.exists()
        log_content = log_file_path.read_text()
        assert "Integration test started" in log_content
        assert "This is a warning" in log_content
        assert "This is an error" in log_content

        # Verify JSON format
        for line in log_content.strip().split("\n"):
            if line.strip():
                log_entry = json.loads(line)
                assert log_entry["name"] == "integration_test"
                assert log_entry["level"] in [
                    "INFO",
                    "WARNING",
                    "ERROR",
                ]  # renamed field
                assert log_entry["app_version"] == "integration_test_version"

TestSetupLogging

Test cases for the setup_logging function.

Source code in app/test_logger.py
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class TestSetupLogging:
    """Test cases for the setup_logging function."""

    def test_setup_logging_creates_log_directory(self):
        """Test that setup_logging creates the log directory if it doesn't exist."""
        with tempfile.TemporaryDirectory() as temp_dir:
            log_file_path = Path(temp_dir) / "logs" / "test.log"

            # Directory shouldn't exist initially
            assert not log_file_path.parent.exists()

            setup_logging(log_file_path)

            # Directory should be created
            assert log_file_path.parent.exists()

    def test_setup_logging_configures_root_logger(self):
        """Test that setup_logging configures the root logger with correct level."""
        with tempfile.TemporaryDirectory() as temp_dir:
            log_file_path = Path(temp_dir) / "test.log"

            setup_logging(log_file_path, logging.DEBUG, "test_version")

            root_logger = logging.getLogger()
            assert root_logger.level == logging.DEBUG

    def test_setup_logging_configures_handlers(self):
        """Test that setup_logging configures both console and file handlers."""
        with tempfile.TemporaryDirectory() as temp_dir:
            log_file_path = Path(temp_dir) / "test.log"

            setup_logging(log_file_path, logging.INFO, "test_version")

            root_logger = logging.getLogger()

            # Should have 2 handlers (console and file)
            assert len(root_logger.handlers) >= 2

            # Find console and file handlers
            console_handler = None
            file_handler = None

            for handler in root_logger.handlers:
                if (
                    isinstance(handler, logging.StreamHandler)
                    and handler.stream == sys.stderr
                ):
                    console_handler = handler
                elif hasattr(handler, "baseFilename"):
                    file_handler = handler

            # Console handler should exist and be set to ERROR level
            assert console_handler is not None
            assert console_handler.level == logging.ERROR

            # File handler should exist and be set to INFO level
            assert file_handler is not None
            assert file_handler.level == logging.INFO

    def test_console_handler_only_shows_errors(self):
        """Test that console handler only shows ERROR and CRITICAL messages."""
        with tempfile.TemporaryDirectory() as temp_dir:
            log_file_path = Path(temp_dir) / "test.log"

            # Use StringIO to capture stderr output
            from io import StringIO

            captured_stderr = StringIO()

            # Temporarily replace sys.stderr before setting up logging
            original_stderr = sys.stderr
            sys.stderr = captured_stderr

            try:
                setup_logging(log_file_path, logging.DEBUG, "test_version")
                logger = logging.getLogger("test")

                # Log messages at different levels
                logger.debug("Debug message")
                logger.info("Info message")
                logger.warning("Warning message")
                logger.error("Error message")
                logger.critical("Critical message")

                # Force handlers to flush
                for handler in logging.getLogger().handlers:
                    handler.flush()

                # Get captured output
                stderr_output = captured_stderr.getvalue()

                # Only ERROR and CRITICAL should appear on console
                assert "Error message" in stderr_output
                assert "Critical message" in stderr_output
                assert "Debug message" not in stderr_output
                assert "Info message" not in stderr_output
                assert "Warning message" not in stderr_output

            finally:
                # Restore original stderr
                sys.stderr = original_stderr

    def test_file_handler_logs_info_and_above(self):
        """Test that file handler logs INFO and above messages when set to INFO level."""
        with tempfile.TemporaryDirectory() as temp_dir:
            log_file_path = Path(temp_dir) / "test.log"

            # Set logging to INFO level (not DEBUG) to test INFO+ filtering
            setup_logging(log_file_path, logging.INFO, "test_version")

            logger = logging.getLogger("test")

            # Log messages at different levels
            logger.debug("Debug message")
            logger.info("Info message")
            logger.warning("Warning message")
            logger.error("Error message")
            logger.critical("Critical message")

            # Force handlers to flush
            for handler in logging.getLogger().handlers:
                handler.flush()

            # Read log file content
            if log_file_path.exists():
                log_content = log_file_path.read_text()

                # Should contain INFO, WARNING, ERROR, CRITICAL but not DEBUG
                assert "Info message" in log_content
                assert "Warning message" in log_content
                assert "Error message" in log_content
                assert "Critical message" in log_content
                assert "Debug message" not in log_content

    def test_log_format_is_json(self):
        """Test that log messages are formatted as JSON."""
        with tempfile.TemporaryDirectory() as temp_dir:
            log_file_path = Path(temp_dir) / "test.log"

            setup_logging(log_file_path, logging.INFO, "test_version")

            logger = logging.getLogger("test")
            logger.info("Test JSON format")

            # Force handlers to flush
            for handler in logging.getLogger().handlers:
                handler.flush()

            # Read log file and verify JSON format
            if log_file_path.exists():
                log_content = log_file_path.read_text().strip()
                if log_content:
                    # Each line should be valid JSON
                    for line in log_content.split("\n"):
                        if line.strip():
                            try:
                                log_entry = json.loads(line)
                                assert "timestamp" in log_entry  # renamed from asctime
                                assert "name" in log_entry
                                assert "level" in log_entry  # renamed from levelname
                                assert "message" in log_entry
                                assert "process_id" in log_entry
                                assert "thread_id" in log_entry
                                assert "app_version" in log_entry
                            except json.JSONDecodeError:
                                pytest.fail(f"Log line is not valid JSON: {line}")
test_console_handler_only_shows_errors()

Test that console handler only shows ERROR and CRITICAL messages.

Source code in app/test_logger.py
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def test_console_handler_only_shows_errors(self):
    """Test that console handler only shows ERROR and CRITICAL messages."""
    with tempfile.TemporaryDirectory() as temp_dir:
        log_file_path = Path(temp_dir) / "test.log"

        # Use StringIO to capture stderr output
        from io import StringIO

        captured_stderr = StringIO()

        # Temporarily replace sys.stderr before setting up logging
        original_stderr = sys.stderr
        sys.stderr = captured_stderr

        try:
            setup_logging(log_file_path, logging.DEBUG, "test_version")
            logger = logging.getLogger("test")

            # Log messages at different levels
            logger.debug("Debug message")
            logger.info("Info message")
            logger.warning("Warning message")
            logger.error("Error message")
            logger.critical("Critical message")

            # Force handlers to flush
            for handler in logging.getLogger().handlers:
                handler.flush()

            # Get captured output
            stderr_output = captured_stderr.getvalue()

            # Only ERROR and CRITICAL should appear on console
            assert "Error message" in stderr_output
            assert "Critical message" in stderr_output
            assert "Debug message" not in stderr_output
            assert "Info message" not in stderr_output
            assert "Warning message" not in stderr_output

        finally:
            # Restore original stderr
            sys.stderr = original_stderr
test_file_handler_logs_info_and_above()

Test that file handler logs INFO and above messages when set to INFO level.

Source code in app/test_logger.py
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def test_file_handler_logs_info_and_above(self):
    """Test that file handler logs INFO and above messages when set to INFO level."""
    with tempfile.TemporaryDirectory() as temp_dir:
        log_file_path = Path(temp_dir) / "test.log"

        # Set logging to INFO level (not DEBUG) to test INFO+ filtering
        setup_logging(log_file_path, logging.INFO, "test_version")

        logger = logging.getLogger("test")

        # Log messages at different levels
        logger.debug("Debug message")
        logger.info("Info message")
        logger.warning("Warning message")
        logger.error("Error message")
        logger.critical("Critical message")

        # Force handlers to flush
        for handler in logging.getLogger().handlers:
            handler.flush()

        # Read log file content
        if log_file_path.exists():
            log_content = log_file_path.read_text()

            # Should contain INFO, WARNING, ERROR, CRITICAL but not DEBUG
            assert "Info message" in log_content
            assert "Warning message" in log_content
            assert "Error message" in log_content
            assert "Critical message" in log_content
            assert "Debug message" not in log_content
test_log_format_is_json()

Test that log messages are formatted as JSON.

Source code in app/test_logger.py
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def test_log_format_is_json(self):
    """Test that log messages are formatted as JSON."""
    with tempfile.TemporaryDirectory() as temp_dir:
        log_file_path = Path(temp_dir) / "test.log"

        setup_logging(log_file_path, logging.INFO, "test_version")

        logger = logging.getLogger("test")
        logger.info("Test JSON format")

        # Force handlers to flush
        for handler in logging.getLogger().handlers:
            handler.flush()

        # Read log file and verify JSON format
        if log_file_path.exists():
            log_content = log_file_path.read_text().strip()
            if log_content:
                # Each line should be valid JSON
                for line in log_content.split("\n"):
                    if line.strip():
                        try:
                            log_entry = json.loads(line)
                            assert "timestamp" in log_entry  # renamed from asctime
                            assert "name" in log_entry
                            assert "level" in log_entry  # renamed from levelname
                            assert "message" in log_entry
                            assert "process_id" in log_entry
                            assert "thread_id" in log_entry
                            assert "app_version" in log_entry
                        except json.JSONDecodeError:
                            pytest.fail(f"Log line is not valid JSON: {line}")
test_setup_logging_configures_handlers()

Test that setup_logging configures both console and file handlers.

Source code in app/test_logger.py
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def test_setup_logging_configures_handlers(self):
    """Test that setup_logging configures both console and file handlers."""
    with tempfile.TemporaryDirectory() as temp_dir:
        log_file_path = Path(temp_dir) / "test.log"

        setup_logging(log_file_path, logging.INFO, "test_version")

        root_logger = logging.getLogger()

        # Should have 2 handlers (console and file)
        assert len(root_logger.handlers) >= 2

        # Find console and file handlers
        console_handler = None
        file_handler = None

        for handler in root_logger.handlers:
            if (
                isinstance(handler, logging.StreamHandler)
                and handler.stream == sys.stderr
            ):
                console_handler = handler
            elif hasattr(handler, "baseFilename"):
                file_handler = handler

        # Console handler should exist and be set to ERROR level
        assert console_handler is not None
        assert console_handler.level == logging.ERROR

        # File handler should exist and be set to INFO level
        assert file_handler is not None
        assert file_handler.level == logging.INFO
test_setup_logging_configures_root_logger()

Test that setup_logging configures the root logger with correct level.

Source code in app/test_logger.py
33
34
35
36
37
38
39
40
41
def test_setup_logging_configures_root_logger(self):
    """Test that setup_logging configures the root logger with correct level."""
    with tempfile.TemporaryDirectory() as temp_dir:
        log_file_path = Path(temp_dir) / "test.log"

        setup_logging(log_file_path, logging.DEBUG, "test_version")

        root_logger = logging.getLogger()
        assert root_logger.level == logging.DEBUG
test_setup_logging_creates_log_directory()

Test that setup_logging creates the log directory if it doesn't exist.

Source code in app/test_logger.py
20
21
22
23
24
25
26
27
28
29
30
31
def test_setup_logging_creates_log_directory(self):
    """Test that setup_logging creates the log directory if it doesn't exist."""
    with tempfile.TemporaryDirectory() as temp_dir:
        log_file_path = Path(temp_dir) / "logs" / "test.log"

        # Directory shouldn't exist initially
        assert not log_file_path.parent.exists()

        setup_logging(log_file_path)

        # Directory should be created
        assert log_file_path.parent.exists()