mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 18:15:28 +03:00
Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 07ac2af71e | |||
| c4b3aa7db8 | |||
| 61ef6a6466 | |||
| 477f3a7dec | |||
| adf3431388 |
+8
-2
@@ -11,7 +11,13 @@ from argenta.router import Router
|
||||
from argenta.orchestrator import Orchestrator
|
||||
|
||||
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
|
||||
|
||||
|
||||
@@ -20,7 +20,7 @@ def command_help(response: Response):
|
||||
print(response.invalid_value_flags.get_flags())
|
||||
|
||||
|
||||
@work_router.command(Command('run', 'Run All'))
|
||||
@work_router.command('run')
|
||||
def command_start_solving(response: Response):
|
||||
print(response.status)
|
||||
print(response.undefined_flags.get_flags())
|
||||
|
||||
+3
-3
@@ -1,7 +1,7 @@
|
||||
[project]
|
||||
name = "argenta"
|
||||
version = "1.0.0-beta1"
|
||||
description = "Python library for creating TUI"
|
||||
version = "1.0.0-beta2"
|
||||
description = "Python library for building modular CLI applications"
|
||||
authors = [{ name = "kolo", email = "kolo.is.main@gmail.com" }]
|
||||
requires-python = ">=3.11, <4.0"
|
||||
readme = "README.md"
|
||||
@@ -9,7 +9,7 @@ license = { text = "MIT" }
|
||||
dependencies = [
|
||||
"rich (>=14.0.0,<15.0.0)",
|
||||
"art (>=6.4,<7.0)",
|
||||
"pyreadline3 (>=3.5.4,<4.0.0)",
|
||||
"pyreadline3>=3.5.4",
|
||||
]
|
||||
|
||||
[dependency-groups]
|
||||
|
||||
@@ -238,7 +238,7 @@ class BaseApp:
|
||||
:return: None
|
||||
"""
|
||||
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'
|
||||
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] '
|
||||
@@ -282,7 +282,7 @@ class BaseApp:
|
||||
for message in self._messages_on_startup:
|
||||
self._print_func(message)
|
||||
if self._messages_on_startup:
|
||||
print('\n\n')
|
||||
print('\n')
|
||||
|
||||
if not self._repeat_command_groups_description:
|
||||
self._print_command_group_description()
|
||||
|
||||
@@ -36,6 +36,9 @@ class BaseFlag:
|
||||
"""
|
||||
return self._prefix
|
||||
|
||||
def __eq__(self, other) -> bool:
|
||||
return self.get_string_entity() == other.get_string_entity()
|
||||
|
||||
|
||||
class Flag(BaseFlag):
|
||||
def __init__(self, name: str,
|
||||
@@ -110,3 +113,6 @@ class InputFlag(BaseFlag):
|
||||
"""
|
||||
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()
|
||||
|
||||
|
||||
@@ -58,6 +58,18 @@ class BaseFlags(Generic[FlagType]):
|
||||
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
|
||||
|
||||
|
||||
@@ -39,7 +39,7 @@ class Command(BaseCommand):
|
||||
"""
|
||||
super().__init__(trigger)
|
||||
self._registered_flags: Flags = flags if isinstance(flags, Flags) else Flags(flags) if isinstance(flags, Flag) else Flags()
|
||||
self._description = f'Description for "{self._trigger}" command' if not description else description
|
||||
self._description = f'Very useful command' if not description else description
|
||||
self._aliases = aliases if isinstance(aliases, list) else []
|
||||
|
||||
def get_registered_flags(self) -> Flags:
|
||||
|
||||
@@ -1,8 +1,15 @@
|
||||
from argenta.command.flags.models import ValidInputFlags, UndefinedInputFlags, InvalidValueInputFlags
|
||||
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(),
|
||||
|
||||
@@ -1,12 +1,15 @@
|
||||
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 argenta.command import Command
|
||||
from argenta.command.models import InputCommand
|
||||
from argenta.response import Response, Status
|
||||
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,
|
||||
TooManyTransferredArgsException,
|
||||
RequiredArgumentNotPassedException,
|
||||
@@ -26,13 +29,15 @@ class Router:
|
||||
self._ignore_command_register: bool = False
|
||||
|
||||
|
||||
def command(self, command: Command) -> Callable:
|
||||
def command(self, command: Command | str) -> Callable:
|
||||
"""
|
||||
Public. Registers handler
|
||||
:param command: Registered command
|
||||
:return: decorated handler as Callable
|
||||
"""
|
||||
self._validate_command(command)
|
||||
if isinstance(command, str):
|
||||
command = Command(command)
|
||||
|
||||
def command_decorator(func):
|
||||
Router._validate_func_args(func)
|
||||
@@ -73,9 +78,7 @@ class Router:
|
||||
response: Response = Response()
|
||||
if handle_command.get_registered_flags().get_flags():
|
||||
if input_command_flags.get_flags():
|
||||
flags, status = self._validate_input_flags(handle_command, input_command_flags)
|
||||
response.valid_flags, response.undefined_flags, response.invalid_value_flags = flags
|
||||
response.status = status
|
||||
response: Response = self._structuring_input_flags(handle_command, input_command_flags)
|
||||
command_handler.handling(response)
|
||||
else:
|
||||
response.status = Status.ALL_FLAGS_VALID
|
||||
@@ -92,15 +95,12 @@ class Router:
|
||||
|
||||
|
||||
@staticmethod
|
||||
def _validate_input_flags(handled_command: Command, input_flags: InputFlags) -> tuple[tuple[ValidInputFlags,
|
||||
UndefinedInputFlags,
|
||||
InvalidValueInputFlags],
|
||||
Status]:
|
||||
def _structuring_input_flags(handled_command: Command, input_flags: InputFlags) -> Response:
|
||||
"""
|
||||
Private. Validates flags of input command
|
||||
:param handled_command: entity of the handled command
|
||||
:param input_flags:
|
||||
:return: is flags of input command valid as bool
|
||||
:return: entity of response as Response
|
||||
"""
|
||||
valid_input_flags: ValidInputFlags = ValidInputFlags()
|
||||
invalid_value_input_flags: InvalidValueInputFlags = InvalidValueInputFlags()
|
||||
@@ -124,25 +124,32 @@ class Router:
|
||||
else:
|
||||
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
|
||||
def _validate_command(command: Command) -> None:
|
||||
def _validate_command(command: Command | str) -> None:
|
||||
"""
|
||||
Private. Validates the command registered in handler
|
||||
:param command: validated command
|
||||
:return: None if command is valid else raise exception
|
||||
"""
|
||||
match type(command).__name__:
|
||||
case 'Command':
|
||||
command_name: str = command.get_trigger()
|
||||
if command_name.find(' ') != -1:
|
||||
raise TriggerContainSpacesException()
|
||||
|
||||
flags: Flags = command.get_registered_flags()
|
||||
if flags:
|
||||
flags_name: list = [x.get_string_entity().lower() for x in flags]
|
||||
if len(set(flags_name)) < len(flags_name):
|
||||
raise RepeatedFlagNameException()
|
||||
case 'str':
|
||||
if command.find(' ') != -1:
|
||||
raise TriggerContainSpacesException()
|
||||
|
||||
|
||||
@staticmethod
|
||||
@@ -158,10 +165,19 @@ class Router:
|
||||
elif len(transferred_args) == 0:
|
||||
raise RequiredArgumentNotPassedException()
|
||||
|
||||
arg_annotation: Type = get_annotations(func)[transferred_args[0]]
|
||||
if not arg_annotation is Response:
|
||||
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)
|
||||
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)
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -1,32 +1,35 @@
|
||||
import _io
|
||||
from unittest.mock import patch, MagicMock
|
||||
import unittest
|
||||
from unittest import TestCase
|
||||
import io
|
||||
import re
|
||||
|
||||
from argenta.app import App
|
||||
from argenta.command import Command
|
||||
from argenta.router import Router
|
||||
from argenta.command.flag.models import Flags, InputFlags
|
||||
from argenta.command.flags.models import Flags
|
||||
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("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
def test(response: Response):
|
||||
print('test command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
@@ -37,9 +40,10 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_incorrect_command2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
def test(response: Response):
|
||||
print('test command')
|
||||
|
||||
app = App(ignore_command_register=False,
|
||||
@@ -47,7 +51,7 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
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()
|
||||
|
||||
@@ -58,74 +62,80 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_unregistered_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
print(f'test command')
|
||||
def test(response: Response):
|
||||
print(f'test command with undefined flag: {response.undefined_flags.get_flag('help').get_string_entity()}')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
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("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_unregistered_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
print('test command')
|
||||
def test(response: Response):
|
||||
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,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
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("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_one_correct_flag_an_one_incorrect_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
flags = Flags(PredefinedFlags.HOST)
|
||||
|
||||
@router.command(Command('test', flags=flags))
|
||||
def test(args: InputFlags):
|
||||
print(f'connecting to host {args.get_flag('host').get_value()}')
|
||||
def test(response: Response):
|
||||
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,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
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("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_one_correct_command_and_one_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
def test(response: Response):
|
||||
print(f'test command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
@@ -136,20 +146,21 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_two_correct_commands_and_one_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
def test(response: Response):
|
||||
print(f'test command')
|
||||
|
||||
@router.command(Command('more'))
|
||||
def test():
|
||||
def test(response: Response):
|
||||
print(f'more command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
@@ -160,16 +171,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_incorrect_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
def test(response: Response):
|
||||
print(f'test command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.set_invalid_input_flags_handler(lambda command: print(f'Incorrect flag syntax: "{command}"'))
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
@@ -180,16 +192,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_empty_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
def test(response: Response):
|
||||
print(f'test command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.set_empty_command_handler(lambda: print('Empty input command'))
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
@@ -200,16 +213,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_repeated_flags(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test', flags=PredefinedFlags.PORT))
|
||||
def test(args: InputFlags):
|
||||
def test(response: Response):
|
||||
print('test command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.set_repeated_input_flags_handler(lambda command: print(f'Repeated input flags: "{command}"'))
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
|
||||
@@ -1,31 +1,35 @@
|
||||
import _io
|
||||
from unittest.mock import patch, MagicMock
|
||||
import unittest
|
||||
from unittest import TestCase
|
||||
import io
|
||||
import re
|
||||
|
||||
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.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
|
||||
|
||||
|
||||
|
||||
class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
class TestSystemHandlerNormalWork(TestCase):
|
||||
@patch("builtins.input", side_effect=["test", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
def test(response):
|
||||
print('test command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
@@ -36,16 +40,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
def test(response):
|
||||
print('test command')
|
||||
|
||||
app = App(ignore_command_register=True,
|
||||
override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
@@ -56,16 +61,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_custom_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
flag = Flag('help', '--', False)
|
||||
|
||||
@router.command(Command('test', flags=flag))
|
||||
def test(args: InputFlags):
|
||||
print(f'\nhelp for {args.get_flag('help').get_name()} flag\n')
|
||||
def test(response: Response):
|
||||
print(f'\nhelp for {response.valid_flags.get_flag('help').get_name()} flag\n')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
@@ -75,16 +81,18 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_custom_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
flag = Flag('port', '--', re.compile(r'^\d{1,5}$'))
|
||||
|
||||
@router.command(Command('test', flags=flag))
|
||||
def test(args: InputFlags):
|
||||
print(f'flag value for {args.get_flag('port').get_name()} flag : {args.get_flag('port').get_value()}')
|
||||
def test(response: Response):
|
||||
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,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
@@ -95,16 +103,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_default_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
flag = PredefinedFlags.SHORT_HELP
|
||||
|
||||
@router.command(Command('test', flags=flag))
|
||||
def test(args: InputFlags):
|
||||
print(f'help for {args.get_flag('H').get_name()} flag')
|
||||
def test(response: Response):
|
||||
print(f'help for {response.valid_flags.get_flag('H').get_name()} flag')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
@@ -115,17 +124,18 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_default_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
flag = PredefinedFlags.INFO
|
||||
|
||||
@router.command(Command('test', flags=flag))
|
||||
def test(args: InputFlags):
|
||||
if args.get_flag('info'):
|
||||
def test(response: Response):
|
||||
if response.valid_flags.get_flag('info'):
|
||||
print('info about test command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
@@ -136,16 +146,17 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_default_flag3(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
flag = PredefinedFlags.HOST
|
||||
|
||||
@router.command(Command('test', flags=flag))
|
||||
def test(args: InputFlags):
|
||||
print(f'connecting to host {args[0].get_value()}')
|
||||
def test(response: Response):
|
||||
print(f'connecting to host {response.valid_flags.get_flag('host').get_value()}')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
@@ -156,16 +167,18 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_two_flags(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
flags = Flags(PredefinedFlags.HOST, PredefinedFlags.PORT)
|
||||
|
||||
@router.command(Command('test', flags=flags))
|
||||
def test(args: InputFlags):
|
||||
print(f'connecting to host {args[0].get_value()} and port {args[1].get_value()}')
|
||||
def test(response: Response):
|
||||
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,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
@@ -176,19 +189,20 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_two_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
def test(response):
|
||||
print(f'test command')
|
||||
|
||||
@router.command(Command('some'))
|
||||
def test2():
|
||||
def test2(response):
|
||||
print(f'some command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
@@ -199,23 +213,24 @@ class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_three_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
def test(response):
|
||||
print(f'test command')
|
||||
|
||||
@router.command(Command('some'))
|
||||
def test():
|
||||
def test(response):
|
||||
print(f'some command')
|
||||
|
||||
@router.command(Command('more'))
|
||||
def test():
|
||||
def test(response):
|
||||
print(f'more command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.run_polling()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
|
||||
@@ -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.exceptions import (UnprocessedInputFlagException,
|
||||
RepeatedInputFlagsException,
|
||||
EmptyInputCommandException)
|
||||
|
||||
import unittest
|
||||
import re
|
||||
|
||||
|
||||
class TestInputCommand(unittest.TestCase):
|
||||
@@ -23,25 +25,37 @@ class TestInputCommand(unittest.TestCase):
|
||||
with self.assertRaises(EmptyInputCommandException):
|
||||
InputCommand.parse('')
|
||||
|
||||
def test_validate_correct_input_flag1(self):
|
||||
def test_validate_valid_input_flag1(self):
|
||||
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')))
|
||||
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):
|
||||
command = Command('some', flags=Flags(Flag('test')))
|
||||
self.assertEqual(command.validate_input_flag(InputFlag('more')), False)
|
||||
def test_validate_undefined_input_flag1(self):
|
||||
command = Command('some', flags=Flag('test'))
|
||||
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')))
|
||||
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')
|
||||
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):
|
||||
cmd = InputCommand.parse('ssh --host 192.168.0.3')
|
||||
|
||||
@@ -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 re
|
||||
|
||||
@@ -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.command import Command
|
||||
from argenta.router.exceptions import (TriggerContainSpacesException,
|
||||
@@ -24,113 +26,95 @@ class TestRouter(unittest.TestCase):
|
||||
with self.assertRaises(RepeatedFlagNameException):
|
||||
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.set_invalid_input_flag_handler(lambda flag: None)
|
||||
self.assertEqual(router._validate_input_flags(Command('cmd'), InputFlags(InputFlag('ssh'))), False)
|
||||
cmd = Command('cmd')
|
||||
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.set_invalid_input_flag_handler(lambda flag: None)
|
||||
self.assertEqual(router._validate_input_flags(Command('cmd'), InputFlags(InputFlag('ssh', value='some'))), False)
|
||||
cmd = Command('cmd')
|
||||
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.set_invalid_input_flag_handler(lambda flag: None)
|
||||
command = Command('cmd', flags=Flag('port'))
|
||||
cmd = Command('cmd', flags=Flag('port'))
|
||||
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.set_invalid_input_flag_handler(lambda flag: None)
|
||||
command = Command('cmd', flags=Flag('ssh', possible_values=False))
|
||||
input_flags = InputFlags(InputFlag('ssh', value='some3'))
|
||||
self.assertEqual(router._validate_input_flags(command, input_flags), False)
|
||||
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.set_invalid_input_flag_handler(lambda flag: None)
|
||||
command = Command('cmd', flags=Flag('ssh', possible_values=re.compile(r'some[1-5]$')))
|
||||
input_flags = InputFlags(InputFlag('ssh', value='some40'))
|
||||
self.assertEqual(router._validate_input_flags(command, input_flags), False)
|
||||
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.set_invalid_input_flag_handler(lambda flag: None)
|
||||
command = Command('cmd', flags=Flag('ssh', possible_values=['example']))
|
||||
input_flags = InputFlags(InputFlag('ssh', value='example2'))
|
||||
self.assertEqual(router._validate_input_flags(command, input_flags), False)
|
||||
self.assertEqual(router._structuring_input_flags(command, input_flags).invalid_value_flags, InvalidValueInputFlags(InputFlag('ssh', value='example2')))
|
||||
|
||||
def test_validate_incorrect_input_flag7(self):
|
||||
router = Router()
|
||||
router.set_invalid_input_flag_handler(lambda flag: None)
|
||||
command = Command('cmd', flags=Flag('ssh', possible_values=['example']))
|
||||
input_flags = InputFlags(InputFlag('ssh'))
|
||||
self.assertEqual(router._validate_input_flags(command, input_flags), False)
|
||||
|
||||
def test_validate_correct_input_flag1(self):
|
||||
def test_structuring_input_flags7(self):
|
||||
command = Command('cmd', flags=Flag('port'))
|
||||
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']))
|
||||
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]$')))
|
||||
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))
|
||||
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):
|
||||
command = Command('cmd', flags=Flag('port'))
|
||||
def handler():
|
||||
pass
|
||||
with self.assertRaises(RequiredArgumentNotPassedException):
|
||||
Router()._validate_func_args(command, handler)
|
||||
Router()._validate_func_args(handler)
|
||||
|
||||
def test_validate_incorrect_func_args2(self):
|
||||
command = Command('cmd', flags=Flag('port'))
|
||||
def handler(args, kwargs):
|
||||
pass
|
||||
with self.assertRaises(TooManyTransferredArgsException):
|
||||
Router()._validate_func_args(command, handler)
|
||||
|
||||
def test_validate_incorrect_func_args3(self):
|
||||
command = Command('cmd')
|
||||
def handler(args):
|
||||
pass
|
||||
with self.assertRaises(TooManyTransferredArgsException):
|
||||
Router()._validate_func_args(command, handler)
|
||||
Router()._validate_func_args(handler)
|
||||
|
||||
def test_get_router_aliases(self):
|
||||
router = Router()
|
||||
@router.command(Command('some', aliases=['test', 'case']))
|
||||
def handler():
|
||||
def handler(response):
|
||||
pass
|
||||
self.assertListEqual(router.get_aliases(), ['test', 'case'])
|
||||
|
||||
def test_get_router_aliases2(self):
|
||||
router = Router()
|
||||
@router.command(Command('some', aliases=['test', 'case']))
|
||||
def handler():
|
||||
def handler(response):
|
||||
pass
|
||||
@router.command(Command('ext', aliases=['more', 'foo']))
|
||||
def handler2():
|
||||
def handler2(response):
|
||||
pass
|
||||
self.assertListEqual(router.get_aliases(), ['test', 'case', 'more', 'foo'])
|
||||
|
||||
def test_get_router_aliases3(self):
|
||||
router = Router()
|
||||
@router.command(Command('some'))
|
||||
def handler():
|
||||
def handler(response):
|
||||
pass
|
||||
self.assertListEqual(router.get_aliases(), [])
|
||||
|
||||
|
||||
Reference in New Issue
Block a user