hold-slayer/tests/test_logging_config.py

"""
Structured-logging tests.

The two things worth guarding are the ones that are easy to break silently:
uvicorn's access logger must actually route through our formatter (it sets
`propagate = False` and brings its own handler), and its arg tuple must land as
real JSON fields rather than a pre-formatted string.
"""

import json
import logging
import logging.config

import pytest

from config import Settings
from core.logging_config import JSONFormatter, configure_logging


def _record(name="test.logger", level=logging.INFO, msg="hello", args=None, **extra):
    record = logging.LogRecord(
        name=name, level=level, pathname=__file__, lineno=1, msg=msg, args=args, exc_info=None
    )
    for key, value in extra.items():
        setattr(record, key, value)
    return record


def _emit(record):
    return json.loads(JSONFormatter().format(record))


class TestJSONFormatter:
    def test_emits_single_line_json_with_core_fields(self):
        out = JSONFormatter().format(_record())
        assert "\n" not in out
        payload = json.loads(out)
        assert payload["msg"] == "hello"
        assert payload["level"] == "INFO"
        assert payload["logger"] == "test.logger"

    def test_timestamp_is_utc_rfc3339_with_date(self):
        # logging's default asctime is local-time and date-less, which is
        # exactly what makes text logs hard to correlate in Loki.
        ts = _emit(_record())["ts"]
        assert ts.endswith("+00:00")
        assert "T" in ts

    def test_interpolates_message_args(self):
        assert _emit(_record(msg="call %s ended", args=("abc123",)))["msg"] == "call abc123 ended"

    def test_extra_fields_are_promoted(self):
        payload = _emit(_record(call_id="c-1", duration=12.5))
        assert payload["call_id"] == "c-1"
        assert payload["duration"] == 12.5

    def test_uvicorn_color_message_is_dropped(self):
        # Uvicorn ships an ANSI-coloured copy of the message via extra=; letting
        # it through puts escape codes in Loki.
        payload = _emit(
            _record(name="uvicorn.error", msg="Started", color_message="\x1b[36mStarted\x1b[0m")
        )
        assert "color_message" not in payload
        assert "\x1b" not in json.dumps(payload)

    def test_non_json_native_extra_is_stringified(self):
        payload = _emit(_record(obj=object()))
        assert isinstance(payload["obj"], str)

    def test_secretstr_extra_stays_masked(self):
        # Config secrets are SecretStr precisely so an accidental log can't leak
        # them; stringification must preserve that.
        from pydantic import SecretStr

        payload = _emit(_record(secret=SecretStr("hs_pat_supersecret")))
        assert "supersecret" not in json.dumps(payload)

    def test_exception_is_captured(self):
        try:
            raise ValueError("boom")
        except ValueError:
            import sys

            record = _record(level=logging.ERROR, msg="failed")
            record.exc_info = sys.exc_info()
        payload = _emit(record)
        assert "ValueError: boom" in payload["exc"]

    def test_thread_name_included_only_off_main_thread(self):
        # Which execution context logged a line is the first question when
        # debugging a call across the asyncio/Sippy/PJSUA2 boundary.
        on_main = _record()
        on_main.threadName = "MainThread"
        assert "thread" not in _emit(on_main)

        off_main = _record()
        off_main.threadName = "sippy-ed"
        assert _emit(off_main)["thread"] == "sippy-ed"


class TestAccessLogFields:
    def _access(self, status=200):
        return _emit(
            _record(
                name="uvicorn.access",
                msg='%s - "%s %s HTTP/%s" %d',
                args=("127.0.0.1:5050", "GET", "/api/v1/calls", "1.1", status),
            )
        )

    def test_arg_tuple_becomes_structured_fields(self):
        payload = self._access()
        assert payload["method"] == "GET"
        assert payload["path"] == "/api/v1/calls"
        assert payload["client_addr"] == "127.0.0.1:5050"
        assert payload["http_version"] == "1.1"

    def test_status_code_is_a_number_not_a_string(self):
        # So Loki can range-filter on it (status_code >= 500).
        assert self._access(503)["status_code"] == 503

    def test_access_line_is_not_pre_formatted(self):
        # The whole point: no interpolated request line to regex back apart.
        assert "msg" not in self._access()

    def test_unexpected_arg_shape_falls_back_to_message(self):
        # A logging path that raises would take out the request; degrade instead.
        payload = _emit(_record(name="uvicorn.access", msg="just a string", args=None))
        assert payload["msg"] == "just a string"


class TestConfigureLogging:
    @pytest.fixture(autouse=True)
    def _restore(self):
        root = logging.getLogger()
        saved = root.handlers[:], root.level
        yield
        root.handlers[:] = saved[0]
        root.setLevel(saved[1])

    def _uvicorn_defaults(self):
        """Reproduce what uvicorn does to its loggers at startup."""
        from uvicorn.config import LOGGING_CONFIG

        logging.config.dictConfig(LOGGING_CONFIG)

    def test_takes_over_uvicorn_handlers(self):
        self._uvicorn_defaults()
        access = logging.getLogger("uvicorn.access")
        assert access.propagate is False  # uvicorn's default, the problem

        configure_logging("json", "info")
        assert access.propagate is True
        assert access.handlers == []

    def test_json_mode_installs_json_formatter(self):
        configure_logging("json", "info")
        assert isinstance(logging.getLogger().handlers[0].formatter, JSONFormatter)

    def test_text_mode_does_not(self):
        configure_logging("text", "info")
        assert not isinstance(logging.getLogger().handlers[0].formatter, JSONFormatter)

    def test_is_idempotent(self):
        # Called at import and again in lifespan; a second call must replace the
        # handler, not add one, or every line is emitted twice.
        configure_logging("json", "info")
        configure_logging("json", "info")
        assert len(logging.getLogger().handlers) == 1

    def test_respects_log_level(self):
        configure_logging("json", "warning")
        assert logging.getLogger().level == logging.WARNING


class TestSettings:
    def test_defaults_to_text(self):
        # A JSON-only default would make local development worse.
        assert Settings(database_url="sqlite+aiosqlite:///:memory:").log_format == "text"

    def test_reads_log_format_env(self, monkeypatch):
        monkeypatch.setenv("LOG_FORMAT", "json")
        assert Settings(database_url="sqlite+aiosqlite:///:memory:").log_format == "json"