Feat/core stability

src/cmdcraft/base.py

@@ -1,4 +1,10 @@
-#!/usr/bin/env python3
+# *****************************************************************************
+# Copyright (c) 2024-2026, Antonio Mario Weinsen Junior
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+# *****************************************************************************
 """Base interpreter class."""
 
 import asyncio
@@ -39,6 +45,7 @@ def get_funcs() -> list[str]:
         self._history = []
         self._is_running: bool = False
         self._is_init: bool = False
+        self._shutdown_requested: bool = False
 
     async def init(self) -> None:
         """Init the interpreter object."""
@@ -52,6 +59,7 @@ async def run(self) -> None:
         """Run Prompter main loop."""
         if not self._is_init:
             await self.init()
+        self._shutdown_requested = False
 
     @property
     def is_running(self) -> bool:
@@ -63,6 +71,24 @@ def is_running(self) -> bool:
         """
         return self._is_running
 
+    @property
+    def shutdown_requested(self) -> bool:
+        """Return whether a graceful shutdown was requested."""
+        return self._shutdown_requested
+
+    def request_shutdown(self) -> None:
+        """Request a graceful shutdown."""
+        if self._shutdown_requested:
+            self.output("Forced shutdown requested.")
+            raise SystemExit(130)
+
+        self._shutdown_requested = True
+        self._is_running = False
+        self.output(
+            "Graceful shutdown requested. Waiting for the current operation "
+            "to finish. Press Ctrl-C again to force exit."
+        )
+
     def register_command(self, command: callable, alias: str | None = None) -> Command:
         """Register a command into the interpreter.
 
@@ -102,9 +128,13 @@ async def interpret(self, cmdline: str) -> None:
             if len(input.tokens) < 1:
                 return
             cmd = self._commands.get(input.tokens[0], None)
+            if cmd is None:
+                self.output(f"Unknown command: {input.tokens[0]}")
+                await self.help()
+                return
             await cmd.eval(*input.tokens[1:])
         except TypeError as e:
-            await self.help(cmd)
+            await self.help(input.tokens[0])
             self.output(e)
         except Exception as e:
             self.output(e)

src/cmdcraft/command.py

@@ -119,10 +119,17 @@ def eval(self, *args) -> asyncio.Future:
         args = []
         var_args = []
         for a, p in zip(pos, self._positional.values()):
+            if p.is_enum_type and "=" in a:
+                (name, value) = a.split("=", 1)
+                if name == p.name and value in p.options:
+                    raise ValueError(
+                        f"Parameter '{p.name}' is positional. "
+                        f"Use '{value}' instead of '{a}'."
+                    )
             args.append(p.cast(a))
 
         if self.has_args:
-            var_args = pos[len(self._positional):]
+            var_args = pos[len(self._positional) :]
 
         kwargs = {}
         for kw in kws:

src/cmdcraft/completer.py

@@ -3,7 +3,7 @@
 
 from __future__ import annotations
 
