6krrt/tests/test_logs.py

"""Tests for the router's structured logging.

Written against the real handler and formatter rather than pytest's caplog,
because the thing worth pinning is what actually lands in the journal --
including the priority prefix, which caplog would never see.
"""

import io
import logging
import os

import pytest

import logs


@pytest.fixture(autouse=True)
def isolated_logger(monkeypatch):
    """Every test gets a fresh logger and a clean environment."""
    monkeypatch.delenv(logs.LEVEL_ENV, raising=False)
    monkeypatch.delenv("JOURNAL_STREAM", raising=False)
    logs.set_trace(logs.NO_TRACE)
    yield
    for handler in list(logs.log.handlers):
        logs.log.removeHandler(handler)


def _capture(level="debug", **kwargs):
    buf = io.StringIO()
    logs.configure(level, stream=buf, **kwargs)
    return buf


# --- rendering ------------------------------------------------------------

def test_a_line_is_logfmt_with_the_trace_id_first():
    buf = _capture()
    logs.set_trace("r123456")

    logs.info("route", cat="coding_general", tier=2)

    assert buf.getvalue() == "route id=r123456 cat=coding_general tier=2\n"


def test_values_with_spaces_are_quoted():
    buf = _capture()

    logs.info("filter", detail="held during peak")

    assert 'detail="held during peak"' in buf.getvalue()


def test_a_value_never_breaks_the_line():
    """One event, one line -- a newline in a detail string would forge a record."""
    buf = _capture()

    logs.info("verify", detail="line one\nline two")

    assert buf.getvalue().count("\n") == 1


def test_none_is_a_dash_and_bools_are_binary():
    buf = _capture()

    logs.info("route", pick=None, tools=True, stream=False)

    assert "pick=- tools=1 stream=0" in buf.getvalue()


def test_floats_are_readable_not_repr():
    """Microdollars and kWh, without 0.0007830000000000001."""
    buf = _capture()

    logs.info("dispatch", usd=0.000783123456, kwh=4.75e-05)

    assert "usd=0.000783123" in buf.getvalue()
    assert "kwh=4.75e-05" in buf.getvalue()


def test_an_empty_string_is_visible():
    buf = _capture()

    logs.info("classify", cat="")

    assert 'cat=""' in buf.getvalue()


# --- levels ---------------------------------------------------------------

def test_debug_is_silent_at_info_level():
    buf = _capture("info")

    logs.debug("filter", model="m")
    logs.info("route", pick="m")

    assert "filter" not in buf.getvalue()
    assert "route" in buf.getvalue()


def test_enabled_for_debug_reports_the_level():
    _capture("info")
    assert logs.enabled_for_debug() is False
    _capture("debug")
    assert logs.enabled_for_debug() is True


def test_the_env_var_overrides_the_configured_level(monkeypatch):
    """So a live service can be turned up without editing a tracked file."""
    monkeypatch.setenv(logs.LEVEL_ENV, "debug")
    buf = _capture("warning")

    logs.debug("filter", model="m")

    assert "filter" in buf.getvalue()


def test_a_nonsense_env_level_falls_back_rather_than_crashing(monkeypatch):
    monkeypatch.setenv(logs.LEVEL_ENV, "loud")

    assert logs.resolve_level("warning") == logging.WARNING


# --- journald -------------------------------------------------------------

def test_no_priority_prefix_in_a_terminal():
    """A foreground uvicorn should not print a literal <6> on every line."""
    buf = _capture(journald=False)

    logs.info("route", pick="m")

    assert buf.getvalue().startswith("route ")


def test_priority_prefix_under_journald():
    """This is what makes `journalctl -p warning` mean anything."""
    buf = _capture(journald=True)

    logs.warning("retry", verdict="malformed")
    logs.error("upstream", status=500)
    logs.info("route", pick="m")

    lines = buf.getvalue().splitlines()
    assert lines[0].startswith("<4>")   # warning
    assert lines[1].startswith("<3>")   # error
    assert lines[2].startswith("<6>")   # info


def test_journald_is_detected_by_matching_the_actual_stream(tmp_path, monkeypatch):
    """The env var alone is not enough -- it is inherited by every child.

    A foreground uvicorn started from a systemd-managed session, with output
    redirected to a file, inherited JOURNAL_STREAM and put a literal <7> on
    every line. systemd documents comparing the value against the fd.
    """
    path = tmp_path / "out.log"
    with open(path, "w") as handle:
        stat = os.fstat(handle.fileno())

        monkeypatch.setenv("JOURNAL_STREAM", f"{stat.st_dev}:{stat.st_ino}")
        assert logs.under_journald(handle) is True

        # Same variable, a different stream: this is the inherited case.
        monkeypatch.setenv("JOURNAL_STREAM", f"{stat.st_dev}:{stat.st_ino + 1}")
        assert logs.under_journald(handle) is False


def test_no_journald_variable_means_no_prefix():
    assert logs.under_journald() is False


def test_a_stream_with_no_fileno_is_not_journald():
    """StringIO under test, and anything else without an fd."""
    assert logs.under_journald(io.StringIO()) is False


def test_a_malformed_journald_variable_is_ignored(monkeypatch):
    monkeypatch.setenv("JOURNAL_STREAM", "not-a-device")
    assert logs.under_journald() is False


# --- trace ids ------------------------------------------------------------

def test_a_trace_id_is_stable_until_a_new_one_is_started():
    first = logs.new_trace()

    assert logs.current_trace() == first
    assert logs.new_trace() != first


def test_trace_ids_do_not_collide():
    assert len({logs.new_trace() for _ in range(200)}) == 200


def test_configure_is_idempotent():
    """Re-importing or re-configuring must not double every line."""
    buf = io.StringIO()
    logs.configure("info", stream=buf, journald=False)
    logs.configure("info", stream=buf, journald=False)

    logs.info("route", pick="m")

    assert buf.getvalue().count("route") == 1


def test_the_logger_does_not_propagate():
    """uvicorn configures its own logging; a root handler would double-print."""
    _capture()
    assert logs.log.propagate is False


# --- bound emitters -------------------------------------------------------

def test_a_bound_emitter_ignores_the_ambient_trace():
    """What the streaming path needs: an id that survives a context copy."""
    buf = _capture()
    bound = logs.bind("rfixed1")
    logs.set_trace("rother2")

    bound.info("dispatch", model="m")

    assert "id=rfixed1" in buf.getvalue()


def test_bind_defaults_to_the_current_trace():
    buf = _capture()
    logs.set_trace("rcurrent")

    logs.bind().warning("retry", verdict="malformed")

    assert "id=rcurrent" in buf.getvalue()


def test_a_bound_emitter_respects_the_level():
    buf = _capture("info")

    logs.bind("r1").debug("filter", model="m")

    assert buf.getvalue() == ""