72 Commits

Author SHA1 Message Date
kolo 106ca058be new tests 2025-04-27 14:11:01 +03:00
kolo b5ddfb3b35 new tests 2025-04-27 13:28:11 +03:00
kolo 61e4502e41 work, fix etc. 2025-04-26 22:23:35 +03:00
kolo 9b2fc87e33 release v1.0.0a1 2025-04-25 02:29:44 +03:00
kolo 89f09c42f8 pre-release v1.0.0 2025-04-24 21:26:41 +03:00
kolo 5bcae8fe68 Update README.md 2025-04-24 17:40:53 +03:00
kolo ca58008431 fix 2025-04-23 22:07:57 +03:00
kolo 30974f48eb fix 2025-04-23 22:06:10 +03:00
kolo df4ba080b0 new docs 2025-04-23 22:02:02 +03:00
kolo f93930d712 work 2025-04-23 21:45:04 +03:00
kolo 036c17ec9a last steps work on new docs, full complete write docstring for all objects 2025-04-23 20:54:03 +03:00
kolo 7281fdeabf new app preview img in docs 2025-04-23 20:50:31 +03:00
kolo 051ec6df28 steps 2025-04-20 23:58:15 +03:00
kolo 00a1e11fc1 new docs 2025-04-19 12:13:29 +03:00
kolo 584df9ba69 new docs 2025-04-18 12:50:30 +03:00
kolo a649022f1d work on docstrings 2025-04-15 23:49:51 +03:00
kolo 26a9d8a6da work on 2025-04-15 22:26:20 +03:00
kolo 9522b0161a work on 2025-04-15 01:09:03 +03:00
kolo e189f8d9aa big step 2025-04-14 16:38:53 +03:00
kolo 3ef8707cfa big step 2025-04-14 14:54:17 +03:00
kolo a5fdcab862 work on 2025-04-14 01:03:24 +03:00
kolo ba035881ee work on support args 2025-04-13 19:24:03 +03:00
kolo 34ebe55531 first steps 2025-04-13 14:39:53 +03:00
kolo 01c9d2dc6d first step 2025-04-13 14:12:08 +03:00
kolo a6db733204 some fix 2025-04-13 13:13:03 +03:00
kolo 8506e4ffcf some fix 2025-04-12 18:07:10 +03:00
kolo 04d3329572 0.5.0-beta 2025-04-11 13:12:44 +03:00
kolo 5bfdde4bd9 now command aliases auto added to autocomlete history for their use 2025-04-10 19:23:57 +03:00
kolo e2dd7e4aea 0.5.0-alpha: support autocomplete, aliases for command, fix many bugs and other 2025-04-10 13:12:36 +03:00
kolo d1d644d422 final adding 2025-04-10 00:23:03 +03:00
kolo 8b496aa782 final work on autocomplete 2025-04-09 23:32:21 +03:00
kolo 592d128ef6 work on autocomplete 2025-04-09 12:16:30 +03:00
kolo b44ee227fd fix many bugs 2025-04-08 20:40:52 +03:00
kolo 0dce4a0d9e refactor app 2025-04-08 10:33:45 +03:00
kolo ca6634c6f0 some fix 2025-04-08 10:04:57 +03:00
kolo c1805af420 final work on dividing line 2025-04-08 00:28:17 +03:00
kolo 0e308ce77f work on dividing line 2025-04-07 19:44:51 +03:00
kolo ab1d335f8e work 2025-04-06 18:57:30 +03:00
kolo 1a2e9d1487 successful adding framed text 2025-04-05 09:58:02 +03:00
kolo 76bcba9340 some 2025-04-05 00:47:32 +03:00
kolo ae9795bd53 some fix and final refactor in readme 2025-04-04 01:43:59 +03:00
kolo 7540728f1b fix 2025-04-03 16:16:05 +03:00
kolo 54da63dd03 fix 2025-04-03 01:14:33 +03:00
kolo 8e08d0fe09 some fix 2025-04-03 01:09:03 +03:00
kolo 55b88f7c8a modified tests 2025-04-02 22:03:30 +03:00
kolo 1cd616336f refactor default view 2025-04-02 21:51:20 +03:00
kolo 253790fe2e work 2025-04-02 16:34:43 +03:00
kolo 1c6f896b73 work 2025-04-02 16:02:22 +03:00
kolo 8810e12551 linter workflow 2025-04-02 00:36:47 +03:00
kolo 285007a59a linter workflow 2025-04-02 00:34:34 +03:00
kolo 6edd17646a final workflow 2025-04-02 00:11:19 +03:00
kolo 154ee25dde test2 workflow 2025-04-02 00:09:48 +03:00
kolo 30cf3cfd06 test workflow 2025-04-02 00:06:48 +03:00
kolo 0d98d80919 Create tests.yml 2025-04-02 00:03:48 +03:00
kolo 54992e55cb some changes in readme 2025-04-01 23:49:17 +03:00
kolo cc8135b733 more working 2025-03-31 23:53:49 +03:00
kolo 5c6fa5151a new models, a model is passed to the command handler instead of a dictionary, removal of checks for intersection of processed triggers in handlers and much, much more 2025-03-31 19:14:42 +03:00
kolo 2918bc9f81 refactor, new model e.t.c. 2025-03-31 01:12:01 +03:00
kolo 6e2fbc23e9 adding __all__ in __init__ 2025-03-28 13:11:40 +03:00
kolo 1ec8ea53b4 new models, logic refactor, renaminf constants and other 2025-03-28 00:54:12 +03:00
kolo 4256d67789 v0.4.1 2025-03-27 00:17:40 +03:00
kolo 0246ff4b22 adding new method for App and new system tests 2025-03-24 10:44:49 +03:00
kolo 956febc757 v0.4.0 2025-03-15 20:50:44 +03:00
kolo beafdd0afd new system tests 2025-03-15 18:47:26 +03:00
kolo b172e2cdc3 new system tests 2025-03-15 11:42:12 +03:00
kolo baaf0e25f3 some fix 2025-03-15 00:08:39 +03:00
kolo 09c40978a1 fix 2025-03-12 13:01:15 +03:00
kolo 0f4f48555e fix 2025-03-12 00:51:11 +03:00
kolo d30515c1a2 v0.3.9 2025-03-11 20:44:06 +03:00
kolo 5a6fc1d8ca first step 2025-03-08 13:49:29 +03:00
kolo e5d6ead38e v0.3.8 2025-03-06 01:35:28 +03:00
kolo b61c151e1c v0.3.8 2025-03-06 01:35:18 +03:00
68 changed files with 3678 additions and 1364 deletions
+30
View File
@@ -0,0 +1,30 @@
name: ruff
on:
push:
branches: [ "kolo" ]
pull_request:
branches: [ "kolo" ]
permissions:
contents: read
jobs:
build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- uses: actions/setup-python@v3
with:
python-version: "3.13"
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install ruff
- name: Run linter
run: ruff check ./argenta
+31
View File
@@ -0,0 +1,31 @@
name: tests
on:
push:
branches: [ "kolo" ]
pull_request:
branches: [ "kolo" ]
permissions:
contents: read
jobs:
build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- uses: actions/setup-python@v3
with:
python-version: "3.13"
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install poetry
poetry install
- name: Run tests
run: poetry run python -m unittest discover
+4 -1
View File
@@ -1,6 +1,9 @@
.venv .venv
.idea .idea
dist dist
poetry.lock uv.lock
*__pycache__/ *__pycache__/
*.hist*
build
source
+1203 -319
View File
File diff suppressed because it is too large Load Diff
-1
View File
@@ -1 +0,0 @@
from .entity import App
-258
View File
@@ -1,258 +0,0 @@
from typing import Callable
from inspect import getfullargspec
import re
from ..command.entity import Command
from ..router.entity import Router
from ..command.exceptions import (UnprocessedInputFlagException,
RepeatedInputFlagsException,
EmptyInputCommandException)
from .exceptions import (InvalidRouterInstanceException,
InvalidDescriptionMessagePatternException,
NoRegisteredRoutersException,
NoRegisteredHandlersException,
RepeatedCommandInDifferentRoutersException,
IncorrectNumberOfHandlerArgsException)
class App:
def __init__(self,
prompt: str = 'Enter a command',
initial_message: str = '\nHello, I am Argenta\n',
farewell_message: str = '\nGoodBye\n',
invalid_input_flags_message: str = 'Invalid input flags',
exit_command: str = 'Q',
exit_command_description: str = 'Exit command',
system_points_title: str = 'System points:',
ignore_exit_command_register: bool = True,
ignore_command_register: bool = False,
line_separate: str = '',
command_group_description_separate: str = '',
repeat_command_groups: bool = True,
print_func: Callable[[str], None] = print) -> None:
self.prompt = prompt
self.print_func = print_func
self.exit_command = exit_command
self.exit_command_description = exit_command_description
self.system_points_title = system_points_title
self.ignore_exit_command_register = ignore_exit_command_register
self.farewell_message = farewell_message
self.initial_message = initial_message
self.invalid_input_flags_message = invalid_input_flags_message
self.line_separate = line_separate
self.command_group_description_separate = command_group_description_separate
self.ignore_command_register = ignore_command_register
self.repeat_command_groups = repeat_command_groups
self._routers: list[Router] = []
self._description_message_pattern: str = '[{command}] *=*=* {description}'
self._registered_router_entities: list[dict[str, str | list[dict[str, Callable[[], None] | Command]] | Router]] = []
self._invalid_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'Incorrect flag syntax: "{raw_command}"')
self._repeated_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'Repeated input flags: "{raw_command}"')
self._empty_input_command_handler: Callable[[], None] = lambda: print_func(f'Empty input command')
self._unknown_command_handler: Callable[[Command], None] = lambda command: print_func(f"Unknown command: {command.get_string_entity()}")
def start_polling(self) -> None:
self._validate_number_of_routers()
self._validate_included_routers()
self._validate_all_router_commands()
self.print_func(self.initial_message)
if not self.repeat_command_groups:
self._print_command_group_description()
self.print_func(self.prompt)
while True:
if self.repeat_command_groups:
self._print_command_group_description()
self.print_func(self.prompt)
raw_command: str = input()
try:
input_command: Command = Command.parse_input_command(raw_command=raw_command)
except UnprocessedInputFlagException:
self.print_func(self.line_separate)
self._invalid_input_flags_handler(raw_command)
self.print_func(self.line_separate)
if not self.repeat_command_groups:
self.print_func(self.prompt)
continue
except RepeatedInputFlagsException:
self.print_func(self.line_separate)
self._repeated_input_flags_handler(raw_command)
self.print_func(self.line_separate)
if not self.repeat_command_groups:
self.print_func(self.prompt)
continue
except EmptyInputCommandException:
self.print_func(self.line_separate)
self._empty_input_command_handler()
self.print_func(self.line_separate)
if not self.repeat_command_groups:
self.print_func(self.prompt)
continue
self._check_command_for_exit_command(input_command.get_trigger())
self.print_func(self.line_separate)
is_unknown_command: bool = self._check_is_command_unknown(input_command)
if is_unknown_command:
self.print_func(self.line_separate)
self.print_func(self.command_group_description_separate)
if not self.repeat_command_groups:
self.print_func(self.prompt)
continue
for router in self._routers:
router.input_command_handler(input_command)
self.print_func(self.line_separate)
self.print_func(self.command_group_description_separate)
if not self.repeat_command_groups:
self.print_func(self.prompt)
def set_initial_message(self, message: str) -> None:
self.initial_message: str = message
def set_farewell_message(self, message: str) -> None:
self.farewell_message: str = message
def set_description_message_pattern(self, pattern: str) -> None:
first_check = re.match(r'.*{command}.*', pattern)
second_check = re.match(r'.*{description}.*', pattern)
if bool(first_check) and bool(second_check):
self._description_message_pattern: str = pattern
else:
raise InvalidDescriptionMessagePatternException(pattern)
def set_invalid_input_flags_handler(self, handler: Callable[[str], None]) -> None:
args = getfullargspec(handler).args
if len(args) != 1:
raise IncorrectNumberOfHandlerArgsException()
else:
self._invalid_input_flags_handler = handler
def set_repeated_input_flags_handler(self, handler: Callable[[str], None]) -> None:
args = getfullargspec(handler).args
if len(args) != 1:
raise IncorrectNumberOfHandlerArgsException()
else:
self._repeated_input_flags_handler = handler
def set_unknown_command_handler(self, handler: Callable[[str], None]) -> None:
args = getfullargspec(handler).args
if len(args) != 1:
raise IncorrectNumberOfHandlerArgsException()
else:
self._unknown_command_handler = handler
def set_empty_command_handler(self, handler: Callable[[str], None]) -> None:
args = getfullargspec(handler).args
if len(args) != 1:
raise IncorrectNumberOfHandlerArgsException()
else:
self._empty_input_command_handler = handler
def include_router(self, router: Router) -> None:
if not isinstance(router, Router):
raise InvalidRouterInstanceException()
router.set_ignore_command_register(self.ignore_command_register)
self._routers.append(router)
command_entities: list[dict[str, Callable[[], None] | Command]] = router.get_command_entities()
self._registered_router_entities.append({'name': router.get_name(),
'title': router.get_title(),
'entity': router,
'commands': command_entities})
def _validate_number_of_routers(self) -> None:
if not self._routers:
raise NoRegisteredRoutersException()
def _validate_included_routers(self) -> None:
for router in self._routers:
if not router.get_command_entities():
raise NoRegisteredHandlersException(router.get_name())
def _validate_all_router_commands(self) -> None:
for idx in range(len(self._registered_router_entities)):
current_router: Router = self._registered_router_entities[idx]['entity']
routers_without_current_router = self._registered_router_entities.copy()
routers_without_current_router.pop(idx)
current_router_all_commands: list[str] = current_router.get_all_commands()
for router_entity in routers_without_current_router:
if len(set(current_router_all_commands).intersection(set(router_entity['entity'].get_all_commands()))) > 0:
raise RepeatedCommandInDifferentRoutersException()
if self.ignore_command_register:
if len(set([x.lower() for x in current_router_all_commands]).intersection(set([x.lower() for x in router_entity['entity'].get_all_commands()]))) > 0:
raise RepeatedCommandInDifferentRoutersException()
def _check_command_for_exit_command(self, command: str):
if command.lower() == self.exit_command.lower():
if self.ignore_exit_command_register:
self.print_func(self.farewell_message)
exit(0)
else:
if command == self.exit_command:
self.print_func(self.farewell_message)
exit(0)
def _check_is_command_unknown(self, command: Command):
registered_router_entities: list[dict[str, str | list[dict[str, Callable[[], None] | Command]] | Router]] = self._registered_router_entities
for router_entity in registered_router_entities:
for command_entity in router_entity['commands']:
if command_entity['command'].get_trigger().lower() == command.get_trigger().lower():
if self.ignore_command_register:
return False
else:
if command_entity['command'].get_trigger() == command.get_trigger():
return False
self._unknown_command_handler(command)
return True
def _print_command_group_description(self):
for router_entity in self._registered_router_entities:
self.print_func(router_entity['title'])
for command_entity in router_entity['commands']:
self.print_func(self._description_message_pattern.format(
command=command_entity['command'].get_trigger(),
description=command_entity['command'].get_description()
)
)
self.print_func(self.command_group_description_separate)
self.print_func(self.system_points_title)
self.print_func(self._description_message_pattern.format(
command=self.exit_command,
description=self.exit_command_description
)
)
self.print_func(self.command_group_description_separate)
-35
View File
@@ -1,35 +0,0 @@
class InvalidRouterInstanceException(Exception):
def __str__(self):
return "Invalid Router Instance"
class InvalidDescriptionMessagePatternException(Exception):
def __init__(self, pattern: str):
self.pattern = pattern
def __str__(self):
return ("Invalid Description Message Pattern\n"
"Correct pattern example: [{command}] *=*=* {description}\n"
"The pattern must contain two variables: `command` and `description` - description of the command\n"
f"Your pattern: {self.pattern}")
class NoRegisteredRoutersException(Exception):
def __str__(self):
return "No Registered Router Found"
class NoRegisteredHandlersException(Exception):
def __init__(self, router_name):
self.router_name = router_name
def __str__(self):
return f"No Registered Handlers Found For '{self.router_name}'"
class RepeatedCommandInDifferentRoutersException(Exception):
def __str__(self):
return "Commands in different handlers cannot be repeated"
class IncorrectNumberOfHandlerArgsException(Exception):
def __str__(self):
return "Incorrect Input Flags Handler has incorrect number of arguments"
-1
View File
@@ -1 +0,0 @@
from .entity import Command
-108
View File
@@ -1,108 +0,0 @@
from .params.flag.entity import Flag
from .params.flag.flags_group.entity import FlagsGroup
from .exceptions import (UnprocessedInputFlagException,
RepeatedInputFlagsException,
EmptyInputCommandException)
from typing import Generic, TypeVar
T = TypeVar('T')
class Command(Generic[T]):
def __init__(self, trigger: str,
description: str = None,
flags: Flag | FlagsGroup = None):
self._trigger = trigger
self._description = f'description for "{self._trigger}" command' if not description else description
self._registered_flags: FlagsGroup | None = flags if isinstance(flags, FlagsGroup) else FlagsGroup([flags]) if isinstance(flags, Flag) else flags
self._input_flags: FlagsGroup | None = None
def get_trigger(self) -> str:
return self._trigger
def get_description(self) -> str:
return self._description
def get_registered_flags(self) -> FlagsGroup | None:
return self._registered_flags
def validate_input_flag(self, flag: Flag):
registered_flags: FlagsGroup | None = self.get_registered_flags()
if registered_flags:
if isinstance(registered_flags, Flag):
if registered_flags.get_string_entity() == flag.get_string_entity():
is_valid = registered_flags.validate_input_flag_value(flag.get_value())
if is_valid:
return True
else:
for registered_flag in registered_flags:
if registered_flag.get_string_entity() == flag.get_string_entity():
is_valid = registered_flag.validate_input_flag_value(flag.get_value())
if is_valid:
return True
return False
def _set_input_flags(self, input_flags: FlagsGroup):
self._input_flags = input_flags
def get_input_flags(self) -> FlagsGroup:
return self._input_flags
@staticmethod
def parse_input_command(raw_command: str) -> 'Command[T]':
if not raw_command:
raise EmptyInputCommandException()
list_of_tokens = raw_command.split()
command = list_of_tokens[0]
list_of_tokens.pop(0)
flags: FlagsGroup = FlagsGroup()
current_flag_name = None
current_flag_value = None
for _ in list_of_tokens:
if _.startswith('-'):
flag_prefix_last_symbol_index = _.rfind('-')
if current_flag_name or len(_) < 2 or len(_[:flag_prefix_last_symbol_index]) > 3:
raise UnprocessedInputFlagException()
else:
current_flag_name = _
else:
if not current_flag_name:
raise UnprocessedInputFlagException()
else:
current_flag_value = _
if current_flag_name and current_flag_value:
flag_prefix_last_symbol_index = current_flag_name.rfind('-')
flag_prefix = current_flag_name[:flag_prefix_last_symbol_index+1]
flag_name = current_flag_name[flag_prefix_last_symbol_index+1:]
input_flag = Flag(flag_name=flag_name,
flag_prefix=flag_prefix)
input_flag.set_value(current_flag_value)
all_flags = [x.get_string_entity() for x in flags.get_flags()]
if input_flag.get_string_entity() not in all_flags:
flags.add_flag(input_flag)
else:
raise RepeatedInputFlagsException(input_flag)
current_flag_name = None
current_flag_value = None
if any([current_flag_name, current_flag_value]):
raise UnprocessedInputFlagException()
if len(flags.get_flags()) == 0:
return Command(trigger=command)
else:
input_command = Command(trigger=command)
input_command._set_input_flags(flags)
return input_command
-19
View File
@@ -1,19 +0,0 @@
from .params.flag.entity import Flag
class UnprocessedInputFlagException(Exception):
def __str__(self):
return "Unprocessed Input Flags"
class RepeatedInputFlagsException(Exception):
def __init__(self, flag: Flag):
self.flag = flag
def __str__(self):
return ("Repeated Input Flags\n"
f"Duplicate flag was detected in the input: '{self.flag.get_string_entity()}'")
class EmptyInputCommandException(Exception):
def __str__(self):
return "Input Command is empty"
-2
View File
@@ -1,2 +0,0 @@
from .entity import Flag
from .flags_group.entity import FlagsGroup
-51
View File
@@ -1,51 +0,0 @@
from typing import Literal, Pattern
class Flag:
def __init__(self, flag_name: str,
flag_prefix: Literal['-', '--', '---'] = '--',
ignore_flag_value_register: bool = False,
possible_flag_values: list[str] | Pattern[str] = False):
self._flag_name = flag_name
self._flag_prefix = flag_prefix
self.possible_flag_values = possible_flag_values
self.ignore_flag_value_register = ignore_flag_value_register
self._flag_value = None
def get_string_entity(self):
string_entity: str = self._flag_prefix + self._flag_name
return string_entity
def get_flag_name(self):
return self._flag_name
def get_flag_prefix(self):
return self._flag_prefix
def get_value(self):
return self._flag_value
def set_value(self, value):
self._flag_value = value
def validate_input_flag_value(self, input_flag_value: str):
if isinstance(self.possible_flag_values, Pattern):
is_valid = bool(self.possible_flag_values.match(input_flag_value))
if bool(is_valid):
return True
else:
return False
if isinstance(self.possible_flag_values, list):
if self.ignore_flag_value_register:
if input_flag_value.lower() in [x.lower() for x in self.possible_flag_values]:
return True
else:
return False
else:
if input_flag_value in self.possible_flag_values:
return True
else:
return False
return True
@@ -1,35 +0,0 @@
from argenta.command.params.flag.entity import Flag
class FlagsGroup:
def __init__(self, flags: list[Flag] = None):
self._flags: list[Flag] = [] if not flags else flags
def get_flags(self) -> list[Flag]:
return self._flags
def add_flag(self, flag: Flag):
self._flags.append(flag)
def add_flags(self, flags: list[Flag]):
self._flags.extend(flags)
def unparse_to_dict(self):
result_dict: dict[str, dict] = {}
for flag in self._flags:
result_dict[flag.get_flag_name()] = {
'name': flag.get_flag_name(),
'string_entity': flag.get_string_entity(),
'prefix': flag.get_flag_prefix(),
'value': flag.get_value()
}
return result_dict
def __iter__(self):
return iter(self._flags)
def __next__(self):
return next(iter(self))
def __getitem__(self, item):
return self._flags[item]
-1
View File
@@ -1 +0,0 @@
from .entity import Router
-121
View File
@@ -1,121 +0,0 @@
from typing import Callable, Any
from inspect import getfullargspec
from ..command.entity import Command
from ..command.params.flag.entity import Flag
from ..command.params.flag.flags_group.entity import FlagsGroup
from ..router.exceptions import (RepeatedCommandException,
RepeatedFlagNameException,
TooManyTransferredArgsException,
RequiredArgumentNotPassedException,
IncorrectNumberOfHandlerArgsException)
class Router:
def __init__(self,
title: str = 'Commands group title:',
name: str = 'Default'):
self._title = title
self._name = name
self._command_entities: list[dict[str, Callable[[], None] | Command]] = []
self._ignore_command_register: bool = False
self._not_valid_flag_handler: Callable[[Flag], None] = lambda flag: print(f"Undefined or incorrect input flag: '{flag.get_string_entity()} {flag.get_value()}'")
def command(self, command: Command) -> Callable[[Any], Any]:
self._validate_command(command)
def command_decorator(func):
Router._validate_func_args(command, func)
self._command_entities.append({'handler_func': func,
'command': command})
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
return command_decorator
def set_invalid_input_flag_handler(self, func):
processed_args = getfullargspec(func).args
if len(processed_args) != 1:
raise IncorrectNumberOfHandlerArgsException()
else:
self._not_valid_flag_handler = func
def input_command_handler(self, input_command: Command):
input_command_name: str = input_command.get_trigger()
input_command_flags: FlagsGroup = input_command.get_input_flags()
for command_entity in self._command_entities:
if input_command_name.lower() == command_entity['command'].get_trigger().lower():
if command_entity['command'].get_registered_flags():
if input_command_flags:
for flag in input_command_flags:
is_valid = command_entity['command'].validate_input_flag(flag)
if not is_valid:
self._not_valid_flag_handler(flag)
return
return command_entity['handler_func'](input_command_flags.unparse_to_dict())
else:
return command_entity['handler_func']({})
else:
if input_command_flags:
self._not_valid_flag_handler(input_command_flags[0])
return
else:
return command_entity['handler_func']()
def _validate_command(self, command: Command):
command_name: str = command.get_trigger()
if command_name in self.get_all_commands():
raise RepeatedCommandException()
if self._ignore_command_register:
if command_name.lower() in [x.lower() for x in self.get_all_commands()]:
raise RepeatedCommandException()
flags: FlagsGroup = command.get_registered_flags()
if flags:
flags_name: list = [x.get_string_entity().lower() for x in flags]
if len(set(flags_name)) < len(flags_name):
raise RepeatedFlagNameException()
@staticmethod
def _validate_func_args(command: Command, func: Callable):
registered_args = command.get_registered_flags()
transferred_args = getfullargspec(func).args
if registered_args and transferred_args:
if len(transferred_args) != 1:
raise TooManyTransferredArgsException()
elif registered_args and not transferred_args:
raise RequiredArgumentNotPassedException()
elif not registered_args and transferred_args:
raise TooManyTransferredArgsException()
def set_ignore_command_register(self, ignore_command_register: bool):
self._ignore_command_register = ignore_command_register
def get_command_entities(self) -> list[dict[str, Callable[[], None] | Command]]:
return self._command_entities
def get_name(self) -> str:
return self._name
def get_title(self) -> str:
return self._title
def get_all_commands(self) -> list[str]:
all_commands: list[str] = []
for command_entity in self._command_entities:
all_commands.append(command_entity['command'].get_trigger())
return all_commands
-23
View File
@@ -1,23 +0,0 @@
class RepeatedCommandException(Exception):
def __str__(self):
return "Commands in handler cannot be repeated"
class RepeatedFlagNameException(Exception):
def __str__(self):
return "Repeated flag name in register command"
class TooManyTransferredArgsException(Exception):
def __str__(self):
return "Too many transferred arguments"
class RequiredArgumentNotPassedException(Exception):
def __str__(self):
return "Required argument not passed"
class IncorrectNumberOfHandlerArgsException(Exception):
def __str__(self):
return "Handler has incorrect number of arguments"
Binary file not shown.

