Add logging duplication tests
- 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>
This commit is contained in:
@@ -2,6 +2,9 @@
|
|||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import contextlib
|
import contextlib
|
||||||
|
import io
|
||||||
|
import json
|
||||||
|
import logging
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from unittest.mock import AsyncMock, MagicMock, patch
|
from unittest.mock import AsyncMock, MagicMock, patch
|
||||||
|
|
||||||
@@ -22,6 +25,7 @@ from app.main import (
|
|||||||
from app.middleware.correlation import CorrelationIdMiddleware
|
from app.middleware.correlation import CorrelationIdMiddleware
|
||||||
from app.middleware.rate_limit import RateLimitMiddleware
|
from app.middleware.rate_limit import RateLimitMiddleware
|
||||||
from app.services import setup_service
|
from app.services import setup_service
|
||||||
|
from app.utils.json_formatter import JSONFormatter
|
||||||
|
|
||||||
|
|
||||||
def test_create_app_configures_cors_from_settings() -> None:
|
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)
|
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
|
# Middleware order validation
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|||||||
Reference in New Issue
Block a user