diff --git a/mock/local_test.py b/mock/local_test.py index 7061395..0aaef4e 100644 --- a/mock/local_test.py +++ b/mock/local_test.py @@ -1,19 +1,10 @@ -from argenta import App, DataBridge, Response, Router -from argenta.di import FromDishka -from argenta.di.integration import setup_dishka, _auto_inject_handlers -from argenta.di.providers import SystemProvider -from dishka import make_container +from argenta import Command, Response, Router +from argenta.command import InputCommand -container = make_container() - -Response.patch_by_container(container) - -app = App() router = Router() -@router.command('command') -def handler(res: Response, data_bridge: FromDishka[DataBridge]): - print(data_bridge) - -_auto_inject_handlers(app) -_auto_inject_handlers(app) +@router.command(Command('heLLo')) +def handler(_res: Response) -> None: + print("Hello World!") + +router.finds_appropriate_handler(InputCommand('HellO')) diff --git a/src/argenta/app/models.py b/src/argenta/app/models.py index 1da6c74..9b0d31b 100644 --- a/src/argenta/app/models.py +++ b/src/argenta/app/models.py @@ -26,7 +26,6 @@ from argenta.command.exceptions import ( from argenta.command.models import Command, InputCommand from argenta.response import Response from argenta.router import Router -from argenta.router.defaults import system_router Matches: TypeAlias = list[str] | list[Never] @@ -50,12 +49,12 @@ class BaseApp: self._prompt: str = prompt self._print_func: Printer = print_func self._exit_command: Command = exit_command - self._system_router_title: str = system_router_title self._dividing_line: StaticDividingLine | DynamicDividingLine = dividing_line self._ignore_command_register: bool = ignore_command_register self._repeat_command_groups_printing_description: bool = repeat_command_groups_printing self._override_system_messages: bool = override_system_messages self._autocompleter: AutoCompleter = autocompleter + self.system_router: Router = Router(title=system_router_title) self._farewell_message: str = farewell_message self._initial_message: str = initial_message @@ -75,18 +74,20 @@ class BaseApp: else self._matching_default_triggers_with_routers ) - self._incorrect_input_syntax_handler: NonStandardBehaviorHandler[str] = lambda _: print_func( - f"Incorrect flag syntax: {_}" + self._incorrect_input_syntax_handler: NonStandardBehaviorHandler[str] = ( + lambda _: print_func(f"Incorrect flag syntax: {_}") ) - self._repeated_input_flags_handler: NonStandardBehaviorHandler[str] = lambda _: print_func( - f"Repeated input flags: {_}" + self._repeated_input_flags_handler: NonStandardBehaviorHandler[str] = ( + lambda _: print_func(f"Repeated input flags: {_}") ) - self._empty_input_command_handler: EmptyCommandHandler = lambda: print_func("Empty input command") - self._unknown_command_handler: NonStandardBehaviorHandler[InputCommand] = lambda _: print_func( - f"Unknown command: {_.trigger}" + self._empty_input_command_handler: EmptyCommandHandler = lambda: print_func( + "Empty input command" ) - self._exit_command_handler: NonStandardBehaviorHandler[Response] = lambda _: print_func( - self._farewell_message + self._unknown_command_handler: NonStandardBehaviorHandler[InputCommand] = ( + lambda _: print_func(f"Unknown command: {_.trigger}") + ) + self._exit_command_handler: NonStandardBehaviorHandler[Response] = ( + lambda _: print_func(self._farewell_message) ) def set_description_message_pattern(self, _: DescriptionMessageGenerator, /) -> None: @@ -97,7 +98,9 @@ class BaseApp: """ self._description_message_gen = _ - def set_incorrect_input_syntax_handler(self, _: NonStandardBehaviorHandler[str], /) -> None: + def set_incorrect_input_syntax_handler( + self, _: NonStandardBehaviorHandler[str], / + ) -> None: """ Public. Sets the handler for incorrect flags when entering a command :param _: handler for incorrect flags when entering a command @@ -105,7 +108,9 @@ class BaseApp: """ self._incorrect_input_syntax_handler = _ - def set_repeated_input_flags_handler(self, _: NonStandardBehaviorHandler[str], /) -> None: + def set_repeated_input_flags_handler( + self, _: NonStandardBehaviorHandler[str], / + ) -> None: """ Public. Sets the handler for repeated flags when entering a command :param _: handler for repeated flags when entering a command @@ -113,7 +118,9 @@ class BaseApp: """ self._repeated_input_flags_handler = _ - def set_unknown_command_handler(self, _: NonStandardBehaviorHandler[InputCommand], /) -> None: + def set_unknown_command_handler( + self, _: NonStandardBehaviorHandler[InputCommand], / + ) -> None: """ Public. Sets the handler for unknown commands when entering a command :param _: handler for unknown commands when entering a command @@ -129,7 +136,9 @@ class BaseApp: """ self._empty_input_command_handler = _ - def set_exit_command_handler(self, _: NonStandardBehaviorHandler[Response], /) -> None: + def set_exit_command_handler( + self, _: NonStandardBehaviorHandler[Response], / + ) -> None: """ Public. Sets the handler for exit command when entering a command :param _: handler for exit command when entering a command @@ -164,7 +173,11 @@ class BaseApp: clear_text = re.sub(r"\u001b\[[0-9;]*m", "", text) max_length_line = max([len(line) for line in clear_text.split("\n")]) max_length_line = ( - max_length_line if 10 <= max_length_line <= 80 else 80 if max_length_line > 80 else 10 + max_length_line + if 10 <= max_length_line <= 80 + else 80 + if max_length_line > 80 + else 10 ) self._print_func( @@ -181,11 +194,15 @@ class BaseApp: elif isinstance(self._dividing_line, StaticDividingLine): # pyright: ignore[reportUnnecessaryIsInstance] self._print_func( - self._dividing_line.get_full_static_line(is_override=self._override_system_messages) + self._dividing_line.get_full_static_line( + is_override=self._override_system_messages + ) ) print(text.strip("\n")) self._print_func( - self._dividing_line.get_full_static_line(is_override=self._override_system_messages) + self._dividing_line.get_full_static_line( + is_override=self._override_system_messages + ) ) else: @@ -219,10 +236,14 @@ class BaseApp: """ input_command_trigger = command.trigger if self._ignore_command_register: - if input_command_trigger.lower() in list(self._current_matching_triggers_with_routers.keys()): + if input_command_trigger.lower() in list( + self._current_matching_triggers_with_routers.keys() + ): return False else: - if input_command_trigger in list(self._current_matching_triggers_with_routers.keys()): + if input_command_trigger in list( + self._current_matching_triggers_with_routers.keys() + ): return False return True @@ -245,14 +266,13 @@ class BaseApp: Private. Sets up system router :return: None """ - system_router.title = self._system_router_title - @system_router.command(self._exit_command) + @self.system_router.command(self._exit_command) def _(response: Response) -> None: self._exit_command_handler(response) - system_router.command_register_ignore = self._ignore_command_register - self.registered_routers.add_registered_router(system_router) + self.system_router.command_register_ignore = self._ignore_command_register + self.registered_routers.add_registered_router(self.system_router) def _most_similar_command(self, unknown_command: str) -> str | None: all_commands = list(self._current_matching_triggers_with_routers.keys()) @@ -279,7 +299,9 @@ class BaseApp: :return: None """ self._prompt = f"[italic dim bold]{self._prompt}" - self._initial_message = "\n" + f"[bold red]{text2art(self._initial_message, font='tarty1')}" + "\n" + self._initial_message = ( + "\n" + f"[bold red]{text2art(self._initial_message, font='tarty1')}" + "\n" + ) self._farewell_message = ( "[bold red]\n\n" + str(text2art(self._farewell_message, font="chanky")) # pyright: ignore[reportUnknownArgumentType] @@ -297,14 +319,20 @@ class BaseApp: self._repeated_input_flags_handler = lambda raw_command: self._print_func( f"[red bold]Repeated input flags: {escape(raw_command)}" ) - self._empty_input_command_handler = lambda: self._print_func("[red bold]Empty input command") + self._empty_input_command_handler = lambda: self._print_func( + "[red bold]Empty input command" + ) def unknown_command_handler(command: InputCommand) -> None: cmd_trg: str = command.trigger mst_sim_cmd: str | None = self._most_similar_command(cmd_trg) - first_part_of_text = f"[red]Unknown command:[/red] [blue]{escape(cmd_trg)}[/blue]" + first_part_of_text = ( + f"[red]Unknown command:[/red] [blue]{escape(cmd_trg)}[/blue]" + ) second_part_of_text = ( - ("[red], most similar:[/red] " + ("[blue]" + mst_sim_cmd + "[/blue]")) if mst_sim_cmd else "" + ("[red], most similar:[/red] " + ("[blue]" + mst_sim_cmd + "[/blue]")) + if mst_sim_cmd + else "" ) self._print_func(first_part_of_text + second_part_of_text) @@ -324,9 +352,13 @@ class BaseApp: for trigger in combined: self._matching_default_triggers_with_routers[trigger] = router_entity - self._matching_lower_triggers_with_routers[trigger.lower()] = router_entity + self._matching_lower_triggers_with_routers[trigger.lower()] = ( + router_entity + ) - self._autocompleter.initial_setup(list(self._current_matching_triggers_with_routers.keys())) + self._autocompleter.initial_setup( + list(self._current_matching_triggers_with_routers.keys()) + ) if not self._override_system_messages: self._setup_default_view() @@ -339,9 +371,11 @@ class BaseApp: print("\n") if not self._repeat_command_groups_printing_description: self._print_command_group_description() - + def _process_exist_and_valid_command(self, input_command: InputCommand) -> None: - processing_router = self._current_matching_triggers_with_routers[input_command.trigger.lower()] + processing_router = self._current_matching_triggers_with_routers[ + input_command.trigger.lower() + ] if processing_router.disable_redirect_stdout: dividing_line_unit_part: str = self._dividing_line.get_unit_part() @@ -439,9 +473,10 @@ class App(BaseApp): continue if self._is_exit_command(input_command): - system_router.finds_appropriate_handler(input_command) + self.system_router.finds_appropriate_handler(input_command) self._autocompleter.exit_setup( - list(self._current_matching_triggers_with_routers.keys()), self._ignore_command_register + list(self._current_matching_triggers_with_routers.keys()), + self._ignore_command_register, ) return diff --git a/src/argenta/response/status.py b/src/argenta/response/status.py index c736de0..c156494 100644 --- a/src/argenta/response/status.py +++ b/src/argenta/response/status.py @@ -1,6 +1,7 @@ __all__ = ["ResponseStatus"] from enum import Enum +from typing import Self class ResponseStatus(Enum): @@ -10,7 +11,7 @@ class ResponseStatus(Enum): UNDEFINED_AND_INVALID_FLAGS = "UNDEFINED_AND_INVALID_FLAGS" @classmethod - def from_flags(cls, *, has_invalid_value_flags: bool, has_undefined_flags: bool) -> "ResponseStatus": + def from_flags(cls, *, has_invalid_value_flags: bool, has_undefined_flags: bool) -> Self: key = (has_invalid_value_flags, has_undefined_flags) status_map: dict[tuple[bool, bool], ResponseStatus] = { (True, True): cls.UNDEFINED_AND_INVALID_FLAGS, diff --git a/src/argenta/router/command_handler/entity.py b/src/argenta/router/command_handler/entity.py index b95afd6..51dc1cb 100644 --- a/src/argenta/router/command_handler/entity.py +++ b/src/argenta/router/command_handler/entity.py @@ -7,14 +7,17 @@ from argenta.command import Command from argenta.response import Response +HandlerFunc = Callable[..., None] + + class CommandHandler: - def __init__(self, handler_as_func: Callable[..., None], handled_command: Command): + def __init__(self, handler_as_func: HandlerFunc, handled_command: Command): """ Private. Entity of the model linking the handler and the command being processed :param handler: the handler being called :param handled_command: the command being processed """ - self.handler_as_func: Callable[..., None] = handler_as_func + self.handler_as_func: HandlerFunc = handler_as_func self.handled_command: Command = handled_command def handling(self, response: Response) -> None: @@ -27,12 +30,13 @@ class CommandHandler: class CommandHandlers: - def __init__(self, command_handlers: list[CommandHandler] | None = None): + def __init__(self, command_handlers: tuple[CommandHandler] = tuple()): """ Private. The model that unites all CommandHandler of the routers :param command_handlers: list of CommandHandlers for register """ - self.command_handlers: list[CommandHandler] = command_handlers if command_handlers else [] + self.command_handlers: list[CommandHandler] = list(command_handlers) if command_handlers else [] + self.paired_command_handler_trigger: dict[str, CommandHandler] = {x.handled_command.trigger: x for x in command_handlers} def add_handler(self, command_handler: CommandHandler) -> None: """ @@ -41,6 +45,13 @@ class CommandHandlers: :return: None """ self.command_handlers.append(command_handler) + self.paired_command_handler_trigger[command_handler.handled_command.trigger.lower()] = command_handler + for alias in command_handler.handled_command.aliases: + self.paired_command_handler_trigger[alias.lower()] = command_handler + + def get_command_handler_by_trigger(self, trigger: str): + print(self.paired_command_handler_trigger) + return self.paired_command_handler_trigger.get(trigger) def __iter__(self) -> Iterator[CommandHandler]: return iter(self.command_handlers) diff --git a/src/argenta/router/defaults.py b/src/argenta/router/defaults.py deleted file mode 100644 index b0b96aa..0000000 --- a/src/argenta/router/defaults.py +++ /dev/null @@ -1,5 +0,0 @@ -__all__ = ["system_router"] - -from argenta.router import Router - -system_router = Router(title="System points:") diff --git a/src/argenta/router/entity.py b/src/argenta/router/entity.py index b267a4f..12edb33 100644 --- a/src/argenta/router/entity.py +++ b/src/argenta/router/entity.py @@ -11,7 +11,9 @@ from argenta.command.flag.flags import Flags, InputFlags from argenta.response import Response, ResponseStatus from argenta.router.command_handler.entity import CommandHandler, CommandHandlers from argenta.router.exceptions import ( + RepeatedAliasNameException, RepeatedFlagNameException, + RepeatedTriggerNameException, RequiredArgumentNotPassedException, TriggerContainSpacesException, ) @@ -57,13 +59,8 @@ class Router: redefined_command = command self._validate_command(redefined_command) - - if overlapping := (self.aliases | self.triggers) & redefined_command.aliases: - Console().print(f"\n[b red]WARNING:[/b red] Overlapping trigger or alias: [b blue]{overlapping}[/b blue]") + self._update_routing_keys(redefined_command) - self.aliases.update(redefined_command.aliases) - self.triggers.add(redefined_command.trigger) - def decorator(func: HandlerFunc) -> HandlerFunc: _validate_func_args(func) self.command_handlers.add_handler(CommandHandler(func, redefined_command)) @@ -80,10 +77,22 @@ class Router: command_name: str = command.trigger if command_name.find(" ") != -1: raise TriggerContainSpacesException() + + if command_name.lower() in self.triggers: + raise RepeatedTriggerNameException() + + if overlapping := (self.aliases | self.triggers) & set(map(lambda x: x.lower(), command.aliases)): + raise RepeatedAliasNameException(overlapping) + flags: Flags = command.registered_flags flags_name: list[str] = [flag.string_entity.lower() for flag in flags] if len(set(flags_name)) < len(flags_name): raise RepeatedFlagNameException() + + def _update_routing_keys(self, registered_command: Command): + redefined_command_aliases_in_lower = set(map(lambda x: x.lower(), registered_command.aliases)) + self.aliases.update(redefined_command_aliases_in_lower) + self.triggers.add(registered_command.trigger.lower()) def finds_appropriate_handler(self, input_command: InputCommand) -> None: """ @@ -91,15 +100,15 @@ class Router: :param input_command: input command as InputCommand :return: None """ - input_command_name: str = input_command.trigger + input_command_name: str = input_command.trigger.lower() input_command_flags: InputFlags = input_command.input_flags - for command_handler in self.command_handlers: - handle_command = command_handler.handled_command - if input_command_name.lower() == handle_command.trigger.lower(): - self.process_input_command(input_command_flags, command_handler) - if input_command_name.lower() in handle_command.aliases: - self.process_input_command(input_command_flags, command_handler) + command_handler = self.command_handlers.get_command_handler_by_trigger(input_command_name) + + if not command_handler: + raise RuntimeError(f"Handler for '{input_command.trigger}' command not found!") + else: + self.process_input_command(input_command_flags, command_handler) def process_input_command(self, input_command_flags: InputFlags, command_handler: CommandHandler) -> None: """ @@ -147,13 +156,14 @@ def _structuring_input_flags(handled_command: Command, input_flags: InputFlags) undefined_flags = True status = ResponseStatus.from_flags( - has_invalid_value_flags=invalid_value_flags, has_undefined_flags=undefined_flags + has_invalid_value_flags=invalid_value_flags, + has_undefined_flags=undefined_flags ) return Response(status=status, input_flags=input_flags) -def _validate_func_args(func: Callable[..., None]) -> None: +def _validate_func_args(func: HandlerFunc) -> None: """ Private. Validates the arguments of the handler :param func: entity of the handler func diff --git a/src/argenta/router/exceptions.py b/src/argenta/router/exceptions.py index 6754a37..772c02b 100644 --- a/src/argenta/router/exceptions.py +++ b/src/argenta/router/exceptions.py @@ -11,7 +11,31 @@ class RepeatedFlagNameException(Exception): @override def __str__(self) -> str: return "Repeated registered flag names in register command" + +class RepeatedTriggerNameException(Exception): + """ + Private. Raised when a repeated trigger name is registered + """ + + @override + def __str__(self) -> str: + return "Repeated trigger name in registered commands" + + +class RepeatedAliasNameException(Exception): + """ + Private. Raised when a repeated alias name is registered + """ + @override + def __init__(self, repeated_aliases: set[str]) -> None: + self.repeated_aliases = repeated_aliases + super().__init__() + + @override + def __str__(self) -> str: + return f"Repeated aliases names: {self.repeated_aliases}" + class RequiredArgumentNotPassedException(Exception): """ diff --git a/tests/unit_tests/test_app.py b/tests/unit_tests/test_app.py index 2b99db4..e75d89f 100644 --- a/tests/unit_tests/test_app.py +++ b/tests/unit_tests/test_app.py @@ -1,3 +1,4 @@ +from argenta.router.exceptions import RepeatedAliasNameException import pytest from pytest import CaptureFixture @@ -207,24 +208,17 @@ def test_include_routers_registers_multiple_routers() -> None: assert app.registered_routers.registered_routers == [router, router2] -def test_overlapping_aliases_prints_warning(capsys: CaptureFixture[str]) -> None: - app = App(override_system_messages=True) +def test_overlapping_aliases_raises_exception() -> None: router = Router() @router.command(Command('test', aliases={'alias'})) def handler(_res: Response) -> None: pass - @router.command(Command('test2', aliases={'alias'})) - def handler2(_res: Response) -> None: - pass - - app.include_routers(router) - app._pre_cycle_setup() - - captured = capsys.readouterr() - - assert "Overlapping" in captured.out + with pytest.raises(RepeatedAliasNameException): + @router.command(Command('test2', aliases={'alias'})) + def handler2(_res: Response) -> None: + pass # ============================================================================ diff --git a/tests/unit_tests/test_router.py b/tests/unit_tests/test_router.py index 1ec082f..d6c022a 100644 --- a/tests/unit_tests/test_router.py +++ b/tests/unit_tests/test_router.py @@ -12,6 +12,7 @@ from argenta.router import Router from argenta.router.entity import _structuring_input_flags, _validate_func_args # pyright: ignore[reportPrivateUsage] from argenta.router.exceptions import ( RepeatedFlagNameException, + RepeatedTriggerNameException, RequiredArgumentNotPassedException, TriggerContainSpacesException, ) @@ -26,7 +27,20 @@ def test_validate_command_raises_error_for_trigger_with_spaces() -> None: router = Router() with pytest.raises(TriggerContainSpacesException): router._validate_command(Command(trigger='command with spaces')) - + + +def test_validate_command_raises_error_for_same_trigger() -> None: + router = Router() + + @router.command('comm') + def handler(res: Response): + pass + + with pytest.raises(RepeatedTriggerNameException): + @router.command('comm') + def handler2(res: Response): + pass + def test_validate_command_raises_error_for_repeated_flag_names() -> None: router = Router() @@ -192,6 +206,33 @@ def test_finds_appropriate_handler_executes_handler_by_alias(capsys: CaptureFixt output = capsys.readouterr() assert "Hello World!" in output.out + +def test_finds_appropriate_handler_executes_handler_by_alias_with_differrent_register(capsys: CaptureFixture[str]) -> None: + router = Router() + + @router.command(Command('hello', aliases={'hI'})) + def handler(_res: Response) -> None: + print("Hello World!") + + router.finds_appropriate_handler(InputCommand('HI')) + + output = capsys.readouterr() + + assert "Hello World!" in output.out + + +def test_finds_appropriate_handler_executes_handler_by_trigger_with_differrent_register(capsys: CaptureFixture[str]) -> None: + router = Router() + + @router.command(Command('heLLo')) + def handler(_res: Response) -> None: + print("Hello World!") + + router.finds_appropriate_handler(InputCommand('HellO')) + + output = capsys.readouterr() + + assert "Hello World!" in output.out def test_finds_appropriate_handler_executes_handler_with_flags_by_alias(capsys: CaptureFixture[str]) -> None: