diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a5a68e01..d3f2e8811 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,14 @@ selections report `No history items in range`, while search misses report `No matching history items`; displaying them still succeeds without output. The no-argument `history --run` safety refusal is unchanged. + - `History` now records which entries were loaded before the session started instead of keeping + a plain start index. Entries added after `start_session()` belong to the current session + wherever they are placed, including entries inserted at the front of the history. + `session_start_index` is now computed as the index just past the last persisted entry. + Assigning it marks only the entries already in the list, so assign it or call + `start_session()` after loading entries. After the list is reordered, + `history[history.session_start_index:]` can omit current-session entries; use + `History.is_persisted()` to check an entry instead. - Bug Fixes - Fixed history ranges starting past the end selecting the last command. Invalid indices and bounded ranges selecting no items report `No history items in range` before running, editing, @@ -34,15 +42,20 @@ it directly would then have retained current-session entries hidden from unbounded ranges and searches. Interactive `history` use was unaffected, since `cmd2` only truncates when persisting history at exit. - - Fixed deleting, popping, removing, inserting, or slice-assigning `History` entries before the - current session leaving `session_start_index` stale. In-place filtering recognizes retained - persisted entries by identity, so equal commands from the current session are not hidden. + - Fixed changing a `History` list in place moving commands between the persisted history and the + current session. Deleting, inserting, assigning, reordering (such as `reverse()` and + `sort()`), or repeating entries no longer shows persisted commands as current or hides + current-session commands, and `history --run` and `history --edit` still require `--all` for a + moved persisted command. Equal commands from different sessions remain separate entries. - Fixed `set_window_title()` leaving title changes buffered until the next prompt render. The title is now flushed before the method returns when no prompt is running. While a prompt or `select()` menu is running, it is written on that prompt's event loop so it doesn't interleave with rendering, and output errors are ignored so they can't disrupt the prompt. Titles are still written to a terminal stdout when prompts use a dummy output, such as when stdin isn't a terminal. +- Enhancements + - Added `History.is_persisted()` to check whether a history entry was loaded from a previous + session. ## 4.3.1 (October 24, 2026) diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py index 28ebf48df..f0de939f5 100644 --- a/cmd2/cmd2.py +++ b/cmd2/cmd2.py @@ -5557,7 +5557,7 @@ def do_history(self, args: argparse.Namespace) -> bool | None: self.perror(NO_MATCHING_HISTORY_ITEMS if is_search else NO_HISTORY_ITEMS) return None - if (args.run or args.edit) and not args.all and any(index <= self.history.session_start_index for index in history): + if (args.run or args.edit) and not args.all and any(self.history.is_persisted(item) for item in history.values()): self.perror("Use --all to run or edit commands from previous sessions.") return None diff --git a/cmd2/history.py b/cmd2/history.py index 4fbd2d4f5..08a178485 100644 --- a/cmd2/history.py +++ b/cmd2/history.py @@ -1,7 +1,6 @@ """History management classes.""" import json -import operator import re from collections.abc import ( Callable, @@ -10,9 +9,6 @@ from dataclasses import dataclass from typing import ( Any, - Self, - SupportsIndex, - cast, overload, ) @@ -140,6 +136,25 @@ def from_dict(source_dict: dict[str, Any]) -> "HistoryItem": return HistoryItem(Statement.from_dict(statement_dict)) +class _EntryRef: + """Hashable reference to a history entry that compares by identity rather than by value. + + Unlike id() keys, these are rehashed when a set of them is copied or unpickled, so they keep + referring to the copied entries. + """ + + __slots__ = ("entry",) + + def __init__(self, entry: HistoryItem) -> None: + self.entry = entry + + def __hash__(self) -> int: + return id(self.entry) + + def __eq__(self, other: object) -> bool: + return isinstance(other, _EntryRef) and other.entry is self.entry + + class HistoryIndexError(IndexError): """Raised when a history index is invalid or a bounded range selects no items.""" @@ -169,83 +184,42 @@ class to gain access to the historical record. def __init__(self, seq: Iterable[HistoryItem] = ()) -> None: """Initialize History instances.""" super().__init__(seq) - self.session_start_index = 0 + # Entries that were in the list when the session started. Holding references keeps their + # ids from being reused by new entries after they leave the list. + self._persisted: set[_EntryRef] = set() - def __delitem__(self, key: SupportsIndex | slice, /) -> None: - """Delete entries and move the session boundary past any removed persisted entries.""" - if isinstance(key, slice): - removed_before = sum(index < self.session_start_index for index in range(*key.indices(len(self)))) - else: - key = operator.index(key) - position = key if key >= 0 else len(self) + key - removed_before = int(position < self.session_start_index) - super().__delitem__(key) - self.session_start_index -= removed_before + def is_persisted(self, item: HistoryItem) -> bool: + """Return whether an entry was in the history when the current session started. - @overload - def __setitem__(self, key: SupportsIndex, value: HistoryItem, /) -> None: ... # pragma: no cover + Membership belongs to the entry object rather than its position, so a persisted entry stays + persisted when the list is reordered, filtered, or repeated, and new entries belong to the + current session wherever they are placed. Equal commands from different sessions are separate + entries. To record a persisted command again in the current session, append its statement. - @overload - def __setitem__(self, key: slice, value: Iterable[HistoryItem], /) -> None: ... # pragma: no cover - - def __setitem__(self, key: SupportsIndex | slice, value: HistoryItem | Iterable[HistoryItem], /) -> None: - """Replace entries, retaining persisted item identities when a slice spans the session boundary.""" - if not isinstance(key, slice): - super().__setitem__(key, cast(HistoryItem, value)) - return - start, stop, step = key.indices(len(self)) - replacement = list(cast(Iterable[HistoryItem], value)) - persisted_ids = ( - {id(entry) for entry in self[start : self.session_start_index]} - if step == 1 and start < self.session_start_index < stop - else None - ) - super().__setitem__(key, replacement) - if persisted_ids is not None: - boundary = start - for entry in replacement: - if id(entry) not in persisted_ids: - break - boundary += 1 - self.session_start_index = boundary - elif step == 1 and start < self.session_start_index: - stop = max(start, stop) - self.session_start_index += len(replacement) - (stop - start) - - def insert(self, index: SupportsIndex, value: HistoryItem, /) -> None: - """Insert an entry, shifting the session boundary if the insertion precedes it.""" - index = operator.index(index) - position = min(len(self), max(0, index if index >= 0 else len(self) + index)) - super().insert(index, value) - if position < self.session_start_index: - self.session_start_index += 1 - - def pop(self, index: SupportsIndex = -1, /) -> HistoryItem: - """Remove and return an entry, retaining the session boundary.""" - index = operator.index(index) - position = index if index >= 0 else len(self) + index - result = super().pop(index) - if position < self.session_start_index: - self.session_start_index -= 1 - return result + :param item: the history entry to check + :return: True if the entry is from a previous session + """ + return _EntryRef(item) in self._persisted + + @property + def session_start_index(self) -> int: + """Index just past the last persisted entry. + + Every entry from this index on belongs to the current session. Unless the list has been + reordered, every entry before it is persisted. Assigning ``n`` marks the first ``n`` entries + as persisted and the rest as current. + """ + for index in range(len(self) - 1, -1, -1): + if self.is_persisted(self[index]): + return index + 1 + return 0 - def remove(self, value: HistoryItem, /) -> None: - """Remove the first matching entry, retaining the session boundary.""" - position = super().index(value) - super().pop(position) - if position < self.session_start_index: - self.session_start_index -= 1 - - def __imul__(self, count: SupportsIndex, /) -> Self: - """Repeat entries in place, resetting the boundary if repetition clears the list.""" - count = operator.index(count) - super().__imul__(count) - if count <= 0: - self.session_start_index = 0 - return self + @session_start_index.setter + def session_start_index(self, index: int) -> None: + self._persisted = {_EntryRef(entry) for entry in self[:index]} def start_session(self) -> None: - """Start a new session, thereby setting the next index as the first index in the new session.""" + """Start a new session, marking every entry currently in the history as persisted.""" self.session_start_index = len(self) def _zero_based_index(self, onebased: int | str) -> int: @@ -339,7 +313,7 @@ def span(self, span: str, include_persisted: bool = False) -> dict[int, "History if start < 0: raise HistoryIndexError(NO_HISTORY_ITEMS) else: - start = 0 if include_persisted or end_token else self.session_start_index + start = 0 if end_token: end = min(int(end_token), len(self)) @@ -348,10 +322,11 @@ def span(self, span: str, include_persisted: bool = False) -> dict[int, "History else: end = len(self) - if (start_token or end_token) and (start >= len(self) or end <= start): + bounded = bool(start_token or end_token) + if bounded and (start >= len(self) or end <= start): raise HistoryIndexError(NO_HISTORY_ITEMS) - return self._build_result_dictionary(start, end) + return self._build_result_dictionary(start, end, include_persisted=include_persisted or bounded) def str_search(self, search: str, include_persisted: bool = False) -> dict[int, "HistoryItem"]: """Find history items which contain a given string. @@ -369,8 +344,7 @@ def isin(history_item: HistoryItem) -> bool: inexpanded = sloppy in su.norm_fold(history_item.expanded) return inraw or inexpanded - start = 0 if include_persisted else self.session_start_index - return self._build_result_dictionary(start, len(self), isin) + return self._build_result_dictionary(0, len(self), isin, include_persisted) def regex_search(self, regex: str, include_persisted: bool = False) -> dict[int, "HistoryItem"]: """Find history items which match a given regular expression. @@ -389,15 +363,13 @@ def isin(hi: HistoryItem) -> bool: """Filter function for doing a regular expression search of history.""" return bool(finder.search(hi.raw) or finder.search(hi.expanded)) - start = 0 if include_persisted else self.session_start_index - return self._build_result_dictionary(start, len(self), isin) + return self._build_result_dictionary(0, len(self), isin, include_persisted) def truncate(self, max_length: int) -> None: """Truncate the length of the history, dropping the oldest items if necessary. - The current session's start index is shifted so it still marks the first retained - current-session item. If any current-session items are dropped, all retained items - belong to the current session. + Retained items keep their session membership, so retained current-session items still + appear in unbounded ranges and searches. :param max_length: the maximum length of the history, if zero or negative, all history items will be deleted @@ -410,17 +382,24 @@ def truncate(self, max_length: int) -> None: del self[0:last_element] def _build_result_dictionary( - self, start: int, end: int, filter_func: Callable[[HistoryItem], bool] | None = None + self, + start: int, + end: int, + filter_func: Callable[[HistoryItem], bool] | None = None, + include_persisted: bool = True, ) -> dict[int, "HistoryItem"]: """Build history search results. :param start: start index to search from :param end: end index to stop searching (exclusive). + :param filter_func: optional function that returns True for entries to include + :param include_persisted: if False, skip entries from previous sessions """ results: dict[int, HistoryItem] = {} for index in range(start, end): - if filter_func is None or filter_func(self[index]): - results[index + 1] = self[index] + entry = self[index] + if (include_persisted or not self.is_persisted(entry)) and (filter_func is None or filter_func(entry)): + results[index + 1] = entry return results def to_json(self) -> str: diff --git a/tests/test_history_mutations.py b/tests/test_history_mutations.py index b8963df95..85918e580 100644 --- a/tests/test_history_mutations.py +++ b/tests/test_history_mutations.py @@ -1,10 +1,16 @@ -"""Session selection survives structural changes to the public history list.""" +"""Session membership survives structural changes to the public history list.""" + +import copy +import pickle +from unittest import mock import pytest from cmd2.history import History, HistoryItem from cmd2.parsing import Statement +from .conftest import run_cmd + def item(raw): return HistoryItem(Statement("", raw=raw)) @@ -12,171 +18,187 @@ def item(raw): @pytest.fixture def history(): + """Four persisted entries (old-0..old-3) followed by two current-session entries (new-0, new-1).""" hist = History(item(f"old-{i}") for i in range(4)) hist.start_session() hist.extend(item(f"new-{i}") for i in range(2)) return hist -def assert_session(hist, boundary, expected): - assert hist.session_start_index == boundary - assert [entry.raw for entry in hist.span(":").values()] == expected - assert [entry.raw for entry in hist.str_search("-").values()] == expected - assert [entry.raw for entry in hist.regex_search(".*").values()] == expected - assert list(hist.span(":")) == list(range(boundary + 1, len(hist) + 1)) +def current_session(hist): + """Return current-session entries as {1-based index: raw}, checking every session-aware query agrees. + + Also checks session_start_index: every entry from it on is current, and the entry before it is persisted. + """ + by_span = {index: entry.raw for index, entry in hist.span(":").items()} + assert {index: entry.raw for index, entry in hist.str_search("-").items()} == by_span + assert {index: entry.raw for index, entry in hist.regex_search(".*").items()} == by_span + start = hist.session_start_index + assert not any(hist.is_persisted(entry) for entry in hist[start:]) + assert start == 0 or hist.is_persisted(hist[start - 1]) + return by_span @pytest.mark.parametrize( - ("index", "boundary", "expected"), + ("index", "expected"), [ - (0, 3, ["new-0", "new-1"]), - (-6, 3, ["new-0", "new-1"]), - (-1, 4, ["new-0"]), - (slice(None, 2), 2, ["new-0", "new-1"]), - (slice(2, 5), 2, ["new-1"]), - (slice(None, None, 2), 2, ["new-1"]), - (slice(None, None, -2), 2, ["new-0"]), - (slice(5, 1, -2), 3, ["new-0"]), - (slice(None), 0, []), - (slice(1, 1), 4, ["new-0", "new-1"]), + (0, {4: "new-0", 5: "new-1"}), + (-6, {4: "new-0", 5: "new-1"}), + (-1, {5: "new-0"}), + (slice(None, 2), {3: "new-0", 4: "new-1"}), + (slice(2, 5), {3: "new-1"}), + (slice(None, None, 2), {3: "new-1"}), + (slice(None, None, -2), {3: "new-0"}), + (slice(5, 1, -2), {4: "new-0"}), + (slice(None), {}), + (slice(1, 1), {5: "new-0", 6: "new-1"}), ], ) -def test_delete_tracks_session(history, index, boundary, expected): +def test_delete_keeps_membership_of_remaining_entries(history, index, expected): del history[index] - assert_session(history, boundary, expected) + assert current_session(history) == expected @pytest.mark.parametrize( - ("index", "removed", "boundary", "expected"), + ("index", "removed", "expected"), [ - (0, "old-0", 3, ["new-0", "new-1"]), - (-3, "old-3", 3, ["new-0", "new-1"]), - (-1, "new-1", 4, ["new-0"]), + (0, "old-0", {4: "new-0", 5: "new-1"}), + (-3, "old-3", {4: "new-0", 5: "new-1"}), + (-1, "new-1", {5: "new-0"}), ], ) -def test_pop_tracks_session(history, index, removed, boundary, expected): +def test_pop_keeps_membership_of_remaining_entries(history, index, removed, expected): assert history.pop(index).raw == removed - assert_session(history, boundary, expected) + assert current_session(history) == expected -@pytest.mark.parametrize(("raw", "boundary", "expected"), [("old-1", 3, ["new-0", "new-1"]), ("new-1", 4, ["new-0"])]) -def test_remove_tracks_session(history, raw, boundary, expected): +@pytest.mark.parametrize( + ("raw", "expected"), + [("old-1", {4: "new-0", 5: "new-1"}), ("new-1", {5: "new-0"})], +) +def test_remove_keeps_membership_of_remaining_entries(history, raw, expected): assert history.remove(item(raw)) is None - assert_session(history, boundary, expected) + assert current_session(history) == expected @pytest.mark.parametrize( - ("index", "boundary", "expected"), + ("index", "expected"), [ - (0, 5, ["new-0", "new-1"]), - (3, 5, ["new-0", "new-1"]), - (-3, 5, ["new-0", "new-1"]), - (-99, 5, ["new-0", "new-1"]), - (4, 4, ["inserted-0", "new-0", "new-1"]), - (99, 4, ["new-0", "new-1", "inserted-0"]), - (-1, 4, ["new-0", "inserted-0", "new-1"]), + (0, {1: "inserted-0", 6: "new-0", 7: "new-1"}), + (3, {4: "inserted-0", 6: "new-0", 7: "new-1"}), + (-3, {4: "inserted-0", 6: "new-0", 7: "new-1"}), + (-99, {1: "inserted-0", 6: "new-0", 7: "new-1"}), + (4, {5: "inserted-0", 6: "new-0", 7: "new-1"}), + (99, {5: "new-0", 6: "new-1", 7: "inserted-0"}), + (-1, {5: "new-0", 6: "inserted-0", 7: "new-1"}), ], ) -def test_insert_tracks_session(history, index, boundary, expected): +def test_inserted_entry_is_current_wherever_it_goes(history, index, expected): history.insert(index, item("inserted-0")) - assert_session(history, boundary, expected) + assert current_session(history) == expected @pytest.mark.parametrize( - ("index", "count", "boundary", "expected"), + ("index", "count", "expected"), [ - (slice(0, 1), 2, 5, ["new-0", "new-1"]), - (slice(1, 3), 0, 2, ["new-0", "new-1"]), - (slice(2, 5), 2, 2, ["replacement-0", "replacement-1", "new-1"]), - (slice(2, 5), 4, 2, ["replacement-0", "replacement-1", "replacement-2", "replacement-3", "new-1"]), - (slice(4, 5), 2, 4, ["replacement-0", "replacement-1", "new-1"]), - (slice(None, 4), 1, 1, ["new-0", "new-1"]), - (slice(None), 2, 0, ["replacement-0", "replacement-1"]), - (slice(2, 2), 2, 6, ["new-0", "new-1"]), - (slice(4, 2), 2, 4, ["replacement-0", "replacement-1", "new-0", "new-1"]), - (slice(3, 1), 2, 6, ["new-0", "new-1"]), - (slice(None, None, 2), 3, 4, ["replacement-2", "new-1"]), - (slice(None, None, -2), 3, 4, ["new-0", "replacement-0"]), + (slice(0, 1), 2, {1: "replacement-0", 2: "replacement-1", 6: "new-0", 7: "new-1"}), + (slice(1, 3), 0, {3: "new-0", 4: "new-1"}), + (slice(2, 5), 2, {3: "replacement-0", 4: "replacement-1", 5: "new-1"}), + ( + slice(2, 5), + 4, + {3: "replacement-0", 4: "replacement-1", 5: "replacement-2", 6: "replacement-3", 7: "new-1"}, + ), + (slice(4, 5), 2, {5: "replacement-0", 6: "replacement-1", 7: "new-1"}), + (slice(None, 4), 1, {1: "replacement-0", 2: "new-0", 3: "new-1"}), + (slice(None), 2, {1: "replacement-0", 2: "replacement-1"}), + (slice(2, 2), 2, {3: "replacement-0", 4: "replacement-1", 7: "new-0", 8: "new-1"}), + (slice(4, 2), 2, {5: "replacement-0", 6: "replacement-1", 7: "new-0", 8: "new-1"}), + (slice(3, 1), 2, {4: "replacement-0", 5: "replacement-1", 7: "new-0", 8: "new-1"}), + (slice(None, None, 2), 3, {1: "replacement-0", 3: "replacement-1", 5: "replacement-2", 6: "new-1"}), + (slice(None, None, -2), 3, {2: "replacement-2", 4: "replacement-1", 5: "new-0", 6: "replacement-0"}), ], ) -def test_slice_assignment_tracks_session(history, index, count, boundary, expected): +def test_new_entries_assigned_to_a_slice_are_current(history, index, count, expected): history[index] = (item(f"replacement-{i}") for i in range(count)) - assert_session(history, boundary, expected) - - -def test_self_assignment_preserves_session(history): - history[:] = history - assert_session(history, 4, ["new-0", "new-1"]) - history[1:3] = history - assert_session(history, 8, ["new-0", "new-1"]) + assert current_session(history) == expected -@pytest.mark.parametrize("index", [0, 3, 4, -1]) -def test_index_assignment_preserves_session(history, index): +@pytest.mark.parametrize( + ("index", "expected"), + [ + (0, {1: "replacement-0", 5: "new-0", 6: "new-1"}), + (3, {4: "replacement-0", 5: "new-0", 6: "new-1"}), + (4, {5: "replacement-0", 6: "new-1"}), + (-1, {5: "new-0", 6: "replacement-0"}), + ], +) +def test_new_entry_assigned_to_an_index_is_current(history, index, expected): history[index] = item("replacement-0") - assert_session(history, 4, [entry.raw for entry in history[4:]]) + assert current_session(history) == expected -@pytest.mark.parametrize("count", [0, -1, 1, 2]) -def test_repetition_preserves_session_boundary(history, count): - history *= count - boundary = 0 if count <= 0 else 4 - assert_session(history, boundary, [entry.raw for entry in history[boundary:]]) +def test_new_entry_assigned_into_empty_session_is_current(): + hist = History(item(f"old-{i}") for i in range(4)) + hist.start_session() + hist[:] = [hist[1], item("new-0")] + assert current_session(hist) == {2: "new-0"} -@pytest.mark.parametrize("operation", ["delete", "pop", "remove", "slice_step", "slice_length", "slice_iterator"]) -def test_failed_mutation_preserves_history_and_boundary(history, operation): - original = list(history) +def test_self_assignment_copies_membership(history): + history[:] = history + assert current_session(history) == {5: "new-0", 6: "new-1"} + history[1:3] = history + assert current_session(history) == {6: "new-0", 7: "new-1", 9: "new-0", 10: "new-1"} - def broken_items(): - yield item("replacement-0") - raise RuntimeError("incomplete replacement") - error = IndexError if operation in ("delete", "pop") else RuntimeError if operation == "slice_iterator" else ValueError +def repeat_twice(hist): + hist *= 2 - def mutate(): - if operation == "delete": - del history[-99] - elif operation == "pop": - history.pop(-99) - elif operation == "remove": - history.remove(item("missing-0")) - elif operation == "slice_step": - del history[::0] - elif operation == "slice_length": - history[::2] = [item("replacement-0")] - else: - history[:2] = broken_items() - with pytest.raises(error): - mutate() - assert history == original - assert_session(history, 4, ["new-0", "new-1"]) +def extend_with_itself(hist): + hist += list(hist) -def test_mutations_compose_with_truncate_and_append(history): - history.insert(0, item("inserted-0")) - history.pop(1) - history.truncate(4) - history.append(Statement("", raw="new-2")) - assert_session(history, 2, ["new-0", "new-1", "new-2"]) - history.clear() - history.extend([item("new-3")]) - assert_session(history, 0, ["new-3"]) +@pytest.mark.parametrize( + ("mutate", "expected"), + [ + (lambda hist: hist.reverse(), {1: "new-1", 2: "new-0"}), + (lambda hist: hist.sort(key=lambda entry: entry.raw), {1: "new-0", 2: "new-1"}), + (lambda hist: hist.sort(key=lambda entry: entry.raw, reverse=True), {5: "new-1", 6: "new-0"}), + (repeat_twice, {5: "new-0", 6: "new-1", 11: "new-0", 12: "new-1"}), + (extend_with_itself, {5: "new-0", 6: "new-1", 11: "new-0", 12: "new-1"}), + ], + ids=["reverse", "sort", "sort-descending", "repeat", "extend-with-itself"], +) +def test_moved_or_copied_persisted_entries_stay_out_of_session(history, mutate, expected): + mutate(history) + assert current_session(history) == expected -def test_remove_retains_native_list_dispatch(history): - class CustomHistory(History): - def index(self, *_args): - raise AssertionError("remove must not dispatch to an overridden index") +@pytest.mark.parametrize( + ("count", "expected"), [(0, {1: "new-2"}), (-1, {1: "new-2"}), (1, {5: "new-0", 6: "new-1", 7: "new-2"})] +) +def test_repetition_then_append_keeps_new_entry_current(history, count, expected): + history *= count + history.append(Statement("", raw="new-2")) + assert current_session(history) == expected - def pop(self, *_args): - raise AssertionError("remove must not dispatch to an overridden pop") - custom = CustomHistory(history) - custom.session_start_index = history.session_start_index - custom.remove(item("old-1")) - assert_session(custom, 3, ["new-0", "new-1"]) +@pytest.mark.parametrize( + ("indices", "expected"), + [ + ([3, 1, 4, 5], {3: "new-0", 4: "new-1"}), + ([4, 1, 5], {1: "new-0", 3: "new-1"}), + ([1, 3], {}), + ([0, 0, 4], {3: "new-0"}), + ([4, 0, 5, 1], {1: "new-0", 3: "new-1"}), + ([0, 2, 1, 3, 4, 5], {5: "new-0", 6: "new-1"}), + ], +) +def test_reordering_by_slice_assignment_keeps_membership(history, indices, expected): + history[:] = [history[index] for index in indices] + assert current_session(history) == expected @pytest.mark.parametrize("as_iterator", [False, True]) @@ -194,58 +216,101 @@ def pop(self, *_args): (0, 5, {1, 4}), ], ) -def test_filtering_straddling_slice_matches_deletion(history, start, stop, removed, as_iterator): +def test_filtering_by_slice_assignment_keeps_membership(history, start, stop, removed, as_iterator): original = list(history) - expected = History(original) - expected.session_start_index = history.session_start_index - for index in sorted(removed, reverse=True): - del expected[index] replacement = [entry for index, entry in enumerate(original) if start <= index < stop and index not in removed] + kept = [entry for index, entry in enumerate(original) if index not in removed] history[start:stop] = iter(replacement) if as_iterator else replacement - assert all(actual is wanted for actual, wanted in zip(history, expected, strict=True)) - assert_session(history, expected.session_start_index, [entry.raw for entry in expected.span(":").values()]) + assert all(actual is wanted for actual, wanted in zip(history, kept, strict=True)) + assert current_session(history) == { + position: entry.raw for position, entry in enumerate(kept, start=1) if entry.raw.startswith("new-") + } -def test_filtering_distinguishes_equal_commands_from_different_sessions(history): +def test_equal_commands_from_different_sessions_are_separate_entries(history): current = item(history[0].raw) assert current == history[0] assert current is not history[0] history[4] = current history[:] = history[1:] assert history[3] is current - assert_session(history, 3, ["old-0", "new-1"]) + assert current_session(history) == {4: "old-0", 5: "new-1"} -@pytest.mark.parametrize( - ("indices", "boundary", "current"), - [ - ([3, 1, 4, 5], 2, ["new-0", "new-1"]), - ([4, 1, 5], 0, ["new-0", "old-1", "new-1"]), - ([1, 3], 2, []), - ([0, 0, 4], 2, ["new-0"]), - ([4, 0, 5, 1], 0, ["new-0", "old-0", "new-1", "old-1"]), - ([0, 2, 1, 3, 4, 5], 4, ["new-0", "new-1"]), - ], -) -def test_reordered_straddling_slice_retains_only_leading_persisted_entries(history, indices, boundary, current): - history[:] = [history[index] for index in indices] - assert_session(history, boundary, current) - - -def test_new_equal_entry_in_straddling_slice_starts_current_session(history): +def test_new_equal_entry_in_slice_assignment_is_current(history): new_entry = item(history[0].raw) history[:] = [new_entry, history[1], history[4]] - assert_session(history, 0, ["old-0", "old-1", "new-0"]) + assert current_session(history) == {1: "old-0", 3: "new-0"} + +def test_reappending_persisted_entry_object_keeps_it_persisted(history): + history.append(history[2]) + history.append(history[2].statement) + assert current_session(history) == {5: "new-0", 6: "new-1", 8: "old-2"} -def test_failed_straddling_replacement_preserves_history(history): + +@pytest.mark.parametrize( + "clone", + [copy.deepcopy, lambda hist: pickle.loads(pickle.dumps(hist))], + ids=["deepcopy", "pickle"], +) +def test_cloned_history_keeps_persisted_entries(history, clone): + assert current_session(clone(history)) == {5: "new-0", 6: "new-1"} + + +@pytest.mark.parametrize("operation", ["delete", "pop", "remove", "slice_step", "slice_length", "slice_iterator"]) +def test_failed_mutation_preserves_history_and_membership(history, operation): original = list(history) def broken_items(): - yield original[0] + yield item("replacement-0") raise RuntimeError("incomplete replacement") - with pytest.raises(RuntimeError, match="incomplete replacement"): - history[:] = broken_items() + error = IndexError if operation in ("delete", "pop") else RuntimeError if operation == "slice_iterator" else ValueError + + def mutate(): + if operation == "delete": + del history[-99] + elif operation == "pop": + history.pop(-99) + elif operation == "remove": + history.remove(item("missing-0")) + elif operation == "slice_step": + del history[::0] + elif operation == "slice_length": + history[::2] = [item("replacement-0")] + else: + history[:2] = broken_items() + + with pytest.raises(error): + mutate() assert all(actual is expected for actual, expected in zip(history, original, strict=True)) - assert_session(history, 4, ["new-0", "new-1"]) + assert current_session(history) == {5: "new-0", 6: "new-1"} + + +def test_mutations_compose_with_truncate_clear_and_append(history): + history.insert(0, item("inserted-0")) + history.pop(1) + history.truncate(4) + history.append(Statement("", raw="new-2")) + assert current_session(history) == {3: "new-0", 4: "new-1", 5: "new-2"} + history.clear() + history.extend([item("new-3")]) + assert current_session(history) == {1: "new-3"} + + +@pytest.mark.parametrize( + ("mutate", "selector"), + [(lambda hist: hist.reverse(), "5:6"), (repeat_twice, "7:10")], + ids=["reverse", "repeat"], +) +def test_moved_persisted_commands_still_require_all(base_app, history, monkeypatch, mutate, selector): + base_app.history = history + mutate(history) + replay = mock.Mock(return_value=False) + monkeypatch.setattr(base_app, "runcmds_plus_hooks", replay) + + _out, err = run_cmd(base_app, f"history --run {selector}") + + assert err == ["Use --all to run or edit commands from previous sessions."] + replay.assert_not_called()