mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 10:05:28 +03:00
fix tests and improve perf
This commit is contained in:
+11
-11
@@ -1,14 +1,14 @@
|
||||
from argenta import Command, Response, Router, App, Orchestrator
|
||||
from argenta.command import InputCommand
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||
print('test command')
|
||||
class Figure(ABC):
|
||||
@abstractmethod
|
||||
def draw(self) -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
app = App(override_system_messages=True, print_func=print)
|
||||
app.include_router(router)
|
||||
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}'))
|
||||
orchestrator.start_polling(app)
|
||||
class Rectangle(Figure):
|
||||
def __init__(self, x: int, y: int) -> None:
|
||||
self.x = x
|
||||
self.y = y
|
||||
|
||||
rec = Rectangle(5, 2)
|
||||
@@ -1,16 +1,13 @@
|
||||
from argenta import App, Orchestrator
|
||||
from argenta.app import PredefinedMessages
|
||||
from argenta.orchestrator.argparser import ArgParser, BooleanArgument
|
||||
from argenta.app.dividing_line.models import DynamicDividingLine
|
||||
from mock.mock_app.routers import work_router
|
||||
|
||||
app: App = App(
|
||||
dividing_line=DynamicDividingLine('^'),
|
||||
)
|
||||
argparser = ArgParser([BooleanArgument('some')])
|
||||
orchestrator: Orchestrator = Orchestrator(argparser)
|
||||
orchestrator: Orchestrator = Orchestrator()
|
||||
|
||||
print(argparser.parsed_argspace.get_by_type(BooleanArgument))
|
||||
|
||||
def main():
|
||||
app.include_router(work_router)
|
||||
@@ -22,5 +19,5 @@ def main():
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
if __name__ == "__main__":
|
||||
orchestrator.start_polling(app)
|
||||
main()
|
||||
|
||||
@@ -4,7 +4,14 @@ from argenta.command import Flag, Flags
|
||||
work_router: Router = Router(title="Base points:", disable_redirect_stdout=True)
|
||||
|
||||
|
||||
@work_router.command(Command("hello", flags=Flags(Flag("test")), description="Hello, world!"))
|
||||
@work_router.command(
|
||||
Command(
|
||||
"hello",
|
||||
flags=Flags([
|
||||
Flag("test")
|
||||
]),
|
||||
description="Hello, world!")
|
||||
)
|
||||
def command_help(response: Response):
|
||||
c = input("Enter your name: ")
|
||||
print(f"Hello, {c}!")
|
||||
|
||||
@@ -212,6 +212,11 @@ class BaseApp:
|
||||
return True
|
||||
return False
|
||||
|
||||
def _is_unknown_command(self, input_command: InputCommand) -> bool:
|
||||
if not self.registered_routers.get_router_by_trigger(input_command.trigger.lower()):
|
||||
return True
|
||||
return False
|
||||
|
||||
def _error_handler(self, error: InputCommandException, raw_command: str) -> None:
|
||||
"""
|
||||
Private. Handles parsing errors of the entered command
|
||||
@@ -389,7 +394,6 @@ class App(BaseApp):
|
||||
farewell_message: str = "\nSee you\n",
|
||||
exit_command: Command = DEFAULT_EXIT_COMMAND,
|
||||
system_router_title: str = "System points:",
|
||||
ignore_command_register: bool = True,
|
||||
dividing_line: AVAILABLE_DIVIDING_LINES = DEFAULT_DIVIDING_LINE,
|
||||
repeat_command_groups_printing: bool = False,
|
||||
override_system_messages: bool = False,
|
||||
@@ -450,7 +454,7 @@ class App(BaseApp):
|
||||
self._autocompleter.exit_setup(self.registered_routers.get_triggers())
|
||||
return
|
||||
|
||||
if self.registered_routers.get_router_by_trigger(input_command.trigger.lower()):
|
||||
if self._is_unknown_command(input_command):
|
||||
with redirect_stdout(io.StringIO()) as stdout:
|
||||
self._unknown_command_handler(input_command)
|
||||
stdout_res: str = stdout.getvalue()
|
||||
|
||||
@@ -49,7 +49,6 @@ class CommandHandlers:
|
||||
self.paired_command_handler_trigger[alias.lower()] = command_handler
|
||||
|
||||
def get_command_handler_by_trigger(self, trigger: str) -> CommandHandler | None:
|
||||
print(self.paired_command_handler_trigger)
|
||||
return self.paired_command_handler_trigger.get(trigger)
|
||||
|
||||
def __iter__(self) -> Iterator[CommandHandler]:
|
||||
|
||||
@@ -72,27 +72,6 @@ def test_unknown_command_triggers_unknown_command_handler(monkeypatch: pytest.Mo
|
||||
assert "\nUnknown command: help\n" in output
|
||||
|
||||
|
||||
def test_case_sensitive_command_triggers_unknown_command_handler(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
inputs = iter(["TeSt", "Q"])
|
||||
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||
print('test command')
|
||||
|
||||
app = App(ignore_command_register=False, override_system_messages=True, print_func=print)
|
||||
app.include_router(router)
|
||||
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}'))
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = capsys.readouterr().out
|
||||
|
||||
assert '\nUnknown command: TeSt\n' in output
|
||||
|
||||
|
||||
def test_mixed_valid_and_unknown_commands_handled_correctly(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
inputs = iter(["test", "some", "q"])
|
||||
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||
|
||||
@@ -46,26 +46,6 @@ def test_simple_command_executes_successfully(monkeypatch: pytest.MonkeyPatch, c
|
||||
assert '\ntest command\n' in output
|
||||
|
||||
|
||||
def test_case_insensitive_command_executes_when_enabled(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
inputs = iter(["TeSt", "q"])
|
||||
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||
print('test command')
|
||||
|
||||
app = App(ignore_command_register=True, override_system_messages=True, print_func=print)
|
||||
app.include_router(router)
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = capsys.readouterr().out
|
||||
|
||||
assert '\ntest command\n' in output
|
||||
|
||||
|
||||
def test_two_commands_execute_sequentially(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
inputs = iter(["test", "some", "q"])
|
||||
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||
|
||||
@@ -26,46 +26,21 @@ def test_default_exit_command_uppercase_q_is_recognized() -> None:
|
||||
assert app._is_exit_command(InputCommand('Q')) is True
|
||||
|
||||
|
||||
def test_exit_command_not_recognized_when_case_sensitivity_enabled() -> None:
|
||||
app = App(ignore_command_register=False)
|
||||
assert app._is_exit_command(InputCommand('q')) is False
|
||||
|
||||
|
||||
def test_custom_exit_command_is_recognized() -> None:
|
||||
app = App(exit_command=Command('quit'))
|
||||
assert app._is_exit_command(InputCommand('quit')) is True
|
||||
|
||||
|
||||
def test_custom_exit_command_case_insensitive_by_default() -> None:
|
||||
app = App(exit_command=Command('quit'))
|
||||
assert app._is_exit_command(InputCommand('qUIt')) is True
|
||||
|
||||
|
||||
def test_custom_exit_command_case_sensitive_when_enabled() -> None:
|
||||
app = App(ignore_command_register=False, exit_command=Command('quit'))
|
||||
assert app._is_exit_command(InputCommand('qUIt')) is False
|
||||
|
||||
|
||||
def test_exit_command_alias_is_recognized() -> None:
|
||||
app = App(exit_command=Command('q', aliases={'exit'}))
|
||||
assert app._is_exit_command(InputCommand('exit')) is True
|
||||
|
||||
|
||||
def test_exit_command_alias_case_sensitive_when_enabled() -> None:
|
||||
app = App(exit_command=Command('q', aliases={'exit'}), ignore_command_register=False)
|
||||
assert app._is_exit_command(InputCommand('exit')) is True
|
||||
|
||||
|
||||
def test_non_exit_command_is_not_recognized() -> None:
|
||||
app = App(exit_command=Command('q', aliases={'exit'}))
|
||||
assert app._is_exit_command(InputCommand('quit')) is False
|
||||
|
||||
|
||||
def test_non_exit_command_with_wrong_case_is_not_recognized() -> None:
|
||||
app = App(exit_command=Command('q', aliases={'exit'}), ignore_command_register=False)
|
||||
assert app._is_exit_command(InputCommand('Exit')) is False
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Tests for unknown command detection
|
||||
# ============================================================================
|
||||
@@ -74,31 +49,22 @@ def test_non_exit_command_with_wrong_case_is_not_recognized() -> None:
|
||||
def test_registered_command_is_not_unknown() -> None:
|
||||
app = App()
|
||||
app.set_unknown_command_handler(lambda command: None)
|
||||
app._current_matching_triggers_with_routers = {'fr': Router(), 'tr': Router(), 'de': Router()}
|
||||
router = Router()
|
||||
|
||||
@router.command('fr')
|
||||
def handler(res: Response):
|
||||
pass
|
||||
|
||||
app.include_router(router)
|
||||
assert app._is_unknown_command(InputCommand('fr')) is False
|
||||
|
||||
|
||||
def test_unregistered_command_is_unknown() -> None:
|
||||
app = App()
|
||||
app.set_unknown_command_handler(lambda command: None)
|
||||
app._current_matching_triggers_with_routers = {'fr': Router(), 'tr': Router(), 'de': Router()}
|
||||
assert app._is_unknown_command(InputCommand('cr')) is True
|
||||
|
||||
|
||||
def test_command_with_wrong_case_is_unknown_when_case_sensitivity_enabled() -> None:
|
||||
app = App(ignore_command_register=False)
|
||||
app.set_unknown_command_handler(lambda command: None)
|
||||
app._current_matching_triggers_with_routers = {'Pr': Router(), 'tW': Router(), 'deQW': Router()}
|
||||
assert app._is_unknown_command(InputCommand('pr')) is True
|
||||
|
||||
|
||||
def test_command_with_exact_case_is_not_unknown_when_case_sensitivity_enabled() -> None:
|
||||
app = App(ignore_command_register=False)
|
||||
app.set_unknown_command_handler(lambda command: None)
|
||||
app._current_matching_triggers_with_routers = {'Pr': Router(), 'tW': Router(), 'deQW': Router()}
|
||||
assert app._is_unknown_command(InputCommand('tW')) is False
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Tests for similar command suggestions
|
||||
# ============================================================================
|
||||
@@ -632,7 +598,7 @@ def test_handler_can_be_replaced_multiple_times() -> None:
|
||||
|
||||
def test_handler_receives_correct_parameters() -> None:
|
||||
app = App()
|
||||
received_data = {'trigger': None}
|
||||
received_data: dict[str, None | str] = {'trigger': None}
|
||||
|
||||
def custom_handler(command: InputCommand) -> None:
|
||||
received_data['trigger'] = command.trigger
|
||||
@@ -645,7 +611,7 @@ def test_handler_receives_correct_parameters() -> None:
|
||||
|
||||
def test_exit_handler_receives_response_object() -> None:
|
||||
app = App()
|
||||
received_data = {'response': None}
|
||||
received_data: dict[str, None | Response] = {'response': None}
|
||||
|
||||
def custom_handler(response: Response) -> None:
|
||||
received_data['response'] = response
|
||||
|
||||
Reference in New Issue
Block a user