diff --git a/src/cmdcraft/base.py b/src/cmdcraft/base.py index 57442b1..2816bfa 100644 --- a/src/cmdcraft/base.py +++ b/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) diff --git a/src/cmdcraft/command.py b/src/cmdcraft/command.py index 749d88a..af7895b 100644 --- a/src/cmdcraft/command.py +++ b/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: diff --git a/src/cmdcraft/completer.py b/src/cmdcraft/completer.py index 9dd007b..7834864 100644 --- a/src/cmdcraft/completer.py +++ b/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) diff --git a/src/cmdcraft/parameter.py b/src/cmdcraft/parameter.py index 04e2138..d322ef4 100644 --- a/src/cmdcraft/parameter.py +++ b/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: diff --git a/src/cmdcraft/prompter.py b/src/cmdcraft/prompter.py index 6375aaa..4c92cb1 100644 --- a/src/cmdcraft/prompter.py +++ b/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.""" diff --git a/test/test_base.py b/test/test_base.py new file mode 100644 index 0000000..47ecea9 --- /dev/null +++ b/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] diff --git a/test/test_completer.py b/test/test_completer.py new file mode 100644 index 0000000..6e61fa5 --- /dev/null +++ b/test/test_completer.py @@ -0,0 +1,66 @@ +# ***************************************************************************** +# 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 Completer.""" + +from enum import Enum + +from prompt_toolkit.completion import CompleteEvent +from prompt_toolkit.document import Document + +from cmdcraft.command import Command +from cmdcraft.completer import CommandCompleter + + +class EnumTest(Enum): + """Test enum for completion.""" + + A = 1 + B = 2 + + +async def _test_input(a: EnumTest, *, prompt: str) -> None: + return None + + +def _dynamic_options() -> list[str]: + return ["alpha", "beta"] + + +def _build_completer() -> CommandCompleter: + cmd = Command(_test_input) + cmd.process() + cmd.parameter("prompt").set_dynamic_options(_dynamic_options) + return CommandCompleter(cmd) + + +def test_keyword_value_completion_uses_dynamic_options(): + """Test keyword value completion with dynamic options.""" + completer = _build_completer() + + completions = [ + completion.text + for completion in completer.get_completions( + Document("test_input --prompt="), CompleteEvent() + ) + ] + + assert completions == ["alpha", "beta"] + + +def test_positional_value_assignment_has_no_completions(): + """Test positional parameters do not get keyword-style value completion.""" + completer = _build_completer() + + completions = [ + completion.text + for completion in completer.get_completions( + Document("test_input a="), CompleteEvent() + ) + ] + + assert completions == [] diff --git a/test/test_parameter.py b/test/test_parameter.py index dbbf20c..50c0303 100644 --- a/test/test_parameter.py +++ b/test/test_parameter.py @@ -1,25 +1,34 @@ -#!/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. +# ***************************************************************************** +"""Tests for Parameter.""" -import pytest from enum import Enum +import pytest + from cmdcraft.parameter import Parameter + def test_name(): """Test the name property method.""" - par = Parameter("parameter1", str) assert par.name == "parameter1" + def test_default(): """Test the default property method.""" - par = Parameter("parameter1", int) - assert par.default == None + assert par.default is None par = Parameter("parameter2", str, "default value") assert par.default == "default value" + def test_parameter_str(): """Test method for a `str` parameter.""" par = Parameter("options", str, "0") @@ -27,6 +36,14 @@ def test_parameter_str(): assert par.default == "0" assert par.cast("1001") == "1001" + +def test_parameter_untyped(): + """Test method for an untyped parameter.""" + par = Parameter("options") + assert par.options == [] + assert par.cast("1001") == "1001" + + def test_parameter_int(): """Test method for a `int` parameter.""" par = Parameter("options", int, 0) @@ -34,8 +51,14 @@ def test_parameter_int(): assert par.default == 0 assert par.cast("1001") == 1001 - with pytest.raises(Exception): - assert par.cast("ARG") + with pytest.raises(ValueError) as excinfo: + par.cast("ARG") + + assert ( + str(excinfo.value) + == "Invalid value for parameter 'options': 'ARG'. Expected int." + ) + def test_parameter_enum(): """Test method for a `Enum` parameter.""" @@ -50,6 +73,10 @@ class TestEnum(Enum): assert par.default == TestEnum.OPT_0 assert par.cast("OPT_A") == TestEnum.OPT_A - with pytest.raises(Exception): - assert par.cast("OPT_Z") + with pytest.raises(ValueError) as excinfo: + par.cast("OPT_Z") + assert ( + str(excinfo.value) == "Invalid value for parameter 'options': 'OPT_Z'. " + "Expected one of: OPT_0, OPT_A, OPT_B." + ) diff --git a/test/test_prompter.py b/test/test_prompter.py new file mode 100644 index 0000000..c8e0a61 --- /dev/null +++ b/test/test_prompter.py @@ -0,0 +1,142 @@ +# ***************************************************************************** +# 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 Prompter.""" + +import asyncio + +import pytest + +from cmdcraft.prompter import Prompter + + +class _DummyPrompter(Prompter): + def __init__(self) -> None: + self.outputs = [] + super().__init__() + + def output(self, *args) -> None: + self.outputs.append(" ".join(str(arg) for arg in args)) + + +class _FakeFuture: + def __init__(self, done: bool = False) -> None: + self._done = done + + def done(self) -> bool: + return self._done + + +class _FakeApp: + def __init__(self, running: bool = True) -> None: + self.is_running = running + self.future = _FakeFuture(done=not running) + self.exit_calls = [] + + def exit(self, result=None, exception=None, style: str = "") -> None: + self.exit_calls.append( + {"result": result, "exception": exception, "style": style} + ) + self.is_running = False + self.future = _FakeFuture(done=True) + + +class _FakeSession: + def __init__(self, app: _FakeApp) -> None: + self.app = app + + +class _GracefulInterruptSession(_FakeSession): + def __init__(self, prompt: Prompter, app: _FakeApp) -> None: + super().__init__(app) + self._prompt = prompt + + async def prompt_async(self, *args, **kwargs) -> str: + self._prompt._handle_sigint(2, None) + return "" + + +class _InterruptingSession(_FakeSession): + async def prompt_async(self, *args, **kwargs) -> str: + raise KeyboardInterrupt + + +def test_sigint_graceful_shutdown_exits_active_prompt(): + """Test first Ctrl-C exits the active prompt cleanly.""" + prompt = _DummyPrompter() + app = _FakeApp(running=True) + prompt._session = _FakeSession(app) + prompt._is_running = True + + prompt._handle_sigint(2, None) + + assert prompt.shutdown_requested is True + assert prompt.is_running is False + assert app.exit_calls == [{"result": "", "exception": None, "style": ""}] + + +def test_sigint_graceful_shutdown_without_active_prompt(): + """Test Ctrl-C still shuts down cleanly when no prompt app is active.""" + prompt = _DummyPrompter() + prompt._session = _FakeSession(None) + prompt._is_running = True + + prompt._handle_sigint(2, None) + + assert prompt.shutdown_requested is True + assert prompt.is_running is False + + +def test_sigint_force_shutdown_raises_on_second_press(): + """Test second Ctrl-C forces the loop to exit immediately.""" + prompt = _DummyPrompter() + prompt._session = _FakeSession(_FakeApp(running=False)) + prompt._is_running = True + + prompt._handle_sigint(2, None) + + with pytest.raises(SystemExit) as excinfo: + prompt._handle_sigint(2, None) + + assert excinfo.value.code == 130 + assert prompt.outputs[-1] == "Forced shutdown requested." + + +def test_run_handles_prompt_keyboardinterrupt_gracefully(): + """Test prompt-side Ctrl-C exits without surfacing a traceback.""" + + async def scenario() -> None: + prompt = _DummyPrompter() + prompt._session = _GracefulInterruptSession(prompt, _FakeApp(running=True)) + + await prompt.run() + + assert prompt.shutdown_requested is True + assert prompt.outputs[-1] == ( + "Graceful shutdown requested. Waiting for the current operation to " + "finish. Press Ctrl-C again to force exit." + ) + + asyncio.run(scenario()) + + +def test_run_handles_direct_prompt_keyboardinterrupt_gracefully(): + """Test a direct prompt KeyboardInterrupt still exits gracefully.""" + + async def scenario() -> None: + prompt = _DummyPrompter() + prompt._session = _InterruptingSession(_FakeApp(running=False)) + + await prompt.run() + + assert prompt.shutdown_requested is True + assert prompt.outputs[-1] == ( + "Graceful shutdown requested. Waiting for the current operation to " + "finish. Press Ctrl-C again to force exit." + ) + + asyncio.run(scenario())