checkbot/tests/test_logsetup.py

import logging
import sys

import pytest

from app import logsetup
from app.fileconfig import DEFAULT_LOG_LEVEL, ENV_LOG_LEVEL, load_file_config
from app.logsetup import SecretMaskFilter, parse_level, setup_logging

TOKEN = "123456:SECRET-token_abc"
MX_TOKEN = "syt_matrixSecretValue"


def record(msg, *args, exc_info=None):
    return logging.LogRecord("t", logging.WARNING, __file__, 1, msg, args, exc_info)


def masked(flt, msg, *args, exc_info=None):
    rec = record(msg, *args, exc_info=exc_info)
    assert flt.filter(rec) is True
    return rec


@pytest.fixture
def clean_logging():
    root = logging.getLogger()
    level = root.level
    yield
    for handler in list(root.handlers):
        if getattr(handler, logsetup.HANDLER_MARK, False):
            root.removeHandler(handler)
    root.setLevel(level)
    for name in logsetup.QUIET_LOGGERS:
        logging.getLogger(name).setLevel(logging.NOTSET)


@pytest.mark.parametrize("raw,expected", [
    ("DEBUG", logging.DEBUG), ("debug", logging.DEBUG), (" warning ", logging.WARNING),
    ("ERROR", logging.ERROR), ("INFO", logging.INFO),
])
def test_parse_level_valid(raw, expected):
    assert parse_level(raw) == (expected, True)


@pytest.mark.parametrize("raw", [None, "", "  "])
def test_parse_level_unset_is_default_without_warning(raw):
    assert parse_level(raw) == (logging.INFO, True)


@pytest.mark.parametrize("raw", ["LAUT", "15", "NOTSET", "INFO!"])
def test_parse_level_invalid_falls_back(raw):
    assert parse_level(raw) == (logging.INFO, False)


def test_file_config_reads_log_level(monkeypatch):
    monkeypatch.delenv(ENV_LOG_LEVEL, raising=False)
    assert load_file_config().log_level == DEFAULT_LOG_LEVEL
    monkeypatch.setenv(ENV_LOG_LEVEL, "debug")
    assert load_file_config().log_level == "debug"


def test_setup_sets_level_quiets_httpx_and_is_idempotent(clean_logging):
    setup_logging("DEBUG", list)
    setup_logging("DEBUG", list)
    root = logging.getLogger()
    ours = [h for h in root.handlers if getattr(h, logsetup.HANDLER_MARK, False)]
    assert len(ours) == 1
    assert root.level == logging.DEBUG
    assert logging.getLogger("httpx").level == logging.WARNING
    assert logging.getLogger("httpcore").level == logging.WARNING


def test_setup_invalid_level_warns_and_uses_info(clean_logging, caplog):
    with caplog.at_level(logging.WARNING):
        setup_logging("LAUT", list)
    assert logging.getLogger().level == logging.INFO
    assert "CHECKBOT_LOG_LEVEL" in caplog.text


def test_setup_format_has_no_timestamp(clean_logging, capsys):
    setup_logging("INFO", list)
    logging.getLogger("app.x").info("hallo")
    err = capsys.readouterr().err
    assert err.strip().endswith("INFO app.x: hallo")
    assert err.startswith("INFO ")


def test_filter_masks_known_secret_in_message_and_args():
    flt = SecretMaskFilter(lambda: [TOKEN, MX_TOKEN])
    rec = masked(flt, "fehler %s und %s", f"x {TOKEN}", MX_TOKEN)
    assert rec.getMessage() == "fehler x *** und ***"


def test_filter_masks_url_encoded_secret():
    flt = SecretMaskFilter(lambda: [TOKEN])
    rec = masked(flt, "url /x?t=123456%3ASECRET-token_abc")
    assert "SECRET" not in rec.getMessage()


def test_filter_masks_bearer_and_bot_url_without_known_secret():
    flt = SecretMaskFilter(list)
    rec = masked(flt, "GET https://api.telegram.org/bot999:ABC-def/getMe mit Authorization: Bearer abc.def-123")
    text = rec.getMessage()
    assert "999:ABC-def" not in text
    assert "abc.def-123" not in text
    assert "/bot***/getMe" in text
    assert "Bearer ***" in text


def test_filter_masks_exception_text():
    flt = SecretMaskFilter(lambda: [TOKEN])
    try:
        raise RuntimeError(f"kaputt bei /bot{TOKEN}/sendMessage")
    except RuntimeError:
        rec = masked(flt, "fehlgeschlagen", exc_info=sys.exc_info())
    formatted = logging.Formatter("%(message)s").format(rec)
    assert "SECRET-token" not in formatted
    assert "RuntimeError" in formatted


def test_filter_masks_chained_exception_text():
    flt = SecretMaskFilter(lambda: [MX_TOKEN])
    try:
        try:
            raise ValueError(f"innen {MX_TOKEN}")
        except ValueError as inner:
            raise RuntimeError("außen") from inner
    except RuntimeError:
        rec = masked(flt, "x", exc_info=sys.exc_info())
    assert MX_TOKEN not in logging.Formatter("%(message)s").format(rec)


def test_filter_survives_empty_store_and_none_values():
    flt = SecretMaskFilter(lambda: [])
    assert masked(flt, "nichts %d", 3).getMessage() == "nichts 3"
    flt = SecretMaskFilter(lambda: [None, "", "ab"])
    assert masked(flt, "ab bleibt").getMessage() == "ab bleibt"


def test_filter_survives_provider_error_and_bad_format():
    def broken():
        raise RuntimeError("store kaputt")

    flt = SecretMaskFilter(broken)
    assert masked(flt, "ok").getMessage() == "ok"
    rec = record("zu viele %s %s", "nur eins")
    assert flt.filter(rec) is True
    assert rec.getMessage()


def test_filter_keeps_last_secrets_when_provider_fails_later():
    state = {"fail": False}
    now = [0.0]

    def provider():
        if state["fail"]:
            raise RuntimeError("weg")
        return [TOKEN]

    flt = SecretMaskFilter(provider, ttl=30, clock=lambda: now[0])
    assert masked(flt, TOKEN).getMessage() == "***"
    state["fail"] = True
    now[0] = 31
    assert masked(flt, TOKEN).getMessage() == "***"


def test_filter_caches_and_refreshes_after_ttl():
    calls = []
    now = [100.0]

    def provider():
        calls.append(1)
        return [TOKEN] if len(calls) == 1 else [MX_TOKEN]

    flt = SecretMaskFilter(provider, ttl=30, clock=lambda: now[0])
    masked(flt, "a")
    masked(flt, "b")
    assert len(calls) == 1
    now[0] += 31
    rec = masked(flt, f"{MX_TOKEN} {TOKEN}")
    assert len(calls) == 2
    assert rec.getMessage() == f"*** {TOKEN}"