Implement richer, fully customizable logging; customizable timestamps (#1)
- `@kanta.logemit` handler for completely customizable logging output, with `LogEvent` structure and `kanta.tty.Line` helper to create colorized text and fixed width fields - `configure_logging(diff=False)` to disable diff display globally (supplementing per-transaction `logdiff=False`) - `transaction(extra: Any = ...)` for passing extra strings or custom metadata to logs - `@kanta.clock` to provide user controlled clock for deterministic database outputs - Added a demo script that shows basic functions, migrations, logfmt etc.
This commit is contained in:
@@ -0,0 +1,135 @@
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
import pytest
|
||||
|
||||
from .support import (
|
||||
Data,
|
||||
make_kanta,
|
||||
make_migrations_module,
|
||||
read_changes,
|
||||
read_last_snapshot,
|
||||
)
|
||||
|
||||
T0 = datetime(2026, 1, 1, 12, 0, tzinfo=UTC)
|
||||
|
||||
|
||||
def test_clock_rejects_non_callable(tmp_path, format_config):
|
||||
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
||||
|
||||
with pytest.raises(TypeError, match="must be callable"):
|
||||
kanta.clock(42)
|
||||
|
||||
|
||||
def test_clock_rejects_required_argument(tmp_path, format_config):
|
||||
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
||||
|
||||
with pytest.raises(TypeError, match="must not require arguments"):
|
||||
|
||||
@kanta.clock
|
||||
def fake_now(tz) -> datetime:
|
||||
return T0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_clock_rejects_non_datetime_result(tmp_path, format_config):
|
||||
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
||||
|
||||
@kanta.clock
|
||||
def fake_now() -> datetime:
|
||||
return "noon"
|
||||
|
||||
with pytest.raises(TypeError, match="must return a datetime"):
|
||||
await kanta.open(log=False)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_clock_controls_record_timestamps(tmp_path, format_config):
|
||||
path = tmp_path / "test.db"
|
||||
kanta = make_kanta(path, Data, format_config)
|
||||
current = T0
|
||||
|
||||
@kanta.clock
|
||||
def fake_now() -> datetime:
|
||||
return current
|
||||
|
||||
await kanta.open(log=False)
|
||||
current = T0 + timedelta(hours=1)
|
||||
with kanta.transaction(action="update") as data:
|
||||
data.counter = 1
|
||||
current = T0 + timedelta(hours=2)
|
||||
with kanta.transaction(action="repair", mtime=False) as data:
|
||||
data.counter = 2
|
||||
await kanta.close()
|
||||
|
||||
bootstrap, update, repair = read_changes(path, format_config)
|
||||
assert bootstrap.ts == T0
|
||||
assert bootstrap.m == T0
|
||||
assert update.ts == T0 + timedelta(hours=1)
|
||||
assert update.m == T0 + timedelta(hours=1)
|
||||
# System operation: stamped by the clock, but m is not updated.
|
||||
assert repair.ts == T0 + timedelta(hours=2)
|
||||
assert repair.m is None
|
||||
assert kanta.mtime == T0 + timedelta(hours=1)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_clock_not_read_without_changes(tmp_path, format_config):
|
||||
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
||||
reads = 0
|
||||
|
||||
@kanta.clock
|
||||
def fake_now() -> datetime:
|
||||
nonlocal reads
|
||||
reads += 1
|
||||
return T0
|
||||
|
||||
await kanta.open(log=False) # bootstrap record: one read
|
||||
reads = 0
|
||||
|
||||
with kanta.transaction(action="noop"):
|
||||
pass # no changes, no record, no clock read
|
||||
await kanta.close() # no snapshot written, no clock read
|
||||
|
||||
assert reads == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_clock_controls_migration_and_snapshot_timestamps(
|
||||
tmp_path, format_config
|
||||
):
|
||||
path = tmp_path / "test.db"
|
||||
kanta = make_kanta(path, Data, format_config)
|
||||
|
||||
@kanta.clock
|
||||
def fake_now() -> datetime:
|
||||
return T0
|
||||
|
||||
await kanta.open(log=False)
|
||||
await kanta.close()
|
||||
|
||||
def migrate_v1(d):
|
||||
"""Bump counter"""
|
||||
d["counter"] = 1
|
||||
|
||||
migrations = make_migrations_module("clock_migrations", "migrate_v1", migrate_v1)
|
||||
t1 = T0 + timedelta(days=1)
|
||||
kanta2 = make_kanta(path, Data, format_config, migrations=migrations)
|
||||
|
||||
@kanta2.clock
|
||||
def fake_now2() -> datetime:
|
||||
return t1
|
||||
|
||||
await kanta2.open(log=False)
|
||||
await kanta2.close()
|
||||
|
||||
migrate_records = [
|
||||
r for r in read_changes(path, format_config) if r.a.startswith("migrate:")
|
||||
]
|
||||
assert migrate_records
|
||||
assert all(r.ts == t1 for r in migrate_records)
|
||||
|
||||
snapshot = read_last_snapshot(path, format_config)
|
||||
assert snapshot is not None
|
||||
assert snapshot.ts == t1
|
||||
# mtime is carried forward from the last real modification.
|
||||
assert snapshot.m == T0
|
||||
@@ -1,4 +1,8 @@
|
||||
from kanta.logging import format_diff
|
||||
from kanta.tty import ESC, colors
|
||||
|
||||
_ADD = f"{ESC}{colors.add}m"
|
||||
_DELETE = f"{ESC}{colors.delete}m"
|
||||
|
||||
|
||||
def test_add():
|
||||
@@ -6,6 +10,28 @@ def test_add():
|
||||
assert any("name" in line for line in lines)
|
||||
|
||||
|
||||
def test_add_path_is_green():
|
||||
lines = format_diff({"name": "Alice"}, previous={})
|
||||
assert any(_ADD in line for line in lines)
|
||||
|
||||
|
||||
def test_nested_add_path_final_element_is_green():
|
||||
lines = format_diff({"users": {"alice": 1}}, previous={"users": {}})
|
||||
assert any(_ADD in line and "alice" in line for line in lines)
|
||||
|
||||
|
||||
def test_update_path_not_colored_as_add():
|
||||
lines = format_diff({"name": "Bob"}, previous={"name": "Alice"})
|
||||
assert lines
|
||||
assert all(_ADD not in line for line in lines)
|
||||
|
||||
|
||||
def test_delete_path_not_colored_as_add():
|
||||
lines = format_diff({"$delete": ["old_key"]}, previous={"old_key": 1})
|
||||
assert any(_DELETE in line for line in lines)
|
||||
assert all(_ADD not in line for line in lines)
|
||||
|
||||
|
||||
def test_update():
|
||||
lines = format_diff({"name": "Bob"}, previous={"name": "Alice"})
|
||||
assert any("Bob" in line for line in lines)
|
||||
|
||||
@@ -738,6 +738,26 @@ async def test_transaction_log_false_suppresses_log(tmp_path, format_config, cap
|
||||
assert not info_messages
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_transaction_logdiff_false_logs_header_only(
|
||||
tmp_path, format_config, caplog
|
||||
):
|
||||
path = tmp_path / "test.db"
|
||||
kanta = make_kanta(path, Data, format_config)
|
||||
await kanta.open()
|
||||
|
||||
with caplog.at_level(logging.INFO, logger="kanta.transaction"):
|
||||
with kanta.transaction(action="inc", logdiff=False) as data:
|
||||
data.counter = 1
|
||||
|
||||
await kanta.close()
|
||||
|
||||
messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
|
||||
assert len(messages) == 1
|
||||
assert "inc" in messages[0]
|
||||
assert "counter" not in messages[0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_transaction_log_custom_logger(tmp_path, format_config, caplog):
|
||||
path = tmp_path / "test.db"
|
||||
|
||||
@@ -0,0 +1,362 @@
|
||||
import logging
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
from kanta.logging import (
|
||||
LogEvent,
|
||||
bootstrap_logger,
|
||||
configure_logging,
|
||||
emit_event,
|
||||
log_change,
|
||||
migration_logger,
|
||||
transaction_logger,
|
||||
)
|
||||
from kanta.migrations import MigrationResult
|
||||
from tests.support import (
|
||||
Data,
|
||||
fixed_change,
|
||||
make_kanta,
|
||||
seed_single_change,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_kanta_loggers():
|
||||
yield
|
||||
for name in (
|
||||
"kanta",
|
||||
"kanta.transaction",
|
||||
"kanta.transaction.diff",
|
||||
"kanta.bootstrap",
|
||||
"kanta.migration",
|
||||
):
|
||||
logger = logging.getLogger(name)
|
||||
logger.setLevel(logging.NOTSET)
|
||||
logger.propagate = True
|
||||
logger.disabled = False
|
||||
logger.handlers.clear()
|
||||
|
||||
|
||||
def _change_event(**kwargs) -> LogEvent:
|
||||
return LogEvent(kind="change", logger=transaction_logger, action="update", **kwargs)
|
||||
|
||||
|
||||
def test_emit_event_falsy_return_stops_chain(capsys):
|
||||
logging.getLogger("kanta").handlers.clear()
|
||||
configure_logging()
|
||||
calls = []
|
||||
|
||||
def first(ev):
|
||||
calls.append("first")
|
||||
return None
|
||||
|
||||
def second(ev):
|
||||
calls.append("second")
|
||||
|
||||
emit_event(_change_event(), [first, second])
|
||||
assert calls == ["first"]
|
||||
assert capsys.readouterr().err == ""
|
||||
|
||||
|
||||
def test_emit_event_truthy_return_falls_back_to_default(capsys):
|
||||
logging.getLogger("kanta").handlers.clear()
|
||||
configure_logging()
|
||||
emit_event(_change_event(), [lambda ev: True])
|
||||
assert "update" in capsys.readouterr().err
|
||||
|
||||
|
||||
def test_emit_event_mutation_reaches_later_handlers_and_default(capsys):
|
||||
logging.getLogger("kanta").handlers.clear()
|
||||
configure_logging()
|
||||
calls = []
|
||||
|
||||
def first(ev):
|
||||
calls.append("first")
|
||||
ev.extra = "tgt"
|
||||
return True
|
||||
|
||||
def second(ev):
|
||||
calls.append(("second", ev.extra))
|
||||
return True
|
||||
|
||||
emit_event(_change_event(), [first, second])
|
||||
assert calls == ["first", ("second", "tgt")]
|
||||
assert "tgt" in capsys.readouterr().err
|
||||
|
||||
|
||||
def test_emit_event_handler_error_falls_back_to_default(capsys):
|
||||
logging.getLogger("kanta").handlers.clear()
|
||||
configure_logging()
|
||||
|
||||
def boom(ev):
|
||||
raise RuntimeError("broken")
|
||||
|
||||
emit_event(_change_event(), [boom])
|
||||
assert "update" in capsys.readouterr().err
|
||||
|
||||
|
||||
def test_diff_lines_built_lazily(monkeypatch):
|
||||
def _boom(*args, **kwargs):
|
||||
raise AssertionError("format_diff should not be called")
|
||||
|
||||
monkeypatch.setattr("kanta.logging.format_diff", _boom)
|
||||
ev = _change_event(diff={"counter": 1})
|
||||
emit_event(ev, [lambda ev: None]) # handled without touching the diff
|
||||
monkeypatch.undo()
|
||||
assert len(ev.diff_lines) == 1
|
||||
assert "counter" in ev.diff_lines[0]
|
||||
|
||||
|
||||
def test_default_emit_created_and_migrated(capsys):
|
||||
logging.getLogger("kanta").handlers.clear()
|
||||
configure_logging()
|
||||
emit_event(LogEvent(kind="created", logger=bootstrap_logger, filename="x.kantadb"))
|
||||
emit_event(
|
||||
LogEvent(
|
||||
kind="migrated",
|
||||
logger=migration_logger,
|
||||
filename="x.kantadb",
|
||||
from_version=0,
|
||||
to_version=1,
|
||||
migrations=["migrate_v1 (rename)"],
|
||||
)
|
||||
)
|
||||
err = capsys.readouterr().err
|
||||
assert "Created x.kantadb" in err
|
||||
assert "Migrated x.kantadb v0 -> v1: migrate_v1 (rename)" in err
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logemit_receives_transaction_events(tmp_path, format_config):
|
||||
path = tmp_path / "test.db"
|
||||
kanta = make_kanta(path, Data, format_config)
|
||||
events = []
|
||||
kanta.logemit(lambda ev: events.append(ev) or True)
|
||||
await kanta.open()
|
||||
|
||||
with kanta.transaction(action="inc", user="u1", extra="x") as data:
|
||||
data.counter = 1
|
||||
|
||||
await kanta.close()
|
||||
|
||||
change = events[-1]
|
||||
assert change.kind == "change"
|
||||
assert change.action == "inc"
|
||||
assert change.user == "u1"
|
||||
assert change.extra == "x"
|
||||
assert change.diff == {"counter": 1}
|
||||
assert change.logger.name == "kanta.transaction"
|
||||
|
||||
|
||||
def test_logemit_rejects_classes_and_async(tmp_path, format_config):
|
||||
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
||||
|
||||
class NotAFunction:
|
||||
pass
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
kanta.logemit(NotAFunction)
|
||||
|
||||
async def ahandler(ev):
|
||||
return None
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
kanta.logemit(ahandler)
|
||||
|
||||
|
||||
def _raise(*args, **kwargs):
|
||||
raise RuntimeError("formatting broken")
|
||||
|
||||
|
||||
def test_log_change_never_raises(monkeypatch):
|
||||
monkeypatch.setattr("kanta.logging.format_action_header", _raise)
|
||||
log_change("update", {"counter": 1}, previous={}) # must not raise
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logging_failure_does_not_break_transaction(
|
||||
tmp_path, format_config, monkeypatch
|
||||
):
|
||||
path = tmp_path / "test.db"
|
||||
kanta = make_kanta(path, Data, format_config)
|
||||
kanta.logemit(_raise)
|
||||
monkeypatch.setattr("kanta.logging.format_action_header", _raise)
|
||||
await kanta.open()
|
||||
|
||||
with kanta.transaction(action="inc") as data:
|
||||
data.counter = 1
|
||||
|
||||
await kanta.close()
|
||||
|
||||
kanta2 = make_kanta(path, Data, format_config)
|
||||
kanta2.logemit(_raise)
|
||||
monkeypatch.setattr("kanta.logging.format_action_header", _raise)
|
||||
await kanta2.open()
|
||||
assert kanta2.data.counter == 1
|
||||
await kanta2.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logfmt_failure_falls_back_to_default(tmp_path, format_config, caplog):
|
||||
path = tmp_path / "test.db"
|
||||
kanta = make_kanta(path, Data, format_config)
|
||||
|
||||
@kanta.logfmt
|
||||
def bad(value: str, path: str) -> str | None:
|
||||
raise RuntimeError("broken")
|
||||
|
||||
await kanta.open()
|
||||
with caplog.at_level(logging.INFO, logger="kanta.transaction"):
|
||||
with kanta.transaction(action="inc", user="alice") as data:
|
||||
data.counter = 1
|
||||
await kanta.close()
|
||||
|
||||
assert kanta.data.counter == 1
|
||||
assert "alice" in caplog.text # raw rendering used despite the failure
|
||||
assert "counter" in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logmigr_failure_does_not_break_open(tmp_path, format_config):
|
||||
path = tmp_path / "test.db"
|
||||
seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
|
||||
|
||||
mod = type(sys)("test_migrations_broken_logmigr")
|
||||
|
||||
def migrate_v1(d, kanta):
|
||||
"""Bump counter."""
|
||||
d["counter"] = 2
|
||||
|
||||
mod.__dict__["migrate_v1"] = migrate_v1
|
||||
|
||||
kanta = make_kanta(path, Data, format_config, migrations=mod)
|
||||
|
||||
@kanta.logmigr
|
||||
def bad(summary: MigrationResult) -> None:
|
||||
raise RuntimeError("broken")
|
||||
|
||||
await kanta.open()
|
||||
assert kanta.data.counter == 2
|
||||
await kanta.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_aborted_transaction_emits_event(tmp_path, format_config, caplog):
|
||||
path = tmp_path / "test.db"
|
||||
kanta = make_kanta(path, Data, format_config)
|
||||
events = []
|
||||
kanta.logemit(lambda ev: events.append(ev) or True)
|
||||
await kanta.open()
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger="kanta.transaction"):
|
||||
with pytest.raises(ValueError):
|
||||
with kanta.transaction(action="reset") as data:
|
||||
data.counter = 99
|
||||
raise ValueError("simulated failure")
|
||||
|
||||
await kanta.close()
|
||||
|
||||
aborted = events[-1]
|
||||
assert aborted.kind == "aborted"
|
||||
assert aborted.action == "reset"
|
||||
assert aborted.level == logging.WARNING
|
||||
assert isinstance(aborted.error, ValueError)
|
||||
messages = [r.getMessage() for r in caplog.records if r.levelno == logging.WARNING]
|
||||
assert any("\x1b[1;34mreset" in m for m in messages) # action color, no quotes
|
||||
assert any(" transaction aborted: simulated failure" in m for m in messages)
|
||||
assert kanta.data.counter == 0 # rolled back
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_aborted_transaction_includes_resolved_user(
|
||||
tmp_path, format_config, caplog
|
||||
):
|
||||
path = tmp_path / "test.db"
|
||||
kanta = make_kanta(path, Data, format_config)
|
||||
|
||||
@kanta.logfmt
|
||||
def resolve(value: str, path: str) -> str | None:
|
||||
return "Alice" if value == "u1" else None
|
||||
|
||||
await kanta.open()
|
||||
with caplog.at_level(logging.WARNING, logger="kanta.transaction"):
|
||||
with pytest.raises(ValueError):
|
||||
with kanta.transaction(action="reset", user="u1", extra="exp") as data:
|
||||
data.counter = 99
|
||||
raise ValueError("boom")
|
||||
await kanta.close()
|
||||
|
||||
messages = [r.getMessage() for r in caplog.records if r.levelno == logging.WARNING]
|
||||
assert any("exp" in m for m in messages)
|
||||
assert any(" by " in m and "Alice" in m for m in messages)
|
||||
assert any(" transaction aborted: boom" in m for m in messages)
|
||||
|
||||
|
||||
def test_event_header_covers_all_kinds():
|
||||
created = LogEvent(kind="created", logger=transaction_logger, filename="x.db")
|
||||
assert created.header == "Created x.db"
|
||||
|
||||
migrated = LogEvent(
|
||||
kind="migrated",
|
||||
logger=transaction_logger,
|
||||
filename="x.db",
|
||||
from_version=0,
|
||||
to_version=1,
|
||||
migrations=["migrate_v1 (rename)"],
|
||||
)
|
||||
assert migrated.header == "Migrated x.db v0 -> v1: migrate_v1 (rename)"
|
||||
|
||||
aborted = LogEvent(
|
||||
kind="aborted",
|
||||
logger=transaction_logger,
|
||||
action="reset",
|
||||
user="alice",
|
||||
error=ValueError("boom"),
|
||||
)
|
||||
assert "transaction aborted: boom" in aborted.header
|
||||
assert "alice" in aborted.header
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_event_carries_kanta_instance(tmp_path, format_config):
|
||||
path = tmp_path / "test.db"
|
||||
kanta = make_kanta(path, Data, format_config)
|
||||
events = []
|
||||
kanta.logemit(lambda ev: events.append(ev) or True)
|
||||
await kanta.open()
|
||||
with kanta.transaction(action="inc") as data:
|
||||
data.counter = 1
|
||||
await kanta.close()
|
||||
|
||||
assert events
|
||||
assert all(ev.kanta is kanta for ev in events)
|
||||
|
||||
|
||||
def test_header_is_settable_and_used_by_default_emit(capsys):
|
||||
logging.getLogger("kanta").handlers.clear()
|
||||
configure_logging()
|
||||
|
||||
def restyle(ev):
|
||||
ev.header = f"CUSTOM {ev.action}"
|
||||
return True
|
||||
|
||||
emit_event(_change_event(diff={"counter": 1}, previous={}), [restyle])
|
||||
err = capsys.readouterr().err
|
||||
assert "CUSTOM update" in err
|
||||
assert "counter" in err # default diff routing still applies
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ctx_reachable_from_event(tmp_path, format_config):
|
||||
path = tmp_path / "test.db"
|
||||
kanta = make_kanta(path, Data, format_config)
|
||||
kanta.ctx.connection_id = 7
|
||||
seen = []
|
||||
kanta.logemit(lambda ev: seen.append(ev.kanta.ctx.connection_id) or True)
|
||||
await kanta.open()
|
||||
with kanta.transaction(action="inc") as data:
|
||||
data.counter = 1
|
||||
await kanta.close()
|
||||
|
||||
assert seen and all(connection_id == 7 for connection_id in seen)
|
||||
+70
-2
@@ -2,16 +2,39 @@ import logging
|
||||
|
||||
import pytest
|
||||
|
||||
from kanta.logging import configure_logging, log_change, transaction_logger
|
||||
from kanta.logging import (
|
||||
configure_logging,
|
||||
format_action_header,
|
||||
log_change,
|
||||
)
|
||||
from kanta.tty import ESC
|
||||
|
||||
|
||||
def test_format_action_header():
|
||||
header = format_action_header("update", "alice", "tgt")
|
||||
assert header == (
|
||||
f"{ESC}1;34mupdate{ESC}0m {ESC}38;5;250mtgt{ESC}0m by {ESC}34malice{ESC}0m"
|
||||
)
|
||||
|
||||
|
||||
def test_format_action_header_action_only():
|
||||
assert format_action_header("update") == f"{ESC}1;34mupdate{ESC}0m"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_kanta_loggers():
|
||||
yield
|
||||
for name in ("kanta", "kanta.transaction", "kanta.bootstrap", "kanta.migration"):
|
||||
for name in (
|
||||
"kanta",
|
||||
"kanta.transaction",
|
||||
"kanta.transaction.diff",
|
||||
"kanta.bootstrap",
|
||||
"kanta.migration",
|
||||
):
|
||||
logger = logging.getLogger(name)
|
||||
logger.setLevel(logging.NOTSET)
|
||||
logger.propagate = True
|
||||
logger.disabled = False
|
||||
logger.handlers.clear()
|
||||
|
||||
|
||||
@@ -46,3 +69,48 @@ def test_log_change_no_diff(capsys):
|
||||
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"{ESC}38;5;250mmydb.db{ESC}0m" in captured.err
|
||||
|
||||
|
||||
def test_log_change_log_diff_false(capsys, monkeypatch):
|
||||
kanta_logger = logging.getLogger("kanta")
|
||||
kanta_logger.handlers.clear()
|
||||
configure_logging()
|
||||
|
||||
def _boom(*args, **kwargs):
|
||||
raise AssertionError("format_diff should not be called")
|
||||
|
||||
monkeypatch.setattr("kanta.logging.format_diff", _boom)
|
||||
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_configure_logging_diff_false(capsys):
|
||||
kanta_logger = logging.getLogger("kanta")
|
||||
kanta_logger.handlers.clear()
|
||||
configure_logging(diff=False)
|
||||
log_change("update", {"counter": 5}, previous={})
|
||||
captured = capsys.readouterr()
|
||||
assert "update" in captured.err
|
||||
assert "counter" not in captured.err
|
||||
|
||||
|
||||
def test_configure_logging_diff_true_reenables(capsys):
|
||||
kanta_logger = logging.getLogger("kanta")
|
||||
kanta_logger.handlers.clear()
|
||||
configure_logging(diff=False)
|
||||
configure_logging(diff=True)
|
||||
log_change("update", {"counter": 5}, previous={})
|
||||
captured = capsys.readouterr()
|
||||
assert "counter" in captured.err
|
||||
|
||||
@@ -0,0 +1,74 @@
|
||||
import pytest
|
||||
|
||||
from kanta.tty import ESC, Colors, Line, colors, displaywidth, pad, strip_ansi
|
||||
|
||||
|
||||
def test_strip_ansi():
|
||||
assert strip_ansi(f"{ESC}1;34mhello{ESC}0m") == "hello"
|
||||
|
||||
|
||||
def test_displaywidth_plain_and_ansi():
|
||||
assert displaywidth("hello") == 5
|
||||
assert displaywidth(f"{ESC}38;5;226mhi{ESC}0m") == 2
|
||||
|
||||
|
||||
def test_displaywidth_wide_and_combining_chars():
|
||||
assert displaywidth("你好") == 4
|
||||
assert displaywidth("🚀") == 2
|
||||
assert displaywidth("é") == 1
|
||||
|
||||
|
||||
def test_pad():
|
||||
assert pad("ab", 4) == "ab "
|
||||
assert pad("ab", 4, align="right") == " ab"
|
||||
assert pad("ab", 5, align="center") == " ab "
|
||||
assert pad("abcdef", 4) == "abcdef"
|
||||
assert pad("你好", 6) == "你好 "
|
||||
|
||||
|
||||
def test_line_plain_and_str_conversion():
|
||||
assert str(Line()("n=", 42)) == "n=42"
|
||||
|
||||
|
||||
def test_line_color_auto_resets_on_next_call():
|
||||
assert str(Line().user("Alice")(" by ")) == f"{ESC}34mAlice{ESC}0m by "
|
||||
|
||||
|
||||
def test_line_str_restores_active_color():
|
||||
assert str(Line().user("Alice")) == f"{ESC}34mAlice{ESC}0m"
|
||||
|
||||
|
||||
def test_line_same_color_not_reemitted():
|
||||
assert str(Line().user("a").user("b")) == f"{ESC}34mab{ESC}0m"
|
||||
|
||||
|
||||
def test_line_transition_folds_reset_into_one_sequence():
|
||||
# bold blue -> plain blue: the bold clear rides in the same sequence
|
||||
assert str(Line().action("a").user("b")) == f"{ESC}1;34ma{ESC}0;34mb{ESC}0m"
|
||||
|
||||
|
||||
def test_line_unknown_color_raises():
|
||||
with pytest.raises(AttributeError, match="unknown color"):
|
||||
Line().nosuchcolor("x")
|
||||
|
||||
|
||||
def test_line_palette_addition(monkeypatch):
|
||||
monkeypatch.setattr(colors, "session", "38;5;226", raising=False)
|
||||
assert str(Line().session("3")) == f"{ESC}38;5;226m3{ESC}0m"
|
||||
|
||||
|
||||
def test_line_palette_override_takes_effect(monkeypatch):
|
||||
monkeypatch.setattr(colors, "user", "36")
|
||||
assert str(Line().user("x")) == f"{ESC}36mx{ESC}0m"
|
||||
|
||||
|
||||
def test_line_custom_palette():
|
||||
palette = Colors()
|
||||
palette.brand = "35"
|
||||
assert str(Line(palette).brand("x")) == f"{ESC}35mx{ESC}0m"
|
||||
|
||||
|
||||
def test_line_width_and_align():
|
||||
assert str(Line()("ab", width=4)) == "ab "
|
||||
assert str(Line()("ab", width=4, align="right")) == " ab"
|
||||
assert str(Line().user("ab", width=4)) == f"{ESC}34mab {ESC}0m"
|
||||
Reference in New Issue
Block a user