After

Width:  |  Height:  |  Size: 60 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 58 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 54 KiB

+7
View File
@@ -0,0 +1,7 @@
from argenta.app import App
from argenta.orchestrator import Orchestrator
app = App(repeat_command_groups=True)
orchestrator = Orchestrator()
orchestrator.start_polling(app)
+5 -10
View File
@@ -1,11 +1,6 @@
import re from argenta.app import App
from argenta.command.models import InputCommand
app = App()
def set_description_message_pattern(pattern: str) -> None: app._all_registered_triggers_in_lower = ['fr', 'Tre', 'Pre']
first_check = re.match(r'.*command.*', pattern) print(app._is_unknown_command(InputCommand('fr')))
second_check = re.match(r'.*{description}.*', pattern)
if bool(first_check) and bool(second_check):
print('Success')
set_description_message_pattern('Invalid des{ommand}cription pattern')
+7 -32
View File
@@ -1,47 +1,22 @@
import re
from pprint import pprint
from rich.console import Console from rich.console import Console
from argenta.command.entity import Command from argenta.command import Command
from argenta.command.params.flag.entity import Flag
from argenta.command.params.flag.flags_group.entity import FlagsGroup
from argenta.router import Router from argenta.router import Router
from .handlers_implementation.help_command import help_command
work_router: Router = Router(title='Work nts:') work_router: Router = Router(title='Work points:')
work_router.set_invalid_input_flag_handler(lambda flag: print(f'Invalid input flag: "{flag.get_string_entity()} {flag.get_value()}"'))
settings_router: Router = Router(title='Settings points:')
console = Console() console = Console()
flags = FlagsGroup(flags=[ '''@work_router.command(Command('get', 'Get Help', aliases=['help', 'Get_help']))
Flag(flag_name='host',
flag_prefix='--',
possible_flag_values=re.compile(r'^192.168.\d{1,3}.\d{1,3}$')),
Flag(flag_name='port',
flag_prefix='--', )
])
@work_router.command(Command(trigger='0', description='Get Help'))
def command_help(): def command_help():
help_command() pass
@work_router.command(Command(trigger='P', description='Start Solving', flags=flags)) @work_router.command(Command('run', 'Run All'))
def command_start_solving(args: dict): def command_start_solving():
print('Solving...') pass'''
pprint(args)
#start_solving_command()
@settings_router.command(Command(trigger='G', description='Update WordMath'))
def command_update():
print('Command update')
+15 -27
View File
@@ -1,40 +1,28 @@
from mock.mock_app.handlers.routers import work_router, settings_router from mock.mock_app.handlers.routers import work_router
from art import text2art
from rich.console import Console
from argenta.app import App from argenta.app import App
from argenta.router import Router from argenta.app.defaults import PredefinedMessages
from argenta.command import Command from argenta.app.dividing_line import DynamicDividingLine
from argenta.command.params.flag import Flag, FlagsGroup from argenta.app.autocompleter import AutoCompleter
from argenta.orchestrator import Orchestrator
from argenta.orchestrator.argparser import ArgParser
from argenta.orchestrator.argparser.arguments import BooleanArgument
app: App = App(prompt='[italic white bold]What do you want to do(enter number of action)?', arg_parser = ArgParser(processed_args=[BooleanArgument('repeat')])
line_separate=f'\n{"[bold green]-[/bold green][bold red]-[/bold red]"*25}\n', app: App = App(dividing_line=DynamicDividingLine(),
print_func=Console().print, autocompleter=AutoCompleter('./mock/.hist'))
command_group_description_separate='', orchestrator: Orchestrator = Orchestrator(arg_parser)
repeat_command_groups=True)
def main(): def main():
ascii_name: str = text2art('WordMath', font='nancyj')
initial_greeting: str = f'[bold red]\n\n{ascii_name}'
ascii_goodbye_message: str = text2art('GoodBye', font='small')
goodbye_message: str = f'[bold red]\n{ascii_goodbye_message}{' '*12}made by kolo\n'
app.include_router(work_router) app.include_router(work_router)
app.include_router(settings_router)
app.set_initial_message(initial_greeting) app.add_message_on_startup(PredefinedMessages.USAGE)
app.set_farewell_message(goodbye_message) app.add_message_on_startup(PredefinedMessages.AUTOCOMPLETE)
app.add_message_on_startup(PredefinedMessages.HELP)
app.set_invalid_input_flags_handler(lambda raw_command: print(f"Invalid input flags: {raw_command}")) orchestrator.start_polling(app)
app.set_unknown_command_handler(lambda command: print(f"Unknown command: {command.get_string_entity()}"))
app.set_repeated_input_flags_handler(lambda raw_command: print(f"Repeated input flags: {raw_command}"))
app.set_description_message_pattern('[bold red][{command}][/bold red] [blue]*=*=*[/blue] [bold yellow italic]{description}')
app.start_polling()
if __name__ == "__main__": if __name__ == "__main__":
main() main()
+16 -19
View File
@@ -1,15 +1,21 @@
[project] [project]
name = "argenta" name = "argenta"
version = "0.3.7" version = "1.0.0-alpha2"
description = "python library for creating custom shells" description = "Python library for creating TUI"
authors = [ authors = [{ name = "kolo", email = "kolo.is.main@gmail.com" }]
{name = "kolo", email = "kolo.is.main@gmail.com"} requires-python = ">=3.11, <4.0"
]
license = {text = "MIT"}
readme = "README.md" readme = "README.md"
requires-python = ">=3.11" license = { text = "MIT" }
dependencies = [] # no dependencies dependencies = [
"rich (>=14.0.0,<15.0.0)",
"art (>=6.4,<7.0)",
"pyreadline3 (>=3.5.4,<4.0.0)",
]
[dependency-groups]
dev = [
"pydoc-markdown>=4.8.2,<5",
]
[tool.ruff] [tool.ruff]
exclude = [ exclude = [
@@ -21,16 +27,7 @@ exclude = [
"tests" "tests"
] ]
[build-system] [build-system]
requires = ["poetry-core>=2.0.0,<3.0.0"] requires = ["hatchling"]
build-backend = "poetry.core.masonry.api" build-backend = "hatchling.build"
[tool.poetry.group.dev.dependencies]
art = "^6.4"
rich = "^13.9.4"
numpy = "^2.2.2"
word2number = "^1.1"
numexpr = "^2.10.2"
requests = "^2.32.3"
+3
View File
@@ -0,0 +1,3 @@
__all__ = ["App"]
from argenta.app.models import App
@@ -0,0 +1,4 @@
__all__ = ["AutoCompleter"]
from argenta.app.autocompleter.entity import AutoCompleter
+72
View File
@@ -0,0 +1,72 @@
import os
import readline
class AutoCompleter:
def __init__(self, history_filename: str = False, autocomplete_button: str = 'tab') -> None:
"""
Public. Configures and implements auto-completion of input command
:param history_filename: the name of the file for saving the history of the autocompleter
:param autocomplete_button: the button for auto-completion
:return: None
"""
self.history_filename = history_filename
self.autocomplete_button = autocomplete_button
self.matches: list[str] = []
def complete(self, text, state) -> str | None:
"""
Private. Auto-completion function
:param text: part of the command being entered
:param state: the current cursor position is relative to the beginning of the line
:return: the desired candidate as str or None
"""
matches: list[str] = sorted(cmd for cmd in self.get_history_items() if cmd.startswith(text))
if len(matches) > 1:
common_prefix = matches[0]
for match in matches[1:]:
i = 0
while i < len(common_prefix) and i < len(match) and common_prefix[i] == match[i]:
i += 1
common_prefix = common_prefix[:i]
if state == 0:
readline.insert_text(common_prefix[len(text):])
readline.redisplay()
return None
elif len(matches) == 1:
return matches[0] if state == 0 else None
else:
return None
def initial_setup(self, all_commands: list[str]) -> None:
"""
Private. Initial setup function
:param all_commands: Registered commands for adding them to the autocomplete history
:return: None
"""
if self.history_filename:
if os.path.exists(self.history_filename):
readline.read_history_file(self.history_filename)
else:
for line in all_commands:
readline.add_history(line)
readline.set_completer(self.complete)
readline.set_completer_delims(readline.get_completer_delims().replace(' ', ''))
readline.parse_and_bind(f'{self.autocomplete_button}: complete')
def exit_setup(self) -> None:
"""
Private. Exit setup function
:return: None
"""
if self.history_filename:
readline.write_history_file(self.history_filename)
@staticmethod
def get_history_items() -> list[str] | list:
"""
Private. Returns a list of all commands entered by the user
:return: all commands entered by the user as list[str]
"""
return [readline.get_history_item(i) for i in range(1, readline.get_current_history_length() + 1)]
+12
View File
@@ -0,0 +1,12 @@
from dataclasses import dataclass
@dataclass
class PredefinedMessages:
"""
Public. A dataclass with predetermined messages for quick use
"""
USAGE = '[b dim]Usage[/b dim]: [i]<command> <[green]flags[/green]>[/i]'
HELP = '[b dim]Help[/b dim]: [i]<command>[/i] [b red]--help[/b red]'
AUTOCOMPLETE = '[b dim]Autocomplete[/b dim]: [i]<part>[/i] [bold]<tab>'
@@ -0,0 +1,4 @@
__all__ = ["StaticDividingLine", "DynamicDividingLine"]
from argenta.app.dividing_line.models import StaticDividingLine, DynamicDividingLine
+68
View File
@@ -0,0 +1,68 @@
from abc import ABC
class BaseDividingLine(ABC):
def __init__(self, unit_part: str = '-') -> None:
"""
Private. The basic dividing line
:param unit_part: the single part of the dividing line
:return: None
"""
self._unit_part = unit_part
def get_unit_part(self) -> str:
"""
Private. Returns the unit part of the dividing line
:return: unit_part of dividing line as str
"""
if len(self._unit_part) == 0:
return ' '
else:
return self._unit_part[0]
class StaticDividingLine(BaseDividingLine):
def __init__(self, unit_part: str = '-', length: int = 25) -> None:
"""
Public. The static dividing line
:param unit_part: the single part of the dividing line
:param length: the length of the dividing line
:return: None
"""
super().__init__(unit_part)
self.length = length
def get_full_static_line(self, is_override: bool) -> str:
"""
Private. Returns the full line of the dividing line
:param is_override: has the default text layout been redefined
:return: full line of dividing line as str
"""
if is_override:
return f'\n{self.length * self.get_unit_part()}\n'
else:
return f'\n[dim]{self.length * self.get_unit_part()}[/dim]\n'
class DynamicDividingLine(BaseDividingLine):
def __init__(self, unit_part: str = '-') -> None:
"""
Public. The dynamic dividing line
:param unit_part: the single part of the dividing line
:return: None
"""
super().__init__(unit_part)
def get_full_dynamic_line(self, length: int, is_override: bool) -> str:
"""
Private. Returns the full line of the dividing line
:param length: the length of the dividing line
:param is_override: has the default text layout been redefined
:return: full line of dividing line as str
"""
if is_override:
return f'\n{length * self.get_unit_part()}\n'
else:
return f'\n[dim]{self.get_unit_part() * length}[/dim]\n'
+380
View File
@@ -0,0 +1,380 @@
from typing import Callable
from rich.console import Console
from rich.markup import escape
from art import text2art
from contextlib import redirect_stdout
import io
import re
from argenta.command.models import Command, InputCommand
from argenta.router import Router
from argenta.router.defaults import system_router
from argenta.app.autocompleter import AutoCompleter
from argenta.app.dividing_line.models import StaticDividingLine, DynamicDividingLine
from argenta.command.exceptions import (UnprocessedInputFlagException,
RepeatedInputFlagsException,
EmptyInputCommandException,
BaseInputCommandException)
from argenta.app.registered_routers.entity import RegisteredRouters
class BaseApp:
def __init__(self,
prompt: str,
initial_message: str,
farewell_message: str,
exit_command: Command,
system_router_title: str | None,
ignore_command_register: bool,
dividing_line: StaticDividingLine | DynamicDividingLine,
repeat_command_groups: bool,
override_system_messages: bool,
autocompleter: AutoCompleter,
print_func: Callable[[str], None]) -> None:
self._prompt = prompt
self._print_func = print_func
self._exit_command = exit_command
self._system_router_title = system_router_title
self._dividing_line = dividing_line
self._ignore_command_register = ignore_command_register
self._repeat_command_groups_description = repeat_command_groups
self._override_system_messages = override_system_messages
self._autocompleter = autocompleter
self._farewell_message = farewell_message
self._initial_message = initial_message
self._description_message_gen: Callable[[str, str], str] = lambda command, description: f'[{command}] *=*=* {description}'
self._registered_routers: RegisteredRouters = RegisteredRouters()
self._messages_on_startup: list[str] = []
self._all_registered_triggers_in_lower: list[str] = []
self._all_registered_triggers_in_default_case: list[str] = []
self._invalid_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'Incorrect flag syntax: {raw_command}')
self._repeated_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'Repeated input flags: {raw_command}')
self._empty_input_command_handler: Callable[[], None] = lambda: print_func('Empty input command')
self._unknown_command_handler: Callable[[InputCommand], None] = lambda command: print_func(f"Unknown command: {command.get_trigger()}")
self._exit_command_handler: Callable[[], None] = lambda: print_func(self._farewell_message)
def set_description_message_pattern(self, pattern: Callable[[str, str], str]) -> None:
"""
Public. Sets the output pattern of the available commands
:param pattern: output pattern of the available commands
:return: None
"""
self._description_message_gen: Callable[[str, str], str] = pattern
def set_invalid_input_flags_handler(self, handler: Callable[[str], None]) -> None:
"""
Public. Sets the handler for incorrect flags when entering a command
:param handler: handler for incorrect flags when entering a command
:return: None
"""
self._invalid_input_flags_handler = handler
def set_repeated_input_flags_handler(self, handler: Callable[[str], None]) -> None:
"""
Public. Sets the handler for repeated flags when entering a command
:param handler: handler for repeated flags when entering a command
:return: None
"""
self._repeated_input_flags_handler = handler
def set_unknown_command_handler(self, handler: Callable[[str], None]) -> None:
"""
Public. Sets the handler for unknown commands when entering a command
:param handler: handler for unknown commands when entering a command
:return: None
"""
self._unknown_command_handler = handler
def set_empty_command_handler(self, handler: Callable[[], None]) -> None:
"""
Public. Sets the handler for empty commands when entering a command
:param handler: handler for empty commands when entering a command
:return: None
"""
self._empty_input_command_handler = handler
def set_exit_command_handler(self, handler: Callable[[], None]) -> None:
"""
Public. Sets the handler for exit command when entering a command
:param handler: handler for exit command when entering a command
:return: None
"""
self._exit_command_handler = handler
def _print_command_group_description(self) -> None:
"""
Private. Prints the description of the available commands
:return: None
"""
for registered_router in self._registered_routers:
if registered_router.get_title():
self._print_func(registered_router.get_title())
for command_handler in registered_router.get_command_handlers():
self._print_func(self._description_message_gen(
command_handler.get_handled_command().get_trigger(),
command_handler.get_handled_command().get_description()))
self._print_func('')
def _print_framed_text(self, text: str) -> None:
"""
Private. Outputs text by framing it in a static or dynamic split strip
:param text: framed text
:return: None
"""
if isinstance(self._dividing_line, StaticDividingLine):
self._print_func(self._dividing_line.get_full_static_line(self._override_system_messages))
self._print_func(text)
self._print_func(self._dividing_line.get_full_static_line(self._override_system_messages))
elif isinstance(self._dividing_line, DynamicDividingLine):
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
self._print_func(self._dividing_line.get_full_dynamic_line(max_length_line, self._override_system_messages))
print(text.strip('\n'))
self._print_func(self._dividing_line.get_full_dynamic_line(max_length_line, self._override_system_messages))
def _is_exit_command(self, command: InputCommand) -> bool:
"""
Private. Checks if the given command is an exit command
:param command: command to check
:return: is it an exit command or not as bool
"""
if self._ignore_command_register:
if command.get_trigger().lower() == self._exit_command.get_trigger().lower():
return True
elif command.get_trigger().lower() in [x.lower() for x in self._exit_command.get_aliases()]:
return True
else:
if command.get_trigger() == self._exit_command.get_trigger():
return True
elif command.get_trigger() in self._exit_command.get_aliases():
return True
return False
def _is_unknown_command(self, command: InputCommand) -> bool:
"""
Private. Checks if the given command is an unknown command
:param command: command to check
:return: is it an unknown command or not as bool
"""
input_command_trigger = command.get_trigger()
if self._ignore_command_register:
if input_command_trigger.lower() in self._all_registered_triggers_in_lower:
return False
else:
if input_command_trigger in self._all_registered_triggers_in_default_case:
return False
return True
def _error_handler(self, error: BaseInputCommandException, raw_command: str) -> None:
"""
Private. Handles parsing errors of the entered command
:param error: error being handled
:param raw_command: the raw input command
:return: None
"""
match error:
case UnprocessedInputFlagException():
self._invalid_input_flags_handler(raw_command)
case RepeatedInputFlagsException():
self._repeated_input_flags_handler(raw_command)
case EmptyInputCommandException():
self._empty_input_command_handler()
def _setup_system_router(self) -> None:
"""
Private. Sets up system router
:return: None
"""
system_router.set_title(self._system_router_title)
@system_router.command(self._exit_command)
def exit_command():
self._exit_command_handler()
if system_router not in self._registered_routers.get_registered_routers():
system_router.set_command_register_ignore(self._ignore_command_register)
self._registered_routers.add_registered_router(system_router)
def _setup_default_view(self) -> None:
"""
Private. Sets up default app view
:return: None
"""
if not self._override_system_messages:
self._initial_message = f'\n[bold red]{text2art(self._initial_message, font="tarty1")}\n\n'
self._farewell_message = (f'[bold red]\n{text2art(f"\n{self._farewell_message}\n", font="chanky")}[/bold red]\n'
f'[red i]github.com/koloideal/Argenta[/red i] | [red bold i]made by kolo[/red bold i]\n')
self._description_message_gen = lambda command, description: (f'[bold red]{escape("[" + command + "]")}[/bold red] '
f'[blue dim]*=*=*[/blue dim] '
f'[bold yellow italic]{escape(description)}')
self._invalid_input_flags_handler = lambda raw_command: self._print_func(f'[red bold]Incorrect flag syntax: {escape(raw_command)}')
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._unknown_command_handler = lambda command: self._print_func(f"[red bold]Unknown command: {escape(command.get_trigger())}")
def _pre_cycle_setup(self) -> None:
"""
Private. Configures various aspects of the application before the start of the cycle
:return: None
"""
self._setup_default_view()
self._setup_system_router()
for router_entity in self._registered_routers:
self._all_registered_triggers_in_default_case.extend(router_entity.get_triggers())
self._all_registered_triggers_in_default_case.extend(router_entity.get_aliases())
self._all_registered_triggers_in_lower.extend([x.lower() for x in router_entity.get_triggers()])
self._all_registered_triggers_in_lower.extend([x.lower() for x in router_entity.get_aliases()])
self._autocompleter.initial_setup(self._all_registered_triggers_in_lower)
self._print_func(self._initial_message)
for message in self._messages_on_startup:
self._print_func(message)
if self._messages_on_startup:
print('\n\n')
if not self._repeat_command_groups_description:
self._print_command_group_description()
class App(BaseApp):
def __init__(self,
prompt: str = '[italic dim bold]What do you want to do?\n',
initial_message: str = '\nArgenta\n',
farewell_message: str = '\nSee you\n',
exit_command: Command = Command('Q', 'Exit command'),
system_router_title: str | None = 'System points:',
ignore_command_register: bool = True,
dividing_line: StaticDividingLine | DynamicDividingLine = StaticDividingLine(),
repeat_command_groups: bool = True,
override_system_messages: bool = False,
autocompleter: AutoCompleter = AutoCompleter(),
print_func: Callable[[str], None] = Console().print) -> None:
"""
Public. The essence of the application itself.
Configures and manages all aspects of the behavior and presentation of the user interacting with the user
:param prompt: displayed before entering the command
:param initial_message: displayed at the start of the app
:param farewell_message: displayed at the end of the app
:param exit_command: the entity of the command that will be terminated when entered
:param system_router_title: system router title
:param ignore_command_register: whether to ignore the case of the entered commands
:param dividing_line: the entity of the dividing line
:param repeat_command_groups: whether to repeat the available commands and their description
:param override_system_messages: whether to redefine the default formatting of system messages
:param autocompleter: the entity of the autocompleter
:param print_func: system messages text output function
:return: None
"""
super().__init__(prompt=prompt,
initial_message=initial_message,
farewell_message=farewell_message,
exit_command=exit_command,
system_router_title=system_router_title,
ignore_command_register=ignore_command_register,
dividing_line=dividing_line,
repeat_command_groups=repeat_command_groups,
override_system_messages=override_system_messages,
autocompleter=autocompleter,
print_func=print_func)
def run_polling(self) -> None:
"""
Private. Starts the user input processing cycle
:return: None
"""
self._pre_cycle_setup()
while True:
if self._repeat_command_groups_description:
self._print_command_group_description()
raw_command: str = Console().input(self._prompt)
try:
input_command: InputCommand = InputCommand.parse(raw_command=raw_command)
except BaseInputCommandException as error:
with redirect_stdout(io.StringIO()) as f:
self._error_handler(error, raw_command)
res: str = f.getvalue()
self._print_framed_text(res)
continue
if self._is_exit_command(input_command):
system_router.finds_appropriate_handler(input_command)
self._autocompleter.exit_setup()
return
if self._is_unknown_command(input_command):
with redirect_stdout(io.StringIO()) as f:
self._unknown_command_handler(input_command)
res: str = f.getvalue()
self._print_framed_text(res)
continue
with redirect_stdout(io.StringIO()) as f:
for registered_router in self._registered_routers:
registered_router.finds_appropriate_handler(input_command)
res: str = f.getvalue()
self._print_framed_text(res)
if not self._repeat_command_groups_description:
self._print_func(self._prompt)
def include_router(self, router: Router) -> None:
"""
Public. Registers the router in the application
:param router: registered router
:return: None
"""
router.set_command_register_ignore(self._ignore_command_register)
self._registered_routers.add_registered_router(router)
def include_routers(self, *routers: Router) -> None:
"""
Public. Registers the routers in the application
:param routers: registered routers
:return: None
"""
for router in routers:
self.include_router(router)
def add_message_on_startup(self, message: str) -> None:
"""
Public. Adds a message that will be displayed when the application is launched
:param message: the message being added
:return: None
"""
self._messages_on_startup.append(message)
@@ -0,0 +1,40 @@
from argenta.router import Router
class RegisteredRouters:
def __init__(self, registered_routers: list[Router] = None) -> None:
"""
Private. Combines registered routers
:param registered_routers: list of the registered routers
:return: None
"""
self._registered_routers = registered_routers if registered_routers else []
def get_registered_routers(self) -> list[Router]:
"""
Private. Returns the registered routers
:return: registered routers as list[Router]
"""
return self._registered_routers
def add_registered_router(self, router: Router) -> None:
"""
Private. Adds a new registered router
:param router: registered router
:return: None
"""
self._registered_routers.append(router)
def add_registered_routers(self, *routers: Router) -> None:
"""
Private. Adds new registered routers
:param routers: registered routers
:return: None
"""
self._registered_routers.extend(routers)
def __iter__(self):
return iter(self._registered_routers)
def __next__(self):
return next(iter(self._registered_routers))
+3
View File
@@ -0,0 +1,3 @@
__all__ = ["Command"]
from argenta.command.models import Command
+35
View File
@@ -0,0 +1,35 @@
from argenta.command.flag.models import InputFlag, Flag
class BaseInputCommandException(Exception):
"""
Private. Base exception class for all exceptions raised when parse input command
"""
pass
class UnprocessedInputFlagException(BaseInputCommandException):
"""
Private. Raised when an unprocessed input flag is detected
"""
def __str__(self):
return "Unprocessed Input Flags"
class RepeatedInputFlagsException(BaseInputCommandException):
"""
Private. Raised when repeated input flags are detected
"""
def __init__(self, flag: Flag | InputFlag):
self.flag = flag
def __str__(self):
return ("Repeated Input Flags\n"
f"Duplicate flag was detected in the input: '{self.flag.get_string_entity()}'")
class EmptyInputCommandException(BaseInputCommandException):
"""
Private. Raised when an empty input command is detected
"""
def __str__(self):
return "Input Command is empty"
+4
View File
@@ -0,0 +1,4 @@
__all__ = ('InputFlags', 'InputFlag', 'Flag', 'Flags')
from argenta.command.flag.models import InputFlags, InputFlag, Flags, Flag
+24
View File
@@ -0,0 +1,24 @@
from dataclasses import dataclass
from argenta.command.flag.models import Flag
import re
@dataclass
class PredefinedFlags:
"""
Public. A dataclass with predefined flags and most frequently used flags for quick use
"""
HELP = Flag(name='help', possible_values=False)
SHORT_HELP = Flag(name='H', prefix='-', possible_values=False)
INFO = Flag(name='info', possible_values=False)
SHORT_INFO = Flag(name='I', prefix='-', possible_values=False)
ALL = Flag(name='all', possible_values=False)
SHORT_ALL = Flag(name='A', prefix='-', possible_values=False)
HOST = Flag(name='host', possible_values=re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$'))
SHORT_HOST = Flag(name='H', prefix='-', possible_values=re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$'))
PORT = Flag(name='port', possible_values=re.compile(r'^\d{1,5}$'))
SHORT_PORT = Flag(name='P', prefix='-', possible_values=re.compile(r'^\d{1,5}$'))
+218
View File
@@ -0,0 +1,218 @@
from typing import Literal, Pattern
from abc import ABC, abstractmethod
class BaseFlag(ABC):
def __init__(self, name: str,
prefix: Literal['-', '--', '---'] = '--') -> None:
"""
Private. Base class for flags
:param name: the name of the flag
:param prefix: the prefix of the flag
:return: None
"""
self._name = name
self._prefix = prefix
def get_string_entity(self) -> str:
"""
Public. Returns a string representation of the flag
:return: string representation of the flag as str
"""
string_entity: str = self._prefix + self._name
return string_entity
def get_name(self) -> str:
"""
Public. Returns the name of the flag
:return: the name of the flag as str
"""
return self._name
def get_prefix(self) -> str:
"""
Public. Returns the prefix of the flag
:return: the prefix of the flag as str
"""
return self._prefix
class InputFlag(BaseFlag):
def __init__(self, name: str,
prefix: Literal['-', '--', '---'] = '--',
value: str = None):
"""
Public. The entity of the flag of the entered command
:param name: the name of the input flag
:param prefix: the prefix of the input flag
:param value: the value of the input flag
:return: None
"""
super().__init__(name, prefix)
self._flag_value = value
def get_value(self) -> str | None:
"""
Public. Returns the value of the flag
:return: the value of the flag as str
"""
return self._flag_value
def set_value(self, value):
"""
Private. Sets the value of the flag
:param value: the fag value to set
:return: None
"""
self._flag_value = value
class Flag(BaseFlag):
def __init__(self, name: str,
prefix: Literal['-', '--', '---'] = '--',
possible_values: list[str] | Pattern[str] | False = True) -> None:
"""
Public. The entity of the flag being registered for subsequent processing
:param name: The name of the flag
:param prefix: The prefix of the flag
:param possible_values: The possible values of the flag, if False then the flag cannot have a value
:return: None
"""
super().__init__(name, prefix)
self.possible_values = possible_values
def validate_input_flag_value(self, input_flag_value: str | None):
"""
Private. Validates the input flag value
:param input_flag_value: The input flag value to validate
:return: whether the entered flag is valid as bool
"""
if self.possible_values is False:
if input_flag_value is None:
return True
else:
return False
elif isinstance(self.possible_values, Pattern):
if isinstance(input_flag_value, str):
is_valid = bool(self.possible_values.match(input_flag_value))
if bool(is_valid):
return True
else:
return False
else:
return False
elif isinstance(self.possible_values, list):
if input_flag_value in self.possible_values:
return True
else:
return False
else:
return True
class BaseFlags(ABC):
"""
Private. Base class for groups of flags
"""
__slots__ = ('_flags',)
@abstractmethod
def get_flags(self):
"""
Public. Returns a list of flags
:return: list of flags
"""
pass
@abstractmethod
def add_flag(self, flag: Flag | InputFlag):
"""
Public. Adds a flag to the list of flags
:param flag: flag to add
:return: None
"""
pass
@abstractmethod
def add_flags(self, flags: list[Flag] | list[InputFlag]):
"""
Public. Adds a list of flags to the list of flags
:param flags: list of flags to add
:return: None
"""
pass
@abstractmethod
def get_flag(self, name: str):
"""
Public. Returns the flag entity by its name or None if not found
:param name: the name of the flag to get
:return: entity of the flag or None
"""
pass
def __iter__(self):
return iter(self._flags)
def __next__(self):
return next(iter(self))
def __getitem__(self, item):
return self._flags[item]
class Flags(BaseFlags, ABC):
def __init__(self, *flags: Flag):
"""
Public. A model that combines the registered flags
:param flags: the flags that will be registered
:return: None
"""
self._flags = flags if flags else []
def get_flags(self) -> list[Flag]:
return self._flags
def add_flag(self, flag: Flag):
self._flags.append(flag)
def add_flags(self, flags: list[Flag]):
self._flags.extend(flags)
def get_flag(self, name: str) -> Flag | None:
if name in [flag.get_name() for flag in self._flags]:
return list(filter(lambda flag: flag.get_name() == name, self._flags))[0]
else:
return None
class InputFlags(BaseFlags, ABC):
def __init__(self, *flags: InputFlag):
"""
Public. A model that combines the input flags of the input command
:param flags: all input flags
:return: None
"""
self._flags = flags if flags else []
def get_flags(self) -> list[InputFlag]:
return self._flags
def add_flag(self, flag: InputFlag):
self._flags.append(flag)
def add_flags(self, flags: list[InputFlag]):
self._flags.extend(flags)
def get_flag(self, name: str) -> InputFlag | None:
if name in [flag.get_name() for flag in self._flags]:
return list(filter(lambda flag: flag.get_name() == name, self._flags))[0]
else:
return None
+164
View File
@@ -0,0 +1,164 @@
from argenta.command.flag.models import Flag, InputFlag, Flags, InputFlags
from argenta.command.exceptions import (UnprocessedInputFlagException,
RepeatedInputFlagsException,
EmptyInputCommandException)
from typing import Generic, TypeVar, cast, Literal
InputCommandType = TypeVar('InputCommandType')
class BaseCommand:
def __init__(self, trigger: str) -> None:
"""
Private. Base class for all commands
:param trigger: A string trigger, which, when entered by the user, indicates that the input corresponds to the command
"""
self._trigger = trigger
def get_trigger(self) -> str:
"""
Public. Returns the trigger of the command
:return: the trigger of the command as str
"""
return self._trigger
class Command(BaseCommand):
def __init__(self, trigger: str,
description: str = None,
flags: Flag | Flags = None,
aliases: list[str] = None):
"""
Public. The command that can and should be registered in the Router
:param trigger: A string trigger, which, when entered by the user, indicates that the input corresponds to the command
:param description: the description of the command
:param flags: processed commands
:param aliases: string synonyms for the main trigger
"""
super().__init__(trigger)
self._registered_flags: Flags = flags if isinstance(flags, Flags) else Flags(flags) if isinstance(flags, Flag) else Flags()
self._description = f'Description for "{self._trigger}" command' if not description else description
self._aliases = aliases if isinstance(aliases, list) else []
def get_registered_flags(self) -> Flags:
"""
Private. Returns the registered flags
:return: the registered flags as Flags
"""
return self._registered_flags
def get_aliases(self) -> list[str] | list:
"""
Public. Returns the aliases of the command
:return: the aliases of the command as list[str] | list
"""
return self._aliases
def validate_input_flag(self, flag: InputFlag) -> bool:
"""
Private. Validates the input flag
:param flag: input flag for validation
:return: is input flag valid as bool
"""
registered_flags: Flags | None = self.get_registered_flags()
if registered_flags:
if isinstance(registered_flags, Flag):
if registered_flags.get_string_entity() == flag.get_string_entity():
is_valid = registered_flags.validate_input_flag_value(flag.get_value())
if is_valid:
return True
else:
for registered_flag in registered_flags:
if registered_flag.get_string_entity() == flag.get_string_entity():
is_valid = registered_flag.validate_input_flag_value(flag.get_value())
if is_valid:
return True
return False
def get_description(self) -> str:
"""
Private. Returns the description of the command
:return: the description of the command as str
"""
return self._description
class InputCommand(BaseCommand, Generic[InputCommandType]):
def __init__(self, trigger: str,
input_flags: InputFlag | InputFlags = None):
"""
Private. The model of the input command, after parsing
:param trigger:the trigger of the command
:param input_flags: the input flags
:return: None
"""
super().__init__(trigger)
self._input_flags: InputFlags = input_flags if isinstance(input_flags, InputFlags) else InputFlags(input_flags) if isinstance(input_flags, InputFlag) else InputFlags()
def _set_input_flags(self, input_flags: InputFlags) -> None:
"""
Private. Sets the input flags
:param input_flags: the input flags to set
:return: None
"""
self._input_flags = input_flags
def get_input_flags(self) -> InputFlags:
"""
Private. Returns the input flags
:return: the input flags as InputFlags
"""
return self._input_flags
@staticmethod
def parse(raw_command: str) -> InputCommandType:
"""
Private. Parse the raw input command
:param raw_command: raw input command
:return: model of the input command, after parsing as InputCommand
"""
if not raw_command:
raise EmptyInputCommandException()
list_of_tokens = raw_command.split()
command = list_of_tokens.pop(0)
input_flags: InputFlags = InputFlags()
current_flag_name, current_flag_value = None, None
for k, _ in enumerate(list_of_tokens):
if _.startswith('-'):
if current_flag_name or len(_) < 2 or len(_[:_.rfind('-')]) > 3:
raise UnprocessedInputFlagException()
current_flag_name = _
else:
if not current_flag_name:
raise UnprocessedInputFlagException()
current_flag_value = _
if current_flag_name:
if not len(list_of_tokens) == k+1:
if not list_of_tokens[k+1].startswith('-'):
continue
input_flag = InputFlag(name=current_flag_name[current_flag_name.rfind('-')+1:],
prefix=cast(Literal['-', '--', '---'],
current_flag_name[:current_flag_name.rfind('-')+1]),
value=current_flag_value)
all_flags = [flag.get_string_entity() for flag in input_flags.get_flags()]
if input_flag.get_string_entity() not in all_flags:
input_flags.add_flag(input_flag)
else:
raise RepeatedInputFlagsException(input_flag)
current_flag_name, current_flag_value = None, None
if any([current_flag_name, current_flag_value]):
raise UnprocessedInputFlagException()
else:
return InputCommand(trigger=command, input_flags=input_flags)
+4
View File
@@ -0,0 +1,4 @@
__all__ = ["Orchestrator"]
from argenta.orchestrator.entity import Orchestrator
@@ -0,0 +1,4 @@
__all__ = ["ArgParser"]
from argenta.orchestrator.argparser.entity import ArgParser
@@ -0,0 +1,6 @@
__all__ = ["BooleanArgument", "PositionalArgument", "OptionalArgument"]
from argenta.orchestrator.argparser.arguments.models import (BooleanArgument,
PositionalArgument,
OptionalArgument)
@@ -0,0 +1,55 @@
from abc import ABC, abstractmethod
from typing import Literal
class BaseArgument(ABC):
"""
Private. Base class for all arguments
"""
@abstractmethod
def get_string_entity(self) -> str:
"""
Public. Returns the string representation of the argument
:return: the string representation as a str
"""
pass
class PositionalArgument(BaseArgument):
def __init__(self, name: str):
"""
Public. Required argument at startup
:param name: name of the argument, must not start with minus (-)
"""
self.name = name
def get_string_entity(self):
return self.name
class OptionalArgument(BaseArgument):
def __init__(self, name: str, prefix: Literal['-', '--', '---'] = '--'):
"""
Public. Optional argument, must have the value
:param name: name of the argument
:param prefix: prefix of the argument
"""
self.name = name
self.prefix = prefix
def get_string_entity(self):
return self.prefix + self.name
class BooleanArgument(BaseArgument):
def __init__(self, name: str, prefix: Literal['-', '--', '---'] = '--'):
"""
Public. Boolean argument, does not require a value
:param name: name of the argument
:param prefix: prefix of the argument
"""
self.name = name
self.prefix = prefix
def get_string_entity(self):
return self.prefix + self.name
@@ -0,0 +1,49 @@
from argparse import ArgumentParser
from argenta.orchestrator.argparser.arguments.models import (BooleanArgument,
OptionalArgument,
PositionalArgument)
class ArgParser:
def __init__(self,
processed_args: list[PositionalArgument | OptionalArgument | BooleanArgument],
name: str = 'Argenta',
description: str = 'Argenta available arguments',
epilog: str = 'github.com/koloideal/Argenta | made by kolo') -> None:
"""
Public. Cmd argument parser and configurator at startup
:param name: the name of the ArgParse instance
:param description: the description of the ArgParse instance
:param epilog: the epilog of the ArgParse instance
:param processed_args: registered and processed arguments
"""
self.name = name
self.description = description
self.epilog = epilog
self.entity: ArgumentParser = ArgumentParser(prog=name, description=description, epilog=epilog)
self.args: list[PositionalArgument | OptionalArgument | BooleanArgument] | None = processed_args
def set_args(self, *args: PositionalArgument | OptionalArgument | BooleanArgument) -> None:
"""
Public. Sets the arguments to be processed
:param args: processed arguments
:return: None
"""
self.args.extend(args)
def register_args(self) -> None:
"""
Private. Registers initialized command line arguments
:return: None
"""
if not self.args:
return
for arg in self.args:
if type(arg) is PositionalArgument:
self.entity.add_argument(arg.get_string_entity())
elif type(arg) is OptionalArgument:
self.entity.add_argument(arg.get_string_entity())
elif type(arg) is BooleanArgument:
self.entity.add_argument(arg.get_string_entity(), action='store_true')
+36
View File
@@ -0,0 +1,36 @@
from argparse import Namespace
from argenta.app import App
from argenta.orchestrator.argparser import ArgParser
class Orchestrator:
def __init__(self, arg_parser: ArgParser = False):
"""
Public. An orchestrator and configurator that defines the behavior of an integrated system, one level higher than the App
:param arg_parser: Cmd argument parser and configurator at startup
:return: None
"""
self.arg_parser: ArgParser | False = arg_parser
if arg_parser:
self.arg_parser.register_args()
@staticmethod
def start_polling(app: App) -> None:
"""
Public. Starting the user input processing cycle
:param app: a running application
:return: None
"""
app.run_polling()
def get_input_args(self) -> Namespace | None:
"""
Public. Returns the arguments parsed
:return: None
"""
if self.arg_parser:
return self.arg_parser.entity.parse_args()
else:
return None
+4
View File
@@ -0,0 +1,4 @@
__all__ = ["Router"]
from argenta.router.entity import Router
@@ -0,0 +1,79 @@
from typing import Callable
from argenta.command import Command
from argenta.command.flag import InputFlags
class CommandHandler:
def __init__(self, handler: Callable[[], None] | Callable[[InputFlags], None], 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 = handler
self._handled_command = handled_command
def handling(self, input_flags: InputFlags = None) -> None:
"""
Private. Direct processing of an input command
:param input_flags: the flags of the input command
:return: None
"""
if input_flags is not None:
self._handler(input_flags)
else:
self._handler()
def get_handler(self) -> Callable[[], None] | Callable[[InputFlags], None]:
"""
Private. Returns the handler being called
:return: the handler being called as Callable[[], None] or Callable[[InputFlags], None]
"""
return self._handler
def get_handled_command(self) -> Command:
"""
Private. Returns the command being processed
:return: the command being processed as Command
"""
return self._handled_command
class CommandHandlers:
def __init__(self, command_handlers: list[CommandHandler] = None):
"""
Private. The model that unites all CommandHandler of the routers
:param command_handlers: list of CommandHandlers for register
"""
self.command_handlers = command_handlers if command_handlers else []
def get_handlers(self) -> list[CommandHandler]:
"""
Private. Returns the list of CommandHandlers
:return: the list of CommandHandlers as list[CommandHandler]
"""
return self.command_handlers
def add_handler(self, command_handler: CommandHandler) -> None:
"""
Private. Adds a CommandHandler to the list of CommandHandlers
:param command_handler: CommandHandler to be added
:return: None
"""
self.command_handlers.append(command_handler)
def add_handlers(self, *command_handlers: CommandHandler) -> None:
"""
Private. Extend a many CommandHandler to the list of CommandHandlers
:param command_handlers: many CommandHandler to be added
:return: None
"""
self.command_handlers.extend(command_handlers)
def __iter__(self):
return iter(self.command_handlers)
def __next__(self):
return next(iter(self.command_handlers))
+4
View File
@@ -0,0 +1,4 @@
from argenta.router import Router
system_router = Router(title='System points:')
+204
View File
@@ -0,0 +1,204 @@
from typing import Callable
from inspect import getfullargspec
from argenta.command import Command
from argenta.command.models import InputCommand
from argenta.router.command_handler.entity import CommandHandlers, CommandHandler
from argenta.command.flag.models import Flag, Flags, InputFlags
from argenta.router.exceptions import (RepeatedFlagNameException,
TooManyTransferredArgsException,
RequiredArgumentNotPassedException,
TriggerContainSpacesException)
class Router:
def __init__(self,
title: str = None):
"""
Public. Directly configures and manages handlers
:param title: the title of the router, displayed when displaying the available commands
:return: None
"""
self._title = title
self._command_handlers: CommandHandlers = CommandHandlers()
self._ignore_command_register: bool = False
self._not_valid_flag_handler: Callable[[Flag], None] = lambda flag: print(f"Undefined or incorrect input flag: {flag.get_string_entity()}{(' '+flag.get_value()) if flag.get_value() else ''}")
def command(self, command: Command) -> Callable:
"""
Public. Registers handler
:param command: Registered command
:return: decorated handler as Callable[[Any], Any]
"""
self._validate_command(command)
def command_decorator(func):
Router._validate_func_args(command, func)
self._command_handlers.add_handler(CommandHandler(func, command))
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
return command_decorator
def set_invalid_input_flag_handler(self, func: Callable[[Flag], None]) -> None:
"""
Public. Registers handler for invalid input flag
:param func: registered handler
:return: None
"""
self._not_valid_flag_handler = func
def finds_appropriate_handler(self, input_command: InputCommand) -> None:
"""
Private. Finds the appropriate handler for given input command and passes control to it
:param input_command: input command as InputCommand
:return: None
"""
input_command_name: str = input_command.get_trigger()
input_command_flags: InputFlags = input_command.get_input_flags()
for command_handler in self._command_handlers:
handle_command = command_handler.get_handled_command()
if input_command_name.lower() == handle_command.get_trigger().lower():
self.process_input_command(input_command_flags, command_handler)
if input_command_name.lower() in handle_command.get_aliases():
self.process_input_command(input_command_flags, command_handler)
def process_input_command(self, input_command_flags: InputFlags, command_handler: CommandHandler) -> None:
"""
Private. Processes input command with the appropriate handler
:param input_command_flags: input command flags as InputFlags
:param command_handler: command handler for input command as CommandHandler
:return: None
"""
handle_command = command_handler.get_handled_command()
if handle_command.get_registered_flags().get_flags():
if input_command_flags.get_flags():
if self._validate_input_flags(handle_command, input_command_flags):
command_handler.handling(input_command_flags)
return
else:
command_handler.handling(input_command_flags)
return
else:
if input_command_flags.get_flags():
self._not_valid_flag_handler(input_command_flags[0])
return
else:
command_handler.handling()
return
def _validate_input_flags(self, handled_command: Command, input_flags: InputFlags) -> bool:
"""
Private. Validates flags of input command
:param handled_command: entity of the handled command
:param input_flags:
:return: is flags of input command valid as bool
"""
for flag in input_flags:
is_valid: bool = handled_command.validate_input_flag(flag)
if not is_valid:
self._not_valid_flag_handler(flag)
return False
return True
@staticmethod
def _validate_command(command: Command) -> None:
"""
Private. Validates the command registered in handler
:param command: validated command
:return: None if command is valid else raise exception
"""
command_name: str = command.get_trigger()
if command_name.find(' ') != -1:
raise TriggerContainSpacesException()
flags: Flags = command.get_registered_flags()
if flags:
flags_name: list = [x.get_string_entity().lower() for x in flags]
if len(set(flags_name)) < len(flags_name):
raise RepeatedFlagNameException()
@staticmethod
def _validate_func_args(command: Command, func: Callable) -> None:
"""
Private. Validates the arguments of the handler
:param command: registered command in handler
:param func: entity of the handler func
:return: None if func is valid else raise exception
"""
registered_args = command.get_registered_flags()
transferred_args = getfullargspec(func).args
if registered_args.get_flags() and transferred_args:
if len(transferred_args) != 1:
raise TooManyTransferredArgsException()
elif registered_args.get_flags() and not transferred_args:
raise RequiredArgumentNotPassedException()
elif not registered_args.get_flags() and transferred_args:
raise TooManyTransferredArgsException()
def set_command_register_ignore(self, _: bool) -> None:
"""
Private. Sets the router behavior on the input commands register
:param _: is command register ignore
:return: None
"""
self._ignore_command_register = _
def get_triggers(self) -> list[str]:
"""
Public. Gets registered triggers
:return: registered in router triggers as list[str]
"""
all_triggers: list[str] = []
for command_handler in self._command_handlers:
all_triggers.append(command_handler.get_handled_command().get_trigger())
return all_triggers
def get_aliases(self) -> list[str]:
"""
Public. Gets registered aliases
:return: registered in router aliases as list[str]
"""
all_aliases: list[str] = []
for command_handler in self._command_handlers:
if command_handler.get_handled_command().get_aliases():
all_aliases.extend(command_handler.get_handled_command().get_aliases())
return all_aliases
def get_command_handlers(self) -> CommandHandlers:
"""
Private. Gets registered command handlers
:return: registered command handlers as CommandHandlers
"""
return self._command_handlers
def get_title(self) -> str | None:
"""
Public. Gets title of the router
:return: the title of the router as str or None
"""
return self._title
def set_title(self, title: str) -> None:
"""
Public. Sets the title of the router
:param title: title that will be setted
:return: None
"""
self._title = title
+30
View File
@@ -0,0 +1,30 @@
class RepeatedFlagNameException(Exception):
"""
Private. Raised when a repeated flag name is registered
"""
def __str__(self):
return "Repeated registered flag names in register command"
class TooManyTransferredArgsException(Exception):
"""
Private. Raised when too many arguments are passed
"""
def __str__(self):
return "Too many transferred arguments"
class RequiredArgumentNotPassedException(Exception):
"""
Private. Raised when a required argument is not passed
"""
def __str__(self):
return "Required argument not passed"
class TriggerContainSpacesException(Exception):
"""
Private. Raised when there is a space in the trigger being registered
"""
def __str__(self):
return "Command trigger cannot contain spaces"
View File
@@ -0,0 +1,216 @@
import _io
from unittest.mock import patch, MagicMock
import unittest
import io
import re
from argenta.app import App
from argenta.command import Command
from argenta.router import Router
from argenta.command.flag.models import Flags, InputFlags
from argenta.command.flag.defaults import PredefinedFlags
class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("builtins.input", side_effect=["help", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
@router.command(Command('test'))
def test():
print('test command')
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.get_trigger()}'))
app.run_polling()
output = mock_stdout.getvalue()
self.assertIn("\nUnknown command: help\n", output)
@patch("builtins.input", side_effect=["TeSt", "Q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_incorrect_command2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
@router.command(Command('test'))
def test():
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.get_trigger()}'))
app.run_polling()
output = mock_stdout.getvalue()
self.assertIn('\nUnknown command: TeSt\n', output)
@patch("builtins.input", side_effect=["test --help", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_unregistered_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
@router.command(Command('test'))
def test():
print(f'test command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.run_polling()
output = mock_stdout.getvalue()
self.assertIn('\nUndefined or incorrect input flag: --help\n', output)
@patch("builtins.input", side_effect=["test --port 22", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_unregistered_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
@router.command(Command('test'))
def test():
print('test command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.run_polling()
output = mock_stdout.getvalue()
self.assertIn('\nUndefined or incorrect input flag: --port 22\n', output)
@patch("builtins.input", side_effect=["test --host 192.168.32.1 --port 132", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_one_correct_flag_an_one_incorrect_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
flags = Flags(PredefinedFlags.HOST)
@router.command(Command('test', flags=flags))
def test(args: InputFlags):
print(f'connecting to host {args.get_flag('host').get_value()}')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.run_polling()
output = mock_stdout.getvalue()
self.assertIn('\nUndefined or incorrect input flag: --port 132\n', output)
@patch("builtins.input", side_effect=["test", "some", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_one_correct_command_and_one_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
@router.command(Command('test'))
def test():
print(f'test command')
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.get_trigger()}'))
app.run_polling()
output = mock_stdout.getvalue()
self.assertRegex(output, re.compile(r'\ntest command\n(.|\n)*\nUnknown command: some'))
@patch("builtins.input", side_effect=["test", "some", "more", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_two_correct_commands_and_one_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
@router.command(Command('test'))
def test():
print(f'test command')
@router.command(Command('more'))
def test():
print(f'more command')
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.get_trigger()}'))
app.run_polling()
output = mock_stdout.getvalue()
self.assertRegex(output, re.compile(r'\ntest command\n(.|\n)*\nUnknown command: some\n(.|\n)*\nmore command'))
@patch("builtins.input", side_effect=["test 535 --port", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_incorrect_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
@router.command(Command('test'))
def test():
print(f'test command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.set_invalid_input_flags_handler(lambda command: print(f'Incorrect flag syntax: "{command}"'))
app.run_polling()
output = mock_stdout.getvalue()
self.assertIn("\nIncorrect flag syntax: \"test 535 --port\"\n", output)
@patch("builtins.input", side_effect=["", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_empty_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
@router.command(Command('test'))
def test():
print(f'test command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.set_empty_command_handler(lambda: print('Empty input command'))
app.run_polling()
output = mock_stdout.getvalue()
self.assertIn("\nEmpty input command\n", output)
@patch("builtins.input", side_effect=["test --port 22 --port 33", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_repeated_flags(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
@router.command(Command('test', flags=PredefinedFlags.PORT))
def test(args: InputFlags):
print('test command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.set_repeated_input_flags_handler(lambda command: print(f'Repeated input flags: "{command}"'))
app.run_polling()
output = mock_stdout.getvalue()
self.assertIn("\nRepeated input flags: \"test --port 22 --port 33\"\n", output)
@@ -0,0 +1,222 @@
import _io
from unittest.mock import patch, MagicMock
import unittest
import io
import re
from argenta.app import App
from argenta.command.models import Command
from argenta.router import Router
from argenta.command.flag.models import Flag, Flags, InputFlags
from argenta.command.flag.defaults import PredefinedFlags
class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("builtins.input", side_effect=["test", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
@router.command(Command('test'))
def test():
print('test command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.run_polling()
output = mock_stdout.getvalue()
self.assertIn('\ntest command\n', output)
@patch("builtins.input", side_effect=["TeSt", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
@router.command(Command('test'))
def test():
print('test command')
app = App(ignore_command_register=True,
override_system_messages=True,
print_func=print)
app.include_router(router)
app.run_polling()
output = mock_stdout.getvalue()
self.assertIn('\ntest command\n', output)
@patch("builtins.input", side_effect=["test --help", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_custom_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
flag = Flag('help', '--', False)
@router.command(Command('test', flags=flag))
def test(args: InputFlags):
print(f'\nhelp for {args.get_flag('help').get_name()} flag\n')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.run_polling()
output = mock_stdout.getvalue()
self.assertIn('\nhelp for help flag\n', output)
@patch("builtins.input", side_effect=["test --port 22", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_custom_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
flag = Flag('port', '--', re.compile(r'^\d{1,5}$'))
@router.command(Command('test', flags=flag))
def test(args: InputFlags):
print(f'flag value for {args.get_flag('port').get_name()} flag : {args.get_flag('port').get_value()}')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.run_polling()
output = mock_stdout.getvalue()
self.assertIn('\nflag value for port flag : 22\n', output)
@patch("builtins.input", side_effect=["test -H", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_default_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
flag = PredefinedFlags.SHORT_HELP
@router.command(Command('test', flags=flag))
def test(args: InputFlags):
print(f'help for {args.get_flag('H').get_name()} flag')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.run_polling()
output = mock_stdout.getvalue()
self.assertIn('\nhelp for H flag\n', output)
@patch("builtins.input", side_effect=["test --info", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_default_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
flag = PredefinedFlags.INFO
@router.command(Command('test', flags=flag))
def test(args: InputFlags):
if args.get_flag('info'):
print('info about test command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.run_polling()
output = mock_stdout.getvalue()
self.assertIn('\ninfo about test command\n', output)
@patch("builtins.input", side_effect=["test --host 192.168.0.1", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_default_flag3(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
flag = PredefinedFlags.HOST
@router.command(Command('test', flags=flag))
def test(args: InputFlags):
print(f'connecting to host {args[0].get_value()}')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.run_polling()
output = mock_stdout.getvalue()
self.assertIn('\nconnecting to host 192.168.0.1\n', output)
@patch("builtins.input", side_effect=["test --host 192.168.32.1 --port 132", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_two_flags(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
flags = Flags(PredefinedFlags.HOST, PredefinedFlags.PORT)
@router.command(Command('test', flags=flags))
def test(args: InputFlags):
print(f'connecting to host {args[0].get_value()} and port {args[1].get_value()}')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.run_polling()
output = mock_stdout.getvalue()
self.assertIn('\nconnecting to host 192.168.32.1 and port 132\n', output)
@patch("builtins.input", side_effect=["test", "some", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_two_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
@router.command(Command('test'))
def test():
print(f'test command')
@router.command(Command('some'))
def test2():
print(f'some command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.run_polling()
output = mock_stdout.getvalue()
self.assertRegex(output, re.compile(r'\ntest command\n(.|\n)*\nsome command\n'))
@patch("builtins.input", side_effect=["test", "some", "more", "q"])
@patch("sys.stdout", new_callable=io.StringIO)
def test_input_three_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router()
@router.command(Command('test'))
def test():
print(f'test command')
@router.command(Command('some'))
def test():
print(f'some command')
@router.command(Command('more'))
def test():
print(f'more command')
app = App(override_system_messages=True,
print_func=print)
app.include_router(router)
app.run_polling()
output = mock_stdout.getvalue()
self.assertRegex(output, re.compile(r'\ntest command\n(.|\n)*\nsome command\n(.|\n)*\nmore command'))
-20
View File
@@ -1,20 +0,0 @@
from argenta.app import App
from argenta.app.exceptions import (InvalidDescriptionMessagePatternException,
NoRegisteredRoutersException)
import unittest
class TestApp(unittest.TestCase):
def test_set_invalid_description_message_pattern(self):
with self.assertRaises(InvalidDescriptionMessagePatternException):
App().set_description_message_pattern('Invalid description pattern')
def test_set_invalid_description_message_pattern2(self):
with self.assertRaises(InvalidDescriptionMessagePatternException):
App().set_description_message_pattern('Invalid {desription} description {comand} pattern')
def test_no_registered_router(self):
with self.assertRaises(NoRegisteredRoutersException):
App()._validate_number_of_routers()
-31
View File
@@ -1,31 +0,0 @@
from argenta.command import Command
from argenta.command.exceptions import (UnprocessedInputFlagException,
RepeatedInputFlagsException,
EmptyInputCommandException)
import unittest
class TestCommand(unittest.TestCase):
def test_parse_correct_raw_command(self):
self.assertEqual(Command.parse_input_command('ssh --host 192.168.0.3').get_trigger(), 'ssh')
def test_parse_raw_command_with_flag_name_without_value(self):
with self.assertRaises(UnprocessedInputFlagException):
Command.parse_input_command('ssh --host')
def test_parse_raw_command_without_flag_name_with_value(self):
with self.assertRaises(UnprocessedInputFlagException):
Command.parse_input_command('ssh 192.168.0.3')
def test_parse_raw_command_with_repeated_flag_name(self):
with self.assertRaises(RepeatedInputFlagsException):
Command.parse_input_command('ssh --host 192.168.0.3 --host 172.198.0.43')
def test_parse_empty_raw_command(self):
with self.assertRaises(EmptyInputCommandException):
Command.parse_input_command('')
def test_get_command_description(self):
self.assertEqual(Command(trigger='test', description='test description').get_description(), 'test description')
-75
View File
@@ -1,75 +0,0 @@
from argenta.command.params.flag import Flag
import unittest
import re
class TestFlag(unittest.TestCase):
def test_get_string_entity(self):
self.assertEqual(Flag(flag_name='test').get_string_entity(),
'--test')
def test_get_string_entity2(self):
self.assertEqual(Flag(flag_name='test',
flag_prefix='---').get_string_entity(),
'---test')
def test_get_flag_name(self):
self.assertEqual(Flag(flag_name='test').get_flag_name(),
'test')
def test_get_flag_prefix(self):
self.assertEqual(Flag(flag_name='test').get_flag_prefix(),
'--')
def test_get_flag_prefix2(self):
self.assertEqual(Flag(flag_name='test',
flag_prefix='--').get_flag_prefix(),
'--')
def test_get_flag_value_without_set(self):
self.assertEqual(Flag(flag_name='test').get_value(),
None)
def test_get_flag_value_with_set(self):
flag = Flag(flag_name='test')
flag.set_value('example')
self.assertEqual(flag.get_value(), 'example')
def test_validate_incorrect_flag_value_with_list_of_possible_flag_values(self):
flag = Flag(flag_name='test', possible_flag_values=['1', '2', '3'])
self.assertEqual(flag.validate_input_flag_value('bad value'), False)
def test_validate_correct_flag_value_with_list_of_possible_flag_values(self):
flag = Flag(flag_name='test', possible_flag_values=['1', '2', '3'])
self.assertEqual(flag.validate_input_flag_value('1'), True)
def test_validate_incorrect_flag_value_with_pattern_of_possible_flag_values(self):
flag = Flag(flag_name='test', possible_flag_values=re.compile(r'192.168.\d+.\d+'))
self.assertEqual(flag.validate_input_flag_value('152.123.9.8'), False)
def test_validate_correct_flag_value_with_pattern_of_possible_flag_values(self):
flag = Flag(flag_name='test', possible_flag_values=re.compile(r'192.168.\d+.\d+'))
self.assertEqual(flag.validate_input_flag_value('192.168.9.8'), True)
-49
View File
@@ -1,49 +0,0 @@
from argenta.command.params.flag import Flag, FlagsGroup
import unittest
class TestFlagsGroup(unittest.TestCase):
def test_get_flags(self):
flags = FlagsGroup()
list_of_flags = [
Flag('test1'),
Flag('test2'),
Flag('test3'),
]
flags.add_flags(list_of_flags)
self.assertEqual(flags.get_flags(),
list_of_flags)
def test_add_flag(self):
flags = FlagsGroup()
flags.add_flag(Flag('test'))
self.assertEqual(len(flags.get_flags()), 1)
def test_add_flags(self):
flags = FlagsGroup()
flags.add_flags([Flag('test'), Flag('test2')])
self.assertEqual(len(flags.get_flags()), 2)
def test_unparse_flags_to_dict(self):
list_of_flags = [
Flag('test1'),
Flag('test2'),
Flag('test3'),
]
flags = FlagsGroup(list_of_flags)
serialized_flags = flags.unparse_to_dict()
needed_result = {'test1': {'name': 'test1',
'prefix': '--',
'string_entity': '--test1',
'value': None},
'test2': {'name': 'test2',
'prefix': '--',
'string_entity': '--test2',
'value': None},
'test3': {'name': 'test3',
'prefix': '--',
'string_entity': '--test3',
'value': None}}
self.assertDictEqual(serialized_flags, needed_result)
-126
View File
@@ -1,126 +0,0 @@
from argenta.command.params.flag import FlagsGroup, Flag
from argenta.router import Router
from argenta.command import Command
from argenta.router.exceptions import RepeatedCommandException
import unittest
class TestRouter(unittest.TestCase):
def test_get_router_name(self):
self.assertEqual(Router(name='test name').get_name(), 'test name')
def test_get_router_title(self):
self.assertEqual(Router(title='test title').get_title(), 'test title')
def test_input_correct_command(self):
router = Router()
@router.command(Command(trigger='test'))
def test():
return 'correct result'
self.assertEqual(router.input_command_handler(Command(trigger='test')), 'correct result')
def test_input_command_with_invalid_flag(self):
router = Router()
router.set_invalid_input_flag_handler(lambda x: x)
@router.command(Command(trigger='test'))
def test():
return 'correct result'
input_command = Command(trigger='test')
input_command._set_input_flags(FlagsGroup([Flag('host')]))
self.assertEqual(router.input_command_handler(input_command), None)
def test_input_correct_command_with_one_register_and_ignore_command_register(self):
router = Router()
router.set_ignore_command_register(True)
@router.command(Command(trigger='test'))
def test():
return 'correct result'
self.assertEqual(router.input_command_handler(Command(trigger='test')), 'correct result')
def test_input_correct_command_with_different_register_and_ignore_command_register(self):
router = Router()
router.set_ignore_command_register(True)
@router.command(Command(trigger='test'))
def test():
return 'correct result'
self.assertEqual(router.input_command_handler(Command(trigger='TeSt')), 'correct result')
def test_input_incorrect_command_with_ignore_command_register(self):
router = Router()
router.set_ignore_command_register(True)
@router.command(Command(trigger='test'))
def test():
return 'correct result'
self.assertEqual(router.input_command_handler(Command(trigger='Test2')), None)
def test_register_repeated_commands_with_one_register(self):
router = Router()
@router.command(Command(trigger='test'))
def test():
return 'correct result'
with self.assertRaises(RepeatedCommandException):
@router.command(Command(trigger='test'))
def test():
return 'correct result'
def test_register_commands_with_different_register(self):
router = Router()
@router.command(Command(trigger='test'))
def test():
return 'correct result'
try:
@router.command(Command(trigger='Test'))
def test():
return 'correct result'
except RepeatedCommandException:
self.fail('RepeatedCommandException should not have been thrown')
def test_register_repeated_commands_with_one_register_and_set_ignore_command_register(self):
router = Router()
router.set_ignore_command_register(True)
@router.command(Command(trigger='test'))
def test():
return 'correct result'
with self.assertRaises(RepeatedCommandException):
@router.command(Command(trigger='test'))
def test():
return 'correct result'
def test_register_repeated_commands_with_different_register_and_set_ignore_command_register(self):
router = Router()
router.set_ignore_command_register(True)
@router.command(Command(trigger='test'))
def test():
return 'correct result'
with self.assertRaises(RepeatedCommandException):
@router.command(Command(trigger='Test'))
def test():
return 'correct result'
View File
+73
View File
@@ -0,0 +1,73 @@
from argenta.command.models import InputCommand, Command
from argenta.app import App
import unittest
class MyTestCase(unittest.TestCase):
def test_is_exit_command1(self):
app = App()
self.assertEqual(app._is_exit_command(InputCommand('q')), True)
def test_is_exit_command5(self):
app = App()
self.assertEqual(app._is_exit_command(InputCommand('Q')), True)
def test_is_exit_command2(self):
app = App(ignore_command_register=False)
self.assertEqual(app._is_exit_command(InputCommand('q')), False)
def test_is_exit_command3(self):
app = App(exit_command=Command('quit'))
self.assertEqual(app._is_exit_command(InputCommand('quit')), True)
def test_is_exit_command4(self):
app = App(exit_command=Command('quit'))
self.assertEqual(app._is_exit_command(InputCommand('qUIt')), True)
def test_is_exit_command6(self):
app = App(ignore_command_register=False,
exit_command=Command('quit'))
self.assertEqual(app._is_exit_command(InputCommand('qUIt')), False)
def test_is_unknown_command1(self):
app = App()
app.set_unknown_command_handler(lambda command: None)
app._all_registered_triggers_in_lower = ['fr', 'tr', 'de']
self.assertEqual(app._is_unknown_command(InputCommand('fr')), False)
def test_is_unknown_command2(self):
app = App()
app.set_unknown_command_handler(lambda command: None)
app._all_registered_triggers_in_lower = ['fr', 'tr', 'de']
self.assertEqual(app._is_unknown_command(InputCommand('cr')), True)
def test_is_unknown_command3(self):
app = App(ignore_command_register=False)
app.set_unknown_command_handler(lambda command: None)
app._all_registered_triggers_in_default_case = ['Pr', 'tW', 'deQW']
self.assertEqual(app._is_unknown_command(InputCommand('pr')), True)
def test_is_unknown_command4(self):
app = App(ignore_command_register=False)
app.set_unknown_command_handler(lambda command: None)
app._all_registered_triggers_in_default_case = ['Pr', 'tW', 'deQW']
self.assertEqual(app._is_unknown_command(InputCommand('tW')), False)
+49
View File
@@ -0,0 +1,49 @@
from argenta.command.flag import Flag, InputFlag, Flags
from argenta.command.models import InputCommand, Command
from argenta.command.exceptions import (UnprocessedInputFlagException,
RepeatedInputFlagsException,
EmptyInputCommandException)
import unittest
class TestInputCommand(unittest.TestCase):
def test_parse_correct_raw_command(self):
self.assertEqual(InputCommand.parse('ssh --host 192.168.0.3').get_trigger(), 'ssh')
def test_parse_raw_command_without_flag_name_with_value(self):
with self.assertRaises(UnprocessedInputFlagException):
InputCommand.parse('ssh 192.168.0.3')
def test_parse_raw_command_with_repeated_flag_name(self):
with self.assertRaises(RepeatedInputFlagsException):
InputCommand.parse('ssh --host 192.168.0.3 --host 172.198.0.43')
def test_parse_empty_raw_command(self):
with self.assertRaises(EmptyInputCommandException):
InputCommand.parse('')
def test_validate_correct_input_flag1(self):
command = Command('some', flags=Flag('test'))
self.assertEqual(command.validate_input_flag(InputFlag('test')), True)
def test_validate_correct_input_flag2(self):
command = Command('some', flags=Flags(Flag('test'), Flag('more')))
self.assertEqual(command.validate_input_flag(InputFlag('more')), True)
def test_validate_incorrect_input_flag1(self):
command = Command('some', flags=Flags(Flag('test')))
self.assertEqual(command.validate_input_flag(InputFlag('more')), False)
def test_validate_incorrect_input_flag2(self):
command = Command('some', flags=Flags(Flag('test'), Flag('more')))
self.assertEqual(command.validate_input_flag(InputFlag('case')), False)
def test_validate_incorrect_input_flag3(self):
command = Command('some')
self.assertEqual(command.validate_input_flag(InputFlag('case')), False)
def test_isinstance_parse_correct_raw_command(self):
cmd = InputCommand.parse('ssh --host 192.168.0.3')
self.assertIsInstance(cmd, InputCommand)
+21
View File
@@ -0,0 +1,21 @@
from argenta.app.dividing_line import DynamicDividingLine, StaticDividingLine
import unittest
class TestDividingLine(unittest.TestCase):
def test_get_static_dividing_line_full_line(self):
line = StaticDividingLine('-')
self.assertEqual(line.get_full_static_line(True).count('-'), 25)
def test_get_dynamic_dividing_line_full_line(self):
line = DynamicDividingLine()
self.assertEqual(line.get_full_dynamic_line(20, True).count('-'), 20)
def test_get_dividing_line_unit_part(self):
line = StaticDividingLine('')
self.assertEqual(line.get_unit_part(), ' ')
def test_get_dividing_line2_unit_part(self):
line = StaticDividingLine('+-0987654321!@#$%^&*()_')
self.assertEqual(line.get_unit_part(), '+')
+129
View File
@@ -0,0 +1,129 @@
from argenta.command.flag.models import Flag, InputFlag, InputFlags, Flags
import unittest
import re
class TestFlag(unittest.TestCase):
def test_get_string_entity(self):
self.assertEqual(Flag(name='test').get_string_entity(),
'--test')
def test_get_string_entity2(self):
self.assertEqual(Flag(name='test',
prefix='---').get_string_entity(),
'---test')
def test_get_flag_name(self):
self.assertEqual(Flag(name='test').get_name(),
'test')
def test_get_flag_prefix(self):
self.assertEqual(Flag(name='test').get_prefix(),
'--')
def test_get_flag_prefix2(self):
self.assertEqual(Flag(name='test',
prefix='--').get_prefix(),
'--')
def test_get_flag_value_without_set(self):
self.assertEqual(InputFlag(name='test').get_value(),
None)
def test_get_flag_value_with_set(self):
flag = InputFlag(name='test')
flag.set_value('example')
self.assertEqual(flag.get_value(), 'example')
def test_validate_incorrect_flag_value_with_list_of_possible_flag_values(self):
flag = Flag(name='test', possible_values=['1', '2', '3'])
self.assertEqual(flag.validate_input_flag_value('bad value'), False)
def test_validate_correct_flag_value_with_list_of_possible_flag_values(self):
flag = Flag(name='test', possible_values=['1', '2', '3'])
self.assertEqual(flag.validate_input_flag_value('1'), True)
def test_validate_incorrect_flag_value_with_pattern_of_possible_flag_values(self):
flag = Flag(name='test', possible_values=re.compile(r'192.168.\d+.\d+'))
self.assertEqual(flag.validate_input_flag_value('152.123.9.8'), False)
def test_validate_correct_flag_value_with_pattern_of_possible_flag_values(self):
flag = Flag(name='test', possible_values=re.compile(r'192.168.\d+.\d+'))
self.assertEqual(flag.validate_input_flag_value('192.168.9.8'), True)
def test_validate_correct_empty_flag_value_without_possible_flag_values(self):
flag = Flag(name='test', possible_values=False)
self.assertEqual(flag.validate_input_flag_value(None), True)
def test_validate_correct_empty_flag_value_with_possible_flag_values(self):
flag = Flag(name='test', possible_values=True)
self.assertEqual(flag.validate_input_flag_value(None), True)
def test_validate_incorrect_random_flag_value_without_possible_flag_values(self):
flag = Flag(name='test', possible_values=False)
self.assertEqual(flag.validate_input_flag_value('random value'), False)
def test_validate_correct_random_flag_value_with_possible_flag_values(self):
flag = Flag(name='test', possible_values=True)
self.assertEqual(flag.validate_input_flag_value('random value'), True)
def test_get_input_flag1(self):
flag = InputFlag(name='test')
input_flags = InputFlags(flag)
self.assertEqual(input_flags.get_flag('test'), flag)
def test_get_input_flag2(self):
flag = InputFlag(name='test')
flag2 = InputFlag(name='some')
input_flags = InputFlags(flag, flag2)
self.assertEqual(input_flags.get_flag('some'), flag2)
def test_get_undefined_input_flag(self):
flag = InputFlag(name='test')
flag2 = InputFlag(name='some')
input_flags = InputFlags(flag, flag2)
self.assertEqual(input_flags.get_flag('case'), None)
def test_get_flags(self):
flags = Flags()
list_of_flags = [
Flag('test1'),
Flag('test2'),
Flag('test3'),
]
flags.add_flags(list_of_flags)
self.assertEqual(flags.get_flags(),
list_of_flags)
def test_add_flag(self):
flags = Flags()
flags.add_flag(Flag('test'))
self.assertEqual(len(flags.get_flags()), 1)
def test_add_flags(self):
flags = Flags()
flags.add_flags([Flag('test'), Flag('test2')])
self.assertEqual(len(flags.get_flags()), 2)
+144
View File
@@ -0,0 +1,144 @@
from argenta.command.flag import InputFlags, InputFlag, Flag, Flags
from argenta.router import Router
from argenta.command import Command
from argenta.router.exceptions import (TriggerContainSpacesException,
RepeatedFlagNameException,
TooManyTransferredArgsException,
RequiredArgumentNotPassedException)
import unittest
import re
class TestRouter(unittest.TestCase):
def test_get_router_title(self):
self.assertEqual(Router(title='test title').get_title(), 'test title')
def test_register_command_with_spaces_in_trigger(self):
router = Router()
with self.assertRaises(TriggerContainSpacesException):
router._validate_command(Command(trigger='command with spaces'))
def test_register_command_with_repeated_flags(self):
router = Router()
with self.assertRaises(RepeatedFlagNameException):
router._validate_command(Command(trigger='command', flags=Flags(Flag('test'), Flag('test'))))
def test_validate_incorrect_input_flag1(self):
router = Router()
router.set_invalid_input_flag_handler(lambda flag: None)
self.assertEqual(router._validate_input_flags(Command('cmd'), InputFlags(InputFlag('ssh'))), False)
def test_validate_incorrect_input_flag2(self):
router = Router()
router.set_invalid_input_flag_handler(lambda flag: None)
self.assertEqual(router._validate_input_flags(Command('cmd'), InputFlags(InputFlag('ssh', value='some'))), False)
def test_validate_incorrect_input_flag3(self):
router = Router()
router.set_invalid_input_flag_handler(lambda flag: None)
command = Command('cmd', flags=Flag('port'))
input_flags = InputFlags(InputFlag('ssh', value='some2'))
self.assertEqual(router._validate_input_flags(command, input_flags), False)
def test_validate_incorrect_input_flag4(self):
router = Router()
router.set_invalid_input_flag_handler(lambda flag: None)
command = Command('cmd', flags=Flag('ssh', possible_values=False))
input_flags = InputFlags(InputFlag('ssh', value='some3'))
self.assertEqual(router._validate_input_flags(command, input_flags), False)
def test_validate_incorrect_input_flag5(self):
router = Router()
router.set_invalid_input_flag_handler(lambda flag: None)
command = Command('cmd', flags=Flag('ssh', possible_values=re.compile(r'some[1-5]$')))
input_flags = InputFlags(InputFlag('ssh', value='some40'))
self.assertEqual(router._validate_input_flags(command, input_flags), False)
def test_validate_incorrect_input_flag6(self):
router = Router()
router.set_invalid_input_flag_handler(lambda flag: None)
command = Command('cmd', flags=Flag('ssh', possible_values=['example']))
input_flags = InputFlags(InputFlag('ssh', value='example2'))
self.assertEqual(router._validate_input_flags(command, input_flags), False)
def test_validate_incorrect_input_flag7(self):
router = Router()
router.set_invalid_input_flag_handler(lambda flag: None)
command = Command('cmd', flags=Flag('ssh', possible_values=['example']))
input_flags = InputFlags(InputFlag('ssh'))
self.assertEqual(router._validate_input_flags(command, input_flags), False)
def test_validate_correct_input_flag1(self):
command = Command('cmd', flags=Flag('port'))
input_flags = InputFlags(InputFlag('port', value='some2'))
self.assertEqual(Router()._validate_input_flags(command, input_flags), True)
def test_validate_correct_input_flag2(self):
command = Command('cmd', flags=Flag('port', possible_values=['some2', 'some3']))
input_flags = InputFlags(InputFlag('port', value='some2'))
self.assertEqual(Router()._validate_input_flags(command, input_flags), True)
def test_validate_correct_input_flag3(self):
command = Command('cmd', flags=Flag('ssh', possible_values=re.compile(r'more[1-5]$')))
input_flags = InputFlags(InputFlag('ssh', value='more5'))
self.assertEqual(Router()._validate_input_flags(command, input_flags), True)
def test_validate_correct_input_flag4(self):
command = Command('cmd', flags=Flag('ssh', possible_values=False))
input_flags = InputFlags(InputFlag('ssh'))
self.assertEqual(Router()._validate_input_flags(command, input_flags), True)
def test_validate_incorrect_func_args1(self):
command = Command('cmd', flags=Flag('port'))
def handler():
pass
with self.assertRaises(RequiredArgumentNotPassedException):
Router()._validate_func_args(command, handler)
def test_validate_incorrect_func_args2(self):
command = Command('cmd', flags=Flag('port'))
def handler(args, kwargs):
pass
with self.assertRaises(TooManyTransferredArgsException):
Router()._validate_func_args(command, handler)
def test_validate_incorrect_func_args3(self):
command = Command('cmd')
def handler(args):
pass
with self.assertRaises(TooManyTransferredArgsException):
Router()._validate_func_args(command, handler)
def test_get_router_aliases(self):
router = Router()
@router.command(Command('some', aliases=['test', 'case']))
def handler():
pass
self.assertListEqual(router.get_aliases(), ['test', 'case'])
def test_get_router_aliases2(self):
router = Router()
@router.command(Command('some', aliases=['test', 'case']))
def handler():
pass
@router.command(Command('ext', aliases=['more', 'foo']))
def handler2():
pass
self.assertListEqual(router.get_aliases(), ['test', 'case', 'more', 'foo'])
def test_get_router_aliases3(self):
router = Router()
@router.command(Command('some'))
def handler():
pass
self.assertListEqual(router.get_aliases(), [])