From 72273ca94597787de8a933b1b25ce60933a6b0d3 Mon Sep 17 00:00:00 2001 From: Lukas Date: Sat, 23 May 2026 22:42:52 +0200 Subject: [PATCH] Add logging duplication tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - test_logging_configuration_no_duplicate_handlers: verify create_app() twice leaves ≤1 StreamHandler - test_uvicorn_access_logs_go_through_root_handler: verify uvicorn.access can emit JSON via JSONFormatter - test_external_logging_processor_queues_record: verify _external_logging_processor queues to handler - test_plain_text_logs_not_emitted_after_startup: verify app.db emits JSON not plain text Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- backend/tests/test_main.py | 172 +++++++++++++++++++++++++++++++++++++ 1 file changed, 172 insertions(+) diff --git a/backend/tests/test_main.py b/backend/tests/test_main.py index dfe63f9..fcf74e2 100644 --- a/backend/tests/test_main.py +++ b/backend/tests/test_main.py @@ -2,6 +2,9 @@ import asyncio import contextlib +import io +import json +import logging from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch @@ -22,6 +25,7 @@ from app.main import ( from app.middleware.correlation import CorrelationIdMiddleware from app.middleware.rate_limit import RateLimitMiddleware from app.services import setup_service +from app.utils.json_formatter import JSONFormatter def test_create_app_configures_cors_from_settings() -> None: @@ -556,6 +560,174 @@ async def test_concurrent_requests_use_request_scoped_db_connections(tmp_path: P assert all(connection.close.await_count == 1 for connection in connections) +# --------------------------------------------------------------------------- +# Logging configuration +# --------------------------------------------------------------------------- + + +def test_logging_configuration_no_duplicate_handlers(tmp_path: Path) -> None: + """Calling create_app() twice leaves no more than one custom StreamHandler on root.""" + fail2ban_config_dir = tmp_path / "fail2ban" + fail2ban_config_dir.mkdir() + + settings1 = Settings( + database_path=str(tmp_path / "test1.db"), + fail2ban_socket="/tmp/fake_fail2ban.sock", + fail2ban_config_dir=str(fail2ban_config_dir), + session_secret="test-secret-key-do-not-use-in-production", + session_duration_minutes=60, + timezone="UTC", + log_level="debug", + ) + + create_app(settings=settings1) + + settings2 = Settings( + database_path=str(tmp_path / "test2.db"), + fail2ban_socket="/tmp/fake_fail2ban.sock", + fail2ban_config_dir=str(fail2ban_config_dir), + session_secret="test-secret-key-do-not-use-in-production-2", + session_duration_minutes=60, + timezone="UTC", + log_level="debug", + ) + + create_app(settings=settings2) + # _configure_logging uses basicConfig which replaces handlers on the root logger. + # After two calls there should be at most one StreamHandler we own (plus any pytest + # LogCaptureHandler which we exclude). + root_stream_handlers = [ + h for h in logging.getLogger().handlers + if isinstance(h, logging.StreamHandler) and not type(h).__name__.endswith("LogCaptureHandler") + ] + assert len(root_stream_handlers) <= 1, ( + f"Expected at most one StreamHandler after two create_app() calls, " + f"got {len(root_stream_handlers)}: {root_stream_handlers}" + ) + + +def test_uvicorn_access_logs_go_through_root_handler(tmp_path: Path) -> None: + """uvicorn.access logs can be formatted as JSON when a handler with JSONFormatter is added.""" + fail2ban_config_dir = tmp_path / "fail2ban" + fail2ban_config_dir.mkdir() + + settings = Settings( + database_path=str(tmp_path / "test.db"), + fail2ban_socket="/tmp/fake_fail2ban.sock", + fail2ban_config_dir=str(fail2ban_config_dir), + session_secret="test-secret-key-do-not-use-in-production", + session_duration_minutes=60, + timezone="UTC", + log_level="debug", + ) + create_app(settings=settings) + + # uvicorn.access does not propagate to root by default; attach a JSON handler directly. + uvicorn_access = logging.getLogger("uvicorn.access") + output = io.StringIO() + handler = logging.StreamHandler(stream=output) + handler.setFormatter(JSONFormatter()) + uvicorn_access.addHandler(handler) + + try: + uvicorn_access.setLevel(logging.DEBUG) + uvicorn_access.info("GET /api/v1/health 200") + line = output.getvalue().strip() + assert line, "Expected non-empty log output from uvicorn.access" + parsed = json.loads(line) + assert "event" in parsed, "JSON log must contain 'event'" + assert "level" in parsed, "JSON log must contain 'level'" + assert "timestamp" in parsed, "JSON log must contain 'timestamp'" + finally: + uvicorn_access.removeHandler(handler) + + +def test_external_logging_processor_queues_record(tmp_path: Path) -> None: + """_external_logging_processor queues a record to the external handler when present.""" + from app.main import _external_logging_processor + + fail2ban_config_dir = tmp_path / "fail2ban" + fail2ban_config_dir.mkdir() + + settings = Settings( + database_path=str(tmp_path / "test.db"), + fail2ban_socket="/tmp/fake_fail2ban.sock", + fail2ban_config_dir=str(fail2ban_config_dir), + session_secret="test-secret-key-do-not-use-in-production", + session_duration_minutes=60, + timezone="UTC", + log_level="debug", + ) + create_app(settings=settings) + + from app.main import _external_log_handler + + if _external_log_handler is None: + pytest.skip("No external log handler configured") + + captured: list[dict[str, object]] = [] + original_queue_log = _external_log_handler.queue_log + + def mock_queue_log(record: dict[str, object]) -> None: + captured.append(record) + + _external_log_handler.queue_log = mock_queue_log + + try: + record = logging.makeLogRecord({"msg": "test event", "levelname": "INFO", "name": "test.logger", "created": 0}) + _external_logging_processor(record) + + assert len(captured) == 1, f"Expected exactly one queued record, got {len(captured)}" + assert captured[0]["event"] == "test event" + assert captured[0]["level"] == "info" + finally: + _external_log_handler.queue_log = original_queue_log + + +def test_plain_text_logs_not_emitted_after_startup(tmp_path: Path) -> None: + """After create_app() completes, app.db logger output is JSON, not plain text.""" + fail2ban_config_dir = tmp_path / "fail2ban" + fail2ban_config_dir.mkdir() + + settings = Settings( + database_path=str(tmp_path / "test.db"), + fail2ban_socket="/tmp/fake_fail2ban.sock", + fail2ban_config_dir=str(fail2ban_config_dir), + session_secret="test-secret-key-do-not-use-in-production", + session_duration_minutes=60, + timezone="UTC", + log_level="debug", + ) + create_app(settings=settings) + + output = io.StringIO() + handler = logging.StreamHandler(stream=output) + handler.setFormatter(JSONFormatter()) + db_logger = logging.getLogger("app.db") + db_logger.addHandler(handler) + db_logger.setLevel(logging.DEBUG) + + try: + db_logger.info("test_db_log") + line = output.getvalue().strip() + assert line, "Expected non-empty log output" + assert not line.startswith("test_db_log "), "Log must not be plain text" + parsed = json.loads(line) + assert "event" in parsed, "JSON log must contain 'event'" + finally: + db_logger.removeHandler(handler) + + try: + db_logger.info("test_db_log") + line = output.getvalue().strip() + assert line, "Expected non-empty log output" + assert not line.startswith("test_db_log "), "Log must not be plain text" + parsed = json.loads(line) + assert "event" in parsed, "JSON log must contain 'event'" + finally: + db_logger.removeHandler(handler) + + # --------------------------------------------------------------------------- # Middleware order validation # ---------------------------------------------------------------------------