5 Commits

Author SHA1 Message Date
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
14 changed files with 237 additions and 162 deletions
+8 -2
View File
@@ -11,7 +11,13 @@ from argenta.router import Router
from argenta.orchestrator import Orchestrator from argenta.orchestrator import Orchestrator
from argenta.command.models import InputCommand from argenta.command.models import InputCommand
from argenta.app.utils import most_similar_command import inspect
print(most_similar_command('case', ['case', 'tester', 'poster', 'caser'])) router = Router()
@router.command(Command('some'))
def handler(res: Response) -> Response:
pass
+1 -1
View File
@@ -20,7 +20,7 @@ def command_help(response: Response):
print(response.invalid_value_flags.get_flags()) print(response.invalid_value_flags.get_flags())
@work_router.command(Command('run', 'Run All')) @work_router.command('run')
def command_start_solving(response: Response): def command_start_solving(response: Response):
print(response.status) print(response.status)
print(response.undefined_flags.get_flags()) print(response.undefined_flags.get_flags())
+3 -3
View File
@@ -1,7 +1,7 @@
[project] [project]
name = "argenta" name = "argenta"
version = "1.0.0-beta1" version = "1.0.0-beta2"
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,7 +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] [dependency-groups]
+2 -2
View File
@@ -238,7 +238,7 @@ class BaseApp:
:return: None :return: None
""" """
self._prompt = '[italic dim bold]What do you want to do?\n' 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 = f'\n[bold red]{text2art(self._initial_message, font="tarty1")}\n'
self._farewell_message = (f'[bold red]\n{text2art(f"\n{self._farewell_message}\n", font="chanky")}[/bold red]\n' self._farewell_message = (f'[bold red]\n{text2art(f"\n{self._farewell_message}\n", font="chanky")}[/bold red]\n'
f'[red i]github.com/koloideal/Argenta[/red i] | [red bold i]made by kolo[/red bold i]\n') 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._description_message_gen = lambda command, description: (f'[bold red]{escape("[" + command + "]")}[/bold red] '
@@ -282,7 +282,7 @@ class BaseApp:
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()
+6
View File
@@ -36,6 +36,9 @@ class BaseFlag:
""" """
return self._prefix return self._prefix
def __eq__(self, other) -> bool:
return self.get_string_entity() == other.get_string_entity()
class Flag(BaseFlag): class Flag(BaseFlag):
def __init__(self, name: str, def __init__(self, name: str,
@@ -110,3 +113,6 @@ class InputFlag(BaseFlag):
""" """
self._flag_value = value self._flag_value = value
def __eq__(self, other) -> bool:
return self.get_string_entity() == other.get_string_entity() and self.get_value() == other.get_value()
+12
View File
@@ -58,6 +58,18 @@ class BaseFlags(Generic[FlagType]):
def __getitem__(self, item): def __getitem__(self, item):
return self._flags[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 Flags(BaseFlags[Flag]): pass
+1 -1
View File
@@ -39,7 +39,7 @@ class Command(BaseCommand):
""" """
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 = flags if isinstance(flags, Flags) else Flags(flags) if isinstance(flags, Flag) else Flags()
self._description = f'Description for "{self._trigger}" command' if not description else description self._description = f'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:
+8 -1
View File
@@ -1,8 +1,15 @@
from argenta.command.flags.models import ValidInputFlags, UndefinedInputFlags, InvalidValueInputFlags
from argenta.response.status import Status from argenta.response.status import Status
from argenta.command.flags import (ValidInputFlags,
UndefinedInputFlags,
InvalidValueInputFlags)
class Response: class Response:
__slots__ = ('status',
'valid_flags',
'undefined_flags',
'invalid_value_flags')
def __init__(self, status: Status = None, def __init__(self, status: Status = None,
valid_flags: ValidInputFlags = ValidInputFlags(), valid_flags: ValidInputFlags = ValidInputFlags(),
undefined_flags: UndefinedInputFlags = UndefinedInputFlags(), undefined_flags: UndefinedInputFlags = UndefinedInputFlags(),
+42 -26
View File
@@ -1,12 +1,15 @@
from typing import Callable, Literal, Type from typing import Callable, Literal, Type
from inspect import getfullargspec, get_annotations from inspect import getfullargspec, get_annotations, getsourcefile, getsourcelines
from rich.console import Console 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.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.flags.models import Flags, InputFlags, UndefinedInputFlags, ValidInputFlags, InvalidValueInputFlags from argenta.command.flags.models import (Flags, InputFlags,
UndefinedInputFlags,
ValidInputFlags,
InvalidValueInputFlags)
from argenta.router.exceptions import (RepeatedFlagNameException, from argenta.router.exceptions import (RepeatedFlagNameException,
TooManyTransferredArgsException, TooManyTransferredArgsException,
RequiredArgumentNotPassedException, RequiredArgumentNotPassedException,
@@ -26,13 +29,15 @@ class Router:
self._ignore_command_register: bool = False self._ignore_command_register: bool = False
def command(self, command: Command) -> Callable: def command(self, command: Command | str) -> Callable:
""" """
Public. Registers handler Public. Registers handler
:param command: Registered command :param command: Registered command
:return: decorated handler as Callable :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(func) Router._validate_func_args(func)
@@ -73,9 +78,7 @@ class Router:
response: Response = Response() 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():
flags, status = self._validate_input_flags(handle_command, input_command_flags) response: Response = self._structuring_input_flags(handle_command, input_command_flags)
response.valid_flags, response.undefined_flags, response.invalid_value_flags = flags
response.status = status
command_handler.handling(response) command_handler.handling(response)
else: else:
response.status = Status.ALL_FLAGS_VALID response.status = Status.ALL_FLAGS_VALID
@@ -92,15 +95,12 @@ class Router:
@staticmethod @staticmethod
def _validate_input_flags(handled_command: Command, input_flags: InputFlags) -> tuple[tuple[ValidInputFlags, def _structuring_input_flags(handled_command: Command, input_flags: InputFlags) -> Response:
UndefinedInputFlags,
InvalidValueInputFlags],
Status]:
""" """
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() valid_input_flags: ValidInputFlags = ValidInputFlags()
invalid_value_input_flags: InvalidValueInputFlags = InvalidValueInputFlags() invalid_value_input_flags: InvalidValueInputFlags = InvalidValueInputFlags()
@@ -124,25 +124,32 @@ class Router:
else: else:
status = Status.UNDEFINED_AND_INVALID_FLAGS status = Status.UNDEFINED_AND_INVALID_FLAGS
return (valid_input_flags, undefined_input_flags, invalid_value_input_flags), status 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
@@ -158,10 +165,19 @@ class Router:
elif len(transferred_args) == 0: elif len(transferred_args) == 0:
raise RequiredArgumentNotPassedException() raise RequiredArgumentNotPassedException()
arg_annotation: Type = get_annotations(func)[transferred_args[0]] transferred_arg: str = transferred_args[0]
if not arg_annotation is Response: func_annotations: dict[str, Type] = get_annotations(func)
Console().print(f'\n\n[b red]WARNING:[/b red] [i]The type of argument passed to the handler is [/i][blue]{Response}[/blue],'
f' [i]but[/i] [bold blue]{arg_annotation}[/bold blue] [i]is specified[/i]', highlight=False) 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)
@@ -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_invalid_input_flags_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
+34 -50
View File
@@ -1,4 +1,6 @@
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.response import Response
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,
@@ -24,113 +26,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(), [])