25 Commits

Author SHA1 Message Date
kolo 05379712f4 some fix 2025-05-09 23:33:54 +03:00
kolo ed1cbf0fcf stable version 2025-05-09 23:27:34 +03:00
kolo 471f05369b Merge branch 'dev' of https://github.com/koloideal/Argenta into dev 2025-05-09 23:25:47 +03:00
kolo 13f7e33db1 ruff format 2025-05-09 23:25:21 +03:00
kolo 9a78aa9263 ruff check 2025-05-09 23:24:12 +03:00
kolo 58ccd6b26d Update README.md 2025-05-09 23:20:45 +03:00
kolo 73144f7ba4 Update README.md 2025-05-09 23:19:07 +03:00
kolo 650f4c9036 some fix 2025-05-09 18:23:08 +03:00
kolo 393f5c7d81 fix 2025-05-08 23:43:08 +03:00
kolo 9eb2bb6c46 new imgs 2025-05-08 23:37:42 +03:00
kolo 79b275eac7 some fix 2025-05-08 01:08:11 +03:00
kolo 07ac2af71e some fix 2025-05-07 19:14:19 +03:00
kolo c4b3aa7db8 working 2025-05-07 02:15:42 +03:00
kolo 61ef6a6466 all tests passed 2025-05-06 21:53:53 +03:00
kolo 477f3a7dec starting refactor tests 2025-05-04 16:40:10 +03:00
kolo adf3431388 more beautiful typehints warning 2025-05-04 03:08:54 +03:00
kolo 83955aa046 first beta - adding hints for similar commands, now - feature freezing 2025-05-04 02:13:05 +03:00
kolo 5a17e916eb work on stable major version 2025-04-30 15:48:38 +03:00
kolo 1159dda16e work on Response model 2025-04-30 00:08:49 +03:00
kolo 315508a36e work on Response 2025-04-29 20:40:47 +03:00
kolo 9d6598c4e0 work on Response model 2025-04-29 00:07:32 +03:00
kolo eb43806da6 new model - Response 2025-04-28 02:21:34 +03:00
kolo e076dbf84f new img in docs 2025-04-27 23:43:14 +03:00
kolo 2f090b6b47 new img in docs 2025-04-27 23:42:56 +03:00
kolo c9dbf2bbae fix print framed text with static dividing line 2025-04-27 23:27:08 +03:00
45 changed files with 1044 additions and 1877 deletions
+1 -1
View File
@@ -27,4 +27,4 @@ jobs:
pip install ruff pip install ruff
- name: Run linter - name: Run linter
run: ruff check ./argenta run: ruff check ./src
+1 -1
View File
@@ -1,4 +1,4 @@
.venv *venv
.idea .idea
dist dist
uv.lock uv.lock
+10 -1279
View File
File diff suppressed because it is too large Load Diff
Binary file not shown.

After

Width:  |  Height:  |  Size: 1.5 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.9 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 928 KiB

+46
View File
@@ -0,0 +1,46 @@
<?xml version="1.0" standalone="no"?>
<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 20010904//EN"
"http://www.w3.org/TR/2001/REC-SVG-20010904/DTD/svg10.dtd">
<svg version="1.0" xmlns="http://www.w3.org/2000/svg"
width="809.000000pt" height="809.000000pt" viewBox="0 0 809.000000 809.000000"
preserveAspectRatio="xMidYMid meet">
<g transform="translate(0.000000,809.000000) scale(0.100000,-0.100000)"
fill="#000000" stroke="none">
<path d="M3845 7600 c-683 -38 -1333 -259 -1875 -639 -702 -492 -1223 -1251
-1434 -2089 -83 -330 -111 -602 -103 -980 6 -273 19 -393 67 -632 167 -829
642 -1593 1321 -2122 512 -399 1133 -656 1794 -745 165 -22 666 -25 825 -5
779 99 1451 397 2020 898 687 603 1095 1401 1201 2344 17 147 17 571 0 715
-59 518 -204 977 -442 1402 -180 322 -355 552 -629 823 -518 515 -1164 850
-1887 979 -254 45 -594 66 -858 51z m655 -234 c545 -76 1005 -248 1445 -541
195 -130 336 -245 506 -415 457 -456 758 -987 909 -1605 134 -547 132 -1119
-5 -1665 -149 -593 -445 -1112 -885 -1550 -624 -623 -1414 -966 -2315 -1006
-412 -18 -935 73 -1339 232 -656 259 -1228 720 -1610 1299 -298 450 -461 890
-538 1451 -28 210 -31 620 -4 819 103 786 437 1477 975 2016 149 149 266 249
417 357 315 225 692 405 1059 506 211 58 342 81 675 120 87 10 601 -3 710 -18z"/>
<path d="M3691 6759 c-231 -17 -522 -67 -660 -114 -227 -77 -354 -211 -381
-400 -14 -105 -13 -628 2 -643 9 -9 158 -12 611 -12 598 0 628 -2 649 -34 14
-21 8 -66 -12 -86 -20 -20 -33 -20 -908 -20 -857 0 -892 -1 -967 -20 -294 -75
-500 -321 -599 -715 -49 -195 -61 -309 -60 -585 0 -221 3 -272 23 -385 44
-251 116 -418 232 -540 78 -82 143 -123 254 -162 78 -27 85 -27 351 -31 l272
-4 16 23 c14 20 16 58 16 259 0 334 20 446 107 613 80 153 268 310 447 374
128 45 138 46 881 52 683 7 713 8 786 28 201 56 353 177 437 348 80 162 86
240 74 905 -11 608 -10 600 -84 743 -80 153 -240 271 -453 332 -252 73 -653
101 -1034 74z m-383 -466 c126 -78 147 -245 44 -355 -144 -154 -396 -54 -395
157 0 97 62 187 154 222 54 20 143 10 197 -24z"/>
<path d="M5451 5491 c-20 -20 -21 -30 -21 -283 0 -276 -7 -351 -42 -458 -81
-251 -281 -454 -523 -534 -149 -48 -151 -48 -890 -56 -678 -6 -703 -7 -780
-28 -139 -38 -219 -84 -315 -181 -61 -61 -95 -105 -117 -151 -65 -134 -64
-129 -75 -780 -12 -665 -10 -694 47 -817 122 -265 419 -429 895 -494 127 -18
592 -18 730 -1 377 48 646 169 785 355 57 75 67 94 96 182 20 64 23 94 27 325
4 238 3 257 -15 278 l-18 23 -551 -1 c-399 0 -559 3 -577 11 -51 23 -56 102
-8 126 9 4 439 10 956 13 1044 7 981 2 1125 75 178 89 305 255 386 502 84 257
111 618 73 973 -39 364 -167 655 -351 801 -63 50 -121 80 -213 110 -64 20 -93
23 -337 27 -260 4 -267 3 -287 -17z m-2104 -1698 c78 -52 531 -413 561 -447
17 -19 33 -48 37 -65 15 -71 3 -84 -323 -356 -275 -229 -310 -255 -342 -255
-78 1 -112 75 -61 134 9 10 86 75 171 144 247 201 355 294 358 311 1 9 -15 28
-38 44 -127 89 -503 393 -512 413 -23 50 20 103 82 104 15 0 45 -12 67 -27z
m1388 -1244 c59 -15 129 -77 151 -134 36 -96 0 -204 -88 -262 -36 -24 -51 -28
-118 -28 -67 0 -82 4 -118 28 -153 101 -124 338 48 392 57 17 71 18 125 4z"/>
</g>
</svg>

After

Width:  |  Height:  |  Size: 2.9 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 50 KiB

