From f8a0a8515865bf89bfa0c0a47c29a83b53a6e85e Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Thu, 13 Aug 2026 21:12:33 +0000 Subject: [PATCH] Pretty-print snapshot lines and drop microseconds from CLI timestamps. --- kanta/__main__.py | 490 +++++++++++++++++++++++++++++++++++++++++++++ kanta/replaylog.py | 390 ++++++++++++++++++++++++++++++++++++ kanta/tty.py | 1 + tests/test_cli.py | 44 ++++ 4 files changed, 925 insertions(+) create mode 100644 kanta/__main__.py create mode 100644 kanta/replaylog.py create mode 100644 tests/test_cli.py diff --git a/kanta/__main__.py b/kanta/__main__.py new file mode 100644 index 0000000..91a1c8b --- /dev/null +++ b/kanta/__main__.py @@ -0,0 +1,490 @@ +"""Module-level CLI for reading a kantadb file and printing its change log.""" + +from __future__ import annotations + +import argparse +import asyncio +import importlib +import importlib.util +import logging +import sys +import tempfile +from pathlib import Path +from typing import Any + +import msgspec + +from kanta import Kanta +from kanta.callbacks import InjectionContext +from kanta.exceptions import DatabaseError, DataIntegrityError, ReplayError +from kanta.logging import LogEvent, emit_event, migration_logger +from kanta.replaylog import ( + RangeNotFoundError, + Selection, + SnapshotEvent, + _snapshot_lines, + end_of_file, + record_change_event, + record_label, + replay_events, + scan_events, + select, +) +from kanta.serialization import Serializer, dict_to_struct, struct_to_dict +from kanta.structs import ChangeRecord, Snapshot +from kanta.tty import Line + +EXIT_SUCCESS = 0 +EXIT_GENERIC = 1 +EXIT_RANGE_ERROR = 2 +EXIT_PARSE_ERROR = 10 +EXIT_MIGRATION_ERROR = 20 +EXIT_VALIDATION_ERROR = 21 + +_logger = logging.getLogger(__name__) + + +class _CliError(Exception): + """A user-facing error message paired with a process exit code.""" + + def __init__(self, message: str, code: int = EXIT_GENERIC) -> None: + self.code = code + super().__init__(message) + + +def _import_dotted(path: str) -> Any: + """Import ``module.submodule.Attr`` and return the attribute.""" + if "." not in path: + raise ValueError(f"dotted path must contain a dot: {path!r}") + module_name, attr_name = path.rsplit(".", 1) + module = importlib.import_module(module_name) + try: + return getattr(module, attr_name) + except AttributeError as exc: + raise ImportError(f"{path!r} not found in {module_name!r}") from exc + + +def _import_kanta_object(path: str) -> Any: + """Import a Kanta object by module path. + + If ``path`` names an importable module, look up an object named + ``kanta`` in it; otherwise treat ``path`` as ``module.attr`` referring + directly to the object. + """ + try: + spec = importlib.util.find_spec(path) + except ImportError: + spec = None + if spec is not None: + module = importlib.import_module(path) + try: + return getattr(module, "kanta") + except AttributeError as exc: + raise ImportError( + f"no 'kanta' object found in module {path!r}" + ) from exc + return _import_dotted(path) + + +def _format_ts(dt) -> str: + """Return a local-looking timestamp without a timezone offset or microseconds.""" + return dt.replace(tzinfo=None, microsecond=0).isoformat(sep=" ") + + +def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + prog="kanta", + description="Read a kantadb file and print each change record to the console.", + ) + parser.add_argument( + "file", + help="Path to the kantadb file, or '-' to read from stdin.", + ) + parser.add_argument( + "-d", + "--data", + metavar="MOD", + help="Dotted path to the root data type (e.g. myapp.models.Data).", + ) + parser.add_argument( + "-m", + "--migrations", + metavar="MOD", + help="Dotted path to the migrations module (e.g. myapp.migrations).", + ) + parser.add_argument( + "-k", + "--kanta", + metavar="MOD", + help=( + "Module path to an existing Kanta object to use: either a module" + " containing an object named 'kanta' (e.g. myapp.db) or a dotted" + " path to the object itself (e.g. myapp.db.kanta). Its type," + " migrations, and logfmt/logemit callbacks are used. Cannot be" + " combined with -d or -m." + ), + ) + parser.add_argument( + "-o", + "--output", + help="Write the final replayed state as JSON to this file, or '-' for stdout.", + ) + parser.add_argument( + "-q", + "--quiet", + action="store_true", + help="Suppress normal change/snapshot logs; only print warnings and errors.", + ) + parser.add_argument( + "-r", + "--range", + help=( + "Python-style range to process. Units: plain number = change index," + " lN = line number, sN = snapshot, vN = version. Negative snapshot" + " values count from the end (s-1 is the last snapshot). Use ':' for" + " half-open ranges and '..' for inclusive end ranges. Examples:" + " '2:5', '2..5', 'l10:l20', 's1:s3', 'v0:v2', 's-1:', ':-1', '-1'." + ), + ) + args = parser.parse_args(argv) + if args.kanta and (args.data or args.migrations): + parser.error("-k/--kanta cannot be used together with -d or -m") + return args + + +def _print_change_log( + label: str, + record: ChangeRecord, + previous: dict[str, Any], + current: dict[str, Any], + kanta: Kanta[Any], +) -> None: + """Log a single change record to stderr. + + The record is dispatched as a :class:`LogEvent` through the Kanta + object's logemit handlers; the CLI's own rendering (with the ``l`` + label and timestamp) is the fallback when no handler claims the event. + """ + event = record_change_event(record, previous, current, kanta) + + def render(ev) -> None: + ts = _format_ts(record.ts) + lines = ev.diff_lines + if not lines: + print(f"{label} {ts} {ev.header}", file=sys.stderr) + elif len(lines) == 1: + print(f"{label} {ts} {ev.header}{lines[0]}", file=sys.stderr) + else: + print(f"{label} {ts} {ev.header}", file=sys.stderr) + for line in lines: + print(line, file=sys.stderr) + print(file=sys.stderr) + + emit_event( + event, + kanta._impl.callback_registry.logemit_handlers, + fallback=render, + ) + + +def _format_size(n: int) -> str: + """Return a human-readable byte size.""" + if n < 1024: + return f"{n} B" + if n < 1024 * 1024: + return f"{n / 1024:.1f} kB" + return f"{n / (1024 * 1024):.1f} MB" + + +def _print_snapshot_indicator( + label: str, + snap: Snapshot, + index: int, + serializer: Serializer, +) -> None: + """Print a snapshot indicator line to stderr. + + ``snapshot s`` is rendered in bright white; the version, optional mtime + and data size are printed in normal and dark colors respectively. + """ + ts = _format_ts(snap.ts) + line = Line().snapshot("snapshot").snapshot(f" s{index}") + line.target(f" v{snap.v}") + if snap.m is not None: + line.target(f" {_format_ts(snap.m)}") + size = len(serializer.encode(snap.state)) + line.path_prefix(f" {_format_size(size)}") + print(f"{label} {ts} {line}", file=sys.stderr) + + +async def _log_migration( + kanta: Kanta[Any], + filename: Path, + result, + previous_version: int, + quiet: bool, +) -> None: + """Log an applied migration through the Kanta instance's callbacks. + + Routes to the object's logmigr callbacks when registered (like + :meth:`KantaImpl._handle_migration_log`); otherwise emits a ``migrated`` + event through its logemit handlers, falling back to a stderr line. + """ + registry = kanta._impl.callback_registry + if registry.has("logmigr"): + try: + await registry.invoke( + "logmigr", + InjectionContext(kanta=kanta, migration_result=result), + ) + except Exception: + _logger.exception("logmigr callback failed") + return + if quiet: + return + descriptions = [ + f"{m.name} ({m.description})" for m in result.migrations if m.changed + ] + emit_event( + LogEvent( + kind="migrated", + logger=migration_logger, + kanta=kanta, + filename=str(filename), + from_version=previous_version, + to_version=result.version, + migrations=descriptions, + ), + registry.logemit_handlers, + fallback=lambda ev: print(ev.header, file=sys.stderr), + ) + + +def _get_kanta( + args: argparse.Namespace, filename: Path +) -> tuple[Kanta[Any], bool]: + """Return the Kanta instance to work with, and whether the CLI owns it. + + With ``-k`` the existing object is used as-is (and never closed by us); + otherwise an instance is constructed with an empty dict state. + """ + if args.kanta: + try: + obj = _import_kanta_object(args.kanta) + except (ImportError, ValueError) as exc: + raise _CliError(f"Invalid --kanta value: {exc}") from exc + if not isinstance(obj, Kanta): + raise _CliError( + f"Invalid --kanta value: {args.kanta!r} is not a Kanta object" + ) + return obj, False + try: + return Kanta(filename, {}, type=dict, migrations=args.migrations), True + except Exception as exc: + if args.migrations: + raise _CliError( + f"Migration error: {exc}", EXIT_MIGRATION_ERROR + ) from exc + raise _CliError(f"Failed to initialize database: {exc}") from exc + + +async def _run(args: argparse.Namespace) -> int: + cleanup_path: Path | None = None + if args.file == "-": + content = sys.stdin.buffer.read() + tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".kantadb") + tmp.write(content) + tmp.close() + filename = Path(tmp.name) + cleanup_path = filename + else: + filename = Path(args.file) + if not filename.exists(): + raise _CliError(f"File not found: {filename}") + content = filename.read_bytes() + + data_type: type[Any] | None = None + if args.data: + try: + data_type = _import_dotted(args.data) + except (ImportError, ValueError) as exc: + raise _CliError(f"Invalid --data value: {exc}") from exc + + kanta: Kanta[Any] | None = None + kanta_owned = False + kanta_typed: Kanta[Any] | None = None + try: + kanta, kanta_owned = _get_kanta(args, filename) + if data_type is None and args.kanta and kanta._impl.data_type is not dict: + data_type = kanta._impl.data_type + + # Decode and validate the whole file into positioned events. + try: + events, change_count = scan_events(content, kanta) + except ReplayError as exc: + raise _CliError(str(exc), EXIT_PARSE_ERROR) from exc + except Exception as exc: + raise _CliError( + f"Failed to replay records from {filename}: {exc}", + EXIT_PARSE_ERROR, + ) from exc + + snapshot_line_to_index = { + line: idx for idx, line in enumerate(_snapshot_lines(events)) + } + + # Resolve -r into a line range or a single snapshot selection. + try: + selection = ( + select(args.range, events, change_count) + if args.range is not None + else Selection(0, end_of_file(events)) + ) + except RangeNotFoundError as exc: + raise _CliError(str(exc), EXIT_RANGE_ERROR) from exc + except ValueError as exc: + raise _CliError(f"Invalid --range value: {exc}") from exc + + if selection.snapshot is not None: + snap_event = selection.snapshot + state = snap_event.snap.state + version = snap_event.snap.v + if not args.quiet: + _print_snapshot_indicator( + record_label(snap_event.line_number, snap_event.record_index), + snap_event.snap, + snapshot_line_to_index[snap_event.line_number], + kanta._impl.serializer, + ) + print(file=sys.stderr) + else: + # Replay up to the range end, printing logs within the range. + state = {} + version = 0 + printed = False + for event, previous, current in replay_events( + events, selection.end_line + ): + state = current + version = event.version + if event.line_number < selection.start_line or args.quiet: + continue + label = record_label(event.line_number, event.record_index) + if isinstance(event, SnapshotEvent): + _print_snapshot_indicator( + label, + event.snap, + snapshot_line_to_index[event.line_number], + kanta._impl.serializer, + ) + else: + assert previous is not None + _print_change_log(label, event.record, previous, current, kanta) + printed = True + if printed: + print(file=sys.stderr) + + # Apply optional migrations to the range-end state. + if kanta._impl.migrations is not None: + try: + previous_version = version + result = kanta._impl.migrations.apply(state, version, kanta) + version = result.version + except Exception as exc: + raise _CliError( + f"Migration error: {exc}", EXIT_MIGRATION_ERROR + ) from exc + if version != previous_version: + await _log_migration(kanta, filename, result, previous_version, args.quiet) + + output_state: dict[str, Any] + if data_type is not None: + try: + data = dict_to_struct( + state, data_type, serializer=kanta._impl.serializer + ) + except ( + msgspec.ValidationError, + msgspec.DecodeError, + TypeError, + ValueError, + ) as exc: + raise _CliError( + f"Validation error: {exc}", EXIT_VALIDATION_ERROR + ) from exc + + if args.kanta: + # The file was already fully decoded and validated above with + # the object's own serializer, and its migrations were applied + # to the state; no need to re-open through a new instance. + print(f"{data}", file=sys.stderr) + output_state = struct_to_dict( + data, serializer=kanta._impl.serializer + ) + else: + kanta_typed = Kanta( + filename, data, type=data_type, migrations=args.migrations + ) + try: + await kanta_typed.open(create=False, readonly=True, log=False) + print(f"{data}", file=sys.stderr) + except (msgspec.ValidationError, msgspec.DecodeError) as exc: + raise _CliError( + f"Validation error: {exc}", EXIT_VALIDATION_ERROR + ) from exc + except DataIntegrityError as exc: + raise _CliError( + f"Parse error: {exc}", EXIT_PARSE_ERROR + ) from exc + except DatabaseError as exc: + if not args.migrations or exc.cause_type == "ReplayError": + raise _CliError( + f"Parse error: {exc}", EXIT_PARSE_ERROR + ) from exc + raise _CliError( + f"Migration error: {exc}", EXIT_MIGRATION_ERROR + ) from exc + except Exception as exc: + if args.migrations: + raise _CliError( + f"Migration error: {exc}", EXIT_MIGRATION_ERROR + ) from exc + raise _CliError( + f"Failed to open {filename}: {exc}" + ) from exc + output_state = kanta_typed._impl.statedict + else: + output_state = state + + if args.output: + try: + out_bytes = msgspec.json.encode(output_state) + if args.output == "-": + sys.stdout.buffer.write(out_bytes) + else: + Path(args.output).write_bytes(out_bytes) + except Exception as exc: # pragma: no cover + raise _CliError(f"Failed to write output: {exc}") from exc + + return EXIT_SUCCESS + finally: + if kanta_typed is not None: + await kanta_typed.close() + if kanta is not None and kanta_owned: + await kanta.close() + if cleanup_path is not None: + cleanup_path.unlink(missing_ok=True) + + +def main(argv: list[str] | None = None) -> int: + """Entry point for ``python -m kanta``.""" + args = _parse_args(argv) + try: + return asyncio.run(_run(args)) + except _CliError as exc: + print(exc, file=sys.stderr) + return exc.code + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/kanta/replaylog.py b/kanta/replaylog.py new file mode 100644 index 0000000..f311307 --- /dev/null +++ b/kanta/replaylog.py @@ -0,0 +1,390 @@ +"""Line-oriented replay and range selection for kantadb files. + +Support machinery for the ``python -m kanta`` CLI: decoding a file into +positioned events, resolving ``-r`` range specifications to line numbers, +replaying state over a line range, and building change log events. Internal +for now; not part of the public API. +""" + +from __future__ import annotations + +import copy +import dataclasses +from collections.abc import Iterator +from typing import TYPE_CHECKING, Any, Union + +import msgspec + +from kanta.callbacks import InjectionContext +from kanta.diff import patch_state +from kanta.exceptions import ReplayError +from kanta.logging import _USER_PATH, LogEvent, transaction_logger +from kanta.structs import ChangeRecord, Snapshot + +if TYPE_CHECKING: + from kanta import Kanta + + +@dataclasses.dataclass +class SnapshotEvent: + """A snapshot record positioned in the file.""" + + line_number: int + byte_pos: int + record_index: int + snap: Snapshot + + @property + def version(self) -> int: + return self.snap.v + + +@dataclasses.dataclass +class ChangeEvent: + """A change record positioned in the file.""" + + line_number: int + byte_pos: int + record_index: int + record: ChangeRecord + + @property + def version(self) -> int: + return self.record.v + + +Event = Union[SnapshotEvent, ChangeEvent] + + +class RangeNotFoundError(Exception): + """A single-item range specification that does not exist in the file. + + The message is fully formatted for display, including the offending + input and how many items of that kind the file contains. + """ + + +@dataclasses.dataclass +class Selection: + """A resolved range specification. + + Either a ``[start_line, end_line)`` line range, or a single snapshot + (``snapshot`` set), used to show the snapshot state without replaying + further records. + """ + + start_line: int + end_line: int + snapshot: SnapshotEvent | None = None + + +def record_label(line_number: int, record_index: int) -> str: + """Return a padded record label based on line number, falling back to record index.""" + number = line_number if line_number else record_index + return f"l{number:<3}" + + +def scan_events(content: bytes, kanta: Kanta[Any]) -> tuple[list[Event], int]: + """Decode all records, validating snapshot consistency. + + Uses the Kanta instance's serializer and framer. Returns the events in + file order and the number of change records. Raises :class:`ReplayError` + with a located, display-ready message on decode failures or when a + snapshot does not match the replayed state. + """ + impl = kanta._impl + state: dict[str, Any] = {} + events: list[Event] = [] + change_count = 0 + for is_snapshot, payload, line_number, byte_pos in impl.framer.iter_records( + content, 0 + ): + record_index = len(events) + 1 + label = record_label(line_number, record_index) + try: + if is_snapshot: + snap = impl.serializer.decode(payload, type=Snapshot) + if record_index > 1 and state != snap.state: + raise ReplayError( + f"Snapshot mismatch at {label}: replayed state" + " does not equal the snapshot state.", + line_number=line_number, + byte_pos=byte_pos, + record_type="snapshot", + ) + state = snap.state + events.append( + SnapshotEvent(line_number, byte_pos, record_index, snap) + ) + else: + record = impl.serializer.decode(payload, type=ChangeRecord) + state = patch_state(state, record.diff) + events.append( + ChangeEvent(line_number, byte_pos, record_index, record) + ) + change_count += 1 + except msgspec.DecodeError as exc: + raise ReplayError( + f"Parse error at {label}: {exc}", + line_number=line_number, + byte_pos=byte_pos, + ) from exc + return events, change_count + + +def replay_events( + events: list[Event], end_line: int +) -> Iterator[tuple[Event, dict[str, Any] | None, dict[str, Any]]]: + """Replay events with line numbers below ``end_line``. + + Yields ``(event, previous, state)`` per event: ``previous`` is the state + before a change (``None`` for snapshots) and ``state`` the state after + the event. + """ + state: dict[str, Any] = {} + for event in events: + if event.line_number >= end_line: + break + if isinstance(event, SnapshotEvent): + state = event.snap.state + yield event, None, state + else: + previous = copy.deepcopy(state) + state = patch_state(state, event.record.diff) + yield event, previous, state + + +def record_change_event( + record: ChangeRecord, + previous: dict[str, Any], + current: dict[str, Any], + kanta: Kanta[Any], +) -> LogEvent: + """Build a change :class:`LogEvent` for a replayed record. + + The Kanta instance's logfmt callbacks are used for value formatting and + for resolving the user/actor name; the event can then be dispatched with + :func:`kanta.logging.emit_event` and the instance's logemit handlers. + """ + registry = kanta._impl.callback_registry + logfmt = registry.build_logfmt( + InjectionContext( + kanta=kanta, + previous_state=previous, + current_state=current, + ) + ) + user = record.u + if user is not None: + resolved = logfmt(user, _USER_PATH) + if resolved is not None: + user = resolved + return LogEvent( + kind="change", + logger=transaction_logger, + kanta=kanta, + action=record.a, + user=user, + diff=record.diff, + previous=previous, + logfmt=logfmt, + ) + + +def _plural(count: int, word: str) -> str: + """Return e.g. ``1 snapshot`` or ``2 snapshots``.""" + return f"{count} {word}{'' if count == 1 else 's'}" + + +def end_of_file(events: list[Event]) -> int: + """Return the sentinel line number just past the last line of the file.""" + return events[-1].line_number + 1 if events else 0 + + +def _change_lines(events: list[Event]) -> list[int]: + """Return the line numbers of all change records, in file order.""" + return [e.line_number for e in events if isinstance(e, ChangeEvent)] + + +def _snapshot_lines(events: list[Event]) -> list[int]: + """Return the line numbers addressed by s0, s1, ... + + If the file begins with a snapshot, s0 is that snapshot (l1) and s1 is + the next snapshot. Otherwise the file begins with change records (empty + initial state): s0 is l0, the position before the start of the file, and + s1 is the first snapshot. + """ + lines = [e.line_number for e in events if isinstance(e, SnapshotEvent)] + if events and isinstance(events[0], SnapshotEvent): + return lines + return [0, *lines] + + +def _version_lines(events: list[Event]) -> dict[int, int]: + """Map each version to the line where it first appears; v0 is l0.""" + lines: dict[int, int] = {0: 0} + for event in events: + lines.setdefault(event.version, event.line_number) + return lines + + +def _event_at_line(events: list[Event], line: int) -> Event | None: + """Return the event whose file line number exactly matches ``line``.""" + for event in events: + if event.line_number == line: + return event + return None + + +def _parse_bound(bound_str: str) -> tuple[str, int | None]: + """Parse a range bound with optional unit prefix (l, s, v) or change index.""" + if not bound_str: + return "change", None + unit_map = {"l": "line", "s": "snapshot", "v": "version"} + if bound_str[0] in unit_map: + unit = unit_map[bound_str[0]] + rest = bound_str[1:] + if not rest: + raise ValueError(f"empty value in {bound_str!r}") + return unit, int(rest) + return "change", int(bound_str) + + +def _bound_to_line( + unit: str, + value: int | None, + events: list[Event], + total: int, + is_start: bool, +) -> int: + """Convert a range bound to a line number. + + Out-of-range values are truncated to l0 (before the first line) or to + the line just past the end of the file rather than erroring; a missing + bound means the corresponding file end. + """ + eof = end_of_file(events) + if value is None: + return 0 if is_start else eof + + if unit == "change": + lines = _change_lines(events) + if value < 0: + value = total + value + value = max(0, min(value, total)) + return lines[value] if value < total else eof + + if unit == "line": + if value < 0: + raise ValueError("line numbers do not support negative indexing") + return value + + if unit == "snapshot": + lines = _snapshot_lines(events) + idx = len(lines) + value if value < 0 else value + if idx < 0: + return 0 + return lines[idx] if idx < len(lines) else eof + + if unit == "version": + if value < 0: + raise ValueError("version numbers do not support negative indexing") + return _version_lines(events).get(value, eof) + + raise ValueError(f"unknown range unit: {unit}") + + +def _resolve_range( + range_str: str, events: list[Event], total: int +) -> tuple[int, int]: + """Parse a range string into a [start_line, end_line) line range.""" + sep = ".." if ".." in range_str else ":" + start_str, end_str = range_str.split(sep, 1) + start_unit, start_val = _parse_bound(start_str) + end_unit, end_val = _parse_bound(end_str) + start_line = _bound_to_line( + start_unit, start_val, events, total, is_start=True + ) + end_line = _bound_to_line(end_unit, end_val, events, total, is_start=False) + # ``..`` makes the end bound inclusive. + if sep == ".." and end_val is not None: + end_line += 1 + return min(start_line, end_line), end_line + + +def _negative_check(unit: str, value: int) -> None: + if value < 0: + raise ValueError(f"{unit} numbers do not support negative indexing") + + +def select(spec: str, events: list[Event], total: int) -> Selection: + """Resolve a range specification against the scanned events. + + ``total`` is the number of change records. Returns a :class:`Selection`: + a line range, or a single snapshot for snapshot selections (``sN``, or + ``lN`` pointing at a snapshot). Ranges truncate out-of-bounds values; + a single index must exist and raises :class:`RangeNotFoundError` + otherwise. Syntax errors raise :class:`ValueError`. + """ + if ":" in spec or ".." in spec: + start_line, end_line = _resolve_range(spec, events, total) + return Selection(start_line, end_line) + + # A single index must exist; out-of-bounds is an error. + unit, value = _parse_bound(spec) + if value is None: + raise ValueError("single bound must not be empty") + + if unit == "snapshot": + lines = _snapshot_lines(events) + idx = len(lines) + value if value < 0 else value + n_snapshots = sum(isinstance(e, SnapshotEvent) for e in events) + count = _plural(n_snapshots, "snapshot") + if not 0 <= idx < len(lines): + raise RangeNotFoundError( + f"Snapshot {spec!r} not found in file ({count})" + ) + event = _event_at_line(events, lines[idx]) + if event is None: + # s0 with an empty initial state (l0): not a real record, so it + # cannot be selected as a single item. + raise RangeNotFoundError( + f"Snapshot {spec!r} not found in file: the file starts" + f" with an empty initial state ({count})" + ) + assert isinstance(event, SnapshotEvent) + return Selection(event.line_number, event.line_number + 1, event) + + if unit == "line": + _negative_check(unit, value) + event = _event_at_line(events, value) + if event is None: + n_lines = events[-1].line_number if events else 0 + raise RangeNotFoundError( + f"Line {spec!r} not found in file ({_plural(n_lines, 'line')})" + ) + if isinstance(event, SnapshotEvent): + return Selection(value, value + 1, event) + return Selection(value, value + 1) + + if unit == "version": + _negative_check(unit, value) + lines = _version_lines(events) + if value not in lines: + raise RangeNotFoundError( + f"Version {spec!r} not found in file" + f" ({_plural(len(lines), 'version')})" + ) + start_line = lines[value] + later = [line for line in lines.values() if line > start_line] + return Selection(start_line, min(later) if later else end_of_file(events)) + + lines = _change_lines(events) + idx = total + value if value < 0 else value + if not 0 <= idx < total: + raise RangeNotFoundError( + f"Change index {spec!r} not found in file" + f" ({_plural(total, 'change')})" + ) + end_line = lines[idx + 1] if idx + 1 < total else end_of_file(events) + return Selection(lines[idx], end_line) diff --git a/kanta/tty.py b/kanta/tty.py index d32db67..560de50 100644 --- a/kanta/tty.py +++ b/kanta/tty.py @@ -70,6 +70,7 @@ class Colors: action = "1;34" # Bold blue for the action name user = "34" # Blue for the user display target = "38;5;250" # White for the extra/target display + snapshot = "97" # Bright white for snapshot indicator text sep = "38;5;242" # Dark grey for separators path_prefix = "38;5;242" # Dark grey for the leading part of a dotted path path_final = "38;5;250" # White for the final path element diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..d5daff9 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,44 @@ +"""Tests for the ``python -m kanta`` CLI output formatting.""" + +from datetime import UTC, datetime + +from kanta.__main__ import _format_ts, main +from kanta.serialization import JsonSerializer +from kanta.serialization.framing import LineFramer +from kanta.structs import ChangeRecord, Snapshot + + +def test_format_ts_strips_microseconds(): + """Timestamps are rendered without microsecond precision.""" + dt = datetime(2026, 8, 12, 10, 6, 52, 375398, tzinfo=UTC) + assert _format_ts(dt) == "2026-08-12 10:06:52" + + +def test_cli_snapshot_line_format(tmp_path, capsys): + """Snapshot lines are timestamped and colored with metadata.""" + path = tmp_path / "test.kantadb" + ts = datetime(2026, 8, 12, 10, 6, 52, 375398, tzinfo=UTC) + mtime = datetime(2026, 8, 12, 9, 0, 0, tzinfo=UTC) + serializer = JsonSerializer() + framer = LineFramer() + + snapshot = Snapshot(ts=ts, v=1, m=mtime, state={"counter": 5}) + change = ChangeRecord(ts=ts, a="inc", v=1, u="user1", diff={"counter": 6}) + data = ( + framer.frame_snapshot(serializer.encode(snapshot), record_offset=0) + + framer.frame_change(serializer.encode(change), record_offset=0) + ) + path.write_bytes(data) + + code = main([str(path)]) + assert code == 0 + + err = capsys.readouterr().err + # No microsecond precision anywhere. + assert "10:06:52" in err + assert "10:06:52.375398" not in err + + # Snapshot line: bright white snapshot/sN, white version/mtime, dark size. + assert "\x1b[97msnapshot s0" in err + assert "\x1b[38;5;250m v1 2026-08-12 09:00:00" in err + assert "\x1b[38;5;242m 13 B" in err