Implement kanta --grep (intelligent search string) with highlight marks.
This commit is contained in:
+69
-8
@@ -19,7 +19,14 @@ 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.grep import GrepPattern, evaluate
|
||||
from kanta.logging import (
|
||||
LogEvent,
|
||||
emit_event,
|
||||
format_action_header,
|
||||
format_diff,
|
||||
migration_logger,
|
||||
)
|
||||
from kanta.replaylog import (
|
||||
RangeNotFoundError,
|
||||
Selection,
|
||||
@@ -239,6 +246,31 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
" '2:5', '2..5', 'l10:l20', 's1:s3', 'v0:v2', 's-1:', ':-1', '-1'."
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
"-g",
|
||||
"--grep",
|
||||
action="append",
|
||||
metavar="PATTERN",
|
||||
help=(
|
||||
"Print only change records matching PATTERN, structurally and"
|
||||
" case-insensitively; matched regions get a yellow background,"
|
||||
" and on a match the whole record is printed, not just the"
|
||||
" matching line. Repeatable: every pattern must match somewhere"
|
||||
" in the same record, but different patterns may match different"
|
||||
" lines of it. A bare pattern matches the action or user as a"
|
||||
" substring, a dotted path by element (each element in full,"
|
||||
" unless it uses wildcards: 'users' matches 'users' anywhere but"
|
||||
" not 'foousers', 'us*' does), or a value (strings by substring,"
|
||||
" other values in full: 'true' matches a boolean, 'tru' does"
|
||||
" not). The 'path=value' form requires the path and the value"
|
||||
" to match within the same change line; use '=value' or 'path='"
|
||||
" to match values or paths only. With -k, logfmt-prettified"
|
||||
" values and users match alongside the raw ones. Examples:"
|
||||
" --grep alice, --grep 'users.*.email', --grep"
|
||||
" 'users.alice.admin=true', --grep create_user --grep"
|
||||
" '@example.com'."
|
||||
),
|
||||
)
|
||||
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")
|
||||
@@ -251,24 +283,33 @@ def _print_change_log(
|
||||
previous: dict[str, Any],
|
||||
current: dict[str, Any],
|
||||
kanta: Kanta[Any],
|
||||
highlight: Any = None,
|
||||
) -> 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<N>``
|
||||
label and timestamp) is the fallback when no handler claims the event.
|
||||
``highlight`` is an optional :class:`kanta.grep.GrepHighlighter` with
|
||||
the record's matched regions, applied to the fallback rendering.
|
||||
"""
|
||||
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}")
|
||||
elif len(lines) == 1:
|
||||
_print(f"{label} {ts} {ev.header}{lines[0]}")
|
||||
if highlight is not None:
|
||||
header = format_action_header(
|
||||
ev.action or "", ev.user, ev.extra, highlight=highlight
|
||||
)
|
||||
lines = format_diff(ev.diff, ev.previous, ev.logfmt, highlight=highlight)
|
||||
else:
|
||||
_print(f"{label} {ts} {ev.header}")
|
||||
header, lines = ev.header, ev.diff_lines
|
||||
if not lines:
|
||||
_print(f"{label} {ts} {header}")
|
||||
elif len(lines) == 1:
|
||||
_print(f"{label} {ts} {header}{lines[0]}")
|
||||
else:
|
||||
_print(f"{label} {ts} {header}")
|
||||
for line in lines:
|
||||
_print(line)
|
||||
_print()
|
||||
@@ -476,6 +517,8 @@ async def _run(args: argparse.Namespace) -> int:
|
||||
except ValueError as exc:
|
||||
raise _CliError(f"Invalid --range value: {exc}") from exc
|
||||
|
||||
grep_patterns = [GrepPattern.parse(p) for p in args.grep or ()]
|
||||
|
||||
if selection.snapshot is not None:
|
||||
snap_event = selection.snapshot
|
||||
state = snap_event.snap.state
|
||||
@@ -508,7 +551,25 @@ async def _run(args: argparse.Namespace) -> int:
|
||||
)
|
||||
else:
|
||||
assert previous is not None
|
||||
_print_change_log(label, event.record, previous, current, kanta)
|
||||
highlight = None
|
||||
if grep_patterns:
|
||||
# Build the same logfmt the rendering uses, so both
|
||||
# raw and prettified values are matched.
|
||||
logfmt = kanta._impl.callback_registry.build_logfmt(
|
||||
InjectionContext(
|
||||
kanta=kanta,
|
||||
previous_state=previous,
|
||||
current_state=current,
|
||||
)
|
||||
)
|
||||
highlight = evaluate(
|
||||
event.record, previous, grep_patterns, logfmt=logfmt
|
||||
)
|
||||
if highlight is None:
|
||||
continue
|
||||
_print_change_log(
|
||||
label, event.record, previous, current, kanta, highlight
|
||||
)
|
||||
printed = True
|
||||
if printed:
|
||||
_print()
|
||||
|
||||
+433
@@ -0,0 +1,433 @@
|
||||
"""Structural grep matching and match highlighting for change records.
|
||||
|
||||
Backs the ``--grep`` option of the ``python -m kanta`` CLI. Patterns
|
||||
match against the structure of a transaction — its action, user, and the
|
||||
dotted paths and values of its diff — never against rendered output
|
||||
text. Matching is case-insensitive.
|
||||
|
||||
Path terms match element-wise: each dotted element of the term must match
|
||||
a whole path element (``users`` matches ``users`` anywhere in the path
|
||||
but not ``foousers``), unless the element uses shell wildcards
|
||||
(``us*``). The term's elements match as a contiguous sequence, so
|
||||
``users.*.email`` matches the path ``users.alice.email``. Only the
|
||||
matched elements are highlighted.
|
||||
|
||||
Value terms match strings by substring and all other values (booleans,
|
||||
numbers, null) only in full: ``true`` matches a boolean but ``tru`` does
|
||||
not. A term with wildcards matches the whole value text. Matched
|
||||
substrings — or the whole scalar — are highlighted. When a logfmt
|
||||
formatter (``-k``) prettifies a value or the user, both the raw and the
|
||||
prettified form are matched.
|
||||
|
||||
A matched record carries a :class:`GrepHighlighter`, which the
|
||||
:kanta.logging formatters consult to wrap exactly the matched regions
|
||||
with a yellow background.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import fnmatch
|
||||
import json
|
||||
from collections.abc import Iterator
|
||||
from typing import Any
|
||||
|
||||
from kanta.logging import _USER_PATH, _collect_changes, _get_nested
|
||||
from kanta.serialization.base import unmarshal
|
||||
from kanta.structs import ChangeRecord
|
||||
from kanta.tty import ANSI_RE, ESC, strip_ansi
|
||||
|
||||
_GLOB_CHARS = frozenset("*?[")
|
||||
|
||||
_MARK_BG = f"{ESC}48;5;220m" # yellow background (xterm256 #ffd700) for matches
|
||||
_UNMARK_BG = f"{ESC}49m" # back to the default background, foreground untouched
|
||||
|
||||
|
||||
def mark_spans(styled: str, spans: list[tuple[int, int]]) -> str:
|
||||
"""Wrap the given visible-text spans of a styled string with the mark color.
|
||||
|
||||
``spans`` are ``(start, end)`` offsets into the visible text of
|
||||
*styled*; ANSI sequences are not counted. Overlapping and adjacent
|
||||
spans are merged first, so overlapping matches from different patterns
|
||||
produce one continuous highlight. The set/clear codes are inserted at
|
||||
the mapped positions in *styled*; only the background attribute is
|
||||
touched, leaving foreground colors intact.
|
||||
"""
|
||||
spans = _merge_spans(spans)
|
||||
if not spans:
|
||||
return styled
|
||||
out: list[str] = []
|
||||
plain_pos = 0
|
||||
prev_end = 0
|
||||
for match in ANSI_RE.finditer(styled):
|
||||
out.append(_wrap_run(styled[prev_end : match.start()], plain_pos, spans))
|
||||
plain_pos += match.start() - prev_end
|
||||
out.append(match.group(0))
|
||||
prev_end = match.end()
|
||||
out.append(_wrap_run(styled[prev_end:], plain_pos, spans))
|
||||
return "".join(out)
|
||||
|
||||
|
||||
def _merge_spans(spans: list[tuple[int, int]]) -> list[tuple[int, int]]:
|
||||
"""Return *spans* sorted, with overlapping and adjacent spans merged."""
|
||||
merged: list[list[int]] = []
|
||||
for start, end in sorted(spans):
|
||||
if start >= end:
|
||||
continue
|
||||
if merged and start <= merged[-1][1]:
|
||||
merged[-1][1] = max(end, merged[-1][1])
|
||||
else:
|
||||
merged.append([start, end])
|
||||
return [(start, end) for start, end in merged]
|
||||
|
||||
|
||||
def _wrap_run(run: str, plain_start: int, spans: list[tuple[int, int]]) -> str:
|
||||
"""Wrap the intersections of *spans* with one escape-free text run."""
|
||||
if not run:
|
||||
return run
|
||||
out: list[str] = []
|
||||
pos = 0
|
||||
for start, end in spans:
|
||||
s = max(start - plain_start, 0)
|
||||
e = min(end - plain_start, len(run))
|
||||
if s >= e or e <= pos:
|
||||
continue
|
||||
out.append(run[pos:s])
|
||||
out.append(f"{_MARK_BG}{run[s:e]}{_UNMARK_BG}")
|
||||
pos = e
|
||||
out.append(run[pos:])
|
||||
return "".join(out)
|
||||
|
||||
|
||||
def _find_spans(text: str, needle: str) -> list[tuple[int, int]]:
|
||||
"""Return visible-text spans of every case-insensitive occurrence of *needle*."""
|
||||
if not needle:
|
||||
return []
|
||||
haystack = strip_ansi(text).lower()
|
||||
needle = needle.lower()
|
||||
spans = []
|
||||
pos = 0
|
||||
while (found := haystack.find(needle, pos)) >= 0:
|
||||
end = found + len(needle)
|
||||
spans.append((found, end))
|
||||
pos = end
|
||||
return spans
|
||||
|
||||
|
||||
@dataclasses.dataclass(frozen=True)
|
||||
class GrepPattern:
|
||||
"""One parsed ``--grep`` pattern.
|
||||
|
||||
The bare form (``term`` set) matches the action, the user, or any
|
||||
dotted path or value in the diff. The ``path=value`` form
|
||||
(``path_term`` and ``value_term`` set) requires both sides to match
|
||||
within the same change line; either side may be left empty to match
|
||||
values only (``=value``) or paths only (``path=``).
|
||||
"""
|
||||
|
||||
raw: str
|
||||
term: str | None = None
|
||||
path_term: str | None = None
|
||||
value_term: str | None = None
|
||||
|
||||
@classmethod
|
||||
def parse(cls, raw: str) -> GrepPattern:
|
||||
"""Parse a pattern, splitting the ``path=value`` form on the first ``=``."""
|
||||
if "=" in raw:
|
||||
path_term, value_term = raw.split("=", 1)
|
||||
return cls(raw, path_term=path_term, value_term=value_term)
|
||||
return cls(raw, term=raw)
|
||||
|
||||
|
||||
@dataclasses.dataclass(frozen=True)
|
||||
class _ValueMark:
|
||||
"""A value-side match.
|
||||
|
||||
``needle`` is the text to locate in the displayed value ("" = a match
|
||||
that marks nothing, e.g. from an empty term). ``whole`` marks the
|
||||
entire displayed value: used when the raw value matched but a logfmt
|
||||
formatter displays something else, so no needle can be located.
|
||||
"""
|
||||
|
||||
needle: str = ""
|
||||
whole: bool = False
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class _Entry:
|
||||
"""One matchable ``(path, value)`` line of a record's flattened diff.
|
||||
|
||||
``anchor`` is set for deleted content: the deleted path whose line is
|
||||
displayed for this entry (the entry itself may sit below it).
|
||||
"""
|
||||
|
||||
path: list[str]
|
||||
raw: Any
|
||||
text: str
|
||||
anchor: list[str] | None = None
|
||||
|
||||
|
||||
def _element_matches(pattern: str, element: str) -> bool:
|
||||
"""Match one path element: in full, or as a glob when it uses wildcards."""
|
||||
pattern = pattern.casefold()
|
||||
element = element.casefold()
|
||||
if any(char in pattern for char in _GLOB_CHARS):
|
||||
return fnmatch.fnmatchcase(element, pattern)
|
||||
return element == pattern
|
||||
|
||||
|
||||
def _match_path(term: str, path: list[str]) -> frozenset[int] | None:
|
||||
"""Match a dotted term against *path* as a contiguous element sequence.
|
||||
|
||||
Returns the indices of the matched elements, or ``None``. An empty
|
||||
term matches anything and marks no elements.
|
||||
"""
|
||||
if not term:
|
||||
return frozenset()
|
||||
patterns = term.split(".")
|
||||
for start in range(len(path) - len(patterns) + 1):
|
||||
if all(
|
||||
_element_matches(pattern, path[start + i])
|
||||
for i, pattern in enumerate(patterns)
|
||||
):
|
||||
return frozenset(range(start, start + len(patterns)))
|
||||
return None
|
||||
|
||||
|
||||
def _match_value(term: str, raw: Any, text: str) -> _ValueMark | None:
|
||||
"""Match a term against a value.
|
||||
|
||||
String values match by substring; containers do not match (their
|
||||
leaves are matched individually); all other values match only in
|
||||
full. A term with wildcards matches the whole value text.
|
||||
"""
|
||||
if not term:
|
||||
return _ValueMark()
|
||||
needle = term.casefold()
|
||||
haystack = text.casefold()
|
||||
if any(char in needle for char in _GLOB_CHARS):
|
||||
return _ValueMark(text) if fnmatch.fnmatchcase(haystack, needle) else None
|
||||
if isinstance(raw, str):
|
||||
return _ValueMark(term) if needle in haystack else None
|
||||
if isinstance(raw, (dict, list)):
|
||||
return None
|
||||
return _ValueMark(text) if needle == haystack else None
|
||||
|
||||
|
||||
def _dual_mark(term: str, raw: Any, text: str, pretty: str | None) -> _ValueMark | None:
|
||||
"""Match a term against both the raw and the prettified form of a value.
|
||||
|
||||
*pretty* is the logfmt-resolved display text, which is what the log
|
||||
shows when set. A match on the displayed form locates its needle
|
||||
there; a match on the raw form alone marks the whole displayed value.
|
||||
"""
|
||||
raw_mark = _match_value(term, raw, text)
|
||||
if pretty is None:
|
||||
return raw_mark
|
||||
pretty_mark = _match_value(term, pretty, pretty)
|
||||
if pretty_mark is not None:
|
||||
return pretty_mark
|
||||
if raw_mark is not None and (raw_mark.needle or raw_mark.whole):
|
||||
return _ValueMark(whole=True)
|
||||
return raw_mark
|
||||
|
||||
|
||||
def _value_text(value: Any) -> str:
|
||||
"""Render a value as matchable text, following the display conventions."""
|
||||
if value is None:
|
||||
return "null"
|
||||
if isinstance(value, bool):
|
||||
return "true" if value else "false"
|
||||
if isinstance(value, (int, float)):
|
||||
return str(value)
|
||||
if isinstance(value, str):
|
||||
return value
|
||||
if isinstance(value, (dict, list)):
|
||||
try:
|
||||
return json.dumps(value, default=str)
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
return str(value)
|
||||
|
||||
|
||||
def _leaf_entries(
|
||||
path: list[str], value: Any, anchor: list[str] | None
|
||||
) -> Iterator[_Entry]:
|
||||
"""Yield an entry for every value under *value*.
|
||||
|
||||
Added, replaced or deleted containers are a single change line in the
|
||||
log but hold many values; descending into them lets patterns match
|
||||
their content. List elements are addressed by index (``users.0``).
|
||||
"""
|
||||
items: Iterator[tuple[str, Any]]
|
||||
if isinstance(value, dict):
|
||||
items = ((str(key), item) for key, item in value.items())
|
||||
elif isinstance(value, list):
|
||||
items = ((str(index), item) for index, item in enumerate(value))
|
||||
else:
|
||||
return
|
||||
for key, item in items:
|
||||
item_path = [*path, key]
|
||||
yield _Entry(item_path, item, _value_text(item), anchor)
|
||||
yield from _leaf_entries(item_path, item, anchor)
|
||||
|
||||
|
||||
def _record_entries(record: ChangeRecord, previous: dict | None) -> list[_Entry]:
|
||||
"""Flatten a record's diff into matchable entries.
|
||||
|
||||
Uses the same traversal as the change-log rendering, so paths match
|
||||
what the log shows; deleted paths carry their previous value.
|
||||
"""
|
||||
changes: list[tuple[str, list[str], Any]] = []
|
||||
_collect_changes(unmarshal(record.diff), [], changes, previous)
|
||||
entries: list[_Entry] = []
|
||||
for change_type, path, value in changes:
|
||||
anchor = None
|
||||
if change_type == "delete":
|
||||
value = _get_nested(previous, path)
|
||||
anchor = path
|
||||
entries.append(_Entry(path, value, _value_text(value), anchor))
|
||||
entries.extend(_leaf_entries(path, value, anchor))
|
||||
return entries
|
||||
|
||||
|
||||
class GrepHighlighter:
|
||||
"""The matched regions of one record, wrapping rendered text on demand.
|
||||
|
||||
Implements the highlighter hook of the :mod:`kanta.logging`
|
||||
formatters: ``path`` for path elements, ``value`` for values,
|
||||
``meta`` for header fields and ``delete`` for deletion markers.
|
||||
All wrapping goes through :func:`mark_spans`, so overlapping matches
|
||||
merge into one highlight.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._lit_paths: set[str] = set()
|
||||
self._lit_deletes: set[str] = set()
|
||||
self._value_marks: dict[str, list[_ValueMark]] = {}
|
||||
self._meta_marks: dict[str, list[_ValueMark]] = {}
|
||||
|
||||
def _light_elements(self, path: list[str], indices) -> None:
|
||||
for i in indices:
|
||||
self._lit_paths.add(".".join(path[: i + 1]))
|
||||
|
||||
def add(
|
||||
self,
|
||||
entry: _Entry,
|
||||
elements: frozenset[int] | None,
|
||||
vmark: _ValueMark | None,
|
||||
) -> None:
|
||||
"""Record one entry's match: lit element indices and/or a value mark."""
|
||||
marked = vmark is not None and (vmark.needle or vmark.whole)
|
||||
if entry.anchor is None:
|
||||
if elements:
|
||||
self._light_elements(entry.path, elements)
|
||||
if marked:
|
||||
key = ".".join(entry.path)
|
||||
self._value_marks.setdefault(key, []).append(vmark)
|
||||
return
|
||||
# Deleted content: only the anchor path line is displayed. Light
|
||||
# the genuinely matched elements within it; a match on the removed
|
||||
# value or below the anchor marks the deletion marker (✗) instead.
|
||||
if elements:
|
||||
shown = {i for i in elements if i < len(entry.anchor)}
|
||||
if shown:
|
||||
self._light_elements(entry.path, shown)
|
||||
if len(shown) != len(elements):
|
||||
self._lit_deletes.add(".".join(entry.anchor))
|
||||
if marked:
|
||||
self._lit_deletes.add(".".join(entry.anchor))
|
||||
|
||||
def add_meta(self, field: str, mark: _ValueMark) -> None:
|
||||
"""Record a header match on ``field`` (``"action"`` or ``"user"``)."""
|
||||
if mark.needle or mark.whole:
|
||||
self._meta_marks.setdefault(field, []).append(mark)
|
||||
|
||||
def path(self, text: str, path: str) -> str:
|
||||
"""Wrap a rendered path element when its element matched."""
|
||||
if text and path in self._lit_paths:
|
||||
return mark_spans(text, [(0, len(strip_ansi(text)))])
|
||||
return text
|
||||
|
||||
def delete(self, text: str, path: str) -> str:
|
||||
"""Wrap the deletion marker when the removed content matched."""
|
||||
if text and path in self._lit_deletes:
|
||||
return mark_spans(text, [(0, len(strip_ansi(text)))])
|
||||
return text
|
||||
|
||||
@staticmethod
|
||||
def _apply_marks(text: str, marks: list[_ValueMark]) -> str:
|
||||
if not text or not marks:
|
||||
return text
|
||||
spans: list[tuple[int, int]] = []
|
||||
for mark in marks:
|
||||
if mark.whole:
|
||||
spans.append((0, len(strip_ansi(text))))
|
||||
else:
|
||||
spans.extend(_find_spans(text, mark.needle))
|
||||
return mark_spans(text, spans)
|
||||
|
||||
def value(self, text: str, path: str) -> str:
|
||||
"""Wrap the matched regions of a rendered value."""
|
||||
return self._apply_marks(text, self._value_marks.get(path, []))
|
||||
|
||||
def meta(self, text: str, field: str) -> str:
|
||||
"""Wrap the matched regions of a rendered header field."""
|
||||
return self._apply_marks(text, self._meta_marks.get(field, []))
|
||||
|
||||
|
||||
def evaluate(
|
||||
record: ChangeRecord,
|
||||
previous: dict | None,
|
||||
patterns: list[GrepPattern],
|
||||
logfmt: Any = None,
|
||||
) -> GrepHighlighter | None:
|
||||
"""Match *patterns* against a record, returning its matched regions.
|
||||
|
||||
Returns ``None`` when any pattern matches nowhere in the transaction.
|
||||
Otherwise every pattern contributed its matches — action, user, or
|
||||
change lines — to the returned highlighter; different patterns may
|
||||
match different lines of the same record.
|
||||
|
||||
``logfmt`` is the optional composed logfmt callable; when given, both
|
||||
the raw and the prettified form of each value (and of the user) are
|
||||
matched.
|
||||
"""
|
||||
entries = _record_entries(record, previous)
|
||||
highlighter = GrepHighlighter()
|
||||
for pattern in patterns:
|
||||
matched = False
|
||||
if pattern.term is not None:
|
||||
mark = _match_value(pattern.term, record.a, record.a)
|
||||
if mark is not None:
|
||||
highlighter.add_meta("action", mark)
|
||||
matched = True
|
||||
if record.u:
|
||||
pretty_user = (
|
||||
logfmt(record.u, _USER_PATH) if logfmt is not None else None
|
||||
)
|
||||
mark = _dual_mark(pattern.term, record.u, record.u, pretty_user)
|
||||
if mark is not None:
|
||||
highlighter.add_meta("user", mark)
|
||||
matched = True
|
||||
for entry in entries:
|
||||
pretty = None
|
||||
if logfmt is not None:
|
||||
pretty = logfmt(entry.raw, ".".join(entry.path))
|
||||
if pattern.term is not None:
|
||||
elements = _match_path(pattern.term, entry.path)
|
||||
vmark = _dual_mark(pattern.term, entry.raw, entry.text, pretty)
|
||||
if elements is None and vmark is None:
|
||||
continue
|
||||
else:
|
||||
elements = _match_path(pattern.path_term or "", entry.path)
|
||||
vmark = _dual_mark(
|
||||
pattern.value_term or "", entry.raw, entry.text, pretty
|
||||
)
|
||||
if elements is None or vmark is None:
|
||||
continue
|
||||
matched = True
|
||||
highlighter.add(entry, elements, vmark)
|
||||
if not matched:
|
||||
return None
|
||||
return highlighter
|
||||
+83
-32
@@ -212,56 +212,80 @@ def _format_value(
|
||||
*,
|
||||
max_len: int = 60,
|
||||
logfmt: Callable[[Any, str], str | None] | None = None,
|
||||
highlight: Any = None,
|
||||
) -> str:
|
||||
"""Format a value for display, truncating if needed."""
|
||||
"""Format a value for display, truncating if needed.
|
||||
|
||||
``highlight`` is an optional hook with ``path(text, path)`` and
|
||||
``value(text, path)`` methods (see :class:`kanta.grep.GrepHighlighter`);
|
||||
it wraps matched keys and scalar values, and recurses into containers.
|
||||
"""
|
||||
if logfmt is not None:
|
||||
resolved = logfmt(value, path)
|
||||
if resolved is not None:
|
||||
if highlight is not None:
|
||||
resolved = highlight.value(resolved, path)
|
||||
return resolved
|
||||
|
||||
def keyed(key: Any, key_path: str) -> str:
|
||||
display = _format_value(key, key_path, max_len=30, logfmt=logfmt)
|
||||
if highlight is not None:
|
||||
display = highlight.path(display, key_path)
|
||||
return display
|
||||
|
||||
if value is None:
|
||||
return "null"
|
||||
if isinstance(value, bool):
|
||||
return "true" if value else "false"
|
||||
if isinstance(value, (int, float)):
|
||||
return str(value)
|
||||
if isinstance(value, str):
|
||||
value = _UNSAFE_CHARS.sub("", value)
|
||||
if len(value) > max_len:
|
||||
return value[: max_len - 1] + _dim_ellipsis()
|
||||
return value
|
||||
if isinstance(value, dict):
|
||||
text = "null"
|
||||
elif isinstance(value, bool):
|
||||
text = "true" if value else "false"
|
||||
elif isinstance(value, (int, float)):
|
||||
text = str(value)
|
||||
elif isinstance(value, str):
|
||||
text = _UNSAFE_CHARS.sub("", value)
|
||||
if len(text) > max_len:
|
||||
text = text[: max_len - 1] + _dim_ellipsis()
|
||||
elif isinstance(value, dict):
|
||||
if not value:
|
||||
return "{}"
|
||||
all_true = all(v is True for v in value.values())
|
||||
parts = []
|
||||
for k, v in value.items():
|
||||
key_path = _join_path(path, str(k))
|
||||
key_display = _format_value(k, key_path, max_len=30, logfmt=logfmt)
|
||||
key_display = keyed(k, key_path)
|
||||
if all_true:
|
||||
parts.append(key_display)
|
||||
else:
|
||||
val_display = _format_value(v, key_path, max_len=30, logfmt=logfmt)
|
||||
val_display = _format_value(
|
||||
v, key_path, max_len=30, logfmt=logfmt, highlight=highlight
|
||||
)
|
||||
parts.append(f"{key_display}: {val_display}")
|
||||
return "{" + ", ".join(parts) + "}"
|
||||
if isinstance(value, list):
|
||||
elif isinstance(value, list):
|
||||
if not value:
|
||||
return "[]"
|
||||
parts = []
|
||||
for i, v in enumerate(value):
|
||||
item_path = _join_path(path, str(i))
|
||||
parts.append(_format_value(v, item_path, max_len=30, logfmt=logfmt))
|
||||
parts.append(
|
||||
_format_value(
|
||||
v, item_path, max_len=30, logfmt=logfmt, highlight=highlight
|
||||
)
|
||||
)
|
||||
return "[" + ", ".join(parts) + "]"
|
||||
text = str(value)
|
||||
if len(text) > max_len:
|
||||
text = text[: max_len - 1] + _dim_ellipsis()
|
||||
else:
|
||||
text = str(value)
|
||||
if len(text) > max_len:
|
||||
text = text[: max_len - 1] + _dim_ellipsis()
|
||||
if highlight is not None:
|
||||
text = highlight.value(text, path)
|
||||
return text
|
||||
|
||||
|
||||
def _format_path_components(
|
||||
path: list[str], logfmt: Callable[[Any, str], str | None] | None
|
||||
path: list[str],
|
||||
logfmt: Callable[[Any, str], str | None] | None,
|
||||
highlight: Any = None,
|
||||
) -> list[str]:
|
||||
"""Return path components after applying formatters."""
|
||||
"""Return path components after applying formatters and match highlights."""
|
||||
if not path:
|
||||
return []
|
||||
result = []
|
||||
@@ -272,6 +296,8 @@ def _format_path_components(
|
||||
resolved = logfmt(component, prefix_path)
|
||||
if resolved is not None:
|
||||
display = resolved
|
||||
if highlight is not None:
|
||||
display = highlight.path(display, prefix_path)
|
||||
result.append(display)
|
||||
return result
|
||||
|
||||
@@ -280,12 +306,13 @@ def _format_path(
|
||||
path: list[str],
|
||||
logfmt: Callable[[Any, str], str | None] | None,
|
||||
final_color: str = "path_final",
|
||||
highlight: Any = None,
|
||||
) -> str:
|
||||
"""Format a path as dot notation with prefix in dark grey, final colored.
|
||||
|
||||
*final_color* names a color in the :data:`kanta.tty.colors` palette.
|
||||
"""
|
||||
components = _format_path_components(path, logfmt)
|
||||
components = _format_path_components(path, logfmt, highlight)
|
||||
if not components:
|
||||
return ""
|
||||
line = Line()
|
||||
@@ -380,25 +407,32 @@ def _format_change_lines(
|
||||
path: list[str],
|
||||
value: Any,
|
||||
logfmt: Callable[[Any, str], str | None] | None = None,
|
||||
highlight: Any = None,
|
||||
) -> list[str]:
|
||||
"""Format a single change as one or more lines."""
|
||||
if change_type == "delete":
|
||||
components = _format_path_components(path, logfmt)
|
||||
components = _format_path_components(path, logfmt, highlight)
|
||||
line = Line()(" ")
|
||||
if len(components) > 1:
|
||||
line.path_prefix(".".join(components[:-1]) + ".")
|
||||
line.delete(components[-1], " ✗")
|
||||
marker = "✗"
|
||||
if highlight is not None:
|
||||
marker = highlight.delete(marker, ".".join(path))
|
||||
line.delete(components[-1], " ", marker)
|
||||
return [str(line)]
|
||||
|
||||
if change_type == "add":
|
||||
path_str = _format_path(path, logfmt, final_color="add")
|
||||
path_str = _format_path(path, logfmt, final_color="add", highlight=highlight)
|
||||
if isinstance(value, dict) and value:
|
||||
lines = [str(Line()(" ", path_str, " ").sep("="))]
|
||||
base_path = ".".join(path)
|
||||
keys = []
|
||||
for k in value:
|
||||
key_path = _join_path(base_path, str(k))
|
||||
keys.append((k, _format_value(k, key_path, max_len=30, logfmt=logfmt)))
|
||||
key_display = _format_value(k, key_path, max_len=30, logfmt=logfmt)
|
||||
if highlight is not None:
|
||||
key_display = highlight.path(key_display, key_path)
|
||||
keys.append((k, key_display))
|
||||
field_width = max(displaywidth(kd) for _, kd in keys)
|
||||
field_width = max(field_width, 12)
|
||||
# Each item line is " {key:{field_width}}: {value}"; budget the
|
||||
@@ -407,7 +441,9 @@ def _format_change_lines(
|
||||
formatted_items = []
|
||||
for (k, key_display), v in zip(keys, value.values()):
|
||||
key_path = _join_path(base_path, str(k))
|
||||
v_str = _format_value(v, key_path, max_len=value_width, logfmt=logfmt)
|
||||
v_str = _format_value(
|
||||
v, key_path, max_len=value_width, logfmt=logfmt, highlight=highlight
|
||||
)
|
||||
formatted_items.append((key_display, v_str))
|
||||
return lines + [
|
||||
str(
|
||||
@@ -417,11 +453,13 @@ def _format_change_lines(
|
||||
)
|
||||
for k, v in formatted_items
|
||||
]
|
||||
value_str = _format_value(value, ".".join(path), logfmt=logfmt)
|
||||
value_str = _format_value(
|
||||
value, ".".join(path), logfmt=logfmt, highlight=highlight
|
||||
)
|
||||
return [str(Line()(" ", path_str, " ").sep("=")(" ", value_str))]
|
||||
|
||||
value_str = _format_value(value, ".".join(path), logfmt=logfmt)
|
||||
path_str = _format_path(path, logfmt=logfmt)
|
||||
value_str = _format_value(value, ".".join(path), logfmt=logfmt, highlight=highlight)
|
||||
path_str = _format_path(path, logfmt=logfmt, highlight=highlight)
|
||||
return [str(Line()(" ", path_str, " ").sep("=")(" ", value_str))]
|
||||
|
||||
|
||||
@@ -429,6 +467,7 @@ def format_diff(
|
||||
diff: dict,
|
||||
previous: dict | None = None,
|
||||
logfmt: Callable[[Any, str], str | None] | None = None,
|
||||
highlight: Any = None,
|
||||
) -> list[str]:
|
||||
"""Format a JSON diff as human-readable lines.
|
||||
|
||||
@@ -439,6 +478,8 @@ def format_diff(
|
||||
``path`` is a dot-notation string; ``"$user"`` is used for the
|
||||
transaction actor. If the callable returns ``None``, default
|
||||
formatting is used.
|
||||
highlight: Optional match highlighter hook (see
|
||||
:class:`kanta.grep.GrepHighlighter`) wrapping matched regions.
|
||||
|
||||
Returns a list of formatted lines (without newlines).
|
||||
"""
|
||||
@@ -448,7 +489,7 @@ def format_diff(
|
||||
return []
|
||||
lines = []
|
||||
for change_type, path, value in changes:
|
||||
lines.extend(_format_change_lines(change_type, path, value, logfmt))
|
||||
lines.extend(_format_change_lines(change_type, path, value, logfmt, highlight))
|
||||
return lines
|
||||
|
||||
|
||||
@@ -456,12 +497,22 @@ def format_action_header(
|
||||
action: str,
|
||||
user: str | None = None,
|
||||
extra: Any = None,
|
||||
highlight: Any = None,
|
||||
) -> str:
|
||||
"""Format the default action header line."""
|
||||
"""Format the default action header line.
|
||||
|
||||
``highlight`` is an optional hook with a ``meta(text, field)`` method
|
||||
(see :class:`kanta.grep.GrepHighlighter`) wrapping matched regions of
|
||||
the action and user fields.
|
||||
"""
|
||||
if highlight is not None:
|
||||
action = highlight.meta(action, "action")
|
||||
line = Line().action(action)
|
||||
if extra is not None and (extra := f"{extra}"):
|
||||
line(" ").target(extra)
|
||||
if user is not None and (user := f"{user}"):
|
||||
if highlight is not None:
|
||||
user = highlight.meta(user, "user")
|
||||
line(" by ").user(user)
|
||||
return str(line)
|
||||
|
||||
|
||||
@@ -0,0 +1,610 @@
|
||||
"""Tests for structural ``--grep`` matching and match highlighting."""
|
||||
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from kanta.__main__ import main
|
||||
from kanta.grep import (
|
||||
GrepPattern,
|
||||
_Entry,
|
||||
_find_spans,
|
||||
_match_path,
|
||||
_match_value,
|
||||
evaluate,
|
||||
mark_spans,
|
||||
)
|
||||
from kanta.logging import _USER_PATH
|
||||
from kanta.serialization import JsonSerializer
|
||||
from kanta.serialization.framing import LineFramer
|
||||
from kanta.structs import ChangeRecord, Snapshot
|
||||
from kanta.tty import strip_ansi
|
||||
|
||||
TS = datetime(2026, 1, 1, tzinfo=UTC)
|
||||
|
||||
MARK = "\x1b[48;5;220m"
|
||||
UNMARK = "\x1b[49m"
|
||||
|
||||
|
||||
def test_match_path_whole_elements_anywhere():
|
||||
assert _match_path("users", ["users"]) == frozenset({0})
|
||||
assert _match_path("users", ["data", "users", "alice"]) == frozenset({1})
|
||||
# Partial element matches require explicit wildcards.
|
||||
assert _match_path("users", ["foousers"]) is None
|
||||
assert _match_path("us*rs", ["foousers"]) is None
|
||||
assert _match_path("*users", ["foousers"]) == frozenset({0})
|
||||
|
||||
|
||||
def test_match_path_contiguous_element_sequence():
|
||||
assert _match_path("alice.email", ["users", "alice", "email"]) == frozenset({1, 2})
|
||||
# The sequence must be contiguous.
|
||||
assert _match_path("users.email", ["users", "alice", "email"]) is None
|
||||
assert _match_path("users.*.email", ["users", "alice", "email"]) == frozenset(
|
||||
{0, 1, 2}
|
||||
)
|
||||
# Matching is case-insensitive and empty terms match without marking.
|
||||
assert _match_path("USERS", ["users"]) == frozenset({0})
|
||||
assert _match_path("", ["anything"]) == frozenset()
|
||||
|
||||
|
||||
def test_match_value_substring_for_strings_full_for_scalars():
|
||||
assert _match_value("lice", "alice@example.com", "alice@example.com")
|
||||
assert not _match_value("bob", "alice@example.com", "alice@example.com")
|
||||
# Booleans, numbers and null match only in full.
|
||||
assert _match_value("true", True, "true")
|
||||
assert not _match_value("tru", True, "true")
|
||||
assert _match_value("42", 42, "42")
|
||||
assert not _match_value("4", 42, "42")
|
||||
assert _match_value("null", None, "null")
|
||||
# Wildcards match the whole value text of any type.
|
||||
assert _match_value("tru*", True, "true")
|
||||
assert _match_value("4*", 42, "42")
|
||||
# Containers do not match; their leaves are matched individually.
|
||||
assert not _match_value("email", {"email": "a@b.c"}, '{"email": "a@b.c"}')
|
||||
# An empty term matches anything.
|
||||
assert _match_value("", True, "true")
|
||||
|
||||
|
||||
def test_mark_spans_merges_overlaps_and_adjacents():
|
||||
assert mark_spans("alice@example.com", [(0, 5), (3, 11)]) == (
|
||||
f"{MARK}alice@examp{UNMARK}le.com"
|
||||
)
|
||||
assert mark_spans("aab", [(0, 1), (1, 3)]) == f"{MARK}aab{UNMARK}"
|
||||
assert mark_spans("ab", [(1, 1), (2, 2)]) == "ab"
|
||||
|
||||
|
||||
def test_mark_spans_offsets_into_styled_text():
|
||||
styled = "\x1b[32mal\x1b[0mlice"
|
||||
# A span crossing an escape sequence wraps each escape-free run.
|
||||
assert mark_spans(styled, [(0, 3)]) == (
|
||||
f"\x1b[32m{MARK}al{UNMARK}\x1b[0m{MARK}l{UNMARK}ice"
|
||||
)
|
||||
# A span covering everything wraps each escape-free run separately.
|
||||
assert mark_spans(styled, [(0, 6)]) == (
|
||||
f"\x1b[32m{MARK}al{UNMARK}\x1b[0m{MARK}lice{UNMARK}"
|
||||
)
|
||||
assert mark_spans(styled, []) == styled
|
||||
|
||||
|
||||
def test_find_spans_case_insensitive_occurrences():
|
||||
assert _find_spans("Alice likes ALICE", "alice") == [(0, 5), (12, 17)]
|
||||
assert _find_spans("nope", "alice") == []
|
||||
assert _find_spans("anything", "") == []
|
||||
|
||||
|
||||
def test_pattern_parse_forms():
|
||||
bare = GrepPattern.parse("alice")
|
||||
assert bare.term == "alice" and bare.path_term is None
|
||||
|
||||
pair = GrepPattern.parse("users.alice.age=30")
|
||||
assert pair.term is None
|
||||
assert pair.path_term == "users.alice.age"
|
||||
assert pair.value_term == "30"
|
||||
|
||||
value_only = GrepPattern.parse("=alice@example.com")
|
||||
assert value_only.path_term == "" and value_only.value_term == "alice@example.com"
|
||||
|
||||
path_only = GrepPattern.parse("users.alice=")
|
||||
assert path_only.path_term == "users.alice" and path_only.value_term == ""
|
||||
|
||||
# Split on the first '=' only; the value may contain '='.
|
||||
multi = GrepPattern.parse("key=a=b")
|
||||
assert multi.path_term == "key" and multi.value_term == "a=b"
|
||||
|
||||
|
||||
def _patterns(*raws: str) -> list[GrepPattern]:
|
||||
return [GrepPattern.parse(raw) for raw in raws]
|
||||
|
||||
|
||||
def test_evaluate_bare_term_against_path_value_action_user():
|
||||
record = ChangeRecord(
|
||||
ts=TS, a="create_user", u="admin", diff={"users": {"alice": {"age": 30}}}
|
||||
)
|
||||
assert evaluate(record, {}, _patterns("users.alice"))
|
||||
assert evaluate(record, {}, _patterns("30"))
|
||||
assert evaluate(record, {}, _patterns("CREATE_user"))
|
||||
assert evaluate(record, {}, _patterns("admin"))
|
||||
assert evaluate(record, {}, _patterns("age=30"))
|
||||
assert not evaluate(record, {}, _patterns("bob"))
|
||||
assert not evaluate(record, {}, _patterns("3")) # not a full number match
|
||||
|
||||
|
||||
def test_evaluate_unescapes_dollar_keys():
|
||||
record = ChangeRecord(ts=TS, a="set", diff={"$$config": 5})
|
||||
assert evaluate(record, {}, _patterns("$config=5"))
|
||||
|
||||
|
||||
def test_evaluate_path_value_form_requires_same_line():
|
||||
record = ChangeRecord(ts=TS, a="set", diff={"a": {"x": 1}, "b": {"y": 2}})
|
||||
previous = {"a": {"x": 0}, "b": {"y": 0}}
|
||||
assert evaluate(record, previous, _patterns("a.x=1"))
|
||||
# 'a' matches one line's path, '2' another line's value: no match.
|
||||
assert not evaluate(record, previous, _patterns("a=2"))
|
||||
|
||||
|
||||
def test_evaluate_all_patterns_same_record_any_line():
|
||||
record = ChangeRecord(ts=TS, a="set", diff={"a": {"x": 1}, "b": {"y": 2}})
|
||||
previous = {"a": {"x": 0}, "b": {"y": 0}}
|
||||
assert evaluate(record, previous, _patterns("a.x", "=2"))
|
||||
assert not evaluate(record, previous, _patterns("a.x", "=2", "missing"))
|
||||
|
||||
|
||||
def test_evaluate_matches_deleted_content_by_previous_value():
|
||||
record = ChangeRecord(ts=TS, a="delete_user", diff={"users": {"$delete": "alice"}})
|
||||
previous = {"users": {"alice": {"email": "alice@example.com"}}}
|
||||
assert evaluate(record, previous, _patterns("alice@example.com"))
|
||||
assert evaluate(record, previous, _patterns("users.alice.email"))
|
||||
assert not evaluate(record, {}, _patterns("alice@example.com"))
|
||||
|
||||
|
||||
def test_highlighter_marks_exactly_the_matched_regions():
|
||||
record = ChangeRecord(
|
||||
ts=TS,
|
||||
a="set",
|
||||
diff={
|
||||
"users": {"alice": {"email": "same@x.com"}, "bob": {"email": "same@x.com"}}
|
||||
},
|
||||
)
|
||||
previous = {
|
||||
"users": {"alice": {"email": "old@x.com"}, "bob": {"email": "same@x.com"}}
|
||||
}
|
||||
hl = evaluate(record, previous, _patterns("users.alice.email"))
|
||||
assert hl is not None
|
||||
# Path-side match: the matched elements are lit, the value is not.
|
||||
assert hl.path("alice", "users.alice") == f"{MARK}alice{UNMARK}"
|
||||
assert hl.value("same@x.com", "users.alice.email") == "same@x.com"
|
||||
# Bob's identical value is a different path: untouched.
|
||||
assert hl.path("bob", "users.bob") == "bob"
|
||||
assert hl.value("same@x.com", "users.bob.email") == "same@x.com"
|
||||
|
||||
hl = evaluate(record, previous, _patterns("=same@x.com"))
|
||||
assert hl is not None
|
||||
# Both lines genuinely match the value: both are marked.
|
||||
assert hl.value("same@x.com", "users.alice.email") == f"{MARK}same@x.com{UNMARK}"
|
||||
assert hl.value("same@x.com", "users.bob.email") == f"{MARK}same@x.com{UNMARK}"
|
||||
assert hl.path("alice", "users.alice") == "alice"
|
||||
|
||||
|
||||
def test_highlighter_merges_overlapping_needles_from_different_patterns():
|
||||
record = ChangeRecord(ts=TS, a="set", diff={"email": "alice@example.com"})
|
||||
hl = evaluate(record, {}, _patterns("alice", "lice@exam"))
|
||||
assert hl is not None
|
||||
assert hl.value("alice@example.com", "email") == (
|
||||
f"{MARK}alice@exam{UNMARK}ple.com"
|
||||
)
|
||||
|
||||
|
||||
def test_highlighter_marks_delete_marker_on_removed_content_match():
|
||||
record = ChangeRecord(ts=TS, a="delete_user", diff={"users": {"$delete": "bob"}})
|
||||
previous = {"users": {"bob": {"email": "bob@example.com"}}}
|
||||
hl = evaluate(record, previous, _patterns("bob@example.com"))
|
||||
assert hl is not None
|
||||
# The matched content is gone: the deletion marker is lit, not the path.
|
||||
assert hl.path("users", "users") == "users"
|
||||
assert hl.path("bob", "users.bob") == "bob"
|
||||
assert hl.delete("✗", "users.bob") == f"{MARK}✗{UNMARK}"
|
||||
assert hl.delete("✗", "users.alice") == "✗"
|
||||
|
||||
# A path-side match lights the genuinely matched elements of the anchor,
|
||||
# and the marker for the element below it.
|
||||
hl = evaluate(record, previous, _patterns("users.bob.email"))
|
||||
assert hl is not None
|
||||
assert hl.path("users", "users") == f"{MARK}users{UNMARK}"
|
||||
assert hl.path("bob", "users.bob") == f"{MARK}bob{UNMARK}"
|
||||
assert hl.delete("✗", "users.bob") == f"{MARK}✗{UNMARK}"
|
||||
|
||||
# A path-side match of one displayed element lights only that element.
|
||||
hl = evaluate(record, previous, _patterns("bob"))
|
||||
assert hl is not None
|
||||
assert hl.path("users", "users") == "users"
|
||||
assert hl.path("bob", "users.bob") == f"{MARK}bob{UNMARK}"
|
||||
|
||||
|
||||
def test_highlighter_meta_marks_action_and_user():
|
||||
record = ChangeRecord(ts=TS, a="create_user", u="admin", diff={"x": 1})
|
||||
hl = evaluate(record, {}, _patterns("create", "ADM"))
|
||||
assert hl is not None
|
||||
assert hl.meta("create_user", "action") == f"{MARK}create{UNMARK}_user"
|
||||
assert hl.meta("admin", "user") == f"{MARK}adm{UNMARK}in"
|
||||
assert hl.meta("extra", "other") == "extra"
|
||||
|
||||
|
||||
def test_entry_dataclass_holds_anchor_for_deletes():
|
||||
entry = _Entry(["a", "b"], 1, "1", anchor=["a"])
|
||||
assert entry.anchor == ["a"]
|
||||
|
||||
|
||||
def _write_db(path, changes, state=None):
|
||||
serializer = JsonSerializer()
|
||||
framer = LineFramer()
|
||||
snapshot = Snapshot(ts=TS, v=1, state=state or {})
|
||||
data = framer.frame_snapshot(serializer.encode(snapshot), record_offset=0)
|
||||
for change in changes:
|
||||
data += framer.frame_change(serializer.encode(change), record_offset=0)
|
||||
path.write_bytes(data)
|
||||
|
||||
|
||||
def _sample_changes():
|
||||
return [
|
||||
ChangeRecord(
|
||||
ts=TS,
|
||||
a="create_alice",
|
||||
u="admin",
|
||||
diff={
|
||||
"users": {
|
||||
"alice": {"email": "alice@example.com", "admin": True},
|
||||
}
|
||||
},
|
||||
),
|
||||
ChangeRecord(
|
||||
ts=TS,
|
||||
a="update_alice",
|
||||
u="admin",
|
||||
diff={"users": {"alice": {"age": 30}}},
|
||||
),
|
||||
ChangeRecord(
|
||||
ts=TS,
|
||||
a="create_bob",
|
||||
u="bob",
|
||||
diff={"users": {"bob": {"email": "bob@example.com"}}},
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
def _run_cli(tmp_path, capsys, monkeypatch, changes, *args, color=False):
|
||||
if color:
|
||||
monkeypatch.setenv("FORCE_COLOR", "1")
|
||||
monkeypatch.delenv("NO_COLOR", raising=False)
|
||||
else:
|
||||
monkeypatch.setenv("NO_COLOR", "1")
|
||||
monkeypatch.delenv("FORCE_COLOR", raising=False)
|
||||
path = tmp_path / "test.kantadb"
|
||||
_write_db(path, changes)
|
||||
code = main([str(path), *args])
|
||||
assert code == 0
|
||||
return capsys.readouterr().err
|
||||
|
||||
|
||||
def test_cli_grep_filters_records(tmp_path, capsys, monkeypatch):
|
||||
"""Only matching change records are printed."""
|
||||
err = _run_cli(tmp_path, capsys, monkeypatch, _sample_changes(), "--grep", "alice")
|
||||
assert "create_alice" in err
|
||||
assert "update_alice" in err
|
||||
assert "create_bob" not in err
|
||||
assert "bob@example.com" not in err
|
||||
|
||||
|
||||
def test_cli_grep_prints_entire_transaction(tmp_path, capsys, monkeypatch):
|
||||
"""A match prints the whole record, not just the matching line."""
|
||||
err = _run_cli(
|
||||
tmp_path, capsys, monkeypatch, _sample_changes(), "--grep", "users.alice.email"
|
||||
)
|
||||
assert "create_alice" in err
|
||||
# Non-matching lines of the same record are printed too.
|
||||
assert "admin" in err and "true" in err
|
||||
# The update record does not contain the path.
|
||||
assert "update_alice" not in err
|
||||
|
||||
|
||||
def test_cli_grep_repeated_patterns_must_all_match(tmp_path, capsys, monkeypatch):
|
||||
"""Repeated --grep options are ANDed within the same record."""
|
||||
err = _run_cli(
|
||||
tmp_path,
|
||||
capsys,
|
||||
monkeypatch,
|
||||
_sample_changes(),
|
||||
"--grep",
|
||||
"alice",
|
||||
"--grep",
|
||||
"=alice@example.com",
|
||||
)
|
||||
assert "create_alice" in err
|
||||
# update_alice matches 'alice' but has no email value.
|
||||
assert "update_alice" not in err
|
||||
|
||||
err = _run_cli(
|
||||
tmp_path,
|
||||
capsys,
|
||||
monkeypatch,
|
||||
_sample_changes(),
|
||||
"--grep",
|
||||
"alice",
|
||||
"--grep",
|
||||
"bob",
|
||||
)
|
||||
assert "create_alice" not in err
|
||||
assert "create_bob" not in err
|
||||
|
||||
|
||||
def test_cli_grep_no_match_exits_zero(tmp_path, capsys, monkeypatch):
|
||||
"""No matching records is not an error; snapshot lines still print."""
|
||||
err = _run_cli(tmp_path, capsys, monkeypatch, _sample_changes(), "--grep", "nobody")
|
||||
assert "snapshot s0" in err
|
||||
assert "create_alice" not in err
|
||||
assert "create_bob" not in err
|
||||
|
||||
|
||||
def test_cli_grep_path_value_forms(tmp_path, capsys, monkeypatch):
|
||||
"""The 'path=value', 'path=' and '=value' forms restrict the match side."""
|
||||
err = _run_cli(
|
||||
tmp_path, capsys, monkeypatch, _sample_changes(), "--grep", "users.*.admin=true"
|
||||
)
|
||||
assert "create_alice" in err
|
||||
assert "create_bob" not in err
|
||||
|
||||
err = _run_cli(
|
||||
tmp_path, capsys, monkeypatch, _sample_changes(), "--grep", "users.alice.age="
|
||||
)
|
||||
assert "update_alice" in err
|
||||
assert "create_alice" not in err
|
||||
|
||||
err = _run_cli(
|
||||
tmp_path, capsys, monkeypatch, _sample_changes(), "--grep", "=bob@example.com"
|
||||
)
|
||||
assert "create_bob" in err
|
||||
assert "create_alice" not in err
|
||||
|
||||
|
||||
def test_cli_grep_path_elements_match_in_full(tmp_path, capsys, monkeypatch):
|
||||
"""'users' does not match a 'foousers' element; 'us*' does."""
|
||||
changes = [ChangeRecord(ts=TS, a="trap", diff={"foousers": {"note": "x"}})]
|
||||
err = _run_cli(tmp_path, capsys, monkeypatch, changes, "--grep", "users")
|
||||
assert "trap" not in err
|
||||
|
||||
err = _run_cli(tmp_path, capsys, monkeypatch, changes, "--grep", "*users")
|
||||
assert "trap" in err
|
||||
|
||||
|
||||
def _shared_email_changes():
|
||||
return [
|
||||
ChangeRecord(
|
||||
ts=TS,
|
||||
a="create_alice",
|
||||
u="admin",
|
||||
diff={"users": {"alice": {"email": "shared@example.com", "admin": True}}},
|
||||
),
|
||||
ChangeRecord(
|
||||
ts=TS,
|
||||
a="create_bob",
|
||||
u="admin",
|
||||
diff={"users": {"bob": {"email": "shared@example.com"}}},
|
||||
),
|
||||
ChangeRecord(
|
||||
ts=TS,
|
||||
a="delete_bob",
|
||||
u="admin",
|
||||
diff={"users": {"$delete": "bob"}},
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
def _lines_with(err: str, text: str) -> list[str]:
|
||||
return [line for line in err.splitlines() if text in strip_ansi(line)]
|
||||
|
||||
|
||||
def test_cli_highlight_marks_exactly_the_matched_regions(tmp_path, capsys, monkeypatch):
|
||||
"""A path-side match lights only the matched elements, not the value."""
|
||||
err = _run_cli(
|
||||
tmp_path,
|
||||
capsys,
|
||||
monkeypatch,
|
||||
_shared_email_changes(),
|
||||
"--grep",
|
||||
"users.alice.email",
|
||||
color=True,
|
||||
)
|
||||
(alice_line,) = _lines_with(err, "shared@example.com")
|
||||
assert f"{MARK}alice{UNMARK}" in alice_line
|
||||
assert f"{MARK}email{UNMARK}" in alice_line
|
||||
# The value itself did not match, so it is not highlighted.
|
||||
assert f"{MARK}shared@example.com{UNMARK}" not in alice_line
|
||||
(header_line,) = _lines_with(err, "users =")
|
||||
assert f"{MARK}users{UNMARK}" in header_line
|
||||
assert "create_bob" not in err
|
||||
|
||||
|
||||
def test_cli_highlight_value_matches_on_all_matching_lines(
|
||||
tmp_path, capsys, monkeypatch
|
||||
):
|
||||
"""A value-side match lights the value on every line that genuinely matched."""
|
||||
err = _run_cli(
|
||||
tmp_path,
|
||||
capsys,
|
||||
monkeypatch,
|
||||
_shared_email_changes(),
|
||||
"--grep",
|
||||
"=shared@example.com",
|
||||
color=True,
|
||||
)
|
||||
lines = _lines_with(err, "shared@example.com")
|
||||
assert len(lines) == 2 # alice's and bob's records both matched
|
||||
for line in lines:
|
||||
assert f"{MARK}shared@example.com{UNMARK}" in line
|
||||
# The deletion of bob matched by its previous (removed) value: the
|
||||
# deletion marker is lit, not the deleted path.
|
||||
(delete_line,) = _lines_with(err, "✗")
|
||||
assert f"{MARK}✗{UNMARK}" in delete_line
|
||||
assert f"{MARK}bob{UNMARK}" not in delete_line
|
||||
assert f"{MARK}users{UNMARK}" not in delete_line
|
||||
|
||||
|
||||
def test_cli_highlight_merges_overlapping_matches(tmp_path, capsys, monkeypatch):
|
||||
"""Overlapping matches from different patterns form one continuous mark."""
|
||||
err = _run_cli(
|
||||
tmp_path,
|
||||
capsys,
|
||||
monkeypatch,
|
||||
_shared_email_changes(),
|
||||
"--grep",
|
||||
"shared",
|
||||
"--grep",
|
||||
"red@exam",
|
||||
color=True,
|
||||
)
|
||||
lines = _lines_with(err, "shared@example.com")
|
||||
assert len(lines) == 2 # both records genuinely match both patterns
|
||||
for line in lines:
|
||||
assert f"{MARK}shared@exam{UNMARK}ple.com" in line
|
||||
|
||||
|
||||
def test_cli_highlight_full_scalar_and_header_fields(tmp_path, capsys, monkeypatch):
|
||||
"""Scalars are marked in full; matched action/user substrings are marked."""
|
||||
err = _run_cli(
|
||||
tmp_path,
|
||||
capsys,
|
||||
monkeypatch,
|
||||
_shared_email_changes(),
|
||||
"--grep",
|
||||
"admin=true",
|
||||
"--grep",
|
||||
"create_al",
|
||||
color=True,
|
||||
)
|
||||
(line,) = _lines_with(err, "shared@example.com")
|
||||
assert f"{MARK}true{UNMARK}" in line
|
||||
(header,) = _lines_with(err, "create_alice")
|
||||
assert f"{MARK}create_al{UNMARK}ice" in header
|
||||
# Only the record matching both patterns is printed.
|
||||
assert "create_bob" not in err
|
||||
assert "delete_bob" not in err
|
||||
|
||||
|
||||
def test_evaluate_matches_prettified_and_raw_value_forms():
|
||||
"""With logfmt, both the raw and the prettified form of a value match."""
|
||||
record = ChangeRecord(ts=TS, a="set", diff={"when": 1767225600})
|
||||
|
||||
def logfmt(value, path):
|
||||
return "2026-01-01" if value == 1767225600 else None
|
||||
|
||||
# The prettified form matches; the needle is located in the display text.
|
||||
hl = evaluate(record, {}, _patterns("2026"), logfmt=logfmt)
|
||||
assert hl is not None
|
||||
assert hl.value("2026-01-01", "when") == f"{MARK}2026{UNMARK}-01-01"
|
||||
# The raw form matches too; its needle is absent from the displayed
|
||||
# text, so the whole displayed value is marked.
|
||||
hl = evaluate(record, {}, _patterns("1767225600"), logfmt=logfmt)
|
||||
assert hl is not None
|
||||
assert hl.value("2026-01-01", "when") == f"{MARK}2026-01-01{UNMARK}"
|
||||
# Without logfmt the raw value is matched and marked precisely.
|
||||
hl = evaluate(record, {}, _patterns("1767225600"))
|
||||
assert hl.value("1767225600", "when") == f"{MARK}1767225600{UNMARK}"
|
||||
# Neither form matches.
|
||||
assert evaluate(record, {}, _patterns("1999"), logfmt=logfmt) is None
|
||||
|
||||
|
||||
def test_evaluate_matches_prettified_user():
|
||||
"""The user field matches both the raw id and the logfmt-resolved name."""
|
||||
|
||||
def logfmt(value, path):
|
||||
return "Alice Admin" if path == _USER_PATH else None
|
||||
|
||||
record = ChangeRecord(ts=TS, a="set", u="u123", diff={"x": 1})
|
||||
hl = evaluate(record, {}, _patterns("alice"), logfmt=logfmt)
|
||||
assert hl is not None
|
||||
assert hl.meta("Alice Admin", "user") == f"{MARK}Alice{UNMARK} Admin"
|
||||
# A raw-only user match marks the whole displayed name.
|
||||
hl = evaluate(record, {}, _patterns("u123"), logfmt=logfmt)
|
||||
assert hl is not None
|
||||
assert hl.meta("Alice Admin", "user") == f"{MARK}Alice Admin{UNMARK}"
|
||||
# The action is never prettified.
|
||||
assert hl.meta("set", "action") == "set"
|
||||
|
||||
|
||||
def test_evaluate_without_logfmt_matches_raw_only():
|
||||
record = ChangeRecord(ts=TS, a="set", u="u123", diff={"when": 1767225600})
|
||||
assert evaluate(record, {}, _patterns("1767225600"))
|
||||
assert evaluate(record, {}, _patterns("2026")) is None
|
||||
assert evaluate(record, {}, _patterns("alice")) is None
|
||||
|
||||
|
||||
def test_cli_grep_matches_prettified_forms_with_kanta_object(
|
||||
tmp_path, capsys, monkeypatch
|
||||
):
|
||||
"""End to end: a -k object's logfmt formatter doubles the match surface."""
|
||||
db_path = tmp_path / "test.kantadb"
|
||||
module = tmp_path / "dbmod.py"
|
||||
module.write_text(
|
||||
"from typing import Any\n"
|
||||
"from kanta import Kanta\n"
|
||||
f"kanta = Kanta({str(db_path)!r}, {{}}, type=dict)\n"
|
||||
"@kanta.logfmt\n"
|
||||
"def pretty(value: Any, path: str) -> str | None:\n"
|
||||
" if path == 'when':\n"
|
||||
" return 'Nov 3, 2025'\n"
|
||||
" if path == '$user':\n"
|
||||
" return 'Alice Admin'\n"
|
||||
" return None\n"
|
||||
)
|
||||
changes = [ChangeRecord(ts=TS, a="set", u="u123", diff={"when": 1767225600})]
|
||||
kanta_args = ("-k", str(module))
|
||||
|
||||
# A term matching only the prettified value finds the record.
|
||||
err = _run_cli(
|
||||
tmp_path, capsys, monkeypatch, changes, *kanta_args, "--grep", "nov", color=True
|
||||
)
|
||||
(line,) = _lines_with(err, "Nov 3, 2025")
|
||||
assert f"{MARK}Nov{UNMARK} 3, 2025" in line
|
||||
|
||||
# A term matching only the raw value marks the whole prettified display.
|
||||
err = _run_cli(
|
||||
tmp_path,
|
||||
capsys,
|
||||
monkeypatch,
|
||||
changes,
|
||||
*kanta_args,
|
||||
"--grep",
|
||||
"1767225600",
|
||||
color=True,
|
||||
)
|
||||
(line,) = _lines_with(err, "Nov 3, 2025")
|
||||
assert f"{MARK}Nov 3, 2025{UNMARK}" in line
|
||||
|
||||
# The prettified user matches, and the raw id marks the whole display.
|
||||
err = _run_cli(
|
||||
tmp_path,
|
||||
capsys,
|
||||
monkeypatch,
|
||||
changes,
|
||||
*kanta_args,
|
||||
"--grep",
|
||||
"alice",
|
||||
color=True,
|
||||
)
|
||||
(header,) = _lines_with(err, "Alice Admin")
|
||||
assert f"{MARK}Alice{UNMARK} Admin" in header
|
||||
err = _run_cli(
|
||||
tmp_path,
|
||||
capsys,
|
||||
monkeypatch,
|
||||
changes,
|
||||
*kanta_args,
|
||||
"--grep",
|
||||
"u123",
|
||||
color=True,
|
||||
)
|
||||
(header,) = _lines_with(err, "Alice Admin")
|
||||
assert f"{MARK}Alice Admin{UNMARK}" in header
|
||||
|
||||
# Without -k there is no prettified form to match.
|
||||
err = _run_cli(tmp_path, capsys, monkeypatch, changes, "--grep", "nov")
|
||||
assert not _lines_with(err, "u123")
|
||||
err = _run_cli(tmp_path, capsys, monkeypatch, changes, "--grep", "1767225600")
|
||||
assert _lines_with(err, "u123")
|
||||
Reference in New Issue
Block a user