Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions cmd2/cmd2.py
Original file line number Diff line number Diff line change
Expand Up @@ -5722,6 +5722,7 @@ def _persist_history(self) -> None:
except ModuleNotFoundError: # pragma: no cover
import bz2 as compress_lib # type: ignore[no-redef]

# Truncate a copy so live history numbering and the session boundary are unchanged.
history = History(self.history)
history.truncate(self._persistent_history_length)
history_json = history.to_json()
Expand Down
102 changes: 28 additions & 74 deletions tests/test_history.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
import pytest

import cmd2
from cmd2.history import HistoryFormatError, HistoryIndexError
from cmd2.history import History, HistoryFormatError, HistoryIndexError
from cmd2.parsing import Statement

from .conftest import (
Expand Down Expand Up @@ -1745,26 +1745,6 @@ def test_history_populates_pt(hist_file) -> None:
# we assume that the atexit module will call this method
# properly
#
def test_persist_history_ensure_no_error_if_no_histfile(base_app, capsys) -> None:
# make sure if there is no persistent history file and someone
# calls the private method call that we don't get an error
base_app._persist_history()
out, err = capsys.readouterr()
assert not out
assert not err


def test_persist_history_permission_error(hist_file, mocker, capsys) -> None:
app = cmd2.Cmd(persistent_history_file=hist_file)
run_cmd(app, "help")
mock_open = mocker.patch("builtins.open")
mock_open.side_effect = PermissionError
app._persist_history()
out, err = capsys.readouterr()
assert not out
assert "Cannot write persistent history file" in err


@pytest.fixture
def persistence_app(base_app, hist, tmp_path):
base_app.history = hist
Expand All @@ -1774,15 +1754,10 @@ def persistence_app(base_app, hist, tmp_path):


def read_persisted_history(app):
try:
import lzma as compress_lib
except ModuleNotFoundError:
import bz2 as compress_lib

from cmd2.history import History
import lzma

with open(app.persistent_history_file, "rb") as stream:
return History.from_json(compress_lib.decompress(stream.read()).decode("utf-8"))
return History.from_json(lzma.decompress(stream.read()).decode("utf-8"))


def history_state(app):
Expand All @@ -1792,11 +1767,33 @@ def history_state(app):
tuple(id(item) for item in history),
history.to_json(),
history.session_start_index,
history.span(":"),
history.str_search("third"),
)


def test_persist_history_ensure_no_error_if_no_histfile(base_app, capsys) -> None:
# make sure if there is no persistent history file and someone
# calls the private method call that we don't get an error
base_app._persist_history()
out, err = capsys.readouterr()
assert not out
assert not err


def test_persist_history_permission_error(hist_file, mocker, capsys) -> None:
app = cmd2.Cmd(persistent_history_file=hist_file)
run_cmd(app, "help")
run_cmd(app, "shortcuts")
app._persistent_history_length = 1
before = history_state(app)
mock_open = mocker.patch("builtins.open")
mock_open.side_effect = PermissionError
app._persist_history()
out, err = capsys.readouterr()
assert not out
assert "Cannot write persistent history file" in err
assert history_state(app) == before


@pytest.mark.parametrize("limit", [-1, 0, 1, 2, 4, 6])
def test_persist_history_preserves_live_state(persistence_app, limit) -> None:
app = persistence_app
Expand All @@ -1810,17 +1807,7 @@ def test_persist_history_preserves_live_state(persistence_app, limit) -> None:
assert history_state(app) == before


def test_persist_history_empty(persistence_app) -> None:
app = persistence_app
app.history.clear()
app._persistent_history_length = 2
before = history_state(app)
app._persist_history()
assert not read_persisted_history(app)
assert history_state(app) == before


def test_persist_history_repeat_append_and_reload(persistence_app, mocker) -> None:
def test_persist_history_repeat_and_append(persistence_app) -> None:
app = persistence_app
app._persistent_history_length = 2
before = history_state(app)
Expand All @@ -1833,36 +1820,3 @@ def test_persist_history_repeat_append_and_reload(persistence_app, mocker) -> No
app._persist_history()
assert [item.raw for item in read_persisted_history(app)] == ["fourth", "fifth café"]
assert history_state(app) == before
assert app.history.get(1).raw == "first"
assert list(app.history.span(":")) == [3, 4, 5]

# The reload is temporary and must not register a later write to a deleted file.
mocker.patch("atexit.register")
reloaded = cmd2.Cmd(persistent_history_file=app.persistent_history_file)
assert [item.raw for item in reloaded.history] == ["fourth", "fifth café"]
assert reloaded.history.session_start_index == 2
assert reloaded.main_session.history.get_strings() == ["fourth", "fifth café"]
assert history_state(app) == before


def test_persist_history_no_file_preserves_live_state(persistence_app, mocker) -> None:
app = persistence_app
app.persistent_history_file = ""
app._persistent_history_length = 1
before = history_state(app)
open_mock = mocker.patch("builtins.open")
app._persist_history()
open_mock.assert_not_called()
assert history_state(app) == before


def test_persist_history_write_error_preserves_live_state(persistence_app, mocker) -> None:
app = persistence_app
app._persistent_history_length = 1
before = history_state(app)
mocker.patch("builtins.open", side_effect=PermissionError("synthetic write failure"))
error = mocker.patch.object(app, "perror")
app._persist_history()
assert "Cannot write persistent history file" in error.call_args.args[0]
assert "synthetic write failure" in error.call_args.args[0]
assert history_state(app) == before
Loading