+23 -3
View File
@@ -1,7 +1,27 @@
from mock.mock_app.handlers.routers import work_router
from argenta.app import App from argenta.app import App
from argenta.app.defaults import PredefinedMessages
from argenta.app.autocompleter import AutoCompleter
from argenta.orchestrator import Orchestrator from argenta.orchestrator import Orchestrator
from argenta.orchestrator.argparser import ArgParser
from argenta.orchestrator.argparser.arguments import BooleanArgument
app = App(repeat_command_groups=True)
orchestrator = Orchestrator() arg_parser = ArgParser(processed_args=[BooleanArgument("repeat")])
orchestrator.start_polling(app) app: App = App(autocompleter=AutoCompleter(".hist"))
orchestrator: Orchestrator = Orchestrator()
def main():
app.include_router(work_router)
app.add_message_on_startup(PredefinedMessages.USAGE)
app.add_message_on_startup(PredefinedMessages.AUTOCOMPLETE)
app.add_message_on_startup(PredefinedMessages.HELP)
orchestrator.start_polling(app)
if __name__ == "__main__":
main()
+87 -4
View File
@@ -1,6 +1,89 @@
from argenta.app import App from argenta.app import App
from argenta.command.models import InputCommand from argenta.app.autocompleter import AutoCompleter
from argenta.router import Router
from argenta.command import Command
from argenta.orchestrator import Orchestrator
from argenta.app.dividing_line import DynamicDividingLine
from argenta.response import Response
import platform
import psutil
import os
import subprocess
import socket
app = App() # Маршрутизатор для работы с файлами
app._all_registered_triggers_in_lower = ['fr', 'Tre', 'Pre'] file_router = Router("Файловые операции")
print(app._is_unknown_command(InputCommand('fr')))
@file_router.command(Command("list", "Список файлов"))
def list_files(response: Response):
files = os.listdir()
for file in files:
print(file)
@file_router.command(Command("size", "Размер файла"))
def file_size(response: Response):
file_name = input("Введите имя файла: ")
if os.path.exists(file_name):
size = os.path.getsize(file_name)
print(f"Размер файла {file_name}: {size} байт")
else:
print(f"Файл {file_name} не найден")
# Маршрутизатор для системных операций
system_router = Router("Системные операции")
@system_router.command(Command("info", "Информация о системе"))
def system_info(response: Response):
print(f"Система: {platform.system()}")
print(f"Версия: {platform.version()}")
print(f"Архитектура: {platform.architecture()}")
print(f"Процессор: {platform.processor()}")
@system_router.command(Command("memory", "Информация о памяти"))
def memory_info(response: Response):
memory = psutil.virtual_memory()
print(f"Всего памяти: {memory.total / (1024**3):.2f} ГБ")
print(f"Доступно: {memory.available / (1024**3):.2f} ГБ")
print(f"Использовано: {memory.used / (1024**3):.2f} ГБ ({memory.percent}%)")
# Маршрутизатор для сетевых операций
network_router = Router("Сетевые операции")
@network_router.command(Command("ping", "Проверка доступности хоста"))
def ping_host(response: Response):
host = input("Введите имя хоста: ")
print(f"Пингую {host}...")
subprocess.run(["ping", "-c", "4", host])
@network_router.command(Command("ip", "Показать IP-адреса"))
def show_ip(response: Response):
hostname = socket.gethostname()
print(f"Имя хоста: {hostname}")
print(f"IP-адрес: {socket.gethostbyname(hostname)}")
# Создание приложения и регистрация маршрутизаторов
app = App(
prompt="System> ",
initial_message="Pingator",
dividing_line=DynamicDividingLine("*"),
autocompleter=AutoCompleter(".hist", "e"),
)
# Добавляем все маршрутизаторы
app.include_routers(file_router, system_router, network_router)
# Добавляем сообщение при запуске
app.add_message_on_startup("Для просмотра доступных команд нажмите Enter")
# Запускаем приложение
orchestrator = Orchestrator()
orchestrator.start_polling(app)
@@ -5,6 +5,8 @@ console = Console()
def help_command(): def help_command():
console.print("[italic bold]The main functionality of the script is to convert an expression from a string " console.print(
"to a mathematical one and then calculate this expression. " "[italic bold]The main functionality of the script is to convert an expression from a string "
"Project GitHub: https://github.com/koloideal/WordMath[/italic bold]") "to a mathematical one and then calculate this expression. "
"Project GitHub: https://github.com/koloideal/WordMath[/italic bold]"
)
+23 -10
View File
@@ -1,22 +1,35 @@
from rich.console import Console from rich.console import Console
from argenta.command import Command from argenta.command import Command
from argenta.command.flag.defaults import PredefinedFlags
from argenta.command.flags import Flags
from argenta.response import Response
from argenta.router import Router from argenta.router import Router
work_router: Router = Router(title='Work points:') work_router: Router = Router(title="Work points:")
console = Console() console = Console()
@work_router.command(Command('get', 'Get Help', aliases=['help', 'Get_help'])) @work_router.command(
def command_help(): Command(
pass "get",
"Get Help",
aliases=["help", "Get_help"],
@work_router.command(Command('run', 'Run All')) flags=Flags(PredefinedFlags.PORT, PredefinedFlags.HOST),
def command_start_solving(): )
pass )
def command_help(response: Response):
print(response.status)
print(response.undefined_flags.get_flags())
print(response.valid_flags.get_flags())
print(response.invalid_value_flags.get_flags())
@work_router.command("run")
def command_start_solving(response: Response):
print(response.status)
print(response.undefined_flags.get_flags())
print(response.valid_flags.get_flags())
print(response.invalid_value_flags.get_flags())
+7 -3
View File
@@ -9,9 +9,12 @@ from argenta.orchestrator.argparser import ArgParser
from argenta.orchestrator.argparser.arguments import BooleanArgument from argenta.orchestrator.argparser.arguments import BooleanArgument
arg_parser = ArgParser(processed_args=[BooleanArgument('repeat')]) arg_parser = ArgParser(processed_args=[BooleanArgument("repeat")])
app: App = App(dividing_line=DynamicDividingLine(), app: App = App(
autocompleter=AutoCompleter('./mock/.hist')) dividing_line=DynamicDividingLine(),
autocompleter=AutoCompleter(),
repeat_command_groups=False,
)
orchestrator: Orchestrator = Orchestrator(arg_parser) orchestrator: Orchestrator = Orchestrator(arg_parser)
@@ -24,5 +27,6 @@ def main():
orchestrator.start_polling(app) orchestrator.start_polling(app)
if __name__ == "__main__": if __name__ == "__main__":
main() main()
+3 -8
View File
@@ -1,7 +1,7 @@
[project] [project]
name = "argenta" name = "argenta"
version = "1.0.0-alpha4" version = "1.0.1"
description = "Python library for creating TUI" description = "Python library for building modular CLI applications"
authors = [{ name = "kolo", email = "kolo.is.main@gmail.com" }] authors = [{ name = "kolo", email = "kolo.is.main@gmail.com" }]
requires-python = ">=3.11, <4.0" requires-python = ">=3.11, <4.0"
readme = "README.md" readme = "README.md"
@@ -9,12 +9,7 @@ license = { text = "MIT" }
dependencies = [ dependencies = [
"rich (>=14.0.0,<15.0.0)", "rich (>=14.0.0,<15.0.0)",
"art (>=6.4,<7.0)", "art (>=6.4,<7.0)",
"pyreadline3 (>=3.5.4,<4.0.0)", "pyreadline3>=3.5.4",
]
[dependency-groups]
dev = [
"pydoc-markdown>=4.8.2,<5",
] ]
[tool.ruff] [tool.ruff]
+30 -11
View File
@@ -1,9 +1,12 @@
import os import os
import readline import readline
from typing import Never
class AutoCompleter: class AutoCompleter:
def __init__(self, history_filename: str = False, autocomplete_button: str = 'tab') -> None: def __init__(
self, history_filename: str = False, autocomplete_button: str = "tab"
) -> None:
""" """
Public. Configures and implements auto-completion of input command 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 history_filename: the name of the file for saving the history of the autocompleter
@@ -12,7 +15,6 @@ class AutoCompleter:
""" """
self.history_filename = history_filename self.history_filename = history_filename
self.autocomplete_button = autocomplete_button self.autocomplete_button = autocomplete_button
self.matches: list[str] = []
def _complete(self, text, state) -> str | None: def _complete(self, text, state) -> str | None:
""" """
@@ -21,16 +23,22 @@ class AutoCompleter:
:param state: the current cursor position is relative to the beginning of the line :param state: the current cursor position is relative to the beginning of the line
:return: the desired candidate as str or None :return: the desired candidate as str or None
""" """
matches: list[str] = sorted(cmd for cmd in self.get_history_items() if cmd.startswith(text)) matches: list[str] = sorted(
cmd for cmd in self.get_history_items() if cmd.startswith(text)
)
if len(matches) > 1: if len(matches) > 1:
common_prefix = matches[0] common_prefix = matches[0]
for match in matches[1:]: for match in matches[1:]:
i = 0 i = 0
while i < len(common_prefix) and i < len(match) and common_prefix[i] == match[i]: while (
i < len(common_prefix)
and i < len(match)
and common_prefix[i] == match[i]
):
i += 1 i += 1
common_prefix = common_prefix[:i] common_prefix = common_prefix[:i]
if state == 0: if state == 0:
readline.insert_text(common_prefix[len(text):]) readline.insert_text(common_prefix[len(text) :])
readline.redisplay() readline.redisplay()
return None return None
elif len(matches) == 1: elif len(matches) == 1:
@@ -52,21 +60,32 @@ class AutoCompleter:
readline.add_history(line) readline.add_history(line)
readline.set_completer(self._complete) readline.set_completer(self._complete)
readline.set_completer_delims(readline.get_completer_delims().replace(' ', '')) readline.set_completer_delims(readline.get_completer_delims().replace(" ", ""))
readline.parse_and_bind(f'{self.autocomplete_button}: complete') readline.parse_and_bind(f"{self.autocomplete_button}: complete")
def exit_setup(self) -> None: def exit_setup(self, all_commands: list[str]) -> None:
""" """
Private. Exit setup function Private. Exit setup function
:return: None :return: None
""" """
if self.history_filename: if self.history_filename:
readline.write_history_file(self.history_filename) readline.write_history_file(self.history_filename)
with open(self.history_filename, "r") as history_file:
raw_history = history_file.read()
pretty_history: list[str] = []
for line in set(raw_history.strip().split("\n")):
if line.split()[0] in all_commands:
pretty_history.append(line)
with open(self.history_filename, "w") as history_file:
history_file.write("\n".join(pretty_history))
@staticmethod @staticmethod
def get_history_items() -> list[str] | list: def get_history_items() -> list[str] | list[Never]:
""" """
Private. Returns a list of all commands entered by the user Private. Returns a list of all commands entered by the user
:return: all commands entered by the user as list[str] :return: all commands entered by the user as list[str] | list[Never]
""" """
return [readline.get_history_item(i) for i in range(1, readline.get_current_history_length() + 1)] return [
readline.get_history_item(i)
for i in range(1, readline.get_current_history_length() + 1)
]
+3 -3
View File
@@ -6,7 +6,7 @@ class PredefinedMessages:
""" """
Public. A dataclass with predetermined messages for quick use 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>'
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>"
+8 -10
View File
@@ -2,7 +2,7 @@ from abc import ABC
class BaseDividingLine(ABC): class BaseDividingLine(ABC):
def __init__(self, unit_part: str = '-') -> None: def __init__(self, unit_part: str = "-") -> None:
""" """
Private. The basic dividing line Private. The basic dividing line
:param unit_part: the single part of the dividing line :param unit_part: the single part of the dividing line
@@ -16,13 +16,13 @@ class BaseDividingLine(ABC):
:return: unit_part of dividing line as str :return: unit_part of dividing line as str
""" """
if len(self._unit_part) == 0: if len(self._unit_part) == 0:
return ' ' return " "
else: else:
return self._unit_part[0] return self._unit_part[0]
class StaticDividingLine(BaseDividingLine): class StaticDividingLine(BaseDividingLine):
def __init__(self, unit_part: str = '-', length: int = 25) -> None: def __init__(self, unit_part: str = "-", length: int = 25) -> None:
""" """
Public. The static dividing line Public. The static dividing line
:param unit_part: the single part of the dividing line :param unit_part: the single part of the dividing line
@@ -39,13 +39,13 @@ class StaticDividingLine(BaseDividingLine):
:return: full line of dividing line as str :return: full line of dividing line as str
""" """
if is_override: if is_override:
return f'\n{self.length * self.get_unit_part()}\n' return f"\n{self.length * self.get_unit_part()}\n"
else: else:
return f'\n[dim]{self.length * self.get_unit_part()}[/dim]\n' return f"\n[dim]{self.length * self.get_unit_part()}[/dim]\n"
class DynamicDividingLine(BaseDividingLine): class DynamicDividingLine(BaseDividingLine):
def __init__(self, unit_part: str = '-') -> None: def __init__(self, unit_part: str = "-") -> None:
""" """
Public. The dynamic dividing line Public. The dynamic dividing line
:param unit_part: the single part of the dividing line :param unit_part: the single part of the dividing line
@@ -61,8 +61,6 @@ class DynamicDividingLine(BaseDividingLine):
:return: full line of dividing line as str :return: full line of dividing line as str
""" """
if is_override: if is_override:
return f'\n{length * self.get_unit_part()}\n' return f"\n{length * self.get_unit_part()}\n"
else: else:
return f'\n[dim]{self.get_unit_part() * length}[/dim]\n' return f"\n[dim]{self.get_unit_part() * length}[/dim]\n"
+199 -111
View File
@@ -11,28 +11,31 @@ from argenta.router import Router
from argenta.router.defaults import system_router from argenta.router.defaults import system_router
from argenta.app.autocompleter import AutoCompleter from argenta.app.autocompleter import AutoCompleter
from argenta.app.dividing_line.models import StaticDividingLine, DynamicDividingLine from argenta.app.dividing_line.models import StaticDividingLine, DynamicDividingLine
from argenta.command.exceptions import (UnprocessedInputFlagException, from argenta.command.exceptions import (
RepeatedInputFlagsException, UnprocessedInputFlagException,
EmptyInputCommandException, RepeatedInputFlagsException,
BaseInputCommandException) EmptyInputCommandException,
BaseInputCommandException,
)
from argenta.app.registered_routers.entity import RegisteredRouters from argenta.app.registered_routers.entity import RegisteredRouters
from argenta.response import Response
class BaseApp: class BaseApp:
def __init__(self, def __init__(
prompt: str, self,
initial_message: str, prompt: str,
farewell_message: str, initial_message: str,
exit_command: Command, farewell_message: str,
system_router_title: str | None, exit_command: Command,
ignore_command_register: bool, system_router_title: str | None,
dividing_line: StaticDividingLine | DynamicDividingLine, ignore_command_register: bool,
repeat_command_groups: bool, dividing_line: StaticDividingLine | DynamicDividingLine,
override_system_messages: bool, repeat_command_groups: bool,
autocompleter: AutoCompleter, override_system_messages: bool,
print_func: Callable[[str], None]) -> None: autocompleter: AutoCompleter,
print_func: Callable[[str], None],
) -> None:
self._prompt = prompt self._prompt = prompt
self._print_func = print_func self._print_func = print_func
self._exit_command = exit_command self._exit_command = exit_command
@@ -46,20 +49,30 @@ class BaseApp:
self._farewell_message = farewell_message self._farewell_message = farewell_message
self._initial_message = initial_message self._initial_message = initial_message
self._description_message_gen: Callable[[str, str], str] = (
self._description_message_gen: Callable[[str, str], str] = lambda command, description: f'[{command}] *=*=* {description}' lambda command, description: f"[{command}] *=*=* {description}"
)
self._registered_routers: RegisteredRouters = RegisteredRouters() self._registered_routers: RegisteredRouters = RegisteredRouters()
self._messages_on_startup: list[str] = [] self._messages_on_startup: list[str] = []
self._all_registered_triggers_in_lower: list[str] = [] self._all_registered_triggers_in_lower: list[str] = []
self._all_registered_triggers_in_default_case: 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._incorrect_input_syntax_handler: Callable[[str], None] = (
self._repeated_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'Repeated input flags: {raw_command}') lambda raw_command: print_func(f"Incorrect flag syntax: {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._repeated_input_flags_handler: Callable[[str], None] = (
self._exit_command_handler: Callable[[], None] = lambda: print_func(self._farewell_message) 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[[Response], None] = (
lambda response: print_func(self._farewell_message)
)
def set_description_message_pattern(self, _: Callable[[str, str], str]) -> None: def set_description_message_pattern(self, _: Callable[[str, str], str]) -> None:
""" """
@@ -69,15 +82,13 @@ class BaseApp:
""" """
self._description_message_gen: Callable[[str, str], str] = _ self._description_message_gen: Callable[[str, str], str] = _
def set_incorrect_input_syntax_handler(self, _: Callable[[str], None]) -> None:
def set_invalid_input_flags_handler(self, _: Callable[[str], None]) -> None:
""" """
Public. Sets the handler for incorrect flags when entering a command Public. Sets the handler for incorrect flags when entering a command
:param _: handler for incorrect flags when entering a command :param _: handler for incorrect flags when entering a command
:return: None :return: None
""" """
self._invalid_input_flags_handler = _ self._incorrect_input_syntax_handler = _
def set_repeated_input_flags_handler(self, _: Callable[[str], None]) -> None: def set_repeated_input_flags_handler(self, _: Callable[[str], None]) -> None:
""" """
@@ -87,7 +98,6 @@ class BaseApp:
""" """
self._repeated_input_flags_handler = _ self._repeated_input_flags_handler = _
def set_unknown_command_handler(self, _: Callable[[str], None]) -> None: def set_unknown_command_handler(self, _: Callable[[str], None]) -> None:
""" """
Public. Sets the handler for unknown commands when entering a command Public. Sets the handler for unknown commands when entering a command
@@ -96,7 +106,6 @@ class BaseApp:
""" """
self._unknown_command_handler = _ self._unknown_command_handler = _
def set_empty_command_handler(self, _: Callable[[], None]) -> None: def set_empty_command_handler(self, _: Callable[[], None]) -> None:
""" """
Public. Sets the handler for empty commands when entering a command Public. Sets the handler for empty commands when entering a command
@@ -105,7 +114,6 @@ class BaseApp:
""" """
self._empty_input_command_handler = _ self._empty_input_command_handler = _
def set_exit_command_handler(self, _: Callable[[], None]) -> None: def set_exit_command_handler(self, _: Callable[[], None]) -> None:
""" """
Public. Sets the handler for exit command when entering a command Public. Sets the handler for exit command when entering a command
@@ -114,21 +122,22 @@ class BaseApp:
""" """
self._exit_command_handler = _ self._exit_command_handler = _
def _print_command_group_description(self) -> None: def _print_command_group_description(self) -> None:
""" """
Private. Prints the description of the available commands Private. Prints the description of the available commands
:return: None :return: None
""" """
for registered_router in self._registered_routers: for registered_router in self._registered_routers:
if registered_router.get_title(): if registered_router.title:
self._print_func(registered_router.get_title()) self._print_func(registered_router.title)
for command_handler in registered_router.get_command_handlers(): for command_handler in registered_router.get_command_handlers():
self._print_func(self._description_message_gen( self._print_func(
self._description_message_gen(
command_handler.get_handled_command().get_trigger(), command_handler.get_handled_command().get_trigger(),
command_handler.get_handled_command().get_description())) command_handler.get_handled_command().get_description(),
self._print_func('') )
)
self._print_func("")
def _print_framed_text(self, text: str) -> None: def _print_framed_text(self, text: str) -> None:
""" """
@@ -137,19 +146,36 @@ class BaseApp:
:return: None :return: None
""" """
if isinstance(self._dividing_line, StaticDividingLine): if isinstance(self._dividing_line, StaticDividingLine):
self._print_func(self._dividing_line.get_full_static_line(self._override_system_messages)) self._print_func(
self._print_func(text) self._dividing_line.get_full_static_line(self._override_system_messages)
self._print_func(self._dividing_line.get_full_static_line(self._override_system_messages)) )
print(text.strip("\n"))
self._print_func(
self._dividing_line.get_full_static_line(self._override_system_messages)
)
elif isinstance(self._dividing_line, DynamicDividingLine): elif isinstance(self._dividing_line, DynamicDividingLine):
clear_text = re.sub(r'\u001b\[[0-9;]*m', '', text) 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([len(line) for line in clear_text.split("\n")])
max_length_line = max_length_line if 10 <= max_length_line <= 80 else 80 if max_length_line > 80 else 10 max_length_line = (
max_length_line
self._print_func(self._dividing_line.get_full_dynamic_line(max_length_line, self._override_system_messages)) if 10 <= max_length_line <= 80
print(text.strip('\n')) else 80
self._print_func(self._dividing_line.get_full_dynamic_line(max_length_line, self._override_system_messages)) 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: def _is_exit_command(self, command: InputCommand) -> bool:
""" """
@@ -158,9 +184,14 @@ class BaseApp:
:return: is it an exit command or not as bool :return: is it an exit command or not as bool
""" """
if self._ignore_command_register: if self._ignore_command_register:
if command.get_trigger().lower() == self._exit_command.get_trigger().lower(): if (
command.get_trigger().lower()
== self._exit_command.get_trigger().lower()
):
return True return True
elif command.get_trigger().lower() in [x.lower() for x in self._exit_command.get_aliases()]: elif command.get_trigger().lower() in [
x.lower() for x in self._exit_command.get_aliases()
]:
return True return True
else: else:
if command.get_trigger() == self._exit_command.get_trigger(): if command.get_trigger() == self._exit_command.get_trigger():
@@ -169,7 +200,6 @@ class BaseApp:
return True return True
return False return False
def _is_unknown_command(self, command: InputCommand) -> bool: def _is_unknown_command(self, command: InputCommand) -> bool:
""" """
Private. Checks if the given command is an unknown command Private. Checks if the given command is an unknown command
@@ -185,8 +215,9 @@ class BaseApp:
return False return False
return True return True
def _error_handler(
def _error_handler(self, error: BaseInputCommandException, raw_command: str) -> None: self, error: BaseInputCommandException, raw_command: str
) -> None:
""" """
Private. Handles parsing errors of the entered command Private. Handles parsing errors of the entered command
:param error: error being handled :param error: error being handled
@@ -195,89 +226,143 @@ class BaseApp:
""" """
match error: match error:
case UnprocessedInputFlagException(): case UnprocessedInputFlagException():
self._invalid_input_flags_handler(raw_command) self._incorrect_input_syntax_handler(raw_command)
case RepeatedInputFlagsException(): case RepeatedInputFlagsException():
self._repeated_input_flags_handler(raw_command) self._repeated_input_flags_handler(raw_command)
case EmptyInputCommandException(): case EmptyInputCommandException():
self._empty_input_command_handler() self._empty_input_command_handler()
def _setup_system_router(self) -> None: def _setup_system_router(self) -> None:
""" """
Private. Sets up system router Private. Sets up system router
:return: None :return: None
""" """
system_router.set_title(self._system_router_title) system_router.title = self._system_router_title
@system_router.command(self._exit_command) @system_router.command(self._exit_command)
def exit_command(): def exit_command(response: Response) -> None:
self._exit_command_handler() self._exit_command_handler(response)
if system_router not in self._registered_routers.get_registered_routers(): if system_router not in self._registered_routers.get_registered_routers():
system_router.set_command_register_ignore(self._ignore_command_register) system_router.set_command_register_ignore(self._ignore_command_register)
self._registered_routers.add_registered_router(system_router) self._registered_routers.add_registered_router(system_router)
def _most_similar_command(self, unknown_command: str) -> str | None:
all_commands = (
self._all_registered_triggers_in_lower
if self._ignore_command_register
else self._all_registered_triggers_in_default_case
)
matches: list[str] | list = sorted(
cmd for cmd in all_commands if cmd.startswith(unknown_command)
)
if not matches:
matches: list[str] | list = sorted(
cmd for cmd in all_commands if unknown_command.startswith(cmd)
)
if len(matches) == 1:
return matches[0]
elif len(matches) > 1:
return sorted(matches, key=lambda cmd: len(cmd))[0]
else:
return None
def _setup_default_view(self) -> None: def _setup_default_view(self) -> None:
""" """
Private. Sets up default app view Private. Sets up default app view
:return: None :return: None
""" """
if not self._override_system_messages: self._prompt = "[italic dim bold]What do you want to do?\n"
self._initial_message = f'\n[bold red]{text2art(self._initial_message, font="tarty1")}\n\n' self._initial_message = (
self._farewell_message = (f'[bold red]\n{text2art(f"\n{self._farewell_message}\n", font="chanky")}[/bold red]\n' f"\n[bold red]{text2art(self._initial_message, font='tarty1')}\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] ' self._farewell_message = (
f'[blue dim]*=*=*[/blue dim] ' f"[bold red]\n{text2art(f'\n{self._farewell_message}\n', font='chanky')}[/bold red]\n"
f'[bold yellow italic]{escape(description)}') f"[red i]github.com/koloideal/Argenta[/red i] | [red bold i]made by kolo[/red bold i]\n"
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._description_message_gen = lambda command, description: (
self._empty_input_command_handler = lambda: self._print_func('[red bold]Empty input command') f"[bold red]{escape('[' + command + ']')}[/bold red] "
self._unknown_command_handler = lambda command: self._print_func(f"[red bold]Unknown command: {escape(command.get_trigger())}") f"[blue dim]*=*=*[/blue dim] "
f"[bold yellow italic]{escape(description)}"
)
self._incorrect_input_syntax_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"
)
def unknown_command_handler(command: InputCommand) -> None:
cmd_trg: str = command.get_trigger()
mst_sim_cmd: str | None = self._most_similar_command(cmd_trg)
first_part_of_text = (
f"[red]Unknown command:[/red] [blue]{escape(cmd_trg)}[/blue]"
)
second_part_of_text = (
("[red], most similar:[/red] " + ("[blue]" + mst_sim_cmd + "[/blue]"))
if mst_sim_cmd
else ""
)
self._print_func(first_part_of_text + second_part_of_text)
self._unknown_command_handler = unknown_command_handler
def _pre_cycle_setup(self) -> None: def _pre_cycle_setup(self) -> None:
""" """
Private. Configures various aspects of the application before the start of the cycle Private. Configures various aspects of the application before the start of the cycle
:return: None :return: None
""" """
self._setup_default_view()
self._setup_system_router() self._setup_system_router()
for router_entity in self._registered_routers: 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(
self._all_registered_triggers_in_default_case.extend(router_entity.get_aliases()) 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(
self._all_registered_triggers_in_lower.extend([x.lower() for x in router_entity.get_aliases()]) [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._autocompleter.initial_setup(self._all_registered_triggers_in_lower)
if not self._override_system_messages:
self._setup_default_view()
self._print_func(self._initial_message) self._print_func(self._initial_message)
for message in self._messages_on_startup: for message in self._messages_on_startup:
self._print_func(message) self._print_func(message)
if self._messages_on_startup: if self._messages_on_startup:
print('\n\n') print("\n")
if not self._repeat_command_groups_description: if not self._repeat_command_groups_description:
self._print_command_group_description() self._print_command_group_description()
class App(BaseApp): class App(BaseApp):
def __init__(self, def __init__(
prompt: str = '[italic dim bold]What do you want to do?\n', self,
initial_message: str = '\nArgenta\n', prompt: str = "What do you want to do?\n",
farewell_message: str = '\nSee you\n', initial_message: str = "\nArgenta\n",
exit_command: Command = Command('Q', 'Exit command'), farewell_message: str = "\nSee you\n",
system_router_title: str | None = 'System points:', exit_command: Command = Command("Q", "Exit command"),
ignore_command_register: bool = True, system_router_title: str | None = "System points:",
dividing_line: StaticDividingLine | DynamicDividingLine = StaticDividingLine(), ignore_command_register: bool = True,
repeat_command_groups: bool = True, dividing_line: StaticDividingLine | DynamicDividingLine = StaticDividingLine(),
override_system_messages: bool = False, repeat_command_groups: bool = True,
autocompleter: AutoCompleter = AutoCompleter(), override_system_messages: bool = False,
print_func: Callable[[str], None] = Console().print) -> None: autocompleter: AutoCompleter = AutoCompleter(),
print_func: Callable[[str], None] = Console().print,
) -> None:
""" """
Public. The essence of the application itself. Public. The essence of the application itself.
Configures and manages all aspects of the behavior and presentation of the user interacting with the user Configures and manages all aspects of the behavior and presentation of the user interacting with the user
@@ -294,18 +379,19 @@ class App(BaseApp):
:param print_func: system messages text output function :param print_func: system messages text output function
:return: None :return: None
""" """
super().__init__(prompt=prompt, super().__init__(
initial_message=initial_message, prompt=prompt,
farewell_message=farewell_message, initial_message=initial_message,
exit_command=exit_command, farewell_message=farewell_message,
system_router_title=system_router_title, exit_command=exit_command,
ignore_command_register=ignore_command_register, system_router_title=system_router_title,
dividing_line=dividing_line, ignore_command_register=ignore_command_register,
repeat_command_groups=repeat_command_groups, dividing_line=dividing_line,
override_system_messages=override_system_messages, repeat_command_groups=repeat_command_groups,
autocompleter=autocompleter, override_system_messages=override_system_messages,
print_func=print_func) autocompleter=autocompleter,
print_func=print_func,
)
def run_polling(self) -> None: def run_polling(self) -> None:
""" """
@@ -320,7 +406,9 @@ class App(BaseApp):
raw_command: str = Console().input(self._prompt) raw_command: str = Console().input(self._prompt)
try: try:
input_command: InputCommand = InputCommand.parse(raw_command=raw_command) input_command: InputCommand = InputCommand.parse(
raw_command=raw_command
)
except BaseInputCommandException as error: except BaseInputCommandException as error:
with redirect_stdout(io.StringIO()) as f: with redirect_stdout(io.StringIO()) as f:
self._error_handler(error, raw_command) self._error_handler(error, raw_command)
@@ -330,7 +418,14 @@ class App(BaseApp):
if self._is_exit_command(input_command): if self._is_exit_command(input_command):
system_router.finds_appropriate_handler(input_command) system_router.finds_appropriate_handler(input_command)
self._autocompleter.exit_setup() if self._ignore_command_register:
self._autocompleter.exit_setup(
self._all_registered_triggers_in_lower
)
else:
self._autocompleter.exit_setup(
self._all_registered_triggers_in_default_case
)
return return
if self._is_unknown_command(input_command): if self._is_unknown_command(input_command):
@@ -346,10 +441,6 @@ class App(BaseApp):
res: str = f.getvalue() res: str = f.getvalue()
self._print_framed_text(res) self._print_framed_text(res)
if not self._repeat_command_groups_description:
self._print_func(self._prompt)
def include_router(self, router: Router) -> None: def include_router(self, router: Router) -> None:
""" """
Public. Registers the router in the application Public. Registers the router in the application
@@ -359,7 +450,6 @@ class App(BaseApp):
router.set_command_register_ignore(self._ignore_command_register) router.set_command_register_ignore(self._ignore_command_register)
self._registered_routers.add_registered_router(router) self._registered_routers.add_registered_router(router)
def include_routers(self, *routers: Router) -> None: def include_routers(self, *routers: Router) -> None:
""" """
Public. Registers the routers in the application Public. Registers the routers in the application
@@ -369,7 +459,6 @@ class App(BaseApp):
for router in routers: for router in routers:
self.include_router(router) self.include_router(router)
def add_message_on_startup(self, message: str) -> None: def add_message_on_startup(self, message: str) -> None:
""" """
Public. Adds a message that will be displayed when the application is launched Public. Adds a message that will be displayed when the application is launched
@@ -377,4 +466,3 @@ class App(BaseApp):
:return: None :return: None
""" """
self._messages_on_startup.append(message) self._messages_on_startup.append(message)
+10 -3
View File
@@ -1,10 +1,11 @@
from argenta.command.flag.models import InputFlag, Flag from argenta.command.flag.models import Flag, InputFlag
class BaseInputCommandException(Exception): class BaseInputCommandException(Exception):
""" """
Private. Base exception class for all exceptions raised when parse input command Private. Base exception class for all exceptions raised when parse input command
""" """
pass pass
@@ -12,6 +13,7 @@ class UnprocessedInputFlagException(BaseInputCommandException):
""" """
Private. Raised when an unprocessed input flag is detected Private. Raised when an unprocessed input flag is detected
""" """
def __str__(self): def __str__(self):
return "Unprocessed Input Flags" return "Unprocessed Input Flags"
@@ -20,16 +22,21 @@ class RepeatedInputFlagsException(BaseInputCommandException):
""" """
Private. Raised when repeated input flags are detected Private. Raised when repeated input flags are detected
""" """
def __init__(self, flag: Flag | InputFlag): def __init__(self, flag: Flag | InputFlag):
self.flag = flag self.flag = flag
def __str__(self): def __str__(self):
return ("Repeated Input Flags\n" return (
f"Duplicate flag was detected in the input: '{self.flag.get_string_entity()}'") "Repeated Input Flags\n"
f"Duplicate flag was detected in the input: '{self.flag.get_string_entity()}'"
)
class EmptyInputCommandException(BaseInputCommandException): class EmptyInputCommandException(BaseInputCommandException):
""" """
Private. Raised when an empty input command is detected Private. Raised when an empty input command is detected
""" """
def __str__(self): def __str__(self):
return "Input Command is empty" return "Input Command is empty"
+2 -2
View File
@@ -1,4 +1,4 @@
__all__ = ('InputFlags', 'InputFlag', 'Flag', 'Flags') __all__ = ["Flag", "InputFlag"]
from argenta.command.flag.models import InputFlags, InputFlag, Flags, Flag from argenta.command.flag.models import Flag, InputFlag
+17 -10
View File
@@ -8,17 +8,24 @@ class PredefinedFlags:
""" """
Public. A dataclass with predefined flags and most frequently used flags for quick use 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) HELP = Flag(name="help", possible_values=False)
SHORT_INFO = Flag(name='I', prefix='-', possible_values=False) SHORT_HELP = Flag(name="H", prefix="-", possible_values=False)
ALL = Flag(name='all', possible_values=False) INFO = Flag(name="info", possible_values=False)
SHORT_ALL = Flag(name='A', prefix='-', possible_values=False) SHORT_INFO = Flag(name="I", 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}$')) ALL = Flag(name="all", possible_values=False)
SHORT_HOST = Flag(name='H', prefix='-', possible_values=re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$')) SHORT_ALL = Flag(name="A", prefix="-", possible_values=False)
PORT = Flag(name='port', possible_values=re.compile(r'^\d{1,5}$')) HOST = Flag(
SHORT_PORT = Flag(name='P', prefix='-', possible_values=re.compile(r'^\d{1,5}$')) 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}$"))
+36 -133
View File
@@ -1,10 +1,8 @@
from typing import Literal, Pattern from typing import Literal, Pattern
from abc import ABC, abstractmethod
class BaseFlag(ABC): class BaseFlag:
def __init__(self, name: str, def __init__(self, name: str, prefix: Literal["-", "--", "---"] = "--") -> None:
prefix: Literal['-', '--', '---'] = '--') -> None:
""" """
Private. Base class for flags Private. Base class for flags
:param name: the name of the flag :param name: the name of the flag
@@ -36,43 +34,17 @@ class BaseFlag(ABC):
""" """
return self._prefix return self._prefix
def __eq__(self, other) -> bool:
return self.get_string_entity() == other.get_string_entity()
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): class Flag(BaseFlag):
def __init__(self, name: str, def __init__(
prefix: Literal['-', '--', '---'] = '--', self,
possible_values: list[str] | Pattern[str] | False = True) -> None: name: str,
prefix: Literal["-", "--", "---"] = "--",
possible_values: list[str] | Pattern[str] | False = True,
) -> None:
""" """
Public. The entity of the flag being registered for subsequent processing Public. The entity of the flag being registered for subsequent processing
:param name: The name of the flag :param name: The name of the flag
@@ -113,106 +85,37 @@ class Flag(BaseFlag):
return True return True
class InputFlag(BaseFlag):
class BaseFlags(ABC): def __init__(
""" self, name: str, prefix: Literal["-", "--", "---"] = "--", value: str = None
Private. Base class for groups of flags ):
"""
__slots__ = ('_flags',)
@abstractmethod
def get_flags(self):
""" """
Public. Returns a list of flags Public. The entity of the flag of the entered command
:return: list of flags :param name: the name of the input flag
""" :param prefix: the prefix of the input flag
pass :param value: the value of the input flag
@abstractmethod
def add_flag(self, flag: Flag | InputFlag):
"""
Public. Adds a flag to the list of flags
:param flag: flag to add
:return: None :return: None
""" """
pass super().__init__(name, prefix)
self._flag_value = value
@abstractmethod def get_value(self) -> str | None:
def add_flags(self, flags: list[Flag] | list[InputFlag]):
""" """
Public. Adds a list of flags to the list of flags Public. Returns the value of the flag
:param flags: list of flags to add :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 :return: None
""" """
pass self._flag_value = value
@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
def __eq__(self, other) -> bool:
return (
self.get_string_entity() == other.get_string_entity()
and self.get_value() == other.get_value()
)
+16
View File
@@ -0,0 +1,16 @@
__all__ = [
"Flags",
"InputFlags",
"UndefinedInputFlags",
"InvalidValueInputFlags",
"ValidInputFlags",
]
from argenta.command.flags.models import (
Flags,
InputFlags,
UndefinedInputFlags,
InvalidValueInputFlags,
ValidInputFlags,
)
+90
View File
@@ -0,0 +1,90 @@
from argenta.command.flag.models import InputFlag, Flag
from typing import Generic, TypeVar
FlagType = TypeVar("FlagType")
class BaseFlags(Generic[FlagType]):
def __init__(self, *flags: FlagType):
"""
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[FlagType]:
"""
Public. Returns a list of flags
:return: list of flags as list[FlagType]
"""
return self._flags
def add_flag(self, flag: FlagType):
"""
Public. Adds a flag to the list of flags
:param flag: flag to add
:return: None
"""
self._flags.append(flag)
def add_flags(self, flags: list[FlagType]):
"""
Public. Adds a list of flags to the list of flags
:param flags: list of flags to add
:return: None
"""
self._flags.extend(flags)
def get_flag(self, name: str) -> FlagType | None:
"""
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
"""
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
def __iter__(self):
return iter(self._flags)
def __next__(self):
return next(iter(self))
def __getitem__(self, item):
return self._flags[item]
def __bool__(self):
return bool(self._flags)
def __eq__(self, other):
if len(self.get_flags()) != len(other.get_flags()):
return False
else:
for flag, other_flag in zip(self.get_flags(), other.get_flags()):
if not flag == other_flag:
return False
return True
class Flags(BaseFlags[Flag]):
pass
class InputFlags(BaseFlags[InputFlag]):
pass
class ValidInputFlags(InputFlags):
pass
class UndefinedInputFlags(InputFlags):
pass
class InvalidValueInputFlags(InputFlags):
pass
+66 -33
View File
@@ -1,11 +1,14 @@
from argenta.command.flag.models import Flag, InputFlag, Flags, InputFlags from argenta.command.flag.models import Flag, InputFlag
from argenta.command.exceptions import (UnprocessedInputFlagException, from argenta.command.flags.models import InputFlags, Flags
RepeatedInputFlagsException, from argenta.command.exceptions import (
EmptyInputCommandException) UnprocessedInputFlagException,
RepeatedInputFlagsException,
EmptyInputCommandException,
)
from typing import Generic, TypeVar, cast, Literal from typing import Generic, TypeVar, cast, Literal
InputCommandType = TypeVar('InputCommandType') InputCommandType = TypeVar("InputCommandType")
class BaseCommand: class BaseCommand:
@@ -25,10 +28,13 @@ class BaseCommand:
class Command(BaseCommand): class Command(BaseCommand):
def __init__(self, trigger: str, def __init__(
description: str = None, self,
flags: Flag | Flags = None, trigger: str,
aliases: list[str] = None): description: str = None,
flags: Flag | Flags = None,
aliases: list[str] = None,
):
""" """
Public. The command that can and should be registered in the Router 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 trigger: A string trigger, which, when entered by the user, indicates that the input corresponds to the command
@@ -37,8 +43,14 @@ class Command(BaseCommand):
:param aliases: string synonyms for the main trigger :param aliases: string synonyms for the main trigger
""" """
super().__init__(trigger) super().__init__(trigger)
self._registered_flags: Flags = flags if isinstance(flags, Flags) else Flags(flags) if isinstance(flags, Flag) else Flags() self._registered_flags: Flags = (
self._description = f'Description for "{self._trigger}" command' if not description else description flags
if isinstance(flags, Flags)
else Flags(flags)
if isinstance(flags, Flag)
else Flags()
)
self._description = "Very useful command" if not description else description
self._aliases = aliases if isinstance(aliases, list) else [] self._aliases = aliases if isinstance(aliases, list) else []
def get_registered_flags(self) -> Flags: def get_registered_flags(self) -> Flags:
@@ -55,7 +67,9 @@ class Command(BaseCommand):
""" """
return self._aliases return self._aliases
def validate_input_flag(self, flag: InputFlag) -> bool: def validate_input_flag(
self, flag: InputFlag
) -> Literal["Undefined", "Valid", "Invalid"]:
""" """
Private. Validates the input flag Private. Validates the input flag
:param flag: input flag for validation :param flag: input flag for validation
@@ -65,16 +79,27 @@ class Command(BaseCommand):
if registered_flags: if registered_flags:
if isinstance(registered_flags, Flag): if isinstance(registered_flags, Flag):
if registered_flags.get_string_entity() == flag.get_string_entity(): if registered_flags.get_string_entity() == flag.get_string_entity():
is_valid = registered_flags.validate_input_flag_value(flag.get_value()) is_valid = registered_flags.validate_input_flag_value(
flag.get_value()
)
if is_valid: if is_valid:
return True return "Valid"
else:
return "Invalid"
else:
return "Undefined"
else: else:
for registered_flag in registered_flags: for registered_flag in registered_flags:
if registered_flag.get_string_entity() == flag.get_string_entity(): if registered_flag.get_string_entity() == flag.get_string_entity():
is_valid = registered_flag.validate_input_flag_value(flag.get_value()) is_valid = registered_flag.validate_input_flag_value(
flag.get_value()
)
if is_valid: if is_valid:
return True return "Valid"
return False else:
return "Invalid"
return "Undefined"
return "Undefined"
def get_description(self) -> str: def get_description(self) -> str:
""" """
@@ -84,10 +109,8 @@ class Command(BaseCommand):
return self._description return self._description
class InputCommand(BaseCommand, Generic[InputCommandType]): class InputCommand(BaseCommand, Generic[InputCommandType]):
def __init__(self, trigger: str, def __init__(self, trigger: str, input_flags: InputFlag | InputFlags = None):
input_flags: InputFlag | InputFlags = None):
""" """
Private. The model of the input command, after parsing Private. The model of the input command, after parsing
:param trigger:the trigger of the command :param trigger:the trigger of the command
@@ -95,7 +118,13 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
:return: None :return: None
""" """
super().__init__(trigger) 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() 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: def _set_input_flags(self, input_flags: InputFlags) -> None:
""" """
@@ -112,7 +141,6 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
""" """
return self._input_flags return self._input_flags
@staticmethod @staticmethod
def parse(raw_command: str) -> InputCommandType: def parse(raw_command: str) -> InputCommandType:
""" """
@@ -130,26 +158,32 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
current_flag_name, current_flag_value = None, None current_flag_name, current_flag_value = None, None
for k, _ in enumerate(list_of_tokens): for k, _ in enumerate(list_of_tokens):
if _.startswith('-'): if _.startswith("-"):
if current_flag_name or len(_) < 2 or len(_[:_.rfind('-')]) > 3: if len(_) < 2 or len(_[: _.rfind("-")]) > 3:
raise UnprocessedInputFlagException() raise UnprocessedInputFlagException()
current_flag_name = _ current_flag_name = _
else: else:
if not current_flag_name: if not current_flag_name or current_flag_value:
raise UnprocessedInputFlagException() raise UnprocessedInputFlagException()
current_flag_value = _ current_flag_value = _
if current_flag_name: if current_flag_name:
if not len(list_of_tokens) == k+1: if not len(list_of_tokens) == k + 1:
if not list_of_tokens[k+1].startswith('-'): if not list_of_tokens[k + 1].startswith("-"):
continue continue
input_flag = InputFlag(name=current_flag_name[current_flag_name.rfind('-')+1:], input_flag = InputFlag(
prefix=cast(Literal['-', '--', '---'], name=current_flag_name[current_flag_name.rfind("-") + 1 :],
current_flag_name[:current_flag_name.rfind('-')+1]), prefix=cast(
value=current_flag_value) 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()] all_flags = [
flag.get_string_entity() for flag in input_flags.get_flags()
]
if input_flag.get_string_entity() not in all_flags: if input_flag.get_string_entity() not in all_flags:
input_flags.add_flag(input_flag) input_flags.add_flag(input_flag)
else: else:
@@ -161,4 +195,3 @@ class InputCommand(BaseCommand, Generic[InputCommandType]):
raise UnprocessedInputFlagException() raise UnprocessedInputFlagException()
else: else:
return InputCommand(trigger=command, input_flags=input_flags) return InputCommand(trigger=command, input_flags=input_flags)
@@ -1,6 +1,8 @@
__all__ = ["BooleanArgument", "PositionalArgument", "OptionalArgument"] __all__ = ["BooleanArgument", "PositionalArgument", "OptionalArgument"]
from argenta.orchestrator.argparser.arguments.models import (BooleanArgument, from argenta.orchestrator.argparser.arguments.models import (
PositionalArgument, BooleanArgument,
OptionalArgument) PositionalArgument,
OptionalArgument,
)
@@ -6,6 +6,7 @@ class BaseArgument(ABC):
""" """
Private. Base class for all arguments Private. Base class for all arguments
""" """
@abstractmethod @abstractmethod
def get_string_entity(self) -> str: def get_string_entity(self) -> str:
""" """
@@ -28,7 +29,7 @@ class PositionalArgument(BaseArgument):
class OptionalArgument(BaseArgument): class OptionalArgument(BaseArgument):
def __init__(self, name: str, prefix: Literal['-', '--', '---'] = '--'): def __init__(self, name: str, prefix: Literal["-", "--", "---"] = "--"):
""" """
Public. Optional argument, must have the value Public. Optional argument, must have the value
:param name: name of the argument :param name: name of the argument
@@ -42,7 +43,7 @@ class OptionalArgument(BaseArgument):
class BooleanArgument(BaseArgument): class BooleanArgument(BaseArgument):
def __init__(self, name: str, prefix: Literal['-', '--', '---'] = '--'): def __init__(self, name: str, prefix: Literal["-", "--", "---"] = "--"):
""" """
Public. Boolean argument, does not require a value Public. Boolean argument, does not require a value
:param name: name of the argument :param name: name of the argument
+22 -12
View File
@@ -1,16 +1,20 @@
from argparse import ArgumentParser from argparse import ArgumentParser
from argenta.orchestrator.argparser.arguments.models import (BooleanArgument, from argenta.orchestrator.argparser.arguments.models import (
OptionalArgument, BooleanArgument,
PositionalArgument) OptionalArgument,
PositionalArgument,
)
class ArgParser: class ArgParser:
def __init__(self, def __init__(
processed_args: list[PositionalArgument | OptionalArgument | BooleanArgument], self,
name: str = 'Argenta', processed_args: list[PositionalArgument | OptionalArgument | BooleanArgument],
description: str = 'Argenta available arguments', name: str = "Argenta",
epilog: str = 'github.com/koloideal/Argenta | made by kolo') -> None: description: str = "Argenta available arguments",
epilog: str = "github.com/koloideal/Argenta | made by kolo",
) -> None:
""" """
Public. Cmd argument parser and configurator at startup Public. Cmd argument parser and configurator at startup
:param name: the name of the ArgParse instance :param name: the name of the ArgParse instance
@@ -22,10 +26,16 @@ class ArgParser:
self.description = description self.description = description
self.epilog = epilog self.epilog = epilog
self.entity: ArgumentParser = ArgumentParser(prog=name, description=description, epilog=epilog) self.entity: ArgumentParser = ArgumentParser(
self.args: list[PositionalArgument | OptionalArgument | BooleanArgument] | None = processed_args prog=name, description=description, epilog=epilog
)
self.args: (
list[PositionalArgument | OptionalArgument | BooleanArgument] | None
) = processed_args
def set_args(self, *args: PositionalArgument | OptionalArgument | BooleanArgument) -> None: def set_args(
self, *args: PositionalArgument | OptionalArgument | BooleanArgument
) -> None:
""" """
Public. Sets the arguments to be processed Public. Sets the arguments to be processed
:param args: processed arguments :param args: processed arguments
@@ -46,4 +56,4 @@ class ArgParser:
elif type(arg) is OptionalArgument: elif type(arg) is OptionalArgument:
self.entity.add_argument(arg.get_string_entity()) self.entity.add_argument(arg.get_string_entity())
elif type(arg) is BooleanArgument: elif type(arg) is BooleanArgument:
self.entity.add_argument(arg.get_string_entity(), action='store_true') self.entity.add_argument(arg.get_string_entity(), action="store_true")
-1
View File
@@ -33,4 +33,3 @@ class Orchestrator:
return self.arg_parser.entity.parse_args() return self.arg_parser.entity.parse_args()
else: else:
return None return None
+5
View File
@@ -0,0 +1,5 @@
__all__ = ["Response", "Status"]
from argenta.response.entity import Response
from argenta.response.status import Status
+29
View File
@@ -0,0 +1,29 @@
from argenta.response.status import Status
from argenta.command.flags import (
ValidInputFlags,
UndefinedInputFlags,
InvalidValueInputFlags,
)
class Response:
__slots__ = ("status", "valid_flags", "undefined_flags", "invalid_value_flags")
def __init__(
self,
status: Status = None,
valid_flags: ValidInputFlags = ValidInputFlags(),
undefined_flags: UndefinedInputFlags = UndefinedInputFlags(),
invalid_value_flags: InvalidValueInputFlags = InvalidValueInputFlags(),
):
"""
Public. The entity of the user input sent to the handler
:param status: the status of the response
:param valid_flags: valid input flags
:param undefined_flags: undefined input flags
:param invalid_value_flags: input flags with invalid values
"""
self.status = status
self.valid_flags = valid_flags
self.undefined_flags = undefined_flags
self.invalid_value_flags = invalid_value_flags
+8
View File
@@ -0,0 +1,8 @@
from enum import Enum
class Status(Enum):
ALL_FLAGS_VALID = "ALL_FLAGS_VALID"
UNDEFINED_FLAGS = "UNDEFINED_FLAGS"
INVALID_VALUE_FLAGS = "INVALID_VALUE_FLAGS"
UNDEFINED_AND_INVALID_FLAGS = "UNDEFINED_AND_INVALID_FLAGS"
+7 -11
View File
@@ -1,12 +1,11 @@
from typing import Callable, Iterator from typing import Callable, Iterator
from argenta.command import Command from argenta.command import Command
from argenta.command.flag import InputFlags from argenta.response import Response
class CommandHandler: class CommandHandler:
def __init__(self, handler: Callable[[], None] | Callable[[InputFlags], None], handled_command: Command): def __init__(self, handler: Callable[[Response], None], handled_command: Command):
""" """
Private. Entity of the model linking the handler and the command being processed Private. Entity of the model linking the handler and the command being processed
:param handler: the handler being called :param handler: the handler being called
@@ -15,21 +14,18 @@ class CommandHandler:
self._handler = handler self._handler = handler
self._handled_command = handled_command self._handled_command = handled_command
def handling(self, input_flags: InputFlags = None) -> None: def handling(self, response: Response) -> None:
""" """
Private. Direct processing of an input command Private. Direct processing of an input command
:param input_flags: the flags of the input command :param response: the entity of response: various groups of flags and status of response
:return: None :return: None
""" """
if input_flags is not None: self._handler(response)
self._handler(input_flags)
else:
self._handler()
def get_handler(self) -> Callable[[], None] | Callable[[InputFlags], None]: def get_handler(self) -> Callable[[Response], None]:
""" """
Private. Returns the handler being called Private. Returns the handler being called
:return: the handler being called as Callable[[], None] or Callable[[InputFlags], None] :return: the handler being called as Callable[[Response], None]
""" """
return self._handler return self._handler
+1 -1
View File
@@ -1,4 +1,4 @@
from argenta.router import Router from argenta.router import Router
system_router = Router(title='System points:') system_router = Router(title="System points:")
+118 -84
View File
@@ -1,58 +1,59 @@
from typing import Callable from typing import Callable, Literal, Type
from inspect import getfullargspec from inspect import getfullargspec, get_annotations, getsourcefile, getsourcelines
from rich.console import Console
from argenta.command import Command from argenta.command import Command
from argenta.command.models import InputCommand from argenta.command.models import InputCommand
from argenta.response import Response, Status
from argenta.router.command_handler.entity import CommandHandlers, CommandHandler from argenta.router.command_handler.entity import CommandHandlers, CommandHandler
from argenta.command.flag.models import Flag, Flags, InputFlags from argenta.command.flags.models import (
from argenta.router.exceptions import (RepeatedFlagNameException, Flags,
TooManyTransferredArgsException, InputFlags,
RequiredArgumentNotPassedException, UndefinedInputFlags,
TriggerContainSpacesException) ValidInputFlags,
InvalidValueInputFlags,
)
from argenta.router.exceptions import (
RepeatedFlagNameException,
TooManyTransferredArgsException,
RequiredArgumentNotPassedException,
TriggerContainSpacesException,
)
class Router: class Router:
def __init__(self, def __init__(self, title: str | None = "Awesome title"):
title: str = None):
""" """
Public. Directly configures and manages handlers Public. Directly configures and manages handlers
:param title: the title of the router, displayed when displaying the available commands :param title: the title of the router, displayed when displaying the available commands
:return: None :return: None
""" """
self._title = title self.title = title
self._command_handlers: CommandHandlers = CommandHandlers() self._command_handlers: CommandHandlers = CommandHandlers()
self._ignore_command_register: bool = False 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 | str) -> Callable:
def command(self, command: Command) -> Callable:
""" """
Public. Registers handler Public. Registers handler
:param command: Registered command :param command: Registered command
:return: decorated handler as Callable[[Any], Any] :return: decorated handler as Callable
""" """
self._validate_command(command) self._validate_command(command)
if isinstance(command, str):
command = Command(command)
def command_decorator(func): def command_decorator(func):
Router._validate_func_args(command, func) Router._validate_func_args(func)
self._command_handlers.add_handler(CommandHandler(func, command)) self._command_handlers.add_handler(CommandHandler(func, command))
def wrapper(*args, **kwargs): def wrapper(*args, **kwargs):
return func(*args, **kwargs) return func(*args, **kwargs)
return wrapper return wrapper
return command_decorator 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: def finds_appropriate_handler(self, input_command: InputCommand) -> None:
""" """
Private. Finds the appropriate handler for given input command and passes control to it Private. Finds the appropriate handler for given input command and passes control to it
@@ -69,8 +70,9 @@ class Router:
if input_command_name.lower() in handle_command.get_aliases(): if input_command_name.lower() in handle_command.get_aliases():
self.process_input_command(input_command_flags, command_handler) self.process_input_command(input_command_flags, command_handler)
def process_input_command(
def process_input_command(self, input_command_flags: InputFlags, command_handler: CommandHandler) -> None: self, input_command_flags: InputFlags, command_handler: CommandHandler
) -> None:
""" """
Private. Processes input command with the appropriate handler Private. Processes input command with the appropriate handler
:param input_command_flags: input command flags as InputFlags :param input_command_flags: input command flags as InputFlags
@@ -78,74 +80,126 @@ class Router:
:return: None :return: None
""" """
handle_command = command_handler.get_handled_command() handle_command = command_handler.get_handled_command()
response: Response = Response()
if handle_command.get_registered_flags().get_flags(): if handle_command.get_registered_flags().get_flags():
if input_command_flags.get_flags(): if input_command_flags.get_flags():
if self._validate_input_flags(handle_command, input_command_flags): response: Response = self._structuring_input_flags(
command_handler.handling(input_command_flags) handle_command, input_command_flags
return )
command_handler.handling(response)
else: else:
command_handler.handling(input_command_flags) response.status = Status.ALL_FLAGS_VALID
return command_handler.handling(response)
else: else:
if input_command_flags.get_flags(): if input_command_flags.get_flags():
self._not_valid_flag_handler(input_command_flags[0]) response.status = Status.UNDEFINED_FLAGS
return response.undefined_flags = UndefinedInputFlags()
response.undefined_flags.add_flags(input_command_flags.get_flags())
command_handler.handling(response)
else: else:
command_handler.handling() response.status = Status.ALL_FLAGS_VALID
return command_handler.handling(response)
@staticmethod
def _validate_input_flags(self, handled_command: Command, input_flags: InputFlags) -> bool: def _structuring_input_flags(
handled_command: Command, input_flags: InputFlags
) -> Response:
""" """
Private. Validates flags of input command Private. Validates flags of input command
:param handled_command: entity of the handled command :param handled_command: entity of the handled command
:param input_flags: :param input_flags:
:return: is flags of input command valid as bool :return: entity of response as Response
""" """
valid_input_flags: ValidInputFlags = ValidInputFlags()
invalid_value_input_flags: InvalidValueInputFlags = InvalidValueInputFlags()
undefined_input_flags: UndefinedInputFlags = UndefinedInputFlags()
for flag in input_flags: for flag in input_flags:
is_valid: bool = handled_command.validate_input_flag(flag) flag_status: Literal["Undefined", "Valid", "Invalid"] = (
if not is_valid: handled_command.validate_input_flag(flag)
self._not_valid_flag_handler(flag) )
return False match flag_status:
return True case "Valid":
valid_input_flags.add_flag(flag)
case "Undefined":
undefined_input_flags.add_flag(flag)
case "Invalid":
invalid_value_input_flags.add_flag(flag)
if (
not invalid_value_input_flags.get_flags()
and not undefined_input_flags.get_flags()
):
status = Status.ALL_FLAGS_VALID
elif (
invalid_value_input_flags.get_flags()
and not undefined_input_flags.get_flags()
):
status = Status.INVALID_VALUE_FLAGS
elif (
not invalid_value_input_flags.get_flags()
and undefined_input_flags.get_flags()
):
status = Status.UNDEFINED_FLAGS
else:
status = Status.UNDEFINED_AND_INVALID_FLAGS
return Response(
invalid_value_flags=invalid_value_input_flags,
valid_flags=valid_input_flags,
status=status,
undefined_flags=undefined_input_flags,
)
@staticmethod @staticmethod
def _validate_command(command: Command) -> None: def _validate_command(command: Command | str) -> None:
""" """
Private. Validates the command registered in handler Private. Validates the command registered in handler
:param command: validated command :param command: validated command
:return: None if command is valid else raise exception :return: None if command is valid else raise exception
""" """
command_name: str = command.get_trigger() match type(command).__name__:
if command_name.find(' ') != -1: case "Command":
raise TriggerContainSpacesException() command_name: str = command.get_trigger()
if command_name.find(" ") != -1:
flags: Flags = command.get_registered_flags() raise TriggerContainSpacesException()
if flags: flags: Flags = command.get_registered_flags()
flags_name: list = [x.get_string_entity().lower() for x in flags] if flags:
if len(set(flags_name)) < len(flags_name): flags_name: list = [x.get_string_entity().lower() for x in flags]
raise RepeatedFlagNameException() if len(set(flags_name)) < len(flags_name):
raise RepeatedFlagNameException()
case "str":
if command.find(" ") != -1:
raise TriggerContainSpacesException()
@staticmethod @staticmethod
def _validate_func_args(command: Command, func: Callable) -> None: def _validate_func_args(func: Callable) -> None:
""" """
Private. Validates the arguments of the handler Private. Validates the arguments of the handler
:param command: registered command in handler
:param func: entity of the handler func :param func: entity of the handler func
:return: None if func is valid else raise exception :return: None if func is valid else raise exception
""" """
registered_args = command.get_registered_flags()
transferred_args = getfullargspec(func).args transferred_args = getfullargspec(func).args
if registered_args.get_flags() and transferred_args: if len(transferred_args) > 1:
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() raise TooManyTransferredArgsException()
elif len(transferred_args) == 0:
raise RequiredArgumentNotPassedException()
transferred_arg: str = transferred_args[0]
func_annotations: dict[str, Type] = get_annotations(func)
if arg_annotation := func_annotations.get(transferred_arg):
if arg_annotation is Response:
pass
else:
file_path: str = getsourcefile(func)
source_line: int = getsourcelines(func)[1] + 1
fprint = Console().print
fprint(
f'\nFile "{file_path}", line {source_line}\n[b red]WARNING:[/b red] [i]The typehint '
f"of argument([green]{transferred_arg}[/green]) passed to the handler is [/i][bold blue]{Response}[/bold blue],"
f" [i]but[/i] [bold blue]{arg_annotation}[/bold blue] [i]is specified[/i]\n",
highlight=False,
)
def set_command_register_ignore(self, _: bool) -> None: def set_command_register_ignore(self, _: bool) -> None:
""" """
@@ -155,7 +209,6 @@ class Router:
""" """
self._ignore_command_register = _ self._ignore_command_register = _
def get_triggers(self) -> list[str]: def get_triggers(self) -> list[str]:
""" """
Public. Gets registered triggers Public. Gets registered triggers
@@ -166,7 +219,6 @@ class Router:
all_triggers.append(command_handler.get_handled_command().get_trigger()) all_triggers.append(command_handler.get_handled_command().get_trigger())
return all_triggers return all_triggers
def get_aliases(self) -> list[str]: def get_aliases(self) -> list[str]:
""" """
Public. Gets registered aliases Public. Gets registered aliases
@@ -178,27 +230,9 @@ class Router:
all_aliases.extend(command_handler.get_handled_command().get_aliases()) all_aliases.extend(command_handler.get_handled_command().get_aliases())
return all_aliases return all_aliases
def get_command_handlers(self) -> CommandHandlers: def get_command_handlers(self) -> CommandHandlers:
""" """
Private. Gets registered command handlers Private. Gets registered command handlers
:return: registered command handlers as CommandHandlers :return: registered command handlers as CommandHandlers
""" """
return self._command_handlers 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
+4
View File
@@ -2,6 +2,7 @@ class RepeatedFlagNameException(Exception):
""" """
Private. Raised when a repeated flag name is registered Private. Raised when a repeated flag name is registered
""" """
def __str__(self): def __str__(self):
return "Repeated registered flag names in register command" return "Repeated registered flag names in register command"
@@ -10,6 +11,7 @@ class TooManyTransferredArgsException(Exception):
""" """
Private. Raised when too many arguments are passed Private. Raised when too many arguments are passed
""" """
def __str__(self): def __str__(self):
return "Too many transferred arguments" return "Too many transferred arguments"
@@ -18,6 +20,7 @@ class RequiredArgumentNotPassedException(Exception):
""" """
Private. Raised when a required argument is not passed Private. Raised when a required argument is not passed
""" """
def __str__(self): def __str__(self):
return "Required argument not passed" return "Required argument not passed"
@@ -26,5 +29,6 @@ class TriggerContainSpacesException(Exception):
""" """
Private. Raised when there is a space in the trigger being registered Private. Raised when there is a space in the trigger being registered
""" """
def __str__(self): def __str__(self):
return "Command trigger cannot contain spaces" return "Command trigger cannot contain spaces"
@@ -1,32 +1,35 @@
import _io import _io
from unittest.mock import patch, MagicMock from unittest.mock import patch, MagicMock
import unittest from unittest import TestCase
import io import io
import re import re
from argenta.app import App from argenta.app import App
from argenta.command import Command from argenta.command import Command
from argenta.router import Router from argenta.router import Router
from argenta.command.flag.models import Flags, InputFlags from argenta.command.flags.models import Flags
from argenta.command.flag.defaults import PredefinedFlags from argenta.command.flag.defaults import PredefinedFlags
from argenta.orchestrator import Orchestrator
from argenta.response import Response
class TestSystemHandlerNormalWork(unittest.TestCase): class TestSystemHandlerNormalWork(TestCase):
@patch("builtins.input", side_effect=["help", "q"]) @patch("builtins.input", side_effect=["help", "q"])
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
@router.command(Command('test')) @router.command(Command('test'))
def test(): def test(response: Response):
print('test command') print('test command')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}')) app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
@@ -37,9 +40,10 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_incorrect_command2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_incorrect_command2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
@router.command(Command('test')) @router.command(Command('test'))
def test(): def test(response: Response):
print('test command') print('test command')
app = App(ignore_command_register=False, app = App(ignore_command_register=False,
@@ -47,7 +51,7 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}')) app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
@@ -58,74 +62,80 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_unregistered_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_correct_command_with_unregistered_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
@router.command(Command('test')) @router.command(Command('test'))
def test(): def test(response: Response):
print(f'test command') print(f'test command with undefined flag: {response.undefined_flags.get_flag('help').get_string_entity()}')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
self.assertIn('\nUndefined or incorrect input flag: --help\n', output) self.assertIn('\ntest command with undefined flag: --help\n', output)
@patch("builtins.input", side_effect=["test --port 22", "q"]) @patch("builtins.input", side_effect=["test --port 22", "q"])
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_unregistered_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_correct_command_with_unregistered_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
@router.command(Command('test')) @router.command(Command('test'))
def test(): def test(response: Response):
print('test command') flag = response.undefined_flags.get_flag("port")
print(f'test command with undefined flag with value: {flag.get_string_entity()} {flag.get_value()}')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
self.assertIn('\nUndefined or incorrect input flag: --port 22\n', output) self.assertIn('\ntest command with undefined flag with value: --port 22\n', output)
@patch("builtins.input", side_effect=["test --host 192.168.32.1 --port 132", "q"]) @patch("builtins.input", side_effect=["test --host 192.168.32.1 --port 132", "q"])
@patch("sys.stdout", new_callable=io.StringIO) @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): def test_input_correct_command_with_one_correct_flag_an_one_incorrect_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
flags = Flags(PredefinedFlags.HOST) flags = Flags(PredefinedFlags.HOST)
@router.command(Command('test', flags=flags)) @router.command(Command('test', flags=flags))
def test(args: InputFlags): def test(response: Response):
print(f'connecting to host {args.get_flag('host').get_value()}') flag = response.undefined_flags.get_flag("port")
print(f'connecting to host with flag: {flag.get_string_entity()} {flag.get_value()}')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
self.assertIn('\nUndefined or incorrect input flag: --port 132\n', output) self.assertIn('\nconnecting to host with flag: --port 132\n', output)
@patch("builtins.input", side_effect=["test", "some", "q"]) @patch("builtins.input", side_effect=["test", "some", "q"])
@patch("sys.stdout", new_callable=io.StringIO) @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): def test_input_one_correct_command_and_one_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
@router.command(Command('test')) @router.command(Command('test'))
def test(): def test(response: Response):
print(f'test command') print(f'test command')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}')) app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
@@ -136,20 +146,21 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO) @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): def test_input_two_correct_commands_and_one_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
@router.command(Command('test')) @router.command(Command('test'))
def test(): def test(response: Response):
print(f'test command') print(f'test command')
@router.command(Command('more')) @router.command(Command('more'))
def test(): def test(response: Response):
print(f'more command') print(f'more command')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}')) app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
@@ -160,16 +171,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_incorrect_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_correct_command_with_incorrect_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
@router.command(Command('test')) @router.command(Command('test'))
def test(): def test(response: Response):
print(f'test command') print(f'test command')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.set_invalid_input_flags_handler(lambda command: print(f'Incorrect flag syntax: "{command}"')) app.set_incorrect_input_syntax_handler(lambda command: print(f'Incorrect flag syntax: "{command}"'))
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
@@ -180,16 +192,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_empty_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_empty_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
@router.command(Command('test')) @router.command(Command('test'))
def test(): def test(response: Response):
print(f'test command') print(f'test command')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.set_empty_command_handler(lambda: print('Empty input command')) app.set_empty_command_handler(lambda: print('Empty input command'))
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
@@ -200,16 +213,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_repeated_flags(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_correct_command_with_repeated_flags(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
@router.command(Command('test', flags=PredefinedFlags.PORT)) @router.command(Command('test', flags=PredefinedFlags.PORT))
def test(args: InputFlags): def test(response: Response):
print('test command') print('test command')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.set_repeated_input_flags_handler(lambda command: print(f'Repeated input flags: "{command}"')) app.set_repeated_input_flags_handler(lambda command: print(f'Repeated input flags: "{command}"'))
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
@@ -1,31 +1,35 @@
import _io import _io
from unittest.mock import patch, MagicMock from unittest.mock import patch, MagicMock
import unittest from unittest import TestCase
import io import io
import re import re
from argenta.app import App from argenta.app import App
from argenta.command.models import Command from argenta.command import Command
from argenta.response import Response
from argenta.router import Router from argenta.router import Router
from argenta.command.flag.models import Flag, Flags, InputFlags from argenta.orchestrator import Orchestrator
from argenta.command.flag import Flag
from argenta.command.flags import Flags
from argenta.command.flag.defaults import PredefinedFlags from argenta.command.flag.defaults import PredefinedFlags
class TestSystemHandlerNormalWork(unittest.TestCase): class TestSystemHandlerNormalWork(TestCase):
@patch("builtins.input", side_effect=["test", "q"]) @patch("builtins.input", side_effect=["test", "q"])
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
@router.command(Command('test')) @router.command(Command('test'))
def test(): def test(response):
print('test command') print('test command')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
@@ -36,16 +40,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_correct_command2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
@router.command(Command('test')) @router.command(Command('test'))
def test(): def test(response):
print('test command') print('test command')
app = App(ignore_command_register=True, app = App(ignore_command_register=True,
override_system_messages=True, override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
@@ -56,16 +61,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_custom_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_correct_command_with_custom_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
flag = Flag('help', '--', False) flag = Flag('help', '--', False)
@router.command(Command('test', flags=flag)) @router.command(Command('test', flags=flag))
def test(args: InputFlags): def test(response: Response):
print(f'\nhelp for {args.get_flag('help').get_name()} flag\n') print(f'\nhelp for {response.valid_flags.get_flag('help').get_name()} flag\n')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
@@ -75,16 +81,18 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_custom_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_correct_command_with_custom_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
flag = Flag('port', '--', re.compile(r'^\d{1,5}$')) flag = Flag('port', '--', re.compile(r'^\d{1,5}$'))
@router.command(Command('test', flags=flag)) @router.command(Command('test', flags=flag))
def test(args: InputFlags): def test(response: Response):
print(f'flag value for {args.get_flag('port').get_name()} flag : {args.get_flag('port').get_value()}') input_flag = response.valid_flags.get_flag('port')
print(f'flag value for {input_flag.get_name()} flag : {input_flag.get_value()}')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
@@ -95,16 +103,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_default_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_correct_command_with_default_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
flag = PredefinedFlags.SHORT_HELP flag = PredefinedFlags.SHORT_HELP
@router.command(Command('test', flags=flag)) @router.command(Command('test', flags=flag))
def test(args: InputFlags): def test(response: Response):
print(f'help for {args.get_flag('H').get_name()} flag') print(f'help for {response.valid_flags.get_flag('H').get_name()} flag')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
@@ -115,17 +124,18 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_default_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_correct_command_with_default_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
flag = PredefinedFlags.INFO flag = PredefinedFlags.INFO
@router.command(Command('test', flags=flag)) @router.command(Command('test', flags=flag))
def test(args: InputFlags): def test(response: Response):
if args.get_flag('info'): if response.valid_flags.get_flag('info'):
print('info about test command') print('info about test command')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
@@ -136,16 +146,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_default_flag3(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_correct_command_with_default_flag3(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
flag = PredefinedFlags.HOST flag = PredefinedFlags.HOST
@router.command(Command('test', flags=flag)) @router.command(Command('test', flags=flag))
def test(args: InputFlags): def test(response: Response):
print(f'connecting to host {args[0].get_value()}') print(f'connecting to host {response.valid_flags.get_flag('host').get_value()}')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
@@ -156,16 +167,18 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_correct_command_with_two_flags(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_correct_command_with_two_flags(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
flags = Flags(PredefinedFlags.HOST, PredefinedFlags.PORT) flags = Flags(PredefinedFlags.HOST, PredefinedFlags.PORT)
@router.command(Command('test', flags=flags)) @router.command(Command('test', flags=flags))
def test(args: InputFlags): def test(response: Response):
print(f'connecting to host {args[0].get_value()} and port {args[1].get_value()}') valid_flags = response.valid_flags
print(f'connecting to host {valid_flags.get_flag('host').get_value()} and port {valid_flags.get_flag('port').get_value()}')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
@@ -176,19 +189,20 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_two_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_two_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
@router.command(Command('test')) @router.command(Command('test'))
def test(): def test(response):
print(f'test command') print(f'test command')
@router.command(Command('some')) @router.command(Command('some'))
def test2(): def test2(response):
print(f'some command') print(f'some command')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
@@ -199,23 +213,24 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
@patch("sys.stdout", new_callable=io.StringIO) @patch("sys.stdout", new_callable=io.StringIO)
def test_input_three_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): def test_input_three_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
router = Router() router = Router()
orchestrator = Orchestrator()
@router.command(Command('test')) @router.command(Command('test'))
def test(): def test(response):
print(f'test command') print(f'test command')
@router.command(Command('some')) @router.command(Command('some'))
def test(): def test(response):
print(f'some command') print(f'some command')
@router.command(Command('more')) @router.command(Command('more'))
def test(): def test(response):
print(f'more command') print(f'more command')
app = App(override_system_messages=True, app = App(override_system_messages=True,
print_func=print) print_func=print)
app.include_router(router) app.include_router(router)
app.run_polling() orchestrator.start_polling(app)
output = mock_stdout.getvalue() output = mock_stdout.getvalue()
+26 -12
View File
@@ -1,10 +1,12 @@
from argenta.command.flag import Flag, InputFlag, Flags from argenta.command.flag import Flag, InputFlag
from argenta.command.flags import Flags
from argenta.command.models import InputCommand, Command from argenta.command.models import InputCommand, Command
from argenta.command.exceptions import (UnprocessedInputFlagException, from argenta.command.exceptions import (UnprocessedInputFlagException,
RepeatedInputFlagsException, RepeatedInputFlagsException,
EmptyInputCommandException) EmptyInputCommandException)
import unittest import unittest
import re
class TestInputCommand(unittest.TestCase): class TestInputCommand(unittest.TestCase):
@@ -23,25 +25,37 @@ class TestInputCommand(unittest.TestCase):
with self.assertRaises(EmptyInputCommandException): with self.assertRaises(EmptyInputCommandException):
InputCommand.parse('') InputCommand.parse('')
def test_validate_correct_input_flag1(self): def test_validate_valid_input_flag1(self):
command = Command('some', flags=Flag('test')) command = Command('some', flags=Flag('test'))
self.assertEqual(command.validate_input_flag(InputFlag('test')), True) self.assertEqual(command.validate_input_flag(InputFlag('test')), 'Valid')
def test_validate_correct_input_flag2(self): def test_validate_valid_input_flag2(self):
command = Command('some', flags=Flags(Flag('test'), Flag('more'))) command = Command('some', flags=Flags(Flag('test'), Flag('more')))
self.assertEqual(command.validate_input_flag(InputFlag('more')), True) self.assertEqual(command.validate_input_flag(InputFlag('more')), 'Valid')
def test_validate_incorrect_input_flag1(self): def test_validate_undefined_input_flag1(self):
command = Command('some', flags=Flags(Flag('test'))) command = Command('some', flags=Flag('test'))
self.assertEqual(command.validate_input_flag(InputFlag('more')), False) self.assertEqual(command.validate_input_flag(InputFlag('more')), 'Undefined')
def test_validate_incorrect_input_flag2(self): def test_validate_undefined_input_flag2(self):
command = Command('some', flags=Flags(Flag('test'), Flag('more'))) command = Command('some', flags=Flags(Flag('test'), Flag('more')))
self.assertEqual(command.validate_input_flag(InputFlag('case')), False) self.assertEqual(command.validate_input_flag(InputFlag('case')), 'Undefined')
def test_validate_incorrect_input_flag3(self): def test_validate_undefined_input_flag3(self):
command = Command('some') command = Command('some')
self.assertEqual(command.validate_input_flag(InputFlag('case')), False) self.assertEqual(command.validate_input_flag(InputFlag('case')), 'Undefined')
def test_invalid_input_flag1(self):
command = Command('some', flags=Flag('test', possible_values=False))
self.assertEqual(command.validate_input_flag(InputFlag('test', value='example')), 'Invalid')
def test_invalid_input_flag2(self):
command = Command('some', flags=Flag('test', possible_values=['some', 'case']))
self.assertEqual(command.validate_input_flag(InputFlag('test', value='slay')), 'Invalid')
def test_invalid_input_flag3(self):
command = Command('some', flags=Flag('test', possible_values=re.compile(r'^ex\d{, 2}op$')))
self.assertEqual(command.validate_input_flag(InputFlag('test', value='example')), 'Invalid')
def test_isinstance_parse_correct_raw_command(self): def test_isinstance_parse_correct_raw_command(self):
cmd = InputCommand.parse('ssh --host 192.168.0.3') cmd = InputCommand.parse('ssh --host 192.168.0.3')
+2 -1
View File
@@ -1,4 +1,5 @@
from argenta.command.flag.models import Flag, InputFlag, InputFlags, Flags from argenta.command.flag import Flag, InputFlag
from argenta.command.flags import InputFlags, Flags
import unittest import unittest
import re import re
+33 -53
View File
@@ -1,4 +1,5 @@
from argenta.command.flag import InputFlags, InputFlag, Flag, Flags from argenta.command.flag import InputFlag, Flag
from argenta.command.flags import Flags, InputFlags, UndefinedInputFlags, InvalidValueInputFlags, ValidInputFlags
from argenta.router import Router from argenta.router import Router
from argenta.command import Command from argenta.command import Command
from argenta.router.exceptions import (TriggerContainSpacesException, from argenta.router.exceptions import (TriggerContainSpacesException,
@@ -11,9 +12,6 @@ import re
class TestRouter(unittest.TestCase): 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): def test_register_command_with_spaces_in_trigger(self):
router = Router() router = Router()
with self.assertRaises(TriggerContainSpacesException): with self.assertRaises(TriggerContainSpacesException):
@@ -24,113 +22,95 @@ class TestRouter(unittest.TestCase):
with self.assertRaises(RepeatedFlagNameException): with self.assertRaises(RepeatedFlagNameException):
router._validate_command(Command(trigger='command', flags=Flags(Flag('test'), Flag('test')))) router._validate_command(Command(trigger='command', flags=Flags(Flag('test'), Flag('test'))))
def test_validate_incorrect_input_flag1(self): def test_structuring_input_flags1(self):
router = Router() router = Router()
router.set_invalid_input_flag_handler(lambda flag: None) cmd = Command('cmd')
self.assertEqual(router._validate_input_flags(Command('cmd'), InputFlags(InputFlag('ssh'))), False) input_flags = InputFlags(InputFlag('ssh'))
self.assertEqual(router._structuring_input_flags(cmd, input_flags).undefined_flags, UndefinedInputFlags(InputFlag('ssh')))
def test_validate_incorrect_input_flag2(self): def test_structuring_input_flags2(self):
router = Router() router = Router()
router.set_invalid_input_flag_handler(lambda flag: None) cmd = Command('cmd')
self.assertEqual(router._validate_input_flags(Command('cmd'), InputFlags(InputFlag('ssh', value='some'))), False) input_flags = InputFlags(InputFlag('ssh', value='some'))
self.assertEqual(router._structuring_input_flags(cmd, input_flags).undefined_flags, UndefinedInputFlags(InputFlag('ssh', value='some')))
def test_validate_incorrect_input_flag3(self): def test_structuring_input_flags3(self):
router = Router() router = Router()
router.set_invalid_input_flag_handler(lambda flag: None) cmd = Command('cmd', flags=Flag('port'))
command = Command('cmd', flags=Flag('port'))
input_flags = InputFlags(InputFlag('ssh', value='some2')) input_flags = InputFlags(InputFlag('ssh', value='some2'))
self.assertEqual(router._validate_input_flags(command, input_flags), False) self.assertEqual(router._structuring_input_flags(cmd, input_flags).undefined_flags, UndefinedInputFlags(InputFlag('ssh', value='some2')))
def test_validate_incorrect_input_flag4(self): def test_structuring_input_flags4(self):
router = Router() router = Router()
router.set_invalid_input_flag_handler(lambda flag: None)
command = Command('cmd', flags=Flag('ssh', possible_values=False)) command = Command('cmd', flags=Flag('ssh', possible_values=False))
input_flags = InputFlags(InputFlag('ssh', value='some3')) input_flags = InputFlags(InputFlag('ssh', value='some3'))
self.assertEqual(router._validate_input_flags(command, input_flags), False) self.assertEqual(router._structuring_input_flags(command, input_flags).invalid_value_flags, InvalidValueInputFlags(InputFlag('ssh', value='some3')))
def test_validate_incorrect_input_flag5(self): def test_structuring_input_flags5(self):
router = Router() 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]$'))) command = Command('cmd', flags=Flag('ssh', possible_values=re.compile(r'some[1-5]$')))
input_flags = InputFlags(InputFlag('ssh', value='some40')) input_flags = InputFlags(InputFlag('ssh', value='some40'))
self.assertEqual(router._validate_input_flags(command, input_flags), False) self.assertEqual(router._structuring_input_flags(command, input_flags).invalid_value_flags, InvalidValueInputFlags(InputFlag('ssh', value='some40')))
def test_validate_incorrect_input_flag6(self): def test_structuring_input_flags6(self):
router = Router() router = Router()
router.set_invalid_input_flag_handler(lambda flag: None)
command = Command('cmd', flags=Flag('ssh', possible_values=['example'])) command = Command('cmd', flags=Flag('ssh', possible_values=['example']))
input_flags = InputFlags(InputFlag('ssh', value='example2')) input_flags = InputFlags(InputFlag('ssh', value='example2'))
self.assertEqual(router._validate_input_flags(command, input_flags), False) self.assertEqual(router._structuring_input_flags(command, input_flags).invalid_value_flags, InvalidValueInputFlags(InputFlag('ssh', value='example2')))
def test_validate_incorrect_input_flag7(self): def test_structuring_input_flags7(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')) command = Command('cmd', flags=Flag('port'))
input_flags = InputFlags(InputFlag('port', value='some2')) input_flags = InputFlags(InputFlag('port', value='some2'))
self.assertEqual(Router()._validate_input_flags(command, input_flags), True) self.assertEqual(Router()._structuring_input_flags(command, input_flags).valid_flags, ValidInputFlags(InputFlag('port', value='some2')))
def test_validate_correct_input_flag2(self): def test_structuring_input_flags8(self):
command = Command('cmd', flags=Flag('port', possible_values=['some2', 'some3'])) command = Command('cmd', flags=Flag('port', possible_values=['some2', 'some3']))
input_flags = InputFlags(InputFlag('port', value='some2')) input_flags = InputFlags(InputFlag('port', value='some2'))
self.assertEqual(Router()._validate_input_flags(command, input_flags), True) self.assertEqual(Router()._structuring_input_flags(command, input_flags).valid_flags, ValidInputFlags(InputFlag('port', value='some2')))
def test_validate_correct_input_flag3(self): def test_structuring_input_flags9(self):
command = Command('cmd', flags=Flag('ssh', possible_values=re.compile(r'more[1-5]$'))) command = Command('cmd', flags=Flag('ssh', possible_values=re.compile(r'more[1-5]$')))
input_flags = InputFlags(InputFlag('ssh', value='more5')) input_flags = InputFlags(InputFlag('ssh', value='more5'))
self.assertEqual(Router()._validate_input_flags(command, input_flags), True) self.assertEqual(Router()._structuring_input_flags(command, input_flags).valid_flags, ValidInputFlags(InputFlag('ssh', value='more5')))
def test_validate_correct_input_flag4(self): def test_structuring_input_flags10(self):
command = Command('cmd', flags=Flag('ssh', possible_values=False)) command = Command('cmd', flags=Flag('ssh', possible_values=False))
input_flags = InputFlags(InputFlag('ssh')) input_flags = InputFlags(InputFlag('ssh'))
self.assertEqual(Router()._validate_input_flags(command, input_flags), True) self.assertEqual(Router()._structuring_input_flags(command, input_flags).valid_flags, ValidInputFlags(InputFlag('ssh')))
def test_validate_incorrect_func_args1(self): def test_validate_incorrect_func_args1(self):
command = Command('cmd', flags=Flag('port'))
def handler(): def handler():
pass pass
with self.assertRaises(RequiredArgumentNotPassedException): with self.assertRaises(RequiredArgumentNotPassedException):
Router()._validate_func_args(command, handler) Router()._validate_func_args(handler)
def test_validate_incorrect_func_args2(self): def test_validate_incorrect_func_args2(self):
command = Command('cmd', flags=Flag('port'))
def handler(args, kwargs): def handler(args, kwargs):
pass pass
with self.assertRaises(TooManyTransferredArgsException): with self.assertRaises(TooManyTransferredArgsException):
Router()._validate_func_args(command, handler) Router()._validate_func_args(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): def test_get_router_aliases(self):
router = Router() router = Router()
@router.command(Command('some', aliases=['test', 'case'])) @router.command(Command('some', aliases=['test', 'case']))
def handler(): def handler(response):
pass pass
self.assertListEqual(router.get_aliases(), ['test', 'case']) self.assertListEqual(router.get_aliases(), ['test', 'case'])
def test_get_router_aliases2(self): def test_get_router_aliases2(self):
router = Router() router = Router()
@router.command(Command('some', aliases=['test', 'case'])) @router.command(Command('some', aliases=['test', 'case']))
def handler(): def handler(response):
pass pass
@router.command(Command('ext', aliases=['more', 'foo'])) @router.command(Command('ext', aliases=['more', 'foo']))
def handler2(): def handler2(response):
pass pass
self.assertListEqual(router.get_aliases(), ['test', 'case', 'more', 'foo']) self.assertListEqual(router.get_aliases(), ['test', 'case', 'more', 'foo'])
def test_get_router_aliases3(self): def test_get_router_aliases3(self):
router = Router() router = Router()
@router.command(Command('some')) @router.command(Command('some'))
def handler(): def handler(response):
pass pass
self.assertListEqual(router.get_aliases(), []) self.assertListEqual(router.get_aliases(), [])