LeoVasanko
5c32e495e9
Change record headers now carry the revision of the state they produce (rN); snapshot indicator lines drop it.
481 lines
17 KiB
Python
481 lines
17 KiB
Python
"""Tests for revision numbers: runtime counter, snapshot ``r`` field, CLI rN."""
|
|
|
|
import asyncio
|
|
import json
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
|
|
from kanta.exceptions import DataIntegrityError
|
|
from kanta.serialization import JsonSerializer
|
|
from kanta.structs import META_FORMAT, ChangeRecord, Snapshot
|
|
from tests.support import Data, make_kanta, read_last_snapshot
|
|
from tests.test_rotation import make_clock, read_all, write_history
|
|
|
|
T0 = datetime(2026, 1, 1, tzinfo=UTC)
|
|
|
|
# The CLI reads files with its own JSON Kanta instance unless -k is given.
|
|
JSON = ("json", JsonSerializer)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_revision_requires_open(tmp_path, format_config):
|
|
kanta = make_kanta(tmp_path / "data.kantadb", Data, format_config)
|
|
with pytest.raises(DataIntegrityError):
|
|
kanta.rev
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_revision_counts_change_records(tmp_path, format_config):
|
|
path = tmp_path / "data.kantadb"
|
|
kanta = make_kanta(path, Data, format_config)
|
|
await kanta.open(log=False)
|
|
# The bootstrap record is the first diff: state is at revision 1.
|
|
assert kanta.rev == 1
|
|
|
|
with kanta.transaction("inc", log=False) as data:
|
|
data.counter += 1
|
|
assert kanta.rev == 2
|
|
|
|
# A no-op transaction queues no record and does not advance the revision.
|
|
with kanta.transaction("noop", log=False):
|
|
pass
|
|
assert kanta.rev == 2
|
|
await kanta.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_revision_restored_on_reopen(tmp_path, format_config):
|
|
path = tmp_path / "data.kantadb"
|
|
kanta = make_kanta(path, Data, format_config)
|
|
await kanta.open(log=False)
|
|
for _ in range(3):
|
|
with kanta.transaction("inc", log=False) as data:
|
|
data.counter += 1
|
|
await kanta.close()
|
|
|
|
kanta2 = make_kanta(path, Data, format_config)
|
|
await kanta2.open(log=False)
|
|
assert kanta2.rev == 4
|
|
await kanta2.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_carries_revision(tmp_path, format_config):
|
|
path = tmp_path / "data.kantadb"
|
|
kanta = make_kanta(path, Data, format_config)
|
|
await kanta.open(log=False)
|
|
for _ in range(2):
|
|
with kanta.transaction("inc", log=False) as data:
|
|
data.counter += 1
|
|
await kanta.flush()
|
|
kanta.request_snapshot()
|
|
kanta._impl.maybe_snapshot()
|
|
await kanta.close()
|
|
|
|
snap = read_last_snapshot(path, format_config)
|
|
assert snap is not None
|
|
assert snap.r == 3
|
|
assert snap.k == META_FORMAT
|
|
|
|
# Reopening from the snapshot keeps the absolute revision.
|
|
kanta2 = make_kanta(path, Data, format_config)
|
|
await kanta2.open(log=False)
|
|
assert kanta2.rev == 3
|
|
with kanta2.transaction("inc", log=False) as data:
|
|
data.counter += 1
|
|
assert kanta2.rev == 4
|
|
await kanta2.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legacy_snapshot_recounted_and_rewritten(tmp_path, format_config):
|
|
"""Meta migration k1 rewrites a k0 file in full (see kanta/metamigr.py)."""
|
|
path = tmp_path / "data.kantadb"
|
|
kanta = make_kanta(path, Data, format_config)
|
|
await kanta.open(log=False)
|
|
for _ in range(2):
|
|
with kanta.transaction("inc", log=False) as data:
|
|
data.counter += 1
|
|
await kanta.flush()
|
|
kanta.request_snapshot()
|
|
kanta._impl.maybe_snapshot()
|
|
await kanta.close()
|
|
|
|
# Simulate a pre-revision database: rewrite the last snapshot without k,
|
|
# with a wrong r, then append one more change record.
|
|
_, serializer_cls = format_config
|
|
serializer = serializer_cls()
|
|
framer = serializer.framer_cls()
|
|
records = read_all(path, format_config)
|
|
out = bytearray()
|
|
for rec in records:
|
|
if isinstance(rec, Snapshot):
|
|
rec = Snapshot(ts=rec.ts, v=rec.v, state=rec.state, m=rec.m, r=0)
|
|
out += framer.frame_snapshot(
|
|
serializer.encode(rec), record_offset=len(out)
|
|
)
|
|
else:
|
|
out += framer.frame_change(
|
|
serializer.encode(rec), record_offset=len(out)
|
|
)
|
|
extra = ChangeRecord(ts=T0, a="inc", v=0, diff={"counter": 3})
|
|
out += framer.frame_change(serializer.encode(extra), record_offset=len(out))
|
|
path.write_bytes(bytes(out))
|
|
|
|
kanta2 = make_kanta(path, Data, format_config)
|
|
await kanta2.open(log=False)
|
|
# bootstrap + 2 changes + trailing change = revision 4, despite r=0 above.
|
|
assert kanta2.rev == 4
|
|
assert kanta2.data.counter == 3
|
|
await kanta2.close()
|
|
|
|
# The file was rewritten in full: all change records preserved in order,
|
|
# the legacy intermediate snapshot dropped, and a k1 closing snapshot
|
|
# carrying the total revision appended.
|
|
records = read_all(path, format_config)
|
|
changes = [r for r in records if isinstance(r, ChangeRecord)]
|
|
assert [c.a for c in changes] == ["bootstrap", "inc", "inc", "inc"]
|
|
snapshots = [r for r in records if isinstance(r, Snapshot)]
|
|
assert len(snapshots) == 1
|
|
assert isinstance(records[-1], Snapshot)
|
|
closing = snapshots[0]
|
|
assert closing.k == META_FORMAT
|
|
assert closing.r == 4
|
|
assert closing.state["counter"] == 3
|
|
|
|
# The next open trusts the stored r (file unchanged this time).
|
|
content = path.read_bytes()
|
|
kanta3 = make_kanta(path, Data, format_config)
|
|
await kanta3.open(log=False)
|
|
assert kanta3.rev == 4
|
|
await kanta3.close()
|
|
assert path.read_bytes() == content
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legacy_migration_preserves_leading_snapshot(tmp_path, format_config):
|
|
"""A legacy file starting with a snapshot keeps it, re-stamped k1."""
|
|
path = tmp_path / "data.kantadb"
|
|
_, serializer_cls = format_config
|
|
serializer = serializer_cls()
|
|
framer = serializer.framer_cls()
|
|
snap_ts = T0 - timedelta(days=40)
|
|
snap_m = T0 - timedelta(days=41)
|
|
leading = Snapshot(ts=snap_ts, v=3, state={"users": {}, "counter": 7}, m=snap_m)
|
|
out = bytearray()
|
|
out += framer.frame_snapshot(serializer.encode(leading), record_offset=0)
|
|
for i in range(2):
|
|
rec = ChangeRecord(ts=T0, a=f"inc{i}", v=3, diff={"counter": 8 + i})
|
|
out += framer.frame_change(serializer.encode(rec), record_offset=len(out))
|
|
path.write_bytes(bytes(out))
|
|
|
|
kanta = make_kanta(path, Data, format_config)
|
|
await kanta.open(log=False)
|
|
# Only the in-file history is countable: 2 change records.
|
|
assert kanta.rev == 2
|
|
assert kanta.data.counter == 9
|
|
await kanta.close()
|
|
|
|
records = read_all(path, format_config)
|
|
assert isinstance(records[0], Snapshot)
|
|
assert records[0].k == META_FORMAT
|
|
assert records[0].r == 0
|
|
assert records[0].ts == snap_ts
|
|
assert records[0].m == snap_m
|
|
assert records[0].v == 3
|
|
assert records[0].state == {"users": {}, "counter": 7}
|
|
changes = [r for r in records if isinstance(r, ChangeRecord)]
|
|
assert [c.a for c in changes] == ["inc0", "inc1"]
|
|
assert isinstance(records[-1], Snapshot)
|
|
assert records[-1].k == META_FORMAT
|
|
assert records[-1].r == 2
|
|
assert records[-1].state["counter"] == 9
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rotation_of_legacy_file_migrates(tmp_path, format_config):
|
|
"""Rotation and meta migration coincide: rotation already produces k1
|
|
snapshots with revisions counted from pre-rotation record positions."""
|
|
path = tmp_path / "data.kantadb"
|
|
_, serializer_cls = format_config
|
|
serializer = serializer_cls()
|
|
framer = serializer.framer_cls()
|
|
|
|
def change(action, day, counter):
|
|
return ChangeRecord(
|
|
ts=T0 + day * timedelta(days=1), a=action, v=0, diff={"counter": counter}
|
|
)
|
|
|
|
legacy_snap = Snapshot(
|
|
ts=T0 - 40 * timedelta(days=1), v=0, state={"users": {}, "counter": 1}, r=0
|
|
)
|
|
out = bytearray()
|
|
out += framer.frame_change(
|
|
serializer.encode(change("bootstrap", -41, 0)), record_offset=len(out)
|
|
)
|
|
out += framer.frame_change(
|
|
serializer.encode(change("day-40", -40, 1)), record_offset=len(out)
|
|
)
|
|
out += framer.frame_snapshot(
|
|
serializer.encode(legacy_snap), record_offset=len(out)
|
|
)
|
|
out += framer.frame_change(
|
|
serializer.encode(change("day-20", -20, 2)), record_offset=len(out)
|
|
)
|
|
out += framer.frame_change(
|
|
serializer.encode(change("day-5", -5, 3)), record_offset=len(out)
|
|
)
|
|
path.write_bytes(bytes(out))
|
|
|
|
cell = [T0]
|
|
kanta = make_kanta(path, Data, format_config, retention=30 * timedelta(days=1))
|
|
kanta.clock(make_clock(cell))
|
|
await kanta.open(log=False)
|
|
assert kanta.rev == 4
|
|
assert kanta.data.counter == 3
|
|
await kanta.close()
|
|
|
|
# The rotation's leading snapshot is k1 with the pre-rotation revision.
|
|
records = read_all(path, format_config)
|
|
assert isinstance(records[0], Snapshot)
|
|
assert records[0].k == META_FORMAT
|
|
assert records[0].r == 2
|
|
assert [r.a for r in records if isinstance(r, ChangeRecord)] == [
|
|
"day-20",
|
|
"day-5",
|
|
]
|
|
|
|
# No further rewriting on the next open.
|
|
content = path.read_bytes()
|
|
kanta2 = make_kanta(path, Data, format_config)
|
|
await kanta2.open(log=False)
|
|
assert kanta2.rev == 4
|
|
await kanta2.close()
|
|
assert path.read_bytes() == content
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legacy_snapshot_readonly_recounts_without_writing(
|
|
tmp_path, format_config
|
|
):
|
|
path = tmp_path / "data.kantadb"
|
|
kanta = make_kanta(path, Data, format_config)
|
|
await kanta.open(log=False)
|
|
with kanta.transaction("inc", log=False) as data:
|
|
data.counter += 1
|
|
await kanta.close()
|
|
|
|
# Inject a legacy snapshot (no k marker) at the start; the whole file is
|
|
# re-framed with a single framer so binary checksums stay valid.
|
|
_, serializer_cls = format_config
|
|
serializer = serializer_cls()
|
|
framer = serializer.framer_cls()
|
|
legacy = Snapshot(ts=T0, v=0, state={"users": {}, "counter": 0}, r=0)
|
|
out = bytearray()
|
|
out += framer.frame_snapshot(serializer.encode(legacy), record_offset=0)
|
|
for rec in read_all(path, format_config):
|
|
out += framer.frame_change(serializer.encode(rec), record_offset=len(out))
|
|
path.write_bytes(bytes(out))
|
|
content = path.read_bytes()
|
|
|
|
kanta2 = make_kanta(path, Data, format_config)
|
|
await kanta2.open(readonly=True, log=False)
|
|
assert kanta2.rev == 2
|
|
await kanta2.close()
|
|
assert path.read_bytes() == content
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rotation_preserves_revisions(tmp_path, format_config):
|
|
path = tmp_path / "data.kantadb"
|
|
await write_history(path, format_config, days=[-40, -20, -5])
|
|
|
|
cell = [T0]
|
|
kanta = make_kanta(path, Data, format_config, retention=30 * timedelta(days=1))
|
|
kanta.clock(make_clock(cell))
|
|
await kanta.open(log=False)
|
|
# bootstrap + 3 changes = revision 4 before and after the rotation.
|
|
assert kanta.rev == 4
|
|
await kanta.close()
|
|
|
|
records = read_all(path, format_config)
|
|
leading = records[0]
|
|
assert isinstance(leading, Snapshot)
|
|
# One change was dropped (bootstrap + day-40 are before the cutoff
|
|
# position... bootstrap is a change record too: 2 records before).
|
|
assert leading.r == 2
|
|
assert leading.k == META_FORMAT
|
|
|
|
kanta2 = make_kanta(path, Data, format_config)
|
|
await kanta2.open(log=False)
|
|
assert kanta2.rev == 4
|
|
with kanta2.transaction("inc", log=False) as data:
|
|
data.counter += 1
|
|
assert kanta2.rev == 5
|
|
await kanta2.close()
|
|
|
|
|
|
def test_cli_revision_snapshot_base_file(tmp_path, capsys):
|
|
"""File = snapshot {r: 2} + one change; the two producing changes are gone."""
|
|
path = tmp_path / "data.kantadb"
|
|
serializer = JsonSerializer()
|
|
framer = serializer.framer_cls()
|
|
snap = Snapshot(
|
|
ts=T0, v=0, state={"foo": 1, "bar": 1}, r=2, k=META_FORMAT
|
|
)
|
|
out = bytearray()
|
|
out += framer.frame_snapshot(serializer.encode(snap), record_offset=0)
|
|
change = ChangeRecord(ts=T0, a="add_baz", v=0, diff={"baz": 1})
|
|
out += framer.frame_change(serializer.encode(change), record_offset=len(out))
|
|
path.write_bytes(bytes(out))
|
|
capsys.readouterr()
|
|
|
|
# r1's producing change is gone and no snapshot holds state 1: error.
|
|
assert _run_cli(path, "-r", "r1") == 2
|
|
assert "r1" in capsys.readouterr().err
|
|
|
|
# r2 prints the snapshot line (it literally carries r: 2) and its state;
|
|
# revisions are shown on change records, not on snapshot indicators.
|
|
assert _run_cli(path, "-r", "r2", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert "snapshot s0" in out.err
|
|
assert "r2" not in out.err
|
|
assert json.loads(out.out) == {"foo": 1, "bar": 1}
|
|
|
|
# r3 is the change, replayed on the snapshot base; state includes baz.
|
|
assert _run_cli(path, "-r", "r3", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert "add_baz" in out.err
|
|
assert " r3" in out.err
|
|
assert json.loads(out.out) == {"foo": 1, "bar": 1, "baz": 1}
|
|
|
|
# Single index and half-open range agree.
|
|
assert _run_cli(path, "-r", "r2") == 0
|
|
single = capsys.readouterr().err
|
|
assert _run_cli(path, "-r", "r2:r3") == 0
|
|
assert capsys.readouterr().err == single
|
|
|
|
|
|
def _run_cli(path, *args):
|
|
from kanta.__main__ import main
|
|
|
|
return main([str(path), *args])
|
|
|
|
|
|
def test_cli_revision_ranges(tmp_path, capsys):
|
|
path = tmp_path / "data.kantadb"
|
|
|
|
async def build():
|
|
kanta = make_kanta(path, Data, JSON)
|
|
await kanta.open(log=False)
|
|
for i in range(3):
|
|
with kanta.transaction(f"inc{i}", log=False) as data:
|
|
data.counter += 1
|
|
await kanta.close()
|
|
|
|
asyncio.run(build())
|
|
capsys.readouterr()
|
|
|
|
# A revision selects the record that produced that state, with the
|
|
# state: r1 and r1:r2 both show only the bootstrap and output state 1.
|
|
assert _run_cli(path, "-r", "r1") == 0
|
|
single = capsys.readouterr().err
|
|
assert "bootstrap" in single
|
|
# Change record headers carry the revision of the state they produce.
|
|
assert " r1" in single
|
|
assert "inc0" not in single
|
|
|
|
assert _run_cli(path, "-r", "r1:r2") == 0
|
|
assert capsys.readouterr().err == single
|
|
|
|
assert _run_cli(path, "-r", "r1", "-o", "-") == 0
|
|
assert json.loads(capsys.readouterr().out) == {"users": {}, "counter": 0}
|
|
|
|
# r1..r2 (inclusive end) adds the next change and outputs state 2.
|
|
assert _run_cli(path, "-r", "r1..r2", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert "inc0" in out.err
|
|
assert json.loads(out.out) == {"users": {}, "counter": 1}
|
|
|
|
# r0 is not meaningful: no record produces the empty initial state.
|
|
assert _run_cli(path, "-r", "r0") == 2
|
|
assert "r0" in capsys.readouterr().err
|
|
|
|
# Range by revision.
|
|
assert _run_cli(path, "-r", "r2:r4") == 0
|
|
err = capsys.readouterr().err
|
|
assert "inc0" in err
|
|
assert "inc1" in err
|
|
assert "inc2" not in err
|
|
|
|
# Negative revisions are rejected.
|
|
assert _run_cli(path, "-r", "r-1") == 1
|
|
assert "negative" in capsys.readouterr().err
|
|
|
|
# Out-of-range single revision is a range error (exit 2).
|
|
assert _run_cli(path, "-r", "r99") == 2
|
|
assert "r99" in capsys.readouterr().err
|
|
|
|
# The same bounds as a range truncate Python-style instead: an empty
|
|
# display and the closest known state, not an error.
|
|
assert _run_cli(path, "-r", "r99:r100", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert "bootstrap" not in out.err
|
|
assert "inc" not in out.err
|
|
assert json.loads(out.out) == {"users": {}, "counter": 3}
|
|
|
|
# Only the end beyond the last revision: truncated to end of file.
|
|
assert _run_cli(path, "-r", "r3:r99", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert "inc1" in out.err
|
|
assert "inc2" in out.err
|
|
assert json.loads(out.out) == {"users": {}, "counter": 3}
|
|
|
|
|
|
def test_cli_revision_after_rotation(tmp_path, capsys):
|
|
path = tmp_path / "data.kantadb"
|
|
|
|
async def build():
|
|
await write_history(path, JSON, days=[-40, -20, -5])
|
|
cell = [T0]
|
|
kanta = make_kanta(
|
|
path, Data, JSON, retention=30 * timedelta(days=1)
|
|
)
|
|
kanta.clock(make_clock(cell))
|
|
await kanta.open(log=False)
|
|
await kanta.close()
|
|
|
|
asyncio.run(build())
|
|
capsys.readouterr()
|
|
|
|
# The leading snapshot holds revision 2 (bootstrap + day-40). Since the
|
|
# change that produced state 2 is gone, r2 shows the snapshot itself and
|
|
# outputs the snapshot state.
|
|
assert _run_cli(path, "-r", "r2", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert "snapshot s0" in out.err
|
|
assert "r2" not in out.err
|
|
assert "day-20" not in out.err
|
|
assert json.loads(out.out) == {"users": {}, "counter": 1}
|
|
|
|
# r1 has no record in this file: a single index does not truncate.
|
|
assert _run_cli(path, "-r", "r1") == 2
|
|
assert "r1" in capsys.readouterr().err
|
|
|
|
# A range truncates like Python slicing instead: r1:r3 starts at the
|
|
# beginning of the file and stops before the r3 record, showing just the
|
|
# leading snapshot.
|
|
assert _run_cli(path, "-r", "r1:r3") == 0
|
|
err = capsys.readouterr().err
|
|
assert "snapshot s0" in err
|
|
assert "r2" not in err
|
|
assert "day-20" not in err
|
|
|
|
# Absolute revisions reach the retained change records.
|
|
assert _run_cli(path, "-r", "r3:r4") == 0
|
|
err = capsys.readouterr().err
|
|
assert "day-20" in err
|
|
assert "snapshot" not in err
|
|
assert "day-5" not in err
|