252 lines
7.4 KiB
Python
252 lines
7.4 KiB
Python
import logging
|
|
|
|
import pytest
|
|
|
|
from kanta import Kanta
|
|
from kanta.callbacks import DictPost, DictPre
|
|
|
|
from .support import Data, make_kanta, read_changes
|
|
|
|
|
|
def test_logheader_rejects_async(tmp_path, format_config):
|
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
|
|
|
with pytest.raises(TypeError, match="must not be async"):
|
|
|
|
@kanta.logheader
|
|
async def header(action: str) -> str:
|
|
return action
|
|
|
|
|
|
def test_logheader_rejects_bad_return_annotation(tmp_path, format_config):
|
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
|
|
|
with pytest.raises(TypeError, match="must return"):
|
|
|
|
@kanta.logheader
|
|
def header(action: str) -> int:
|
|
return 1
|
|
|
|
|
|
def test_logheader_rejects_unknown_annotation(tmp_path, format_config):
|
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
|
|
|
with pytest.raises(TypeError, match="unsupported annotation"):
|
|
|
|
@kanta.logheader
|
|
def header(action: str, bogus: int) -> str:
|
|
return action
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logheader_replaces_default_header(tmp_path, format_config, caplog):
|
|
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
|
|
|
@kanta.logheader
|
|
def header(action: str, user: str | None, extra: dict | None) -> str:
|
|
return f"HDR {action} user={user} session={extra['session']}"
|
|
|
|
await kanta.open(log=False)
|
|
with kanta.transaction(action="update", user="alice", extra={"session": 3}) as data:
|
|
data.counter = 1
|
|
await kanta.close()
|
|
|
|
assert "HDR update user=alice session=3" in caplog.text
|
|
# The diff body is still logged after the custom header.
|
|
assert "counter" in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logheader_default_target_is_filename(tmp_path, format_config, caplog):
|
|
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
|
kanta = make_kanta(tmp_path / "mydb.db", Data, format_config)
|
|
|
|
@kanta.logheader
|
|
def header(action: str, extra: dict | None) -> str:
|
|
return f"target={extra['target']}"
|
|
|
|
await kanta.open(log=False)
|
|
with kanta.transaction(action="update") as data:
|
|
data.counter = 1
|
|
await kanta.close()
|
|
|
|
assert "target=mydb.db" in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logheader_explicit_target_kept(tmp_path, format_config, caplog):
|
|
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
|
kanta = make_kanta(tmp_path / "mydb.db", Data, format_config)
|
|
|
|
@kanta.logheader
|
|
def header(action: str, extra: dict | None) -> str:
|
|
return f"target={extra['target']}"
|
|
|
|
await kanta.open(log=False)
|
|
with kanta.transaction(
|
|
action="update", extra={"target": "Project X (abcd1234)"}
|
|
) as data:
|
|
data.counter = 1
|
|
await kanta.close()
|
|
|
|
assert "target=Project X (abcd1234)" in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logheader_injects_states_and_kanta(tmp_path, format_config, caplog):
|
|
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
|
|
|
@kanta.logheader
|
|
def header(
|
|
action: str,
|
|
previous: DictPre,
|
|
current: DictPost,
|
|
kanta: Kanta,
|
|
) -> str:
|
|
return (
|
|
f"{action} counter {previous.get('counter')}"
|
|
f" -> {current.get('counter')} db={kanta.filename.name}"
|
|
)
|
|
|
|
await kanta.open(log=False)
|
|
with kanta.transaction(action="increment") as data:
|
|
data.counter = 5
|
|
await kanta.close()
|
|
|
|
assert "increment counter 0 -> 5 db=test.db" in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logheader_chain_first_non_none_wins(tmp_path, format_config, caplog):
|
|
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
|
|
|
@kanta.logheader
|
|
def first(action: str) -> str | None:
|
|
return None
|
|
|
|
@kanta.logheader
|
|
def second(action: str) -> str:
|
|
return f"SECOND {action}"
|
|
|
|
await kanta.open(log=False)
|
|
with kanta.transaction(action="update") as data:
|
|
data.counter = 1
|
|
await kanta.close()
|
|
|
|
assert "SECOND update" in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logheader_all_none_falls_back_to_default(
|
|
tmp_path, format_config, caplog
|
|
):
|
|
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
|
|
|
@kanta.logheader
|
|
def header(action: str) -> str | None:
|
|
return None
|
|
|
|
await kanta.open(log=False)
|
|
with kanta.transaction(action="update", user="alice") as data:
|
|
data.counter = 1
|
|
await kanta.close()
|
|
|
|
assert "update" in caplog.text
|
|
assert "alice" in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logheader_receives_formatted_user(tmp_path, format_config, caplog):
|
|
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
|
|
|
@kanta.logfmt(path="$user")
|
|
def resolve_user(value: str) -> str | None:
|
|
return "Alice"
|
|
|
|
@kanta.logheader
|
|
def header(action: str, user: str | None) -> str:
|
|
return f"actor={user}"
|
|
|
|
await kanta.open(log=False)
|
|
with kanta.transaction(action="update", user="uuid-1") as data:
|
|
data.counter = 1
|
|
await kanta.close()
|
|
|
|
assert "actor=Alice" in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logheader_applies_to_bootstrap(tmp_path, format_config, caplog):
|
|
caplog.set_level(logging.INFO, logger="kanta.bootstrap")
|
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
|
|
|
@kanta.logheader
|
|
def header(action: str, extra: dict | None) -> str:
|
|
return f"BOOT {action} target={extra['target']}"
|
|
|
|
await kanta.open()
|
|
await kanta.close()
|
|
|
|
assert "BOOT bootstrap target=test.db" in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transaction_extra_string_in_default_header(
|
|
tmp_path, format_config, caplog
|
|
):
|
|
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
|
|
|
await kanta.open(log=False)
|
|
with kanta.transaction(action="export", extra="mydb.db") as data:
|
|
data.counter = 1
|
|
await kanta.close()
|
|
|
|
assert "export" in caplog.text
|
|
assert "mydb.db" in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extra_is_not_persisted(tmp_path, format_config):
|
|
path = tmp_path / "test.db"
|
|
kanta = make_kanta(path, Data, format_config)
|
|
|
|
await kanta.open(log=False)
|
|
with kanta.transaction(
|
|
action="update", user="alice", extra={"session": 3, "target": "X"}
|
|
) as data:
|
|
data.counter = 1
|
|
await kanta.close()
|
|
|
|
record = read_changes(path, format_config)[-1]
|
|
assert record.a == "update"
|
|
assert record.u == "alice"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transaction_log_dict_toggles(tmp_path, format_config, caplog):
|
|
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
|
|
|
await kanta.open(log=False)
|
|
with kanta.transaction(
|
|
action="myaction", log={"header": True, "diff": False}
|
|
) as data:
|
|
data.counter = 1
|
|
with kanta.transaction(
|
|
action="otheraction", log={"header": False, "diff": True}
|
|
) as data:
|
|
data.counter = 2
|
|
await kanta.close()
|
|
|
|
# First transaction: header only.
|
|
assert "myaction" in caplog.text
|
|
# Second transaction: diff only, no header.
|
|
assert "otheraction" not in caplog.text
|
|
assert "counter" in caplog.text
|