-from typing import Iterable
+from collections.abc import Iterable
 
 from prompt_toolkit.completion import (
     CompleteEvent,
@@ -84,10 +84,10 @@ def _get_value_completions(
             Iterable[Completion]: List of Completions for current prompt.
 
         """
-        (par, arg) = prompt.lstrip("--").split("=")
-        if par not in self._command._pars:
+        (par, arg) = prompt.lstrip("--").split("=", 1)
+        if par not in self._command.keyword_parameters:
             return ()
-        vs = self._command._pars[par].options
+        vs = self._command.keyword_parameters[par].options
         if vs is None:
             return ()
         completer = FuzzyWordCompleter(vs)

src/cmdcraft/parameter.py

@@ -1,4 +1,10 @@
-#!/usr/bin/env python3
+# *****************************************************************************
+# Copyright (c) 2024-2026, Antonio Mario Weinsen Junior
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+# *****************************************************************************
 """Callable wrapper for info extraction."""
 
 from __future__ import annotations
@@ -29,6 +35,34 @@ def __init__(
         self._default = default
         self._dyn_opts = None
 
+    def _supports_runtime_cast(self) -> bool:
+        """Return whether the parameter annotation can be cast at runtime.
+
+        Returns:
+            bool: True if the annotation can be cast at runtime, False otherwise.
+
+        """
+        return isinstance(self._type, type)
+
+    def _is_enum_type(self) -> bool:
+        """Return whether the parameter annotation is an enum type.
+
+        Returns:
+            bool: True if the annotation is an enum type, False otherwise.
+
+        """
+        return self._supports_runtime_cast() and issubclass(self._type, Enum)
+
+    @property
+    def is_enum_type(self) -> bool:
+        """Return whether the parameter annotation is an enum type.
+
+        Returns:
+            bool: True if the annotation is an enum type, False otherwise.
+
+        """
+        return self._is_enum_type()
+
     @property
     def name(self) -> str:
         """Return parameter name.
@@ -59,7 +93,7 @@ def options(self) -> list[str]:
         """
         if self._dyn_opts is not None:
             return self._dyn_opts()
-        elif issubclass(self._type, Enum):
+        elif self._is_enum_type():
             return self._type._member_names_
         return []
 
@@ -73,10 +107,24 @@ def cast(self, value: str) -> any:
             any: The cast value.
 
         """
-        if issubclass(self._type, Enum):
-            return self._type[value]
-        if self._type:
-            return self._type(value)
+        if self._is_enum_type():
+            try:
+                return self._type[value]
+            except KeyError:
+                options = ", ".join(self.options)
+                raise ValueError(
+                    f"Invalid value for parameter '{self.name}': {value!r}. "
+                    f"Expected one of: {options}."
+                ) from None
+        if self._supports_runtime_cast():
+            try:
+                return self._type(value)
+            except (TypeError, ValueError):
+                type_name = getattr(self._type, "__name__", str(self._type))
+                raise ValueError(
+                    f"Invalid value for parameter '{self.name}': {value!r}. "
+                    f"Expected {type_name}."
+                ) from None
         return value
 
     def set_dynamic_options(self, generator: callable) -> None:

src/cmdcraft/prompter.py

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+import signal
+
 from prompt_toolkit import PromptSession
 from prompt_toolkit.completion import NestedCompleter
 
@@ -18,6 +20,7 @@ def __init__(self) -> None:
         """Construct the interpreter object."""
         super().__init__()
         self._session = PromptSession()
+        self._previous_sigint_handler = None
 
     async def init(self) -> None:
         """Init the interpreter object."""
@@ -30,15 +33,63 @@ def completer(self) -> None:
             cmds[name] = CommandCompleter(cmd)
         return NestedCompleter(cmds)
 
+    def _has_active_prompt(self) -> bool:
+        """Return whether a prompt_toolkit application is running."""
+        app = self._session.app
+        if app is None:
+            return False
+        future = getattr(app, "future", None)
+        return (
+            getattr(app, "is_running", False)
+            and future is not None
+            and not future.done()
+        )
+
+    def _exit_active_prompt(self) -> None:
+        """Exit the active prompt without raising a traceback."""
+        if self._has_active_prompt():
+            self._session.app.exit(result="")
+
+    def _handle_sigint(self, _: int, __) -> None:
+        """Handle Ctrl-C with graceful-then-force semantics."""
+        self._exit_active_prompt()
+        self.request_shutdown()
+
+    def _handle_runtime_interrupt(self) -> None:
+        """Handle a KeyboardInterrupt raised inside the runtime loop."""
+        if self.shutdown_requested:
+            raise SystemExit(130) from None
+
+        self.request_shutdown()
+
     async def run(self) -> None:
         """Run Prompter main loop."""
         await super().run()
         self._is_running = True
-        await self.interpret("help")
-        while self.is_running:
-            cmdline = await self._session.prompt_async("> ", completer=self.completer())
-            self._history.append(cmdline)
-            await self.interpret(cmdline)
+        self._previous_sigint_handler = signal.getsignal(signal.SIGINT)
+        signal.signal(signal.SIGINT, self._handle_sigint)
+
+        try:
+            await self.interpret("help")
+            while self.is_running:
+                try:
+                    cmdline = await self._session.prompt_async(
+                        "> ",
+                        completer=self.completer(),
+                        handle_sigint=False,
+                    )
+                    if self.shutdown_requested:
+                        break
+
+                    self._history.append(cmdline)
+                    await self.interpret(cmdline)
+                except KeyboardInterrupt:
+                    self._handle_runtime_interrupt()
+                    break
+        finally:
+            if self._previous_sigint_handler is not None:
+                signal.signal(signal.SIGINT, self._previous_sigint_handler)
+                self._previous_sigint_handler = None
 
     def output(self, *args) -> None:
         """Output command."""

test/test_base.py

@@ -0,0 +1,90 @@
+# *****************************************************************************
+# Copyright (c) 2024-2026, Antonio Mario Weinsen Junior
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+# *****************************************************************************
+"""Tests for BasePrompter."""
+
+import asyncio
+from enum import Enum
+
+from cmdcraft.base import BasePrompter
+
+
+class _DummyPrompter(BasePrompter):
+    def __init__(self) -> None:
+        self.outputs = []
+        super().__init__()
+
+    def output(self, *args) -> None:
+        self.outputs.append(" ".join(str(arg) for arg in args))
+
+
+def test_interpret_unknown_command():
+    """Test handling of unknown commands."""
+    prompt = _DummyPrompter()
+
+    asyncio.run(prompt.interpret("missing"))
+
+    assert prompt.outputs[0] == "Unknown command: missing"
+    assert "Show Cmdcraft interpreter help." in prompt.outputs[1]
+
+
+def test_request_shutdown_escalates_on_second_call():
+    """Test graceful then forced shutdown requests."""
+    prompt = _DummyPrompter()
+    prompt._is_running = True
+
+    prompt.request_shutdown()
+    assert prompt.shutdown_requested is True
+    assert prompt.is_running is False
+    assert prompt.outputs[-1] == (
+        "Graceful shutdown requested. Waiting for the current operation to "
+        "finish. Press Ctrl-C again to force exit."
+    )
+
+    try:
+        prompt.request_shutdown()
+    except SystemExit as exc:
+        assert exc.code == 130
+    else:
+        assert False, "Expected a forced shutdown to raise SystemExit"
+
+    assert prompt.outputs[-1] == "Forced shutdown requested."
+
+
+def test_interpret_positional_enum_assignment_hint():
+    """Test a readable hint for mistaken named positional enum syntax."""
+
+    class EnumTest(Enum):
+        A = 1
+        B = 2
+
+    async def test_input(a: EnumTest) -> None:
+        return None
+
+    prompt = _DummyPrompter()
+    prompt.register_command(test_input)
+
+    asyncio.run(prompt.interpret("test_input a=A"))
+
+    assert prompt.outputs == ["Parameter 'a' is positional. Use 'A' instead of 'a=A'."]
+
+
+def test_interpret_type_error_shows_command_help():
+    """Test TypeError handling shows the failing command help text."""
+
+    async def test_input(required: str) -> None:
+        """Specific command help."""
+        return None
+
+    prompt = _DummyPrompter()
+    prompt.register_command(test_input)
+
+    asyncio.run(prompt.interpret("test_input"))
+
+    assert prompt.outputs[0] == "Specific command help."
+    assert prompt.outputs[1] == ""
+    assert "missing 1 required positional argument" in prompt.outputs[2]