110 lines
3.3 KiB
Python
110 lines
3.3 KiB
Python
import logging
|
|
|
|
import pytest
|
|
|
|
from kanta.logging import (
|
|
_ACTION,
|
|
_RESET,
|
|
_TARGET,
|
|
_USER,
|
|
colorize_header_parts,
|
|
configure_logging,
|
|
log_change,
|
|
)
|
|
|
|
|
|
def test_colorize_header_parts():
|
|
action, user, extra = colorize_header_parts("update", "alice", "tgt")
|
|
assert action == f"{_ACTION}update{_RESET}"
|
|
assert user == f"{_USER}alice{_RESET}"
|
|
assert extra == f"{_TARGET}tgt{_RESET}"
|
|
|
|
|
|
def test_colorize_header_parts_missing_user_and_extra():
|
|
action, user, extra = colorize_header_parts("update")
|
|
assert action == f"{_ACTION}update{_RESET}"
|
|
assert user == ""
|
|
assert extra == ""
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_kanta_loggers():
|
|
yield
|
|
for name in ("kanta", "kanta.transaction", "kanta.bootstrap", "kanta.migration"):
|
|
logger = logging.getLogger(name)
|
|
logger.setLevel(logging.NOTSET)
|
|
logger.propagate = True
|
|
logger.handlers.clear()
|
|
|
|
|
|
def test_configure_logging_defaults():
|
|
kanta_logger = logging.getLogger("kanta")
|
|
configure_logging()
|
|
assert kanta_logger.level == logging.INFO
|
|
assert not kanta_logger.propagate
|
|
assert kanta_logger.handlers
|
|
|
|
|
|
def test_configure_logging_disables_specific_loggers():
|
|
configure_logging(bootstrap=False, migration=False, transaction=False)
|
|
assert not logging.getLogger("kanta.bootstrap").propagate
|
|
assert not logging.getLogger("kanta.migration").propagate
|
|
assert not logging.getLogger("kanta.transaction").propagate
|
|
|
|
|
|
def test_configure_logging_skiproot_false_leaves_kanta_propagation():
|
|
kanta_logger = logging.getLogger("kanta")
|
|
kanta_logger.handlers.clear()
|
|
configure_logging(bootstrap=False, skiproot=False)
|
|
assert kanta_logger.propagate
|
|
assert not kanta_logger.handlers
|
|
assert not logging.getLogger("kanta.bootstrap").propagate
|
|
|
|
|
|
def test_log_change_no_diff(capsys):
|
|
kanta_logger = logging.getLogger("kanta")
|
|
kanta_logger.handlers.clear()
|
|
configure_logging()
|
|
log_change("test", {})
|
|
captured = capsys.readouterr()
|
|
assert "test" in captured.err
|
|
|
|
|
|
def test_log_change_appends_extra_string(capsys):
|
|
kanta_logger = logging.getLogger("kanta")
|
|
kanta_logger.handlers.clear()
|
|
configure_logging()
|
|
log_change("export", {}, extra="mydb.db")
|
|
captured = capsys.readouterr()
|
|
assert "export" in captured.err
|
|
assert f"{_TARGET}mydb.db{_RESET}" in captured.err
|
|
|
|
|
|
def test_log_change_log_diff_false(capsys):
|
|
kanta_logger = logging.getLogger("kanta")
|
|
kanta_logger.handlers.clear()
|
|
configure_logging()
|
|
log_change("update", {"counter": 5}, previous={}, log_diff=False)
|
|
captured = capsys.readouterr()
|
|
assert "update" in captured.err
|
|
assert "counter" not in captured.err
|
|
|
|
|
|
def test_log_change_log_header_false(capsys):
|
|
kanta_logger = logging.getLogger("kanta")
|
|
kanta_logger.handlers.clear()
|
|
configure_logging()
|
|
log_change("update", {"counter": 5}, previous={}, log_header=False)
|
|
captured = capsys.readouterr()
|
|
assert "update" not in captured.err
|
|
assert "counter" in captured.err
|
|
|
|
|
|
def test_log_change_both_disabled_logs_nothing(capsys):
|
|
kanta_logger = logging.getLogger("kanta")
|
|
kanta_logger.handlers.clear()
|
|
configure_logging()
|
|
log_change("update", {"counter": 5}, previous={}, log_header=False, log_diff=False)
|
|
captured = capsys.readouterr()
|
|
assert captured.err == ""
|