diff --git a/command_line_assistant/commands/history.py b/command_line_assistant/commands/history.py index 60622fe..f2d9c34 100644 --- a/command_line_assistant/commands/history.py +++ b/command_line_assistant/commands/history.py @@ -4,26 +4,129 @@ from dasbus.error import DBusError from command_line_assistant.dbus.constants import HISTORY_IDENTIFIER +from command_line_assistant.dbus.structures import HistoryEntry +from command_line_assistant.rendering.decorators.colors import ColorDecorator +from command_line_assistant.rendering.decorators.text import ( + EmojiDecorator, + TextWrapDecorator, +) +from command_line_assistant.rendering.renders.spinner import SpinnerRenderer +from command_line_assistant.rendering.renders.text import TextRenderer +from command_line_assistant.rendering.stream import StdoutStream from command_line_assistant.utils.cli import BaseCLICommand, SubParsersAction logger = logging.getLogger(__name__) +def _initialize_spinner_renderer() -> SpinnerRenderer: + spinner = SpinnerRenderer(message="Loading history", stream=StdoutStream(end="")) + spinner.update(TextWrapDecorator()) + + return spinner + + +def _initialize_qa_renderer(is_assistant: bool = False) -> TextRenderer: + text = TextRenderer(stream=StdoutStream()) + foreground = "lightblue" if is_assistant else "lightgreen" + text.update(ColorDecorator(foreground=foreground)) + text.update(EmojiDecorator("🤖")) + return text + + +def _initialize_text_renderer() -> TextRenderer: + text = TextRenderer(stream=StdoutStream()) + return text + + class HistoryCommand(BaseCLICommand): - def __init__(self, clear: bool) -> None: + def __init__(self, clear: bool, first: bool, last: bool) -> None: self._clear = clear + self._first = first + self._last = last + + self._proxy = HISTORY_IDENTIFIER.get_proxy() + self._user_renderer = _initialize_qa_renderer() + self._assistant_renderer = _initialize_qa_renderer(is_assistant=True) + self._text_renderer = _initialize_text_renderer() + self._spinner_renderer = _initialize_spinner_renderer() super().__init__() def run(self) -> None: - proxy = HISTORY_IDENTIFIER.get_proxy() - if self._clear: - try: - logger.info("Cleaning the history.") - proxy.ClearHistory() - except DBusError as e: - logger.info("Failed to clean the history: %s", e) - raise e + return self._clear_history() + + if self._first: + return self._retrieve_first_conversation() + + if self._last: + return self._retrieve_last_conversation() + + return self._retrieve_all_conversations() + + def _retrieve_all_conversations(self): + """Retrieve and display all conversations from history.""" + try: + logger.info("Getting all conversations from history.") + response = self._proxy.GetHistory() + history = HistoryEntry.from_structure(response) + + if history.entries: + for entry in history.entries: + self._user_renderer.render(f"Query: {entry.query}") + self._assistant_renderer.render(f"Answer: {entry.response}") + self._text_renderer.render(f"Time: {entry.timestamp}") + self._text_renderer.render( + "-" * 50 + ) # Separator between conversations + else: + print("No history found.") + except DBusError as e: + logger.info("Failed to get history: %s", e) + raise e + + def _retrieve_first_conversation(self): + try: + logger.info("Getting first conversation from history.") + response = self._proxy.GetFirstConversation() + history = HistoryEntry.from_structure(response) + if history.entries: + # Display the conversation + entry = history.entries[0] + self._user_renderer.render(f"Query: {entry.query}") + self._assistant_renderer.render(f"Answer: {entry.response}") + self._text_renderer.render(f"Time: {entry.timestamp}") + else: + print("No history found.") + except DBusError as e: + logger.info("Failed to get first conversation: %s", e) + raise e + + def _retrieve_last_conversation(self): + try: + logger.info("Getting last conversation from history.") + response = self._proxy.GetLastConversation() + + # Handle and display the response + history = HistoryEntry.from_structure(response) + if history.entries: + # Display the conversation + entry = history.entries[0] + self._user_renderer.render(f"Query: {entry.query}") + self._assistant_renderer.render(f"Answer: {entry.response}") + self._text_renderer.render(f"Time: {entry.timestamp}") + else: + print("No history found.") + except DBusError as e: + logger.info("Failed to get last conversation: %s", e) + raise e + + def _clear_history(self) -> None: + try: + logger.info("Cleaning the history.") + self._proxy.ClearHistory() + except DBusError as e: + logger.info("Failed to clean the history: %s", e) + raise e def register_subcommand(parser: SubParsersAction): @@ -38,10 +141,22 @@ def register_subcommand(parser: SubParsersAction): help="Manage conversation history", ) history_parser.add_argument( - "--clear", action="store_true", help="Clear the history." + "--clear", + action="store_true", + help="Clear the history.", + ) + history_parser.add_argument( + "--first", + action="store_true", + help="Get the first conversation from history.", + ) + history_parser.add_argument( + "--last", + action="store_true", + help="Get the last conversation from history.", ) history_parser.set_defaults(func=_command_factory) def _command_factory(args: Namespace) -> HistoryCommand: - return HistoryCommand(args.clear) + return HistoryCommand(args.clear, args.first, args.last) diff --git a/command_line_assistant/config/schemas.py b/command_line_assistant/config/schemas.py index 1b152da..2d52c0e 100644 --- a/command_line_assistant/config/schemas.py +++ b/command_line_assistant/config/schemas.py @@ -38,9 +38,8 @@ class HistorySchema: enabled: bool = True file: Union[str, Path] = Path( # type: ignore - "~/.local/share/command-line-assistant/command-line-assistant_history.json" + "/var/lib/command-line-assistant/history.json" ) - max_size: int = 100 def __post_init__(self): self.file: Path = Path(self.file).expanduser() diff --git a/command_line_assistant/dbus/interfaces.py b/command_line_assistant/dbus/interfaces.py index ad1ed28..f3ca084 100644 --- a/command_line_assistant/dbus/interfaces.py +++ b/command_line_assistant/dbus/interfaces.py @@ -9,7 +9,8 @@ HistoryEntry, Message, ) -from command_line_assistant.history import handle_history_read, handle_history_write +from command_line_assistant.history.manager import HistoryManager +from command_line_assistant.history.plugins.local import LocalHistory @dbus_interface(QUERY_IDENTIFIER.interface_name) @@ -19,11 +20,13 @@ class QueryInterface(InterfaceTemplate): @property def RetrieveAnswer(self) -> Structure: """This method is mainly called by the client to retrieve it's answer.""" - llm_response = submit( - self.implementation.query.message, self.implementation.config - ) + query = self.implementation.query.message + llm_response = submit(query, self.implementation.config) message = Message() message.message = llm_response + manager = HistoryManager(self.implementation.config, LocalHistory) + current_history = manager.read() + manager.write(current_history, query, llm_response) return Message.to_structure(message) @emits_properties_changed @@ -34,11 +37,47 @@ def ProcessQuery(self, query: Structure) -> None: @dbus_interface(HISTORY_IDENTIFIER.interface_name) class HistoryInterface(InterfaceTemplate): - @property def GetHistory(self) -> Structure: - history = HistoryEntry() - history.entries = handle_history_read(self.implementation.config) - return history.to_structure(history) + """Get all conversations from history.""" + manager = HistoryManager(self.implementation.config, LocalHistory) + history = manager.read() + + history_entry = HistoryEntry() + if history.history: + [history_entry.set_from_dict(entry.to_dict()) for entry in history.history] + else: + history_entry.entries = [] + + return HistoryEntry.to_structure(history_entry) + + # Add new methods with parameters + def GetFirstConversation(self) -> Structure: + """Get first conversation from history.""" + manager = HistoryManager(self.implementation.config, LocalHistory) + history = manager.read() + history_entry = HistoryEntry() + if history.history: + last_entry = history.history[0] + history_entry.set_from_dict(last_entry.to_dict()) + else: + history_entry.entries = [] + + return HistoryEntry.to_structure(history_entry) + + def GetLastConversation(self) -> Structure: + """Get last conversation from history.""" + manager = HistoryManager(self.implementation.config, LocalHistory) + history = manager.read() + history_entry = HistoryEntry() + + if history.history: + last_entry = history.history[-1] + history_entry.set_from_dict(last_entry.to_dict()) + else: + history_entry.entries = [] + + return HistoryEntry.to_structure(history_entry) def ClearHistory(self) -> None: - handle_history_write(self.implementation.config.history.file, [], "") + manager = HistoryManager(self.implementation.config, LocalHistory) + manager.clear() diff --git a/command_line_assistant/dbus/structures.py b/command_line_assistant/dbus/structures.py index ec16f27..6524499 100644 --- a/command_line_assistant/dbus/structures.py +++ b/command_line_assistant/dbus/structures.py @@ -1,5 +1,5 @@ from dasbus.structure import DBusData -from dasbus.typing import Str +from dasbus.typing import List, Str class Message(DBusData): @@ -18,15 +18,58 @@ def message(self, value: Str) -> None: self._message = value +class HistoryItem(DBusData): + """Represents a single history item with query and response""" + + def __init__(self) -> None: + self._query: Str = "" + self._response: Str = "" + self._timestamp: Str = "" + super().__init__() + + @property + def query(self) -> Str: + return self._query + + @query.setter + def query(self, value: Str) -> None: + self._query = value + + @property + def response(self) -> Str: + return self._response + + @response.setter + def response(self, value: Str) -> None: + self._response = value + + @property + def timestamp(self) -> Str: + return self._timestamp + + @timestamp.setter + def timestamp(self, value: Str) -> None: + self._timestamp = value + + class HistoryEntry(DBusData): def __init__(self) -> None: - self._entries: list[str] = [] + self._entries: List[HistoryItem] = [] super().__init__() @property - def entries(self) -> list[str]: + def entries(self) -> List[HistoryItem]: return self._entries @entries.setter - def entries(self, value: list[str]) -> None: + def entries(self, value: List[HistoryItem]) -> None: + # This handles setting from DBus structure self._entries = value + + def set_from_dict(self, entry: dict) -> None: + """Separate method to handle conversion from history dictionary""" + item = HistoryItem() + item.query = entry["interaction"]["query"]["text"] or "" + item.response = entry["interaction"]["response"]["text"] or "" + item.timestamp = entry["timestamp"] or "" + self._entries.append(item) diff --git a/command_line_assistant/history.py b/command_line_assistant/history.py deleted file mode 100644 index 25a73d2..0000000 --- a/command_line_assistant/history.py +++ /dev/null @@ -1,48 +0,0 @@ -import json -import logging -from pathlib import Path - -from command_line_assistant.config import Config - - -def handle_history_read(config: Config) -> list: - """ - Reads the history from a file and returns it as a list of dictionaries. - """ - if not config.history.enabled: - return [] - - filepath = config.history.file - if not filepath or not filepath.exists(): - logging.warning("History file %s does not exist.", filepath) - logging.warning("File will be created with first response.") - return [] - - max_size = config.history.max_size - history = [] - try: - data = filepath.read_text() - history = json.loads(data) - except json.JSONDecodeError as e: - logging.error("Failed to read history file %s: %s", filepath, e) - return [] - - logging.info("Taking maximum of %s entries from history.", max_size) - return history[:max_size] - - -def handle_history_write(history_file: Path, history: list, response: str) -> None: - """ - Writes the history to a file. - """ - - if not history_file.exists(): - history_file.parent.mkdir(mode=0o755) - - history.append({"role": "assistant", "content": response}) - - try: - data = json.dumps(history) - history_file.write_text(data) - except json.JSONDecodeError as e: - logging.error("Failed to write history file %s: %s", history_file, e) diff --git a/command_line_assistant/history/__init__.py b/command_line_assistant/history/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/command_line_assistant/history/base.py b/command_line_assistant/history/base.py new file mode 100644 index 0000000..bc06b38 --- /dev/null +++ b/command_line_assistant/history/base.py @@ -0,0 +1,59 @@ +import logging +from abc import ABC, abstractmethod +from datetime import UTC, datetime + +from command_line_assistant.config import Config +from command_line_assistant.history.schemas import ( + History, + HistoryEntry, + InteractionData, + QueryData, + ResponseData, +) + +logger = logging.getLogger(__name__) + + +class BaseHistory(ABC): + def __init__(self, config: Config) -> None: + self._config = config + + @abstractmethod + def read(self) -> History: + pass + + @abstractmethod + def write(self, current_history: History, query: str, response: str) -> None: + pass + + @abstractmethod + def clear(self) -> None: + pass + + def _add_new_entry( + self, current_history: History, query: str, response: str + ) -> History: + new_entry = HistoryEntry( + interaction=InteractionData( + query=QueryData(text=query), + response=ResponseData( + text=response, + tokens=len( + response.split() + ), # TODO(r0x0d): Simple token count, replace with actual + ), + ) + ) + + current_history.history.append(new_entry) + current_history.metadata.entry_count = len(current_history.history) + current_history.metadata.last_updated = datetime.now(UTC).strftime( + "%Y-%m-%dT%H:%M:%S.%fZ" + ) + return current_history + + def _check_if_history_is_enabled(self) -> bool: + if not self._config.history.enabled: + logger.info("History disabled. Nothing to return.") + + return self._config.history.enabled diff --git a/command_line_assistant/history/manager.py b/command_line_assistant/history/manager.py new file mode 100644 index 0000000..3996ed3 --- /dev/null +++ b/command_line_assistant/history/manager.py @@ -0,0 +1,96 @@ +from typing import Optional, Type + +from command_line_assistant.config import Config +from command_line_assistant.history.base import BaseHistory +from command_line_assistant.history.schemas import History + + +class HistoryManager: + """Manages history operations by delegating to a specific history implementation. + + Example: + >>> manager = HistoryManager(config, plugin=LocalHistory) + >>> entries = manager.read() + >>> manager.write("How do I check disk space?", "Use df -h command...") + >>> manager.clear() + """ + + def __init__( + self, config: Config, plugin: Optional[Type[BaseHistory]] = None + ) -> None: + """Initialize the history manager. + + Args: + config: Application configuration + plugin: Optional history implementation class. Defaults to LocalHistory + """ + self._config = config + self._plugin: Optional[Type[BaseHistory]] = None + self._instance: Optional[BaseHistory] = None + + # Set initial plugin if provided + if plugin: + self.plugin = plugin + + @property + def plugin(self) -> Optional[Type[BaseHistory]]: + """Get the current plugin class.""" + return self._plugin + + @plugin.setter + def plugin(self, plugin_cls: Type[BaseHistory]) -> None: + """Set and initialize a new plugin. + + Args: + plugin_cls: History implementation class to use + + Raises: + TypeError: If plugin_cls is not a subclass of BaseHistory + """ + if not issubclass(plugin_cls, BaseHistory): + raise TypeError( + f"Plugin must be a subclass of BaseHistory, got {plugin_cls.__name__}" + ) + + self._plugin = plugin_cls + self._instance = plugin_cls(self._config) + + def read(self) -> History: + """Read history entries using the current plugin. + + Returns: + History object containing entries and metadata + + Raises: + RuntimeError: If no plugin is set + """ + if not self._instance: + raise RuntimeError("No history plugin set. Set plugin before operations.") + + return self._instance.read() + + def write(self, current_history: History, query: str, response: str) -> None: + """Write a new history entry using the current plugin. + + Args: + query: The user's query + response: The LLM's response + + Raises: + RuntimeError: If no plugin is set + """ + if not self._instance: + raise RuntimeError("No history plugin set. Set plugin before operations.") + + self._instance.write(current_history, query, response) + + def clear(self) -> None: + """Clear all history entries. + + Raises: + RuntimeError: If no plugin is set + """ + if not self._instance: + raise RuntimeError("No history plugin set. Set plugin before operations.") + + self._instance.clear() diff --git a/command_line_assistant/history/plugins/__init__.py b/command_line_assistant/history/plugins/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/command_line_assistant/history/plugins/local.py b/command_line_assistant/history/plugins/local.py new file mode 100644 index 0000000..6e6625f --- /dev/null +++ b/command_line_assistant/history/plugins/local.py @@ -0,0 +1,51 @@ +import json +import logging + +from command_line_assistant.history.base import BaseHistory +from command_line_assistant.history.schemas import History + +logger = logging.getLogger(__name__) + + +class LocalHistory(BaseHistory): + def read(self) -> History: + """ + Reads the history from a file and returns it as a list of dictionaries. + """ + history = History() + if not self._check_if_history_is_enabled(): + return history + + filepath = self._config.history.file + + try: + data = filepath.read_text() + return History.from_json(data) + except json.JSONDecodeError as e: + logger.error("Failed to read history file %s: %s", filepath, e) + return history + + def write(self, current_history: History, query: str, response: str) -> None: + """ + Writes the history to a file. + """ + if not self._check_if_history_is_enabled(): + return + + filepath = self._config.history.file + final_history = self._add_new_entry(current_history, query, response) + try: + filepath.write_text(final_history.to_json()) + except json.JSONDecodeError as e: + logger.error("Failed to write history file %s: %s", filepath, e) + + def clear(self) -> None: + """Clear all history entries.""" + # Write empty history + current_history = History() + + try: + self._config.history.file.write_text(current_history.to_json()) + logger.info("History cleared successfully") + except Exception as e: + logger.error("Failed to clear history: %s", e) diff --git a/command_line_assistant/history/schemas.py b/command_line_assistant/history/schemas.py new file mode 100644 index 0000000..29dec10 --- /dev/null +++ b/command_line_assistant/history/schemas.py @@ -0,0 +1,116 @@ +import json +import platform +import uuid +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Optional + +from command_line_assistant.constants import VERSION + + +@dataclass +class QueryData: + text: Optional[str] = None + context: Optional[str] = None + role: str = "user" + + +@dataclass +class ResponseData: + text: Optional[str] = None + tokens: Optional[int] = 0 + role: str = "assistant" + + +@dataclass +class InteractionData: + query: QueryData = field(default_factory=QueryData) + response: ResponseData = field(default_factory=ResponseData) + + +@dataclass +class OSInfo: + distribution: str = "RHEL" + version: str = platform.version() + arch: str = platform.machine() + + +@dataclass +class EntryMetadata: + session_id: str = field(default_factory=lambda: str(uuid.uuid4())) + os_info: OSInfo = field(default_factory=OSInfo) + + +@dataclass +class HistoryEntry: + id: str = field(default_factory=lambda: str(uuid.uuid4())) + timestamp: str = field( + default_factory=lambda: datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S.%fZ") + ) + interaction: InteractionData = field(default_factory=InteractionData) + metadata: EntryMetadata = field(default_factory=EntryMetadata) + + def to_dict(self) -> dict: + return { + "id": self.id, + "timestamp": self.timestamp, + "interaction": { + "query": vars(self.interaction.query), + "response": vars(self.interaction.response), + }, + "metadata": { + "session_id": self.metadata.session_id, + "os_info": vars(self.metadata.os_info), + }, + } + + +@dataclass +class HistoryMetadata: + last_updated: str = field( + default_factory=lambda: datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S.%fZ") + ) + version: str = VERSION + entry_count: int = 0 + size_bytes: int = 0 + + +@dataclass +class History: + history: list[HistoryEntry] = field(default_factory=list) + metadata: HistoryMetadata = field(default_factory=HistoryMetadata) + + def to_json(self) -> str: + return json.dumps( + { + "history": [entry.to_dict() for entry in self.history], + "metadata": vars(self.metadata), + }, + indent=2, + ) + + @classmethod + def from_json(cls, json_str: str) -> "History": + data = json.loads(json_str) + history = [] + for entry_data in data["history"]: + query = QueryData(**entry_data["interaction"]["query"]) + response = ResponseData(**entry_data["interaction"]["response"]) + interaction = InteractionData(query=query, response=response) + + os_info = OSInfo(**entry_data["metadata"]["os_info"]) + metadata = EntryMetadata( + session_id=entry_data["metadata"]["session_id"], + os_info=os_info, + ) + + entry = HistoryEntry( + id=entry_data["id"], + timestamp=entry_data["timestamp"], + interaction=interaction, + metadata=metadata, + ) + history.append(entry) + + metadata = HistoryMetadata(**data["metadata"]) + return cls(history=history, metadata=metadata) diff --git a/data/development/config/command_line_assistant/config.toml b/data/development/config/command_line_assistant/config.toml index 2ea07af..373847f 100644 --- a/data/development/config/command_line_assistant/config.toml +++ b/data/development/config/command_line_assistant/config.toml @@ -9,8 +9,6 @@ prompt_separator = "$" [history] enabled = true file = "~/.local/share/command-line-assistant/command-line-assistant_history.json" -# max number of queries in history (including responses) -max_size = 100 [backend] endpoint = "http://localhost:8080" diff --git a/tests/commands/test_history.py b/tests/commands/test_history.py index 9489a78..f1ef349 100644 --- a/tests/commands/test_history.py +++ b/tests/commands/test_history.py @@ -1,81 +1,217 @@ -from unittest.mock import MagicMock, patch +import logging +from unittest.mock import Mock, patch import pytest +from dasbus.error import DBusError from command_line_assistant.commands.history import ( HistoryCommand, - _command_factory, - register_subcommand, + _initialize_qa_renderer, + _initialize_spinner_renderer, + _initialize_text_renderer, ) -from command_line_assistant.dbus.constants import HISTORY_IDENTIFIER +from command_line_assistant.dbus.structures import HistoryEntry, HistoryItem @pytest.fixture -def mock_get_proxy(mock_proxy): - """Mock the get_proxy method.""" - with patch.object(HISTORY_IDENTIFIER, "get_proxy", return_value=mock_proxy): - yield mock_proxy +def mock_proxy(): + """Create a mock DBus proxy.""" + proxy = Mock() + return proxy -def test_history_command_clear_success(mock_get_proxy): - """Test successful history clear operation.""" - cmd = HistoryCommand(clear=True) - cmd.run() +@pytest.fixture +def history_command(mock_proxy): + """Create a HistoryCommand instance with mocked proxy.""" + with patch( + "command_line_assistant.commands.history.HISTORY_IDENTIFIER.get_proxy", + return_value=mock_proxy, + ): + command = HistoryCommand(clear=False, first=False, last=False) + return command + + +@pytest.fixture +def sample_history_entry(): + """Create a sample history entry for testing.""" + entry = HistoryItem() + entry.query = "test query" + entry.response = "test response" + entry.timestamp = "2024-01-01T00:00:00Z" + + history_entry = HistoryEntry() + history_entry.entries = [entry] + return history_entry + + +def test_initialize_renderers(): + """Test initialization of various renderers.""" + spinner = _initialize_spinner_renderer() + qa = _initialize_qa_renderer() + text = _initialize_text_renderer() + + assert spinner is not None + assert qa is not None + assert text is not None + + +def test_retrieve_all_conversations_success( + history_command, mock_proxy, sample_history_entry, caplog +): + """Test retrieving all conversations successfully.""" + mock_proxy.GetHistory.return_value = sample_history_entry.to_structure( + sample_history_entry + ) - mock_get_proxy.ClearHistory.assert_called_once() + with caplog.at_level(logging.INFO): + history_command._retrieve_all_conversations() + assert "Getting all conversations from history" in caplog.text + mock_proxy.GetHistory.assert_called_once() -def test_history_command_clear_failure(mock_get_proxy, caplog): - """Test failed history clear operation.""" - from dasbus.error import DBusError - # Configure mock to raise DBusError - mock_get_proxy.ClearHistory.side_effect = DBusError("Failed to clear history") +def test_retrieve_all_conversations_empty(history_command, mock_proxy, caplog, capsys): + """Test retrieving all conversations when history is empty.""" + empty_history = HistoryEntry() + empty_history.entries = [] + mock_proxy.GetHistory.return_value = empty_history.to_structure(empty_history) - cmd = HistoryCommand(clear=True) + history_command._retrieve_all_conversations() + captured = capsys.readouterr() + assert captured.out == "No history found.\n" + + +def test_retrieve_all_conversations_error(history_command, mock_proxy, caplog): + """Test retrieving all conversations with DBus error.""" + mock_proxy.GetHistory.side_effect = DBusError("test.error", "Test error") with pytest.raises(DBusError): - cmd.run() + with caplog.at_level(logging.INFO): + history_command._retrieve_all_conversations() - assert "Failed to clean the history" in caplog.text + assert "Failed to get history" in caplog.text -def test_history_command_no_clear(mock_get_proxy): - """Test history command without clear flag.""" - cmd = HistoryCommand(clear=False) - cmd.run() +def test_retrieve_first_conversation_success( + history_command, mock_proxy, sample_history_entry +): + """Test retrieving first conversation successfully.""" + mock_proxy.GetFirstConversation.return_value = sample_history_entry.to_structure( + sample_history_entry + ) - mock_get_proxy.ClearHistory.assert_not_called() + history_command._retrieve_first_conversation() + mock_proxy.GetFirstConversation.assert_called_once() -def test_register_subcommand(): - """Test registration of history subcommand.""" - parser = MagicMock() - subparser = MagicMock() - parser.add_parser.return_value = subparser - register_subcommand(parser) +def test_retrieve_first_conversation_empty(history_command, mock_proxy, capsys): + """Test retrieving first conversation when history is empty.""" + empty_history = HistoryEntry() + empty_history.entries = [] + mock_proxy.GetFirstConversation.return_value = empty_history.to_structure( + empty_history + ) - # Verify parser configuration - parser.add_parser.assert_called_once_with( - "history", help="Manage conversation history" + history_command._retrieve_first_conversation() + captured = capsys.readouterr() + assert captured.out == "No history found.\n" + + +def test_retrieve_first_conversation_error(history_command, mock_proxy, caplog): + """Test retrieving first conversation with DBus error.""" + mock_proxy.GetFirstConversation.side_effect = DBusError("test.error", "Test error") + + with pytest.raises(DBusError): + with caplog.at_level(logging.INFO): + history_command._retrieve_first_conversation() + + assert "Failed to get first conversation" in caplog.text + + +def test_retrieve_last_conversation_success( + history_command, mock_proxy, sample_history_entry +): + """Test retrieving last conversation successfully.""" + mock_proxy.GetLastConversation.return_value = sample_history_entry.to_structure( + sample_history_entry ) - # Verify arguments added to subparser - subparser.add_argument.assert_called_once_with( - "--clear", action="store_true", help="Clear the history." + history_command._retrieve_last_conversation() + + mock_proxy.GetLastConversation.assert_called_once() + + +def test_retrieve_last_conversation_empty(history_command, mock_proxy, capsys): + """Test retrieving last conversation when history is empty.""" + empty_history = HistoryEntry() + empty_history.entries = [] + mock_proxy.GetLastConversation.return_value = empty_history.to_structure( + empty_history ) - # Verify defaults set - assert hasattr(subparser, "set_defaults") + history_command._retrieve_last_conversation() + captured = capsys.readouterr() + assert captured.out == "No history found.\n" + + +def test_retrieve_last_conversation_error(history_command, mock_proxy, caplog): + """Test retrieving last conversation with DBus error.""" + mock_proxy.GetLastConversation.side_effect = DBusError("test.error", "Test error") + + with pytest.raises(DBusError): + with caplog.at_level(logging.INFO): + history_command._retrieve_last_conversation() + + assert "Failed to get last conversation" in caplog.text + + +def test_clear_history_success(history_command, mock_proxy, caplog): + """Test clearing history successfully.""" + with caplog.at_level(logging.INFO): + history_command._clear_history() + + assert "Cleaning the history" in caplog.text + mock_proxy.ClearHistory.assert_called_once() + + +def test_clear_history_error(history_command, mock_proxy, caplog): + """Test clearing history with DBus error.""" + mock_proxy.ClearHistory.side_effect = DBusError("test.error", "Test error") + + with pytest.raises(DBusError): + with caplog.at_level(logging.INFO): + history_command._clear_history() + + assert "Failed to clean the history" in caplog.text + + +def test_run_clear(history_command): + """Test run method with clear flag.""" + history_command._clear = True + with patch.object(history_command, "_clear_history") as mock_clear: + history_command.run() + mock_clear.assert_called_once() + + +def test_run_first(history_command): + """Test run method with first flag.""" + history_command._first = True + with patch.object(history_command, "_retrieve_first_conversation") as mock_first: + history_command.run() + mock_first.assert_called_once() -def test_command_factory(): - """Test the command factory creates correct command instance.""" - from argparse import Namespace +def test_run_last(history_command): + """Test run method with last flag.""" + history_command._last = True + with patch.object(history_command, "_retrieve_last_conversation") as mock_last: + history_command.run() + mock_last.assert_called_once() - args = Namespace(clear=True) - cmd = _command_factory(args) - assert isinstance(cmd, HistoryCommand) - assert cmd._clear is True +def test_run_all(history_command): + """Test run method with no flags (all conversations).""" + with patch.object(history_command, "_retrieve_all_conversations") as mock_all: + history_command.run() + mock_all.assert_called_once() diff --git a/tests/config/test_config.py b/tests/config/test_config.py index c512bde..66f0b5f 100644 --- a/tests/config/test_config.py +++ b/tests/config/test_config.py @@ -28,7 +28,6 @@ def get_config_template(tmp_path) -> str: enabled = true file = "{history_file}" # max number of queries in history (including responses) -max_size = 1 [backend] endpoint = "https://localhost" diff --git a/tests/conftest.py b/tests/conftest.py index 2d99afb..c979a14 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -34,6 +34,7 @@ def mock_config(tmp_path): """Fixture to create a mock configuration""" cert_file = tmp_path / "cert.pem" key_file = tmp_path / "key.pem" + history_file = tmp_path / "command_line_assistant" / "history.json" cert_file.write_text("cert") key_file.write_text("key") @@ -47,9 +48,7 @@ def mock_config(tmp_path): endpoint="http://test.endpoint/v1/query", auth=AuthSchema(cert_file=cert_file, key_file=key_file, verify_ssl=False), ), - history=HistorySchema( - enabled=True, file=Path("/tmp/test_history.json"), max_size=100 - ), + history=HistorySchema(enabled=True, file=history_file), logging=LoggingSchema(level="debug"), ) diff --git a/tests/dbus/test_interfaces.py b/tests/dbus/test_interfaces.py index cb24f79..bfc5087 100644 --- a/tests/dbus/test_interfaces.py +++ b/tests/dbus/test_interfaces.py @@ -1,118 +1,178 @@ +import json from unittest.mock import Mock, patch import pytest +from dasbus.server.template import InterfaceTemplate -from command_line_assistant.config import Config from command_line_assistant.dbus.interfaces import ( HistoryInterface, - Message, QueryInterface, ) -from command_line_assistant.dbus.structures import HistoryEntry +from command_line_assistant.dbus.structures import HistoryEntry, Message +from command_line_assistant.history.schemas import History @pytest.fixture -def mock_implementation(): +def mock_implementation(mock_config): + """Create a mock implementation with configuration.""" impl = Mock() - impl.config = Mock(spec=Config) - impl.config.history = Mock() - impl.query = Mock() + impl.config = mock_config + mock_query = Message() + mock_query.message = "test query" + impl.query = mock_query return impl @pytest.fixture def query_interface(mock_implementation): + """Create a QueryInterface instance with mock implementation.""" interface = QueryInterface(mock_implementation) - interface.watch_property = Mock() + assert isinstance(interface, InterfaceTemplate) return interface -class TestQueryInterface: - @patch("command_line_assistant.dbus.interfaces.submit") - def test_retrieve_answer_success( - self, mock_submit, query_interface, mock_implementation - ): - mock_submit.return_value = "test response" - mock_implementation.query.message = "test query" +@pytest.fixture +def history_interface(mock_implementation): + """Create a HistoryInterface instance with mock implementation.""" + interface = HistoryInterface(mock_implementation) + assert isinstance(interface, InterfaceTemplate) + return interface - result = query_interface.RetrieveAnswer - mock_submit.assert_called_once_with("test query", mock_implementation.config) - message = Message.from_structure(result) - assert message.message == "test response" +@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", "context": None, "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" + mock_implementation.config.history.file.parent.mkdir() + mock_implementation.config.history.file.write_text("") + with patch( + "command_line_assistant.dbus.interfaces.submit", return_value=expected_response + ) as mock_submit: + response = query_interface.RetrieveAnswer + + mock_submit.assert_called_once_with( + mock_implementation.query.message, mock_implementation.config + ) - @patch("command_line_assistant.dbus.interfaces.submit") - def test_retrieve_answer_empty_query( - self, mock_submit, query_interface, mock_implementation - ): - mock_submit.return_value = "" - mock_implementation.query.message = "" + reconstructed = Message.from_structure(response) + assert reconstructed.message == expected_response - result = query_interface.RetrieveAnswer - mock_submit.assert_called_once_with("", mock_implementation.config) - message = Message.from_structure(result) - assert message.message == "" +def test_query_interface_process_query(query_interface, mock_implementation): + """Test processing query through query interface.""" + test_query = Message() + test_query.message = "test query" - def test_process_query(self, query_interface, mock_implementation): - test_query = Message() - test_query.message = "test message" - query_structure = Message.to_structure(test_query) + query_interface.ProcessQuery(Message.to_structure(test_query)) - query_interface.ProcessQuery(query_structure) + mock_implementation.process_query.assert_called_once() + processed_query = mock_implementation.process_query.call_args[0][0] + assert isinstance(processed_query, Message) + assert processed_query.message == test_query.message - mock_implementation.process_query.assert_called_once() - processed_message = mock_implementation.process_query.call_args[0][0] - assert isinstance(processed_message, Message) - assert processed_message.message == "test message" +def test_history_interface_get_history( + history_interface, mock_implementation, sample_history_data +): + """Test getting all history through history interface.""" + mock_history = History.from_json(json.dumps(sample_history_data)) -class TestHistoryInterface: - @pytest.fixture - def history_interface(self, mock_implementation): - return HistoryInterface(mock_implementation) + with patch("command_line_assistant.dbus.interfaces.HistoryManager") as mock_manager: + mock_manager.return_value.read.return_value = mock_history + response = history_interface.GetHistory() - @patch("command_line_assistant.dbus.interfaces.handle_history_read") - def test_get_history_success( - self, mock_history_read, history_interface, mock_implementation - ): - test_entries = ["entry1", "entry2"] - mock_history_read.return_value = test_entries + reconstructed = HistoryEntry.from_structure(response) + assert len(reconstructed.entries) == 1 + assert reconstructed.entries[0].query == "test query" + assert reconstructed.entries[0].response == "test response" - result = history_interface.GetHistory - mock_history_read.assert_called_once_with(mock_implementation.config) - history = HistoryEntry.from_structure(result) - assert history.entries == test_entries +def test_history_interface_get_first_conversation( + history_interface, mock_implementation, sample_history_data +): + """Test getting first conversation through history interface.""" + mock_history = History.from_json(json.dumps(sample_history_data)) - @patch("command_line_assistant.dbus.interfaces.handle_history_read") - def test_get_history_empty( - self, mock_history_read, history_interface, mock_implementation - ): - mock_history_read.return_value = [] + with patch("command_line_assistant.dbus.interfaces.HistoryManager") as mock_manager: + mock_manager.return_value.read.return_value = mock_history + response = history_interface.GetFirstConversation() - result = history_interface.GetHistory + reconstructed = HistoryEntry.from_structure(response) + assert len(reconstructed.entries) == 1 + assert reconstructed.entries[0].query == "test query" + assert reconstructed.entries[0].response == "test response" - mock_history_read.assert_called_once_with(mock_implementation.config) - history = HistoryEntry.from_structure(result) - assert history.entries == [] - @patch("command_line_assistant.dbus.interfaces.handle_history_write") - def test_clear_history( - self, mock_history_write, history_interface, mock_implementation - ): - history_interface.ClearHistory() +def test_history_interface_get_last_conversation( + history_interface, mock_implementation, sample_history_data +): + """Test getting first conversation through history interface.""" + mock_history = History.from_json(json.dumps(sample_history_data)) - mock_history_write.assert_called_once_with( - mock_implementation.config.history.file, [], "" - ) + with patch("command_line_assistant.dbus.interfaces.HistoryManager") as mock_manager: + mock_manager.return_value.read.return_value = mock_history + response = history_interface.GetLastConversation() + + reconstructed = HistoryEntry.from_structure(response) + assert len(reconstructed.entries) == 1 + assert reconstructed.entries[0].query == "test query" + assert reconstructed.entries[0].response == "test response" - @patch("command_line_assistant.dbus.interfaces.handle_history_write") - def test_clear_history_with_nonexistent_file( - self, mock_history_write, history_interface, mock_implementation - ): - mock_implementation.config.history.file = "/nonexistent/path" +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: history_interface.ClearHistory() + mock_manager.return_value.clear.assert_called_once() + + +def test_history_interface_empty_history(history_interface, mock_implementation): + """Test handling empty history in all methods.""" + empty_history = History() + + with patch("command_line_assistant.dbus.interfaces.HistoryManager") as mock_manager: + mock_manager.return_value.read.return_value = empty_history - mock_history_write.assert_called_once_with("/nonexistent/path", [], "") + # Test all methods with empty history + for method in [ + history_interface.GetHistory, + history_interface.GetFirstConversation, + history_interface.GetLastConversation, + ]: + response = method() + reconstructed = HistoryEntry.from_structure(response) + assert len(reconstructed.entries) == 0 diff --git a/tests/dbus/test_structures.py b/tests/dbus/test_structures.py index da6a2ff..f757fe6 100644 --- a/tests/dbus/test_structures.py +++ b/tests/dbus/test_structures.py @@ -1,4 +1,4 @@ -from command_line_assistant.dbus.structures import HistoryEntry, Message +from command_line_assistant.dbus.structures import HistoryEntry, HistoryItem, Message def test_message_init(): @@ -19,7 +19,7 @@ def test_history_entry_init(): def test_history_entry_setter(): history = HistoryEntry() - test_entries = ["entry1", "entry2", "entry3"] + test_entries = [HistoryItem(), HistoryItem(), HistoryItem()] history.entries = test_entries assert history.entries == test_entries @@ -38,6 +38,6 @@ def test_message_empty_string(): def test_history_entry_single_entry(): history = HistoryEntry() - history.entries = ["single entry"] + + history.entries = [HistoryItem()] assert len(history.entries) == 1 - assert history.entries[0] == "single entry" diff --git a/tests/history/__init__.py b/tests/history/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/history/plugins/__init__.py b/tests/history/plugins/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/history/plugins/test_local.py b/tests/history/plugins/test_local.py new file mode 100644 index 0000000..5d08206 --- /dev/null +++ b/tests/history/plugins/test_local.py @@ -0,0 +1,176 @@ +import json +import logging +from unittest.mock import patch + +import pytest + +from command_line_assistant.history.plugins.local import LocalHistory +from command_line_assistant.history.schemas import ( + History, + HistoryMetadata, +) + + +@pytest.fixture +def local_history(mock_config): + """Create a LocalHistory instance for testing.""" + 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", "context": None, "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 + + history = local_history.read() + + assert isinstance(history, History) + assert len(history.history) == 0 + assert isinstance(history.metadata, HistoryMetadata) + + +def test_read_existing_history(local_history, sample_history_data): + """Test reading existing history file.""" + with patch("pathlib.Path.read_text") as mock_read: + mock_read.return_value = json.dumps(sample_history_data) + + history = local_history.read() + + assert isinstance(history, History) + assert len(history.history) == 1 + assert history.history[0].interaction.query.text == "test query" + assert history.history[0].interaction.response.text == "test response" + assert history.metadata.entry_count == 1 + + +def test_read_invalid_json(local_history, caplog): + """Test reading invalid JSON history file.""" + with patch("pathlib.Path.read_text") as mock_read: + mock_read.return_value = "invalid json" + + with caplog.at_level(logging.ERROR): + history = local_history.read() + + assert isinstance(history, History) + assert len(history.history) == 0 + assert "Failed to read history file" in caplog.text + + +def test_write_new_entry(local_history): + """Test writing a new history entry.""" + current_history = History() + query = "test query" + response = "test response" + + with patch("pathlib.Path.write_text") as mock_write: + local_history.write(current_history, query, response) + + # Verify write was called + mock_write.assert_called_once() + + # Verify the written content + written_data = json.loads(mock_write.call_args[0][0]) + assert len(written_data["history"]) == 1 + assert written_data["history"][0]["interaction"]["query"]["text"] == query + assert written_data["history"][0]["interaction"]["response"]["text"] == response + + +def test_write_when_history_disabled(local_history): + """Test writing history when history is disabled.""" + local_history._config.history.enabled = False + current_history = History() + + with patch("pathlib.Path.write_text") as mock_write: + local_history.write(current_history, "query", "response") + mock_write.assert_not_called() + + +def test_write_with_error(local_history, caplog): + """Test writing history when an error occurs.""" + current_history = History() + + with patch("pathlib.Path.write_text") as mock_write: + mock_write.side_effect = json.JSONDecodeError("Test error", "doc", 0) + + with caplog.at_level(logging.ERROR): + local_history.write(current_history, "query", "response") + + assert "Failed to write history file" in caplog.text + + +def test_clear_history(local_history): + """Test clearing history.""" + with patch("pathlib.Path.write_text") as mock_write: + local_history.clear() + + mock_write.assert_called_once() + written_data = json.loads(mock_write.call_args[0][0]) + assert len(written_data["history"]) == 0 + assert written_data["metadata"]["entry_count"] == 0 + + +def test_clear_history_with_error(local_history, caplog): + """Test clearing history when an error occurs.""" + with patch("pathlib.Path.write_text") as mock_write: + mock_write.side_effect = Exception("Test error") + + with caplog.at_level(logging.ERROR): + local_history.clear() + + assert "Failed to clear history" in caplog.text + + +def test_check_if_history_is_enabled(local_history): + """Test history enabled check.""" + assert local_history._check_if_history_is_enabled() is True + + local_history._config.history.enabled = False + assert local_history._check_if_history_is_enabled() is False + + +def test_add_new_entry(local_history): + """Test adding a new entry to history.""" + current_history = History() + query = "test query" + response = "test response" + + updated_history = local_history._add_new_entry(current_history, query, response) + + assert len(updated_history.history) == 1 + assert updated_history.history[0].interaction.query.text == query + assert updated_history.history[0].interaction.response.text == response + assert updated_history.metadata.entry_count == 1 diff --git a/tests/history/test_manager.py b/tests/history/test_manager.py new file mode 100644 index 0000000..25ee79b --- /dev/null +++ b/tests/history/test_manager.py @@ -0,0 +1,152 @@ +import pytest + +from command_line_assistant.config import Config +from command_line_assistant.config.schemas import HistorySchema +from command_line_assistant.history.base import BaseHistory +from command_line_assistant.history.manager import HistoryManager +from command_line_assistant.history.schemas import ( + History, + HistoryEntry, + InteractionData, + QueryData, + ResponseData, +) + + +class MockHistoryPlugin(BaseHistory): + def __init__(self, config): + super().__init__(config) + self.read_called = False + self.write_called = False + self.clear_called = False + self._history = History() + + def read(self) -> History: + self.read_called = True + return self._history + + def write(self, current_history: History, query: str, response: str) -> None: + self.write_called = True + entry = HistoryEntry( + interaction=InteractionData( + query=QueryData(text=query), response=ResponseData(text=response) + ) + ) + self._history.history.append(entry) + + def clear(self) -> None: + self.clear_called = True + self._history.history = [] + + +@pytest.fixture +def mock_config(tmp_path): + return Config( + history=HistorySchema(enabled=True, file=tmp_path / "test_history.json") + ) + + +@pytest.fixture +def history_manager(mock_config): + return HistoryManager(mock_config, plugin=MockHistoryPlugin) + + +def test_history_manager_initialization(mock_config): + """Test that HistoryManager initializes correctly""" + manager = HistoryManager(mock_config) + assert manager._config == mock_config + assert manager._plugin is None + assert manager._instance is None + + +def test_history_manager_plugin_setter(mock_config): + """Test setting a valid plugin""" + manager = HistoryManager(mock_config) + manager.plugin = MockHistoryPlugin + assert manager._plugin == MockHistoryPlugin + assert isinstance(manager._instance, MockHistoryPlugin) + + +def test_history_manager_invalid_plugin(mock_config): + """Test setting an invalid plugin""" + manager = HistoryManager(mock_config) + + class InvalidPlugin(BaseHistory): + pass + + with pytest.raises(TypeError): + manager.plugin = InvalidPlugin + + +def test_history_manager_read_without_plugin(mock_config): + """Test reading history without setting a plugin first""" + manager = HistoryManager(mock_config) + with pytest.raises(RuntimeError): + manager.read() + + +def test_history_manager_write_without_plugin(mock_config): + """Test writing history without setting a plugin first""" + manager = HistoryManager(mock_config) + with pytest.raises(RuntimeError): + manager.write(History(), "test query", "test response") + + +def test_history_manager_clear_without_plugin(mock_config): + """Test clearing history without setting a plugin first""" + manager = HistoryManager(mock_config) + with pytest.raises(RuntimeError): + manager.clear() + + +def test_history_manager_read(history_manager): + """Test reading history""" + history = history_manager.read() + assert isinstance(history, History) + assert isinstance(history_manager._instance, MockHistoryPlugin) + assert history_manager._instance.read_called + + +def test_history_manager_write(history_manager): + """Test writing to history""" + test_query = "How do I check disk space?" + test_response = "Use the df command" + + history_manager.write(History(), test_query, test_response) + + assert history_manager._instance.write_called + history = history_manager.read() + assert len(history.history) == 1 + assert history.history[0].interaction.query.text == test_query + assert history.history[0].interaction.response.text == test_response + + +def test_history_manager_clear(history_manager): + """Test clearing history""" + # First write something + history_manager.write(History(), "test query", "test response") + assert len(history_manager.read().history) == 1 + + # Then clear it + history_manager.clear() + assert history_manager._instance.clear_called + assert len(history_manager.read().history) == 0 + + +def test_history_manager_multiple_writes(history_manager): + """Test multiple writes to history""" + entries = [ + ("query1", "response1"), + ("query2", "response2"), + ("query3", "response3"), + ] + + for query, response in entries: + history_manager.write(History(), query, response) + + history = history_manager.read() + assert len(history.history) == len(entries) + + for i, (query, response) in enumerate(entries): + assert history.history[i].interaction.query.text == query + assert history.history[i].interaction.response.text == response diff --git a/tests/history/test_schemas.py b/tests/history/test_schemas.py new file mode 100644 index 0000000..32753af --- /dev/null +++ b/tests/history/test_schemas.py @@ -0,0 +1,211 @@ +import json +from datetime import datetime + +import pytest + +from command_line_assistant.history.schemas import ( + EntryMetadata, + History, + HistoryEntry, + HistoryMetadata, + InteractionData, + OSInfo, + QueryData, + ResponseData, +) + + +def test_query_data_initialization(): + """Test QueryData initialization and defaults""" + query = QueryData() + assert query.text is None + assert query.context is None + assert query.role == "user" + + # Test with values + query = QueryData(text="test query", context="some context", role="custom") + assert query.text == "test query" + assert query.context == "some context" + assert query.role == "custom" + + +def test_response_data_initialization(): + """Test ResponseData initialization and defaults""" + response = ResponseData() + assert response.text is None + assert response.tokens == 0 + assert response.role == "assistant" + + # Test with values + response = ResponseData(text="test response", tokens=42, role="custom") + assert response.text == "test response" + assert response.tokens == 42 + assert response.role == "custom" + + +def test_interaction_data_initialization(): + """Test InteractionData initialization and defaults""" + interaction = InteractionData() + assert isinstance(interaction.query, QueryData) + assert isinstance(interaction.response, ResponseData) + + # Test with custom query and response + query = QueryData(text="test query") + response = ResponseData(text="test response") + interaction = InteractionData(query=query, response=response) + assert interaction.query.text == "test query" + assert interaction.response.text == "test response" + + +def test_os_info_initialization(): + """Test OSInfo initialization and defaults""" + os_info = OSInfo() + assert os_info.distribution == "RHEL" + assert isinstance(os_info.version, str) + assert isinstance(os_info.arch, str) + + # Test with custom values + os_info = OSInfo(distribution="Ubuntu", version="22.04", arch="x86_64") + assert os_info.distribution == "Ubuntu" + assert os_info.version == "22.04" + assert os_info.arch == "x86_64" + + +def test_entry_metadata_initialization(): + """Test EntryMetadata initialization""" + metadata = EntryMetadata() + assert isinstance(metadata.session_id, str) + assert isinstance(metadata.os_info, OSInfo) + + # Verify UUID format + import uuid + + uuid.UUID(metadata.session_id) # Should not raise exception + + +def test_history_entry_initialization(): + """Test HistoryEntry initialization and to_dict method""" + entry = HistoryEntry() + assert isinstance(entry.id, str) + assert isinstance(entry.timestamp, str) + assert isinstance(entry.interaction, InteractionData) + assert isinstance(entry.metadata, EntryMetadata) + + +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_dict = entry.to_dict() + assert isinstance(entry_dict, dict) + assert entry_dict["interaction"]["query"]["text"] == "test query" + assert entry_dict["interaction"]["response"]["text"] == "test response" + assert "id" in entry_dict + assert "timestamp" in entry_dict + assert "metadata" in entry_dict + + +def test_history_metadata_initialization(): + """Test HistoryMetadata initialization""" + metadata = HistoryMetadata() + assert isinstance(metadata.last_updated, str) + assert isinstance(metadata.version, str) + assert metadata.entry_count == 0 + assert metadata.size_bytes == 0 + + +def test_history_initialization(): + """Test History initialization""" + history = History() + assert isinstance(history.history, list) + assert len(history.history) == 0 + assert isinstance(history.metadata, HistoryMetadata) + + +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" + history.history.append(entry) + + # Convert to JSON + json_str = history.to_json() + assert isinstance(json_str, str) + + # Parse JSON string to verify structure + parsed = json.loads(json_str) + assert "history" in parsed + assert "metadata" in parsed + assert len(parsed["history"]) == 1 + + # Convert back from JSON + new_history = History.from_json(json_str) + assert isinstance(new_history, History) + assert len(new_history.history) == 1 + assert new_history.history[0].interaction.query.text == "test query" + assert new_history.history[0].interaction.response.text == "test response" + + +def test_history_with_multiple_entries(): + """Test History with multiple entries""" + history = History() + + # Add multiple entries + entries = [ + ("query1", "response1"), + ("query2", "response2"), + ("query3", "response3"), + ] + + for query_text, response_text in entries: + entry = HistoryEntry() + entry.interaction.query.text = query_text + entry.interaction.response.text = response_text + history.history.append(entry) + + # Verify entries + assert len(history.history) == len(entries) + for i, (query_text, response_text) in enumerate(entries): + assert history.history[i].interaction.query.text == query_text + assert history.history[i].interaction.response.text == response_text + + +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" + history.history.append(entry) + + # Roundtrip through JSON + json_str = history.to_json() + new_history = History.from_json(json_str) + + assert new_history.history[0].interaction.query.text == entry.interaction.query.text + assert ( + new_history.history[0].interaction.response.text + == entry.interaction.response.text + ) + + +@pytest.mark.parametrize("invalid_json", ["", "{}", '{"invalid": "data"}']) +def test_history_from_json_with_invalid_data(invalid_json): + """Test History.from_json with invalid JSON data""" + with pytest.raises((KeyError, json.JSONDecodeError)): + History.from_json(invalid_json) + + +def test_history_entry_timestamp_format(): + """Test that HistoryEntry timestamps are in the correct format""" + entry = HistoryEntry() + # Verify the timestamp is in ISO format + try: + datetime.fromisoformat(entry.timestamp.rstrip("Z")) + except ValueError: + pytest.fail("Timestamp is not in valid ISO format") diff --git a/tests/test_history.py b/tests/test_history.py deleted file mode 100644 index 090d839..0000000 --- a/tests/test_history.py +++ /dev/null @@ -1,94 +0,0 @@ -import json -from pathlib import Path - -import pytest - -from command_line_assistant import history -from command_line_assistant.config import Config, HistorySchema - -#: Mock history conversation for testing -MOCK_HISTORY_CONVERSATION: list[dict] = [ - {"role": "user", "content": "create a file under /etc"}, - { - "role": "assistant", - "content": "Sure, I can help you with that. To create a new file under the `/etc` directory in OpenShift, follow these steps:\n\n1. Log in to your OpenShift cluster using the command `oc login`.\n2. Once logged in, navigate to the project where you want to create the file. You can use the command `oc project ` to change to the desired project.\n3. Create a new file by running the command `oc create -f .yaml`. Replace `.yaml` with the name of the YAML file containing the content of the file you want to create.\n\nFor example, if you want to create a new file named `myfile` with the content `Hello, World!`, you can create the file using the following command:\n```\noc create -f myfile.yaml\n```\nThis will create a new file called `myfile` under the `/etc` directory in your OpenShift project.\n\nKeep in mind that the `oc create` command requires the `oc` command line tool to be installed and configured on your system. If you encounter any issues, please refer to the official OpenShift documentation for troubleshooting steps.", - }, -] - - -class TestHistoryRead: - """Holds the testing functions for reading the history.""" - - def test_not_enabled(self): - config = Config(history=HistorySchema(enabled=False)) - assert not history.handle_history_read(config) - - def test_history_file_missing(self, tmp_path, caplog): - history_file = tmp_path / "non-existing-file.json" - config = Config(history=HistorySchema(file=history_file)) - - assert not history.handle_history_read(config) - assert "File will be created with first response." in caplog.records[-1].message - - def test_history_failed_to_decode_json(self, tmp_path, caplog): - history_file = tmp_path / "non-existing-file.json" - history_file.write_text("not a json") - config = Config(history=HistorySchema(file=history_file)) - - assert not history.handle_history_read(config) - assert "Failed to read history file" in caplog.records[-1].message - - def test_history_read(self, tmp_path): - history_file = tmp_path / "history.json" - history_file.write_text(json.dumps(MOCK_HISTORY_CONVERSATION)) - config = Config(history=HistorySchema(file=history_file)) - - assert history.handle_history_read(config) == MOCK_HISTORY_CONVERSATION - - @pytest.mark.parametrize( - ("multiply", "max_size"), - ( - ( - 10, - 5, - ), - ), - ) - def test_history_over_max_size(self, tmp_path, multiply, max_size): - total_mock_data = MOCK_HISTORY_CONVERSATION * multiply - history_file = tmp_path / "history.json" - history_file.write_text(json.dumps(total_mock_data)) - config = Config(history=HistorySchema(file=history_file, max_size=max_size)) - - history_result = history.handle_history_read(config) - assert len(history_result) == max_size - assert len(history_result) < len(total_mock_data) - - # TODO(r0x0d): Maybe include a test to check which records got back? - - -class TestHistoryWrite: - def test_history_file_missing(self, tmp_path): - history_file = tmp_path / "history" / "non-existing-file.json" - - history.handle_history_write(history_file, [], "test") - assert Path(history_file).exists() - - def test_history_write(self, tmp_path): - expected = [{"role": "assistant", "content": "test"}] - history_file = tmp_path / "history" / "non-existing-file.json" - history.handle_history_write(history_file, [], "test") - - raw_history = Path(history_file).read_text() - assert json.loads(raw_history) == expected - - def test_history_append(self, tmp_path): - expected = MOCK_HISTORY_CONVERSATION.copy() - expected.append({"role": "assistant", "content": "test"}) - - history_file = tmp_path / "history" / "non-existing-file.json" - - history.handle_history_write(history_file, MOCK_HISTORY_CONVERSATION, "test") - - raw_history = Path(history_file).read_text() - assert json.loads(raw_history) == expected