diff --git a/CHANGELOG.md b/CHANGELOG.md index 2bc9b730b..1a5a68e01 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,9 @@ 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 `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 diff --git a/cmd2/history.py b/cmd2/history.py index 8856bb610..4fbd2d4f5 100644 --- a/cmd2/history.py +++ b/cmd2/history.py @@ -1,6 +1,7 @@ """History management classes.""" import json +import operator import re from collections.abc import ( Callable, @@ -9,6 +10,9 @@ from dataclasses import dataclass from typing import ( Any, + Self, + SupportsIndex, + cast, overload, ) @@ -167,6 +171,79 @@ def __init__(self, seq: Iterable[HistoryItem] = ()) -> None: super().__init__(seq) self.session_start_index = 0 + 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 + + @overload + def __setitem__(self, key: SupportsIndex, value: HistoryItem, /) -> None: ... # pragma: no cover + + @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 + + 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 + def start_session(self) -> None: """Start a new session, thereby setting the next index as the first index in the new session.""" self.session_start_index = len(self) @@ -331,8 +408,6 @@ def truncate(self, max_length: int) -> None: elif len(self) > max_length: last_element = len(self) - max_length del self[0:last_element] - # keep the session boundary pointing at the same retained items - self.session_start_index = max(0, self.session_start_index - last_element) def _build_result_dictionary( self, start: int, end: int, filter_func: Callable[[HistoryItem], bool] | None = None diff --git a/tests/test_history_mutations.py b/tests/test_history_mutations.py new file mode 100644 index 000000000..b8963df95 --- /dev/null +++ b/tests/test_history_mutations.py @@ -0,0 +1,251 @@ +"""Session selection survives structural changes to the public history list.""" + +import pytest + +from cmd2.history import History, HistoryItem +from cmd2.parsing import Statement + + +def item(raw): + return HistoryItem(Statement("", raw=raw)) + + +@pytest.fixture +def history(): + 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)) + + +@pytest.mark.parametrize( + ("index", "boundary", "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"]), + ], +) +def test_delete_tracks_session(history, index, boundary, expected): + del history[index] + assert_session(history, boundary, expected) + + +@pytest.mark.parametrize( + ("index", "removed", "boundary", "expected"), + [ + (0, "old-0", 3, ["new-0", "new-1"]), + (-3, "old-3", 3, ["new-0", "new-1"]), + (-1, "new-1", 4, ["new-0"]), + ], +) +def test_pop_tracks_session(history, index, removed, boundary, expected): + assert history.pop(index).raw == removed + assert_session(history, boundary, 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): + assert history.remove(item(raw)) is None + assert_session(history, boundary, expected) + + +@pytest.mark.parametrize( + ("index", "boundary", "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"]), + ], +) +def test_insert_tracks_session(history, index, boundary, expected): + history.insert(index, item("inserted-0")) + assert_session(history, boundary, expected) + + +@pytest.mark.parametrize( + ("index", "count", "boundary", "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"]), + ], +) +def test_slice_assignment_tracks_session(history, index, count, boundary, 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"]) + + +@pytest.mark.parametrize("index", [0, 3, 4, -1]) +def test_index_assignment_preserves_session(history, index): + history[index] = item("replacement-0") + assert_session(history, 4, [entry.raw for entry in history[4:]]) + + +@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:]]) + + +@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 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 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 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"]) + + +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") + + 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("as_iterator", [False, True]) +@pytest.mark.parametrize( + ("start", "stop", "removed"), + [ + (0, 6, {1}), + (0, 6, {0, 2}), + (0, 6, {4}), + (0, 6, {5}), + (0, 6, {0, 4}), + (1, 6, {1, 3}), + (2, 6, {2, 4}), + (3, 5, {3}), + (0, 5, {1, 4}), + ], +) +def test_filtering_straddling_slice_matches_deletion(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] + 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()]) + + +def test_filtering_distinguishes_equal_commands_from_different_sessions(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"]) + + +@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): + new_entry = item(history[0].raw) + history[:] = [new_entry, history[1], history[4]] + assert_session(history, 0, ["old-0", "old-1", "new-0"]) + + +def test_failed_straddling_replacement_preserves_history(history): + original = list(history) + + def broken_items(): + yield original[0] + raise RuntimeError("incomplete replacement") + + with pytest.raises(RuntimeError, match="incomplete replacement"): + history[:] = broken_items() + assert all(actual is expected for actual, expected in zip(history, original, strict=True)) + assert_session(history, 4, ["new-0", "new-1"])