482 lines
16 KiB
Python
482 lines
16 KiB
Python
"""Tests for repeated -r ranges: unit window intersection, plain item slicing."""
|
|
|
|
import asyncio
|
|
import json
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
|
|
from kanta.replaylog import is_time_spec, parse_time_spec
|
|
from kanta.serialization import JsonSerializer
|
|
from kanta.structs import META_FORMAT, ChangeRecord, Snapshot
|
|
from tests.support import Data, make_kanta
|
|
|
|
# The CLI reads files with its own JSON Kanta instance unless -k is given.
|
|
JSON = ("json", JsonSerializer)
|
|
|
|
T0 = datetime(2026, 1, 1, tzinfo=UTC)
|
|
NOW = datetime(2026, 9, 28, 12, 0, 0, tzinfo=UTC)
|
|
|
|
|
|
def _run_cli(path, *args):
|
|
from kanta.__main__ import main
|
|
|
|
return main([str(path), *args])
|
|
|
|
|
|
def _build(path, increments=3):
|
|
async def build():
|
|
kanta = make_kanta(path, Data, JSON)
|
|
await kanta.open(log=False)
|
|
for i in range(increments):
|
|
with kanta.transaction(f"inc{i}", log=False) as data:
|
|
data.counter += 1
|
|
await kanta.close()
|
|
|
|
asyncio.run(build())
|
|
|
|
|
|
def test_plain_range_restricts_dump_to_last_printed(tmp_path, capsys):
|
|
"""The JSON dump holds the state after the last printed item."""
|
|
path = tmp_path / "data.kantadb"
|
|
_build(path)
|
|
capsys.readouterr()
|
|
|
|
# Items are bootstrap, inc0, inc1, inc2: 1:3 prints inc0 and inc1.
|
|
assert _run_cli(path, "-r", "1:3", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert "bootstrap" not in out.err
|
|
assert "inc0" in out.err
|
|
assert "inc1" in out.err
|
|
assert "inc2" not in out.err
|
|
# The dump is the state after inc1, the last printed record.
|
|
assert json.loads(out.out) == {"users": {}, "counter": 2}
|
|
|
|
# Inclusive end.
|
|
assert _run_cli(path, "-r", "0..1") == 0
|
|
err = capsys.readouterr().err
|
|
assert "bootstrap" in err
|
|
assert "inc0" in err
|
|
assert "inc1" not in err
|
|
|
|
|
|
def test_plain_single_index(tmp_path, capsys):
|
|
path = tmp_path / "data.kantadb"
|
|
_build(path)
|
|
capsys.readouterr()
|
|
|
|
assert _run_cli(path, "-r", "0") == 0
|
|
err = capsys.readouterr().err
|
|
assert "bootstrap" in err
|
|
assert "inc0" not in err
|
|
|
|
# Negative index counts from the end of the filtered output.
|
|
assert _run_cli(path, "-r", "-1") == 0
|
|
err = capsys.readouterr().err
|
|
assert "inc2" in err
|
|
assert "inc1" not in err
|
|
|
|
# A single index must exist; out of range is exit 2.
|
|
assert _run_cli(path, "-r", "99") == 2
|
|
assert "99" in capsys.readouterr().err
|
|
|
|
|
|
def test_plain_ranges_intersect(tmp_path, capsys):
|
|
path = tmp_path / "data.kantadb"
|
|
_build(path)
|
|
capsys.readouterr()
|
|
|
|
# [1:) and [:3) intersect to items 1 and 2.
|
|
assert _run_cli(path, "-r", "1:", "-r", ":3") == 0
|
|
err = capsys.readouterr().err
|
|
assert "bootstrap" not in err
|
|
assert "inc0" in err
|
|
assert "inc1" in err
|
|
assert "inc2" not in err
|
|
|
|
# An empty intersection prints nothing, exit 0 (truncation, not error).
|
|
assert _run_cli(path, "-r", ":1", "-r", "3:") == 0
|
|
err = capsys.readouterr().err
|
|
assert "inc" not in err
|
|
assert "bootstrap" not in err
|
|
|
|
|
|
def test_plain_applies_after_grep(tmp_path, capsys):
|
|
"""Plain numbers index into what survives --grep, order-independently."""
|
|
path = tmp_path / "data.kantadb"
|
|
_build(path)
|
|
capsys.readouterr()
|
|
|
|
# Grep leaves inc0..inc2; plain 0 picks the first of those.
|
|
assert _run_cli(path, "--grep", "inc", "-r", "0") == 0
|
|
first = capsys.readouterr().err
|
|
assert "inc0" in first
|
|
assert "inc1" not in first
|
|
assert "bootstrap" not in first
|
|
|
|
assert _run_cli(path, "-r", "0", "--grep", "inc") == 0
|
|
assert capsys.readouterr().err == first
|
|
|
|
# Negative index from the filtered end.
|
|
assert _run_cli(path, "--grep", "inc", "-r", "-1") == 0
|
|
err = capsys.readouterr().err
|
|
assert "inc2" in err
|
|
assert "inc1" not in err
|
|
|
|
|
|
def test_unit_specs_intersect(tmp_path, capsys):
|
|
path = tmp_path / "data.kantadb"
|
|
_build(path)
|
|
capsys.readouterr()
|
|
|
|
# r1:r4 = bootstrap..inc1, r2:r5 = inc0..inc2; intersection is inc0, inc1.
|
|
assert _run_cli(path, "-r", "r1:r4", "-r", "r2:r5", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert "bootstrap" not in out.err
|
|
assert "inc0" in out.err
|
|
assert "inc1" in out.err
|
|
assert "inc2" not in out.err
|
|
# The replay end is the intersected window end (state after inc1).
|
|
assert json.loads(out.out) == {"users": {}, "counter": 2}
|
|
|
|
|
|
def test_unit_and_plain_combine(tmp_path, capsys):
|
|
"""Unit window first, then plain slicing over the window's items."""
|
|
path = tmp_path / "data.kantadb"
|
|
_build(path)
|
|
capsys.readouterr()
|
|
|
|
# Window r2:r4 yields inc0, inc1; plain item 0 prints only inc0.
|
|
assert _run_cli(path, "-r", "r2:r4", "-r", "0", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert "inc0" in out.err
|
|
assert "inc1" not in out.err
|
|
# The dump is the state after the last printed item (inc0).
|
|
assert json.loads(out.out) == {"users": {}, "counter": 1}
|
|
|
|
|
|
def test_context_expands_after_grep(tmp_path, capsys):
|
|
"""Context records come from the whole file, free of --grep."""
|
|
path = tmp_path / "data.kantadb"
|
|
_build(path, increments=5)
|
|
capsys.readouterr()
|
|
|
|
assert _run_cli(path, "--grep", "inc2", "--context", "1") == 0
|
|
err = capsys.readouterr().err
|
|
assert "inc1" in err
|
|
assert "inc2" in err
|
|
assert "inc3" in err
|
|
assert "inc0" not in err
|
|
assert "inc4" not in err
|
|
assert "bootstrap" not in err
|
|
|
|
# Context 0 (default) changes nothing.
|
|
assert _run_cli(path, "--grep", "inc2") == 0
|
|
err = capsys.readouterr().err
|
|
assert "inc1" not in err
|
|
assert "inc2" in err
|
|
|
|
# Context is clamped at the start of the file.
|
|
assert _run_cli(path, "--grep", "bootstrap", "--context", "2") == 0
|
|
err = capsys.readouterr().err
|
|
assert "bootstrap" in err
|
|
assert "inc0" in err
|
|
assert "inc1" in err
|
|
assert "inc2" not in err
|
|
|
|
|
|
def test_context_expands_past_unit_window(tmp_path, capsys):
|
|
"""Context expands whatever a unit selection would show, past the window."""
|
|
path = tmp_path / "data.kantadb"
|
|
_build(path)
|
|
capsys.readouterr()
|
|
|
|
# -r r2 alone shows only inc0 (the record producing state 2); context 1
|
|
# expands past the window to bootstrap and inc1, and the dump follows
|
|
# the last printed record.
|
|
assert _run_cli(path, "-r", "r2", "--context", "1", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert "bootstrap" in out.err
|
|
assert "inc0" in out.err
|
|
assert "inc1" in out.err
|
|
assert "inc2" not in out.err
|
|
assert json.loads(out.out) == {"users": {}, "counter": 2}
|
|
|
|
|
|
def test_context_expands_snapshot_selection(tmp_path, capsys):
|
|
"""-r sN --context N shows the snapshot plus its neighbors."""
|
|
path = tmp_path / "data.kantadb"
|
|
serializer = JsonSerializer()
|
|
framer = serializer.framer_cls()
|
|
out_bytes = bytearray()
|
|
out_bytes += framer.frame_change(
|
|
serializer.encode(ChangeRecord(ts=T0, a="inc0", v=0, diff={"counter": 1})),
|
|
record_offset=len(out_bytes),
|
|
)
|
|
snap = Snapshot(ts=T0, v=0, state={"counter": 1}, r=1, k=META_FORMAT)
|
|
out_bytes += framer.frame_snapshot(
|
|
serializer.encode(snap), record_offset=len(out_bytes)
|
|
)
|
|
out_bytes += framer.frame_change(
|
|
serializer.encode(ChangeRecord(ts=T0, a="inc1", v=0, diff={"counter": 2})),
|
|
record_offset=len(out_bytes),
|
|
)
|
|
path.write_bytes(bytes(out_bytes))
|
|
capsys.readouterr()
|
|
|
|
# Without context, s1 (the file starts with a change, so s0 is l0)
|
|
# shows just the snapshot and its state.
|
|
assert _run_cli(path, "-r", "s1", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert "snapshot s1" in out.err
|
|
assert "inc0" not in out.err
|
|
assert json.loads(out.out) == {"counter": 1}
|
|
|
|
# With context, the selection expands past the snapshot line.
|
|
assert _run_cli(path, "-r", "s1", "--context", "1", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert "inc0" in out.err
|
|
assert "snapshot s1" in out.err
|
|
assert "inc1" in out.err
|
|
assert json.loads(out.out) == {"counter": 2}
|
|
|
|
|
|
def test_context_merges_overlaps(tmp_path, capsys):
|
|
"""Nearby matches share context instead of duplicating records."""
|
|
path = tmp_path / "data.kantadb"
|
|
_build(path, increments=5)
|
|
capsys.readouterr()
|
|
|
|
# Grep leaves inc0..inc4; plain 1:3 selects inc1, inc2; context 1
|
|
# expands to inc0..inc3, each printed exactly once (overlaps merged).
|
|
assert _run_cli(path, "--grep", "inc", "-r", "1:3", "--context", "1") == 0
|
|
err = capsys.readouterr().err
|
|
for action in ["inc0", "inc1", "inc2", "inc3"]:
|
|
assert err.count(action) == 1
|
|
assert "bootstrap" not in err
|
|
assert "inc4" not in err
|
|
|
|
|
|
def test_grep_and_context_shift_dump_to_last_printed(tmp_path, capsys):
|
|
"""Grep and --context are display filters, but the dump follows the
|
|
last printed record, so they move it too."""
|
|
path = tmp_path / "data.kantadb"
|
|
_build(path)
|
|
capsys.readouterr()
|
|
|
|
# Only inc1 prints; the dump is the state after inc1.
|
|
assert _run_cli(path, "--grep", "inc1", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert "inc0" not in out.err
|
|
assert json.loads(out.out) == {"users": {}, "counter": 2}
|
|
|
|
# Context extends the printout past the match; the dump follows it.
|
|
assert _run_cli(path, "--grep", "inc1", "--context", "1", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert "inc2" in out.err
|
|
assert json.loads(out.out) == {"users": {}, "counter": 3}
|
|
|
|
# Nothing printed: the dump falls back to the range-end state.
|
|
assert _run_cli(path, "--grep", "zzz", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert json.loads(out.out) == {"users": {}, "counter": 3}
|
|
|
|
|
|
def _write_timed(path, entries):
|
|
"""Write one change record per (timestamp, action, counter) entry."""
|
|
serializer = JsonSerializer()
|
|
framer = serializer.framer_cls()
|
|
out = bytearray()
|
|
for ts, action, counter in entries:
|
|
out += framer.frame_change(
|
|
serializer.encode(
|
|
ChangeRecord(ts=ts, a=action, v=0, diff={"counter": counter})
|
|
),
|
|
record_offset=len(out),
|
|
)
|
|
path.write_bytes(bytes(out))
|
|
|
|
|
|
def test_is_time_spec_classification():
|
|
for spec in [
|
|
"30s",
|
|
"15min",
|
|
"1h",
|
|
"7d",
|
|
"2w",
|
|
"2026-09-28",
|
|
"2026-09-28T10",
|
|
"2026-09-28T10:30",
|
|
"20260928T103045",
|
|
"2026-09-24/2026-09-28",
|
|
"2026-09-24/",
|
|
"/2026-09-28",
|
|
"2026-13-45", # malformed ISO still classifies as time, then errors
|
|
]:
|
|
assert is_time_spec(spec), spec
|
|
for spec in ["5", "-1", "2:5", "2..5", ":-1", "r3", "s-1", "l10", "v0", "r1:r4"]:
|
|
assert not is_time_spec(spec), spec
|
|
|
|
|
|
def test_parse_time_spec_relative():
|
|
assert parse_time_spec("1h", NOW) == (NOW - timedelta(hours=1), NOW)
|
|
assert parse_time_spec("15min", NOW) == (NOW - timedelta(minutes=15), NOW)
|
|
assert parse_time_spec("30s", NOW) == (NOW - timedelta(seconds=30), NOW)
|
|
assert parse_time_spec("7d", NOW) == (NOW - timedelta(days=7), NOW)
|
|
assert parse_time_spec("2w", NOW) == (NOW - timedelta(weeks=2), NOW)
|
|
|
|
|
|
def test_parse_time_spec_iso_periods():
|
|
assert parse_time_spec("2026-09-28", NOW) == (
|
|
datetime(2026, 9, 28, tzinfo=UTC),
|
|
datetime(2026, 9, 29, tzinfo=UTC),
|
|
)
|
|
assert parse_time_spec("2026-09-28T10", NOW) == (
|
|
datetime(2026, 9, 28, 10, tzinfo=UTC),
|
|
datetime(2026, 9, 28, 11, tzinfo=UTC),
|
|
)
|
|
assert parse_time_spec("2026-09-28T10:30", NOW) == (
|
|
datetime(2026, 9, 28, 10, 30, tzinfo=UTC),
|
|
datetime(2026, 9, 28, 10, 31, tzinfo=UTC),
|
|
)
|
|
assert parse_time_spec("20260928T103045", NOW) == (
|
|
datetime(2026, 9, 28, 10, 30, 45, tzinfo=UTC),
|
|
datetime(2026, 9, 28, 10, 30, 46, tzinfo=UTC),
|
|
)
|
|
|
|
|
|
def test_parse_time_spec_iso_ranges():
|
|
# Inclusive of the end's whole period.
|
|
assert parse_time_spec("2026-09-24/2026-09-28", NOW) == (
|
|
datetime(2026, 9, 24, tzinfo=UTC),
|
|
datetime(2026, 9, 29, tzinfo=UTC),
|
|
)
|
|
# Open ends are unbounded.
|
|
assert parse_time_spec("2026-09-24/", NOW) == (
|
|
datetime(2026, 9, 24, tzinfo=UTC),
|
|
None,
|
|
)
|
|
assert parse_time_spec("/2026-09-28", NOW) == (
|
|
None,
|
|
datetime(2026, 9, 29, tzinfo=UTC),
|
|
)
|
|
with pytest.raises(ValueError, match="invalid time"):
|
|
parse_time_spec("2026-13-45", NOW)
|
|
|
|
|
|
def test_cli_time_iso_filters_items(tmp_path, capsys):
|
|
path = tmp_path / "data.kantadb"
|
|
_write_timed(
|
|
path,
|
|
[
|
|
(datetime(2026, 9, 24, 10, tzinfo=UTC), "inc0", 1),
|
|
(datetime(2026, 9, 28, 8, tzinfo=UTC), "inc1", 2),
|
|
(datetime(2026, 9, 28, 23, tzinfo=UTC), "inc2", 3),
|
|
],
|
|
)
|
|
capsys.readouterr()
|
|
|
|
# A date denotes the full UTC day; the dump follows the last printed item.
|
|
assert _run_cli(path, "-r", "2026-09-28", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert "inc0" not in out.err
|
|
assert "inc1" in out.err
|
|
assert "inc2" in out.err
|
|
assert json.loads(out.out) == {"counter": 3}
|
|
|
|
# Hour precision.
|
|
assert _run_cli(path, "-r", "2026-09-28T08") == 0
|
|
err = capsys.readouterr().err
|
|
assert "inc1" in err
|
|
assert "inc2" not in err
|
|
|
|
# Compact minute precision.
|
|
assert _run_cli(path, "-r", "20260924T1000") == 0
|
|
err = capsys.readouterr().err
|
|
assert "inc0" in err
|
|
assert "inc1" not in err
|
|
|
|
# An inclusive ISO range covers the end's whole period.
|
|
assert _run_cli(path, "-r", "2026-09-24/2026-09-28") == 0
|
|
err = capsys.readouterr().err
|
|
assert "inc0" in err and "inc2" in err
|
|
|
|
# Open-ended range.
|
|
assert _run_cli(path, "-r", "/2026-09-25") == 0
|
|
err = capsys.readouterr().err
|
|
assert "inc0" in err
|
|
assert "inc1" not in err
|
|
|
|
|
|
def test_cli_time_combines_with_plain_and_grep(tmp_path, capsys):
|
|
path = tmp_path / "data.kantadb"
|
|
_write_timed(
|
|
path,
|
|
[
|
|
(datetime(2026, 9, 24, 10, tzinfo=UTC), "old", 1),
|
|
(datetime(2026, 9, 28, 8, tzinfo=UTC), "new_a", 2),
|
|
(datetime(2026, 9, 28, 9, tzinfo=UTC), "new_b", 3),
|
|
],
|
|
)
|
|
capsys.readouterr()
|
|
|
|
# Time filter first, then the plain slice over what remains.
|
|
assert _run_cli(path, "-r", "2026-09-28", "-r", "0", "-o", "-") == 0
|
|
out = capsys.readouterr()
|
|
assert "new_a" in out.err
|
|
assert "new_b" not in out.err
|
|
assert json.loads(out.out) == {"counter": 2}
|
|
|
|
# Time and grep are both restrictive.
|
|
assert _run_cli(path, "-r", "2026-09-28", "--grep", "new_b") == 0
|
|
err = capsys.readouterr().err
|
|
assert "new_a" not in err
|
|
assert "new_b" in err
|
|
|
|
|
|
def test_cli_time_matches_items_not_windows(tmp_path, capsys):
|
|
"""Non-monotonic timestamps: items match individually, not by position."""
|
|
path = tmp_path / "data.kantadb"
|
|
_write_timed(
|
|
path,
|
|
[
|
|
(datetime(2026, 9, 28, 12, tzinfo=UTC), "noon", 1),
|
|
(datetime(2026, 9, 28, 8, tzinfo=UTC), "morning", 2),
|
|
(datetime(2026, 9, 28, 10, tzinfo=UTC), "ten", 3),
|
|
],
|
|
)
|
|
capsys.readouterr()
|
|
|
|
assert _run_cli(path, "-r", "2026-09-28T10") == 0
|
|
err = capsys.readouterr().err
|
|
assert "ten" in err
|
|
assert "noon" not in err
|
|
assert "morning" not in err
|
|
|
|
|
|
def test_cli_time_relative(tmp_path, capsys):
|
|
path = tmp_path / "data.kantadb"
|
|
now = datetime.now(UTC)
|
|
_write_timed(
|
|
path,
|
|
[
|
|
(now - timedelta(hours=2), "old", 1),
|
|
(now - timedelta(minutes=30), "recent", 2),
|
|
],
|
|
)
|
|
capsys.readouterr()
|
|
|
|
assert _run_cli(path, "-r", "1h") == 0
|
|
err = capsys.readouterr().err
|
|
assert "recent" in err
|
|
assert "old" not in err
|
|
|
|
|
|
def test_cli_time_invalid_is_error(tmp_path, capsys):
|
|
path = tmp_path / "data.kantadb"
|
|
_write_timed(path, [(T0, "inc0", 1)])
|
|
capsys.readouterr()
|
|
|
|
assert _run_cli(path, "-r", "2026-13-45") == 1
|
|
assert "Invalid --range value" in capsys.readouterr().err
|