From 9bd01000108769b3d03597e22a42e831ccf8297d Mon Sep 17 00:00:00 2001 From: Rodolfo Olivieri Date: Tue, 7 Jan 2025 16:38:38 -0300 Subject: [PATCH] Include a filter option for history management The --filter accepts a keyword to filter the user history based on that keyword. --- command_line_assistant/commands/history.py | 40 ++++-- command_line_assistant/dbus/interfaces.py | 37 ++++- .../history/plugins/local.py | 3 + command_line_assistant/history/schemas.py | 12 +- command_line_assistant/logger.py | 2 +- tests/commands/test_history.py | 33 +++++ tests/conftest.py | 35 +++++ tests/dbus/test_interfaces.py | 132 +++++++++++++----- tests/history/plugins/test_local.py | 35 ----- tests/history/test_schemas.py | 33 +++-- 10 files changed, 263 insertions(+), 99 deletions(-) diff --git a/command_line_assistant/commands/history.py b/command_line_assistant/commands/history.py index eb91598..447cfa1 100644 --- a/command_line_assistant/commands/history.py +++ b/command_line_assistant/commands/history.py @@ -1,6 +1,7 @@ """Module to handle the history command.""" from argparse import Namespace +from typing import Optional from command_line_assistant.dbus.constants import HISTORY_IDENTIFIER from command_line_assistant.dbus.exceptions import ( @@ -25,7 +26,9 @@ class HistoryCommand(BaseCLICommand): """Class that represents the history command.""" - def __init__(self, clear: bool, first: bool, last: bool) -> None: + def __init__( + self, clear: bool, first: bool, last: bool, filter: Optional[str] = None + ) -> None: """Constructor of the class. Note: @@ -36,10 +39,12 @@ def __init__(self, clear: bool, first: bool, last: bool) -> None: clear (bool): If the history should be cleared first (bool): Retrieve only the first conversation from history last (bool): Retrieve only last conversation from history + filter (Optional[str], optional): Keyword to filter in the user history """ self._clear = clear self._first = first self._last = last + self._filter = filter self._proxy = HISTORY_IDENTIFIER.get_proxy() @@ -67,14 +72,13 @@ def run(self) -> int: try: if self._clear: self._clear_history() - - if self._first: + elif self._first: self._retrieve_first_conversation() - - if self._last: + elif self._last: self._retrieve_last_conversation() - - if not self._last and not self._clear and not self._first: + elif self._filter: + self._retrieve_conversation_filtered(self._filter) + else: self._retrieve_all_conversations() return 0 @@ -100,7 +104,22 @@ def _retrieve_first_conversation(self) -> None: # Display the conversation self._show_history(history.entries) - def _retrieve_last_conversation(self): + def _retrieve_conversation_filtered(self, filter: str) -> None: + """Retrieve the user conversation with keyword filtering. + + Args: + filter (str): Keyword to filter in the user history + """ + self._text_renderer.render("Filtering conversation history.") + response = self._proxy.GetFilteredConversation(filter) + + # Handle and display the response + history = HistoryEntry.from_structure(response) + + # Display the conversation + self._show_history(history.entries) + + def _retrieve_last_conversation(self) -> None: """Retrieve the last conversation in the conversation cache.""" self._text_renderer.render("Getting last conversation from history.") response = self._proxy.GetLastConversation() @@ -165,6 +184,9 @@ def register_subcommand(parser: SubParsersAction): action="store_true", help="Get the last conversation from history.", ) + history_parser.add_argument( + "--filter", help="Search for a specific string of text in the history." + ) history_parser.set_defaults(func=_command_factory) @@ -177,4 +199,4 @@ def _command_factory(args: Namespace) -> HistoryCommand: Returns: HistoryCommand: Return an instance of class """ - return HistoryCommand(args.clear, args.first, args.last) + return HistoryCommand(args.clear, args.first, args.last, args.filter) diff --git a/command_line_assistant/dbus/interfaces.py b/command_line_assistant/dbus/interfaces.py index cb37cd2..fe576a4 100644 --- a/command_line_assistant/dbus/interfaces.py +++ b/command_line_assistant/dbus/interfaces.py @@ -5,7 +5,7 @@ from dasbus.server.interface import dbus_interface from dasbus.server.property import emits_properties_changed from dasbus.server.template import InterfaceTemplate -from dasbus.typing import Structure +from dasbus.typing import Str, Structure from command_line_assistant.daemon.http.query import submit from command_line_assistant.dbus.constants import HISTORY_IDENTIFIER, QUERY_IDENTIFIER @@ -17,6 +17,7 @@ from command_line_assistant.history.plugins.local import LocalHistory audit_logger = logging.getLogger("audit") +logger = logging.getLogger(__name__) @dbus_interface(QUERY_IDENTIFIER.interface_name) @@ -121,6 +122,40 @@ def GetLastConversation(self) -> Structure: return HistoryEntry.to_structure(history_entry) + def GetFilteredConversation(self, filter: Str) -> Structure: + """Get last conversation from history. + + Args: + filter (str): The filter + + Returns: + Structure: A single history entyr in a dbus structure format. + """ + manager = HistoryManager(self.implementation.config, LocalHistory) + history = manager.read() + history_entry = HistoryEntry() + found_entries = [] + + if history.history: + logger.info("Filtering the user history with keyword '%s'", filter) + # We ignore the type in the condition as pyright thinks that "Str" is not "str". + # Pyright is correct about this, but "Str" is a special type for dbus. It will be "str" in the end. + found_entries = [ + entry + for entry in history.history + if ( + filter in entry.interaction.query.text # type: ignore + or filter in entry.interaction.response.text # type: ignore + ) + ] + + logger.info("Found %s entries in the history", len(found_entries)) + # Normalize the entries to send over dbus + _ = [ + history_entry.set_from_dict(entry.to_dict()) for entry in set(found_entries) + ] + return HistoryEntry.to_structure(history_entry) + def ClearHistory(self) -> None: """Clear the user history.""" manager = HistoryManager(self.implementation.config, LocalHistory) diff --git a/command_line_assistant/history/plugins/local.py b/command_line_assistant/history/plugins/local.py index 6d8d8a8..dd8ecc2 100644 --- a/command_line_assistant/history/plugins/local.py +++ b/command_line_assistant/history/plugins/local.py @@ -32,6 +32,7 @@ def read(self) -> History: filepath = self._config.history.file + logger.info("Reading history at %s", filepath) try: data = filepath.read_text() return History.from_json(data) @@ -63,6 +64,7 @@ def write(self, current_history: History, query: str, response: str) -> None: filepath = self._config.history.file final_history = self._add_new_entry(current_history, query, response) + logger.info("Writting user history at %s", filepath) try: filepath.write_text(final_history.to_json()) except json.JSONDecodeError as e: @@ -85,6 +87,7 @@ def clear(self) -> None: # Write empty history current_history = History() filepath = self._config.history.file + logger.info("Clearing history at %s", filepath) try: filepath.write_text(current_history.to_json()) logger.info("History cleared successfully") diff --git a/command_line_assistant/history/schemas.py b/command_line_assistant/history/schemas.py index 22e0876..f3dffc7 100644 --- a/command_line_assistant/history/schemas.py +++ b/command_line_assistant/history/schemas.py @@ -10,7 +10,7 @@ from command_line_assistant.constants import VERSION -@dataclass +@dataclass(frozen=True) class QueryData: """Schema to represent a query emited by the user. @@ -23,7 +23,7 @@ class QueryData: role: str = "user" -@dataclass +@dataclass(frozen=True) class ResponseData: """Schema to represent the LLM response. @@ -38,7 +38,7 @@ class ResponseData: role: str = "assistant" -@dataclass +@dataclass(frozen=True) class InteractionData: """Schema to represent the interaction data between user and LLM. @@ -51,7 +51,7 @@ class InteractionData: response: ResponseData = field(default_factory=ResponseData) -@dataclass +@dataclass(frozen=True) class OSInfo: """Schema to represent the system information @@ -66,7 +66,7 @@ class OSInfo: arch: str = platform.architecture()[0] -@dataclass +@dataclass(frozen=True) class EntryMetadata: """Schema to represent the entry metadata information @@ -79,7 +79,7 @@ class EntryMetadata: os_info: OSInfo = field(default_factory=OSInfo) -@dataclass +@dataclass(frozen=True) class HistoryEntry: """Schema to represent an entry of the history diff --git a/command_line_assistant/logger.py b/command_line_assistant/logger.py index f34a7e0..3857f43 100644 --- a/command_line_assistant/logger.py +++ b/command_line_assistant/logger.py @@ -29,7 +29,7 @@ }, "audit_file": { "class": "logging.FileHandler", - "filename": "/var/log/audit/command-line-assistant.log", + "filename": "/tmp/command-line-assistant.log", "formatter": "audit", "mode": "a", }, diff --git a/tests/commands/test_history.py b/tests/commands/test_history.py index 2546032..ca01c30 100644 --- a/tests/commands/test_history.py +++ b/tests/commands/test_history.py @@ -62,6 +62,39 @@ def test_retrieve_all_conversations_empty(mock_proxy, capsys): assert "No history found.\n" in captured.out +def test_retrieve_conversation_filtered_empty(mock_proxy, capsys): + """Test retrieving first conversation when history is empty.""" + empty_history = HistoryEntry() + mock_proxy.GetFilteredConversation.return_value = empty_history.to_structure( + empty_history + ) + + HistoryCommand( + clear=False, first=True, last=False, filter="missing" + )._retrieve_conversation_filtered(filter="missing") + captured = capsys.readouterr() + assert "No history found.\n" in captured.out + + +def test_retrieve_conversation_filtered_success( + mock_proxy, sample_history_entry, capsys +): + """Test retrieving last conversation successfully.""" + mock_proxy.GetFilteredConversation.return_value = sample_history_entry.to_structure( + sample_history_entry + ) + + HistoryCommand( + clear=False, first=False, last=True, filter="test" + )._retrieve_conversation_filtered(filter="missing") + captured = capsys.readouterr() + mock_proxy.GetFilteredConversation.assert_called_once() + assert ( + "\x1b[92mQuery: test query\x1b[0m\n\x1b[94mAnswer: test response\x1b[0m\n" + in captured.out + ) + + def test_retrieve_first_conversation_success(mock_proxy, sample_history_entry, capsys): """Test retrieving first conversation successfully.""" mock_proxy.GetFirstConversation.return_value = sample_history_entry.to_structure( diff --git a/tests/conftest.py b/tests/conftest.py index 8a9739c..09e3774 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -77,3 +77,38 @@ def mock_proxy(): @pytest.fixture def mock_stream(): return MockStream() + + +@pytest.fixture +def sample_history_data(): + """Create sample history data for testing.""" + return { + "history": [ + { + "id": "test-id", + "timestamp": "2024-01-01T00:00:00Z", + "interaction": { + "query": {"text": "test query", "role": "user"}, + "response": { + "text": "test response", + "tokens": 2, + "role": "assistant", + }, + }, + "metadata": { + "session_id": "test-session", + "os_info": { + "distribution": "RHEL", + "version": "test", + "arch": "x86_64", + }, + }, + } + ], + "metadata": { + "last_updated": "2024-01-01T00:00:00Z", + "version": "0.1.0", + "entry_count": 1, + "size_bytes": 0, + }, + } diff --git a/tests/dbus/test_interfaces.py b/tests/dbus/test_interfaces.py index c616338..87c646d 100644 --- a/tests/dbus/test_interfaces.py +++ b/tests/dbus/test_interfaces.py @@ -39,41 +39,6 @@ def history_interface(mock_implementation): return interface -@pytest.fixture -def sample_history_data(): - """Create sample history data for testing.""" - return { - "history": [ - { - "id": "test-id", - "timestamp": "2024-01-01T00:00:00Z", - "interaction": { - "query": {"text": "test query", "role": "user"}, - "response": { - "text": "test response", - "tokens": 2, - "role": "assistant", - }, - }, - "metadata": { - "session_id": "test-session", - "os_info": { - "distribution": "RHEL", - "version": "test", - "arch": "x86_64", - }, - }, - } - ], - "metadata": { - "last_updated": "2024-01-01T00:00:00Z", - "version": "0.1.0", - "entry_count": 1, - "size_bytes": 0, - }, - } - - def test_query_interface_retrieve_answer(query_interface, mock_implementation): """Test retrieving answer from query interface.""" expected_response = "test response" @@ -153,6 +118,103 @@ def test_history_interface_get_last_conversation( assert reconstructed.entries[0].response == "test response" +def test_history_interface_get_filtered_conversation( + history_interface, mock_implementation, sample_history_data +): + """Test getting filtered conversation through history interface.""" + mock_history = History.from_json(json.dumps(sample_history_data)) + + with patch("command_line_assistant.dbus.interfaces.HistoryManager") as mock_manager: + mock_manager.return_value.read.return_value = mock_history + response = history_interface.GetFilteredConversation(filter="test") + + reconstructed = HistoryEntry.from_structure(response) + assert len(reconstructed.entries) == 1 + assert reconstructed.entries[0].query == "test query" + assert reconstructed.entries[0].response == "test response" + + +def test_history_interface_get_filtered_conversation_duplicate_entries( + history_interface, mock_implementation, sample_history_data +): + """Test getting filtered conversation through duplicate history interface.""" + # Add a new entry manually + sample_history_data["history"].append( + { + "id": "test-id", + "timestamp": "2024-01-01T00:00:00Z", + "interaction": { + "query": {"text": "test query", "role": "user"}, + "response": { + "text": "test response", + "tokens": 2, + "role": "assistant", + }, + }, + "metadata": { + "session_id": "test-session", + "os_info": { + "distribution": "RHEL", + "version": "test", + "arch": "x86_64", + }, + }, + } + ) + mock_history = History.from_json(json.dumps(sample_history_data)) + + with patch("command_line_assistant.dbus.interfaces.HistoryManager") as mock_manager: + mock_manager.return_value.read.return_value = mock_history + response = history_interface.GetFilteredConversation(filter="test") + + reconstructed = HistoryEntry.from_structure(response) + assert len(reconstructed.entries) == 1 + assert reconstructed.entries[0].query == "test query" + assert reconstructed.entries[0].response == "test response" + + +def test_history_interface_get_filtered_conversation_duplicate_entries_not_matching( + history_interface, mock_implementation, sample_history_data +): + """Test getting filtered conversation through duplicated history interface. + + This test will have a duplicated entry, but not matching the "id". This should be enough to be considered a new entry + """ + # Add a new entry manually + sample_history_data["history"].append( + { + "id": "test-other-id", + "timestamp": "2024-01-01T00:00:00Z", + "interaction": { + "query": {"text": "test query", "role": "user"}, + "response": { + "text": "test response", + "tokens": 2, + "role": "assistant", + }, + }, + "metadata": { + "session_id": "test-session", + "os_info": { + "distribution": "RHEL", + "version": "test", + "arch": "x86_64", + }, + }, + } + ) + mock_history = History.from_json(json.dumps(sample_history_data)) + + with patch("command_line_assistant.dbus.interfaces.HistoryManager") as mock_manager: + mock_manager.return_value.read.return_value = mock_history + response = history_interface.GetFilteredConversation(filter="test") + + reconstructed = HistoryEntry.from_structure(response) + assert len(reconstructed.entries) == 2 + assert reconstructed.entries[0].query == "test query" + assert reconstructed.entries[0].response == "test response" + + def test_history_interface_clear_history(history_interface): """Test clearing history through history interface.""" with patch("command_line_assistant.dbus.interfaces.HistoryManager") as mock_manager: diff --git a/tests/history/plugins/test_local.py b/tests/history/plugins/test_local.py index f7112ef..8a276bb 100644 --- a/tests/history/plugins/test_local.py +++ b/tests/history/plugins/test_local.py @@ -21,41 +21,6 @@ def local_history(mock_config): return LocalHistory(mock_config) -@pytest.fixture -def sample_history_data(): - """Create sample history data for testing.""" - return { - "history": [ - { - "id": "test-id", - "timestamp": "2024-01-01T00:00:00Z", - "interaction": { - "query": {"text": "test query", "role": "user"}, - "response": { - "text": "test response", - "tokens": 2, - "role": "assistant", - }, - }, - "metadata": { - "session_id": "test-session", - "os_info": { - "distribution": "RHEL", - "version": "test", - "arch": "x86_64", - }, - }, - } - ], - "metadata": { - "last_updated": "2024-01-01T00:00:00Z", - "version": "0.1.0", - "entry_count": 1, - "size_bytes": 0, - }, - } - - def test_read_when_history_disabled(local_history): """Test reading history when history is disabled.""" local_history._config.history.enabled = False diff --git a/tests/history/test_schemas.py b/tests/history/test_schemas.py index 8e3c9dc..43c9db1 100644 --- a/tests/history/test_schemas.py +++ b/tests/history/test_schemas.py @@ -92,9 +92,11 @@ def test_history_entry_initialization(): def test_history_entry_to_dict(): """Test HistoryEntry to_dict conversion""" - entry = HistoryEntry() - entry.interaction.query.text = "test query" - entry.interaction.response.text = "test response" + entry = HistoryEntry( + interaction=InteractionData( + QueryData("test query"), ResponseData("test response") + ) + ) entry_dict = entry.to_dict() assert isinstance(entry_dict, dict) @@ -126,9 +128,11 @@ def test_history_json_serialization(): """Test History to_json and from_json methods""" # Create a history with some test data history = History() - entry = HistoryEntry() - entry.interaction.query.text = "test query" - entry.interaction.response.text = "test response" + entry = HistoryEntry( + interaction=InteractionData( + QueryData("test query"), ResponseData("test response") + ) + ) history.history.append(entry) # Convert to JSON @@ -161,9 +165,11 @@ def test_history_with_multiple_entries(): ] for query_text, response_text in entries: - entry = HistoryEntry() - entry.interaction.query.text = query_text - entry.interaction.response.text = response_text + entry = HistoryEntry( + interaction=InteractionData( + QueryData(query_text), ResponseData(response_text) + ) + ) history.history.append(entry) # Verify entries @@ -176,9 +182,12 @@ def test_history_with_multiple_entries(): def test_history_json_roundtrip_with_special_characters(): """Test History JSON serialization with special characters""" history = History() - entry = HistoryEntry() - entry.interaction.query.text = "test\nquery with 'special' \"characters\" & symbols" - entry.interaction.response.text = "response\twith\nspecial\rcharacters" + entry = HistoryEntry( + interaction=InteractionData( + QueryData("test\nquery with 'special' \"characters\" & symbols"), + ResponseData("response\twith\nspecial\rcharacters"), + ) + ) history.history.append(entry) # Roundtrip through JSON