Fix query read from stdin

With all the CLI refactors, the query reading from stdin was broken.
This patch aims to fix this and allow queries to be passed or appended
from stdin.

command_line_assistant/commands/query.py

@@ -1,6 +1,7 @@
 """Module to handle the query command."""
 
 from argparse import Namespace
+from typing import Optional
 
 from command_line_assistant.dbus.constants import QUERY_IDENTIFIER
 from command_line_assistant.dbus.exceptions import (
@@ -84,13 +85,15 @@ def _initialize_legal_renderer(write_once: bool = False) -> TextRenderer:
 class QueryCommand(BaseCLICommand):
     """Class that represents the query command."""
 
-    def __init__(self, query_string: str) -> None:
+    def __init__(self, query_string: str, stdin: Optional[str]) -> None:
         """Constructor of the class.
 
         Args:
             query_string (str): The query provided by the user.
+            stdin (Optional[str]): The user redirect input from stdin
         """
         self._query = query_string
+        self._stdin = stdin
 
         self._spinner_renderer: SpinnerRenderer = _initialize_spinner_renderer()
         self._text_renderer: TextRenderer = _initialize_text_renderer()
@@ -106,9 +109,16 @@ def run(self) -> int:
             int: Status code of the execution
         """
         proxy = QUERY_IDENTIFIER.get_proxy()
-        input_query = Message()
-        input_query.message = self._query
 
+        query = self._query
+        if self._stdin:
+            # If query is provided, the message becomes "{query} {stdin}",
+            # otherwise, to avoid submitting `None` as part of the query, let's
+            # default to submit only the stidn.
+            query = f"{query} {self._stdin}" if query else self._stdin
+
+        input_query = Message()
+        input_query.message = query
         output = "Nothing to see here..."
 
         try:
@@ -159,4 +169,8 @@ def _command_factory(args: Namespace) -> QueryCommand:
     Returns:
         QueryCommand: Return an instance of class
     """
-    return QueryCommand(args.query_string)
+    # We may not always have the stdin argument in the namespace.
+    if "stdin" in args:
+        return QueryCommand(args.query_string, args.stdin)
+
+    return QueryCommand(args.query_string, None)

command_line_assistant/initialize.py

@@ -1,11 +1,13 @@
 """Main module for the cli."""
 
 import sys
+from argparse import Namespace
 
 from command_line_assistant.commands import history, query, record
 from command_line_assistant.utils.cli import (
     add_default_command,
     create_argument_parser,
+    read_stdin,
 )
 
 
@@ -17,15 +19,19 @@ def initialize() -> int:
     """
     parser, commands_parser = create_argument_parser()
 
-    # TODO: add autodetection of BaseCLICommand classes in the future so we can just drop
-    # new subcommand python modules into the directory and then loop and call `register_subcommand()`
-    # on each one.
+    # TODO: add autodetection of BaseCLICommand classes in the future so we can
+    # just drop new subcommand python modules into the directory and then loop
+    # and call `register_subcommand()` on each one.
     query.register_subcommand(commands_parser)  # type: ignore
     history.register_subcommand(commands_parser)  # type: ignore
     record.register_subcommand(commands_parser)  # type: ignore
 
-    args = add_default_command(sys.argv)
-    args = parser.parse_args(args)
+    stdin = read_stdin()
+    args = add_default_command(stdin, sys.argv)
+
+    # Small workaround to include the stdin in the namespace object in case it exists.
+    namespace = Namespace(stdin=stdin) if stdin else Namespace()
+    args = parser.parse_args(args, namespace=namespace)
 
     if not hasattr(args, "func"):
         parser.print_help()

command_line_assistant/utils/cli.py

@@ -26,16 +26,17 @@ def run(self) -> int:
         """Entrypoint method for all CLI commands."""
 
 
-def add_default_command(argv: list[str]):
+def add_default_command(stdin: Optional[str], argv: list[str]):
     """Add the default command when none is given
 
     Args:
-        argv: THe arguments passed from the stdin.
+        stdin (str): The input string coming from stdin
+        argv (list[str]): List of arguments from CLI
     """
     args = argv[1:]
 
-    # Early exit if we don't have any argv
-    if not args:
+    # Early exit if we don't have any argv or stdin
+    if not args and not stdin:
         return args
 
     subcommand = _subcommand_used(argv)

data/development/config/command_line_assistant/config.toml

@@ -11,8 +11,7 @@ enabled = true
 file = "~/.local/share/command-line-assistant/command-line-assistant_history.json"
 
 [backend]
-endpoint = "https://rlsapi-rhel-lightspeed--runtime-int.apps.int.spoke.preprod.us-east-1.aws.paas.redat.com"
-# endpoint = "http://localhost:8080"
+endpoint = "http://localhost:8080"
 
 [backend.auth]
 cert_file = "data/development/certificate/fake-certificate.pem"

data/release/xdg/history.json

@@ -1 +1 @@
-{"history": []}
+{"history": [],"metadata": {"last_updated": "2024-12-30T12:04:36.298732Z",  "version": "0.1.0",  "entry_count": 0,  "size_bytes": 0}}

tests/commands/test_query.py

@@ -33,7 +33,7 @@ def mock_dbus_service(mock_proxy):
 def test_query_command_initialization():
     """Test QueryCommand initialization"""
     query = "test query"
-    command = QueryCommand(query)
+    command = QueryCommand(query, None)
     assert command._query == query
 
 
@@ -45,6 +45,7 @@ def test_query_command_initialization():
     [
         ("how to list files?", "Use the ls command"),
         ("what is linux?", "Linux is an operating system"),
+        ("test!@#$%^&*()_+ query", "response with special chars !@#%"),
     ],
 )
 def test_query_command_run(mock_dbus_service, test_input, expected_output, capsys):
@@ -55,7 +56,7 @@ def test_query_command_run(mock_dbus_service, test_input, expected_output, capsy
     mock_dbus_service.RetrieveAnswer = lambda: Message.to_structure(mock_output)
 
     # Create and run command
-    command = QueryCommand(test_input)
+    command = QueryCommand(test_input, None)
     command.run()
 
     # Verify ProcessQuery was called with correct input
@@ -77,7 +78,7 @@ def test_query_command_empty_response(mock_dbus_service, capsys):
     mock_output.message = ""
     mock_dbus_service.RetrieveAnswer = lambda: Message.to_structure(mock_output)
 
-    command = QueryCommand("test query")
+    command = QueryCommand("test query", None)
     command.run()
 
     captured = capsys.readouterr()
@@ -93,7 +94,7 @@ def test_query_command_empty_response(mock_dbus_service, capsys):
 )
 def test_query_command_invalid_inputs(mock_dbus_service, test_args):
     """Test QueryCommand with invalid inputs"""
-    command = QueryCommand(test_args)
+    command = QueryCommand(test_args, None)
     command.run()
     # Verify DBus calls still happen even with invalid input
     mock_dbus_service.ProcessQuery.assert_called_once()
@@ -114,16 +115,32 @@ def test_register_subcommand():
     assert hasattr(args, "func")
 
 
-def test_command_factory():
+@pytest.mark.parametrize(
+    ("query_string", "stdin"),
+    (
+        (
+            "test query",
+            None,
+        ),
+        (None, "stdin"),
+        ("test query", "test stdin"),
+    ),
+)
+def test_command_factory(query_string, stdin):
     """Test _command_factory function"""
 
     from command_line_assistant.commands.query import _command_factory
 
-    args = Namespace(query_string="test query")
+    args = (
+        Namespace(query_string=query_string, stdin=stdin)
+        if stdin
+        else Namespace(query_string=query_string)
+    )
     command = _command_factory(args)
 
     assert isinstance(command, QueryCommand)
-    assert command._query == "test query"
+    assert command._query == query_string
+    assert command._stdin == stdin
 
 
 @pytest.mark.parametrize(
@@ -148,31 +165,9 @@ def test_dbus_error_handling(exception, expected, mock_dbus_service, capsys):
     # Make ProcessQuery raise a DBus error
     mock_dbus_service.ProcessQuery.side_effect = exception
 
-    command = QueryCommand("test query")
+    command = QueryCommand("test query", None)
     command.run()
 
     # Verify error message in stdout
     captured = capsys.readouterr()
     assert expected in captured.out.strip()
-
-
-def test_query_with_special_characters(mock_dbus_service, capsys):
-    """Test query containing special characters"""
-    special_query = "test!@#$%^&*()_+ query"
-    expected_response = "response with special chars !@#$"
-
-    mock_output = Message()
-    mock_output.message = expected_response
-    mock_dbus_service.RetrieveAnswer = lambda: Message.to_structure(mock_output)
-
-    command = QueryCommand(special_query)
-    command.run()
-
-    expected_input = Message()
-    expected_input.message = special_query
-    mock_dbus_service.ProcessQuery.assert_called_once_with(
-        Message.to_structure(expected_input)
-    )
-
-    captured = capsys.readouterr()
-    assert expected_response in captured.out.strip()

tests/test_initialize.py

@@ -13,23 +13,38 @@ def run(self):  # type: ignore
 
 def test_initialize_with_no_args(capsys):
     """Test initialize with no arguments - should print help and return 1"""
-    with patch("sys.argv", ["c"]):
+    with (
+        patch("sys.argv", ["c"]),
+        patch("command_line_assistant.initialize.read_stdin", lambda: None),
+    ):
         result = initialize()
         captured = capsys.readouterr()
 
         assert result == 1
         assert "usage:" in captured.out
 
 
-def test_initialize_with_query_command():
+@pytest.mark.parametrize(
+    ("argv", "stdin"),
+    (
+        (
+            ["c", "query", "test", "query"],
+            None,
+        ),
+        (["c"], "test from stdin"),
+        (["c", "what is this?"], "error in line 1"),
+    ),
+)
+def test_initialize_with_query_command(argv, stdin):
     """Test initialize with query command"""
     mock_command = Mock(return_value=MockCommand())
 
     with (
-        patch("sys.argv", ["c", "query", "test query"]),
+        patch("sys.argv", argv),
         patch("command_line_assistant.commands.query.register_subcommand"),
         patch("command_line_assistant.commands.history.register_subcommand"),
         patch("command_line_assistant.commands.record.register_subcommand"),
+        patch("command_line_assistant.initialize.read_stdin", lambda: stdin),
         patch("argparse.ArgumentParser.parse_args") as mock_parse,
     ):
         mock_parse.return_value.func = mock_command
@@ -48,6 +63,7 @@ def test_initialize_with_history_command():
         patch("command_line_assistant.commands.query.register_subcommand"),
         patch("command_line_assistant.commands.history.register_subcommand"),
         patch("command_line_assistant.commands.record.register_subcommand"),
+        patch("command_line_assistant.initialize.read_stdin", lambda: None),
         patch("argparse.ArgumentParser.parse_args") as mock_parse,
     ):
         mock_parse.return_value.func = mock_command
@@ -66,6 +82,7 @@ def test_initialize_with_record_command():
         patch("command_line_assistant.commands.query.register_subcommand"),
         patch("command_line_assistant.commands.history.register_subcommand"),
         patch("command_line_assistant.commands.record.register_subcommand"),
+        patch("command_line_assistant.initialize.read_stdin", lambda: None),
         patch("argparse.ArgumentParser.parse_args") as mock_parse,
     ):
         mock_parse.return_value.func = mock_command
@@ -79,6 +96,7 @@ def test_initialize_with_version():
     """Test initialize with --version flag"""
     with (
         patch("sys.argv", ["c", "--version"]),
+        patch("command_line_assistant.initialize.read_stdin", lambda: None),
         patch("argparse.ArgumentParser.exit") as mock_exit,
     ):
         initialize()
@@ -87,7 +105,10 @@ def test_initialize_with_version():
 
 def test_initialize_with_help(capsys):
     """Test initialize with --help flag"""
-    with patch("sys.argv", ["c", "--help"]):
+    with (
+        patch("sys.argv", ["c", "--help"]),
+        patch("command_line_assistant.initialize.read_stdin", lambda: None),
+    ):
         with pytest.raises(SystemExit):
             initialize()
 
@@ -113,6 +134,7 @@ def test_initialize_command_selection(argv, expected_command):
 
     with (
         patch("sys.argv", argv),
+        patch("command_line_assistant.initialize.read_stdin", lambda: None),
         patch("command_line_assistant.commands.query.register_subcommand"),
         patch("command_line_assistant.commands.history.register_subcommand"),
         patch("command_line_assistant.commands.record.register_subcommand"),