mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 18:15:28 +03:00
Compare commits
99 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| a6db733204 | |||
| 8506e4ffcf | |||
| 04d3329572 | |||
| 5bfdde4bd9 | |||
| e2dd7e4aea | |||
| d1d644d422 | |||
| 8b496aa782 | |||
| 592d128ef6 | |||
| b44ee227fd | |||
| 0dce4a0d9e | |||
| ca6634c6f0 | |||
| c1805af420 | |||
| 0e308ce77f | |||
| ab1d335f8e | |||
| 1a2e9d1487 | |||
| 76bcba9340 | |||
| ae9795bd53 | |||
| 7540728f1b | |||
| 54da63dd03 | |||
| 8e08d0fe09 | |||
| 55b88f7c8a | |||
| 1cd616336f | |||
| 253790fe2e | |||
| 1c6f896b73 | |||
| 8810e12551 | |||
| 285007a59a | |||
| 6edd17646a | |||
| 154ee25dde | |||
| 30cf3cfd06 | |||
| 0d98d80919 | |||
| 54992e55cb | |||
| cc8135b733 | |||
| 5c6fa5151a | |||
| 2918bc9f81 | |||
| 6e2fbc23e9 | |||
| 1ec8ea53b4 | |||
| 4256d67789 | |||
| 0246ff4b22 | |||
| 956febc757 | |||
| beafdd0afd | |||
| b172e2cdc3 | |||
| baaf0e25f3 | |||
| 09c40978a1 | |||
| 0f4f48555e | |||
| d30515c1a2 | |||
| 5a6fc1d8ca | |||
| e5d6ead38e | |||
| b61c151e1c | |||
| a57ce490c1 | |||
| edbd45f0bf | |||
| d59d274bca | |||
| 2bf2144815 | |||
| 971258728c | |||
| 2c9c8da13c | |||
| 404758bd91 | |||
| 8fbf651223 | |||
| 459c16ec87 | |||
| d9c74310c3 | |||
| 5e6cdc342e | |||
| a378163431 | |||
| fd4f2e1570 | |||
| 3f7c577c29 | |||
| b72fcc6a11 | |||
| 158e5eb75a | |||
| 46cb2f70c5 | |||
| ef242a5732 | |||
| f87f102ced | |||
| ddf2a2fb10 | |||
| eac5358ead | |||
| 76c18ddbff | |||
| 4eeb4eb182 | |||
| 905698384a | |||
| 79ccfbb3b1 | |||
| a63c46a78b | |||
| a3a7cbf2e6 | |||
| a9e545f3d8 | |||
| 37b62fd69b | |||
| 0ae86d0b2b | |||
| ebfd5a80b3 | |||
| 250704fc88 | |||
| 87239f1501 | |||
| 48c117dd72 | |||
| 1d782f6213 | |||
| ddf396a644 | |||
| c10af04280 | |||
| 890d863391 | |||
| 29b184b2ed | |||
| 6d331a57c1 | |||
| b0eb1e3e6c | |||
| 2d088caaaf | |||
| b4b7d5442c | |||
| 684760121c | |||
| dfc3c472ce | |||
| 8a3f742636 | |||
| 213525915d | |||
| 6174de3bc0 | |||
| c78609caca | |||
| 3f88c5278e | |||
| 93438f8e45 |
@@ -0,0 +1,30 @@
|
||||
name: ruff
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [ "kolo" ]
|
||||
pull_request:
|
||||
branches: [ "kolo" ]
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
jobs:
|
||||
build:
|
||||
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v3
|
||||
|
||||
- uses: actions/setup-python@v3
|
||||
with:
|
||||
python-version: "3.13"
|
||||
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
python -m pip install --upgrade pip
|
||||
pip install ruff
|
||||
|
||||
- name: Run linter
|
||||
run: ruff check ./argenta
|
||||
@@ -0,0 +1,31 @@
|
||||
name: tests
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [ "kolo" ]
|
||||
pull_request:
|
||||
branches: [ "kolo" ]
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
jobs:
|
||||
build:
|
||||
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v3
|
||||
|
||||
- uses: actions/setup-python@v3
|
||||
with:
|
||||
python-version: "3.13"
|
||||
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
python -m pip install --upgrade pip
|
||||
pip install poetry
|
||||
poetry install
|
||||
|
||||
- name: Run tests
|
||||
run: poetry run python -m unittest discover
|
||||
+4
-3
@@ -1,6 +1,7 @@
|
||||
.venv
|
||||
.idea
|
||||
argenta/router/__pycache__/
|
||||
argenta/app/__pycache__/
|
||||
argenta/__pycache__/
|
||||
dist
|
||||
poetry.lock
|
||||
*__pycache__/
|
||||
*.hist*
|
||||
|
||||
|
||||
@@ -1,2 +1,583 @@
|
||||
# Argenta
|
||||
Python library for creating cli apps
|
||||
|
||||
---
|
||||
|
||||
## Описание
|
||||
**Argenta** — Python library for creating TUI
|
||||
|
||||

|
||||
Пример внешнего вида TUI, написанного с помощью Argenta
|
||||
|
||||
---
|
||||
|
||||
# Установка
|
||||
```bash
|
||||
pip install argenta
|
||||
```
|
||||
or
|
||||
```bash
|
||||
poetry add argenta
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
# Быстрый старт
|
||||
|
||||
Пример простейшей оболочки с командой без зарегистрированных флагов
|
||||
```python
|
||||
# routers.py
|
||||
from argenta.router import Router
|
||||
from argenta.command import Command
|
||||
|
||||
|
||||
router = Router()
|
||||
|
||||
@router.command(Command("hello"))
|
||||
def handler():
|
||||
print("Hello, world!")
|
||||
```
|
||||
|
||||
```python
|
||||
# main.py
|
||||
from argenta.app import App
|
||||
from routers import router
|
||||
|
||||
app: App = App()
|
||||
|
||||
def main() -> None:
|
||||
app.include_router(router)
|
||||
app.start_polling()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
```
|
||||
Пример оболочки с командой, у которой зарегистрированы флаги
|
||||
|
||||
```python
|
||||
# routers.py
|
||||
import re
|
||||
from argenta.router import Router
|
||||
from argenta.command import Command
|
||||
from argenta.command.flag import Flags, Flag, InputFlags
|
||||
|
||||
router = Router()
|
||||
|
||||
registered_flags = Flags(
|
||||
Flag(name='host',
|
||||
prefix='--',
|
||||
possible_values=re.compile(r'^192.168.\d{1,3}.\d{1,3}$')),
|
||||
Flag('port', '--', re.compile(r'^[0-9]{1,4}$')))
|
||||
|
||||
|
||||
@router.command(Command("hello"))
|
||||
def handler():
|
||||
print("Hello, world!")
|
||||
|
||||
|
||||
@router.command(Command(trigger="ssh",
|
||||
description='connect via ssh',
|
||||
flags=registered_flags))
|
||||
def handler_with_flags(flags: InputFlags):
|
||||
for flag in flags:
|
||||
print(f'Flag name: {flag.get_name()}\n'
|
||||
f'Flag value: {flag.get_value()}')
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
# *classes* :
|
||||
|
||||
---
|
||||
|
||||
## *class* : : `App`
|
||||
Класс, определяющий поведение и состояние оболочки
|
||||
|
||||
### Конструктор
|
||||
```python
|
||||
App(prompt: str = '[italic dim bold]What do you want to do?\n',
|
||||
initial_message: str = '\nArgenta\n',
|
||||
farewell_message: str = '\nSee you\n',
|
||||
exit_command: Command = Command('Q', 'Exit command'),
|
||||
system_points_title: str | None = 'System points:',
|
||||
ignore_command_register: bool = True,
|
||||
dividing_line: StaticDividingLine | DynamicDividingLine = StaticDividingLine(),
|
||||
repeat_command_groups: bool = True,
|
||||
override_system_messages: bool = False,
|
||||
autocompleter: AutoCompleter = AutoCompleter(),
|
||||
print_func: Callable[[str], None] = Console().print)
|
||||
```
|
||||
**Аргументы:**
|
||||
- **name : mean**
|
||||
- `prompt` (`str`): Сообщение перед вводом команды.
|
||||
- `initial_message` (`str`): Приветственное сообщение при запуске.
|
||||
- `farewell_message` (`str`): Сообщение при выходе.
|
||||
- `exit_command` (`Command`): Сущность команды, которая будет отвечать за завершение работы.
|
||||
- `system_points_title` (`str`): Заголовок перед списком системных команд.
|
||||
- `ignore_command_register` (`bool`): Игнорировать регистр всех команд.
|
||||
- `dividing_line` (`StaticDividingLine | DynamicDividingLine`): Разделительная строка.
|
||||
- `repeat_command_groups` (`bool`): Повторять описание команд перед вводом.
|
||||
- `override_system_messages` (`bool`): Переопределить ли дефолтное оформление сообщений ([подробнее см.](#override_defaults))
|
||||
- `autocompleter` (`AutoCompleter`): Сущность автодополнителя ввода.
|
||||
- `print_func` (`Callable[[str], None]`): Функция вывода текста в терминал.
|
||||
|
||||
---
|
||||
|
||||
### ***methods***
|
||||
|
||||
---
|
||||
|
||||
#### **.start_polling() -> `None`**
|
||||
|
||||
*method mean* **::** Запускает цикл обработки ввода
|
||||
|
||||
---
|
||||
|
||||
#### **.include_router(router: Router) -> `None`**
|
||||
|
||||
*param* `router: Router` **::** Регистрируемый роутер
|
||||
*required* **::** True
|
||||
|
||||
*method mean* **::** Регистрирует роутер в оболочке
|
||||
|
||||
---
|
||||
|
||||
#### **.include_routers(\*routers: Router) -> `None`**
|
||||
|
||||
*param* `routers: Router` **::** Неограниченное количество регистрируемых роутеров
|
||||
*required* **::** True
|
||||
|
||||
*method mean* **::** Регистрирует роутер в оболочке
|
||||
|
||||
---
|
||||
|
||||
#### **.set_description_message_pattern(pattern: str) -> `None`**
|
||||
|
||||
*param* `pattern: str` **::** Паттерн описания команды при её выводе в консоль
|
||||
*required* **::** True
|
||||
*example* **::** `"[{command}] *=*=* {description}"`
|
||||
|
||||
*method mean* **::** Устанавливает паттерн описания команд, который будет использован
|
||||
при выводе в консоль
|
||||
|
||||
---
|
||||
|
||||
#### **.add_message_on_startup(message: str) -> `None`**
|
||||
|
||||
*param* `message: str` **::** Сообщение, которое будет выведено при запуске приложения
|
||||
*required* **::** True
|
||||
*example* **::** `Message on startup`
|
||||
|
||||
*method mean* **::** Устанавливает паттерн описания команд, который будет использован
|
||||
при выводе в консоль
|
||||
|
||||
---
|
||||
|
||||
<a name="custom_handler"></a>
|
||||
#### **.repeated_input_flags_handler: `Callable[[str], None])`**
|
||||
|
||||
*example* **::** `lambda raw_command: print_func(f'Repeated input flags: "{raw_command}"')`
|
||||
|
||||
*attr mean* **::** Устанавливает функцию, которой будет передано управление при
|
||||
вводе юзером повторяющихся флагов
|
||||
|
||||
---
|
||||
|
||||
#### **.invalid_input_flags_handler: `Callable[[str], None])`**
|
||||
|
||||
*example* **::** `lambda raw_command: print_func(f'Incorrect flag syntax: "{raw_command}"')`
|
||||
|
||||
*attr mean* **::** Устанавливает функцию, которой будет передано управление при
|
||||
вводе юзером команды с некорректным синтаксисом флагов
|
||||
|
||||
---
|
||||
|
||||
#### **.unknown_command_handler: `Callable[[str], None]`**
|
||||
|
||||
*example* **::** `lambda command: print_func(f"Unknown command: {command.get_string_entity()}")`
|
||||
|
||||
*attr mean* **::** Устанавливает функцию, которой будет передано управление при
|
||||
вводе юзером неизвестной команды
|
||||
|
||||
---
|
||||
|
||||
#### **.empty_command_handler: `Callable[[str], None])`**
|
||||
|
||||
*example* **::** `lambda: print_func(f'Empty input command')`
|
||||
|
||||
*attr mean* **::** Устанавливает функцию, которой будет передано управление при
|
||||
вводе юзером пустой команды
|
||||
|
||||
---
|
||||
|
||||
### Примечания
|
||||
|
||||
- В устанавливаемом паттерне сообщения описания команды необходимы быть два ключевых слова:
|
||||
`command` и `description`, каждое из которых должно быть заключено в фигурные скобки, после обработки
|
||||
паттерна на места этих ключевых слов будут подставлены соответствующие атрибуты команды, при отсутствии
|
||||
этих двух ключевых слов будет вызвано исключение `InvalidDescriptionMessagePatternException`
|
||||
|
||||
- Команды оболочки не должны повторяться, при значении атрибута `ignore_command_register` равным `True`
|
||||
допускается создание обработчиков для разных регистров одинаковых символов в команде, для примера `u` и `U`,
|
||||
при значении атрибута `ignore_command_register` класса `App` равным `False` тот же пример вызывает исключение
|
||||
`RepeatedCommandInDifferentRoutersException`. Исключение вызывается только при наличии пересекающихся команд
|
||||
у __<u>разных</u>__ роутеров
|
||||
|
||||
- Наиболее частые сообщение при запуске предопределены и доступны для быстрого
|
||||
использования: `argenta.app.defaults.PredeterminedMessages`
|
||||
|
||||
<a name="override_defaults"></a>
|
||||
- Если `override_system_messages`=`False`, то при переопределении таких атрибутов как `initial_message` и
|
||||
`farawell_message` будет использовано дефолтное оформление текста, в виде красного ascii арта, при значении
|
||||
`override_system_messages`=`True` системные сообщения будут отображены в точности какими были переданы
|
||||
|
||||
|
||||
|
||||
|
||||
### Исключения
|
||||
|
||||
- `InvalidRouterInstanceException` — Переданный объект в метод `App().include_router()` не является экземпляром класса `Router`.
|
||||
- `InvalidDescriptionMessagePatternException` — Неправильный формат паттерна описания команд.
|
||||
- `IncorrectNumberOfHandlerArgsException` — У обработчика нестандартного поведения зарегистрировано неверное количество аргументов(в большинстве случаев у него должен быть один аргумент).
|
||||
- `NoRegisteredHandlersException` — У роутера нет ни одного обработчика команд.
|
||||
|
||||
---
|
||||
|
||||
## *class* :: `AutoCompleter`
|
||||
Класс, экземпляр которого представляет собой автодополнитель ввода
|
||||
|
||||
### Конструктор
|
||||
```python
|
||||
AutoCompleter(history_filename: str = False,
|
||||
autocomplete_button: str = 'tab')
|
||||
```
|
||||
|
||||
**Аргументы:**
|
||||
- **name : mean**
|
||||
- `history_filename` (`str` | `False`): Путь к файлу, который будет являться или является
|
||||
историй пользовательского ввода, в последующем эти команды будут автодополняться, файл
|
||||
может не существовать при инициализации, тогда он будет создан, при значении аргумента `False`
|
||||
история пользовательского ввода будет существовать только в пределах сессии и не сохраняться в файл
|
||||
- `autocomplete_button` (`str`): Строковое обозначение кнопки на клавиатуре, которая будет
|
||||
использоваться для автодополнения при вводе, по умолчанию `tab`
|
||||
|
||||
---
|
||||
|
||||
## *class* :: `StaticDivideLine`
|
||||
Класс, экземпляр которого представляет собой строковый разделитель фиксированной длины
|
||||
|
||||
### Конструктор
|
||||
```python
|
||||
StaticDivideLine(unit_part: str = '-',
|
||||
length: int = 25)
|
||||
```
|
||||
|
||||
**Аргументы:**
|
||||
- **name : mean**
|
||||
- `unit_part` (`str`): Единичная часть строкового разделителя
|
||||
- `length` (`int`): Длина строкового разделителя
|
||||
|
||||
---
|
||||
|
||||
## *class* :: `DinamicDivideLine`
|
||||
Строковый разделитель динамической длины, которая определяется длиной обрамляемого вывода команды
|
||||
|
||||
### Конструктор
|
||||
```python
|
||||
DinamicDivideLine(unit_part: str = '-')
|
||||
```
|
||||
|
||||
**Аргументы:**
|
||||
- **name : mean**
|
||||
- `unit_part` (`str`): Единичная часть строкового разделителя
|
||||
|
||||
---
|
||||
|
||||
## *class* :: `Router`
|
||||
Класс, который определяет и конфигурирует обработчики команд
|
||||
|
||||
### Конструктор
|
||||
```python
|
||||
Router(title: str | None = None,
|
||||
name: str = 'Default')
|
||||
```
|
||||
|
||||
|
||||
|
||||
**Аргументы:**
|
||||
- **name : mean**
|
||||
- `title` (`str`): Заголовок группы команд.
|
||||
- `name` (`str`): Персональное название роутера
|
||||
|
||||
---
|
||||
|
||||
### ***methods***
|
||||
|
||||
---
|
||||
|
||||
#### **command(command: Command)**
|
||||
|
||||
*param* `command: Command` **::** Экземпляр класса `Command`, который определяет строковый триггер команды,
|
||||
допустимые флаги команды и другое
|
||||
*required* **::** True
|
||||
*example* **::** `Command(command='ssh', description='connect via ssh')`
|
||||
|
||||
*method mean* **::** Декоратор, который регистрирует функцию как обработчик команды
|
||||
|
||||
---
|
||||
|
||||
#### **.get_name() -> `str`**
|
||||
|
||||
*method mean* **::** Возвращает установленное название роутера
|
||||
|
||||
---
|
||||
|
||||
#### **.get_title() -> `str`**
|
||||
|
||||
*method mean* **::** Возвращает установленный заголовок группы команд данного роутера
|
||||
|
||||
---
|
||||
|
||||
|
||||
### Исключения
|
||||
- `RepeatedFlagNameException` - Повторяющиеся зарегистрированные флаги в команде
|
||||
- `TooManyTransferredArgsException` - Слишком много зарегистрированных аргументов у обработчика команды
|
||||
- `RequiredArgumentNotPassedException` - Не зарегистрирован обязательный аргумент у обработчика команды(аргумент, через который будут переданы флаги введённой команды)
|
||||
- `IncorrectNumberOfHandlerArgsException` - У обработчика нестандартного поведения зарегистрировано неверное количество аргументов(в большинстве случаев у него должен быть один аргумент)
|
||||
- `TriggerCannotContainSpacesException` - У регистрируемой команды в триггере содержатся пробелы
|
||||
|
||||
---
|
||||
|
||||
## *class* :: `Command`
|
||||
Класс, экземпляр которого определяет строковый триггер хэндлера и конфигурирует его атрибуты
|
||||
|
||||
### Конструктор
|
||||
```python
|
||||
Command(trigger: str,
|
||||
description: str = None,
|
||||
flags: Flag | Flags = None)
|
||||
```
|
||||
|
||||
**Аргументы:**
|
||||
- **name : mean**
|
||||
- `trigger` (`str`): Строковый триггер
|
||||
- `description` (`str`): Описание команды, которое будет выведено в консоль при запуске оболочки
|
||||
- `flags` (`Flag | Flags`): Флаги, которые будут обработаны при их наличии во вводе юзера
|
||||
|
||||
---
|
||||
|
||||
#### **.get_trigger() -> `str`**
|
||||
|
||||
*method mean* **::** Возвращает строковый триггер экземпляра
|
||||
|
||||
---
|
||||
|
||||
#### **.get_description() -> `str`**
|
||||
|
||||
*method mean* **::** Возвращает описание команды
|
||||
|
||||
---
|
||||
|
||||
#### **.get_registered_flags() -> `Flags | None`**
|
||||
|
||||
*method mean* **::** Возвращает зарегистрированные флаги экземпляра
|
||||
|
||||
---
|
||||
|
||||
### Исключения
|
||||
- `UnprocessedInputFlagException` - Некорректный синтаксис ввода команды
|
||||
- `RepeatedInputFlagsException` - Повторяющиеся флаги во введённой команде
|
||||
- `EmptyInputCommandException` - Введённая команда является пустой(не содержит символов)
|
||||
|
||||
**Примечание**
|
||||
Все вышеуказанные исключения класса `Command` вызываются в рантайме запущенным экземпляром класса
|
||||
`App`, а также по дефолту обрабатываются, при желании можно задать пользовательские
|
||||
обработчики для этих исключений ([подробнее см.](#custom_handler))
|
||||
|
||||
---
|
||||
|
||||
## *class* :: `Flag`
|
||||
Класс, экземпляры которого в большинстве случаев передаются при создании
|
||||
экземпляра класса `Command` для регистрации допустимого флага при вводе юзером команды
|
||||
|
||||
### Конструктор
|
||||
```python
|
||||
Flag(name: str,
|
||||
prefix: typing.Literal['-', '--', '---'] = '-',
|
||||
possible_values: list[str] | typing.Pattern[str] | False = True)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
**Аргументы:**
|
||||
- **name : mean**
|
||||
- `name` (`str`): Имя флага
|
||||
- `prefix` (`Literal['-', '--', '---']`): Префикс команды, допустимым значением является от одного до трёх минусов
|
||||
- `possible_values` (`list[str] | Pattern[str] | bool`): Множество допустимых значений флага, может быть задано
|
||||
списком с допустимыми значениями или регулярным выражением (рекомендуется `re.compile(r'example exspression')`), при значении
|
||||
аргумента `False` у введённого флага не может быть значения, иначе будет вызвано исключение и обработано соответствующим
|
||||
еррор-хэндлером
|
||||
|
||||
---
|
||||
|
||||
### ***methods***
|
||||
|
||||
---
|
||||
|
||||
#### **.get_string_entity() -> `str`**
|
||||
|
||||
*method mean* **::** Возвращает строковое представление флага(префикс + имя)
|
||||
|
||||
---
|
||||
|
||||
#### **.get_name() -> `str`**
|
||||
|
||||
*method mean* **::** Возвращает имя флага
|
||||
|
||||
---
|
||||
|
||||
#### **.get_prefix() -> `str`**
|
||||
|
||||
*method mean* **::** Возвращает префикс флага
|
||||
|
||||
---
|
||||
|
||||
## *class* :: `InputFlag`
|
||||
Класс, экземпляры которого являются введёнными флагами команды, передаётся в хэндлер команды
|
||||
через `InputFlags`
|
||||
|
||||
---
|
||||
|
||||
### Примечания
|
||||
|
||||
- Наиболее часто используемые флаги предопределены и доступны для быстрого использования:
|
||||
`argenta.command.flag.defaults.PredeterminedFlags`
|
||||
|
||||
---
|
||||
|
||||
|
||||
### Конструктор
|
||||
```python
|
||||
InputFlag(name: str,
|
||||
prefix: typing.Literal['-', '--', '---'] = '-',
|
||||
value: str = None)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
**Аргументы:**
|
||||
- **name : mean**
|
||||
- `name` (`str`): Имя флага
|
||||
- `prefix` (`Literal['-', '--', '---']`): Префикс команды, допустимым значением является от одного до трёх минусов
|
||||
- `value` (`str`): Значение введённого флага, если оно есть
|
||||
|
||||
---
|
||||
|
||||
### ***methods***
|
||||
|
||||
---
|
||||
|
||||
#### **.get_value() -> `str | None`**
|
||||
|
||||
*method mean* **::** Возвращает значение введённого флага
|
||||
|
||||
---
|
||||
|
||||
## *class* :: `Flags`
|
||||
Класс, объединяющий список флагов в один объект, используется в качестве
|
||||
передаваемого аргумента `flags` экземпляру класса `Command`, при регистрации
|
||||
хэндлера
|
||||
|
||||
### Конструктор
|
||||
```python
|
||||
Flags(*flags: Flag)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
**Аргументы:**
|
||||
- **name : mean**
|
||||
- `*flags` (`Flag`): Неограниченное количество передаваемых флагов
|
||||
|
||||
---
|
||||
|
||||
### ***methods***
|
||||
|
||||
---
|
||||
|
||||
#### **.get_flags() -> `list[Flag]`**
|
||||
|
||||
*method mean* **::** Возвращает зарегистрированные флаги
|
||||
|
||||
---
|
||||
|
||||
#### **.add_flag(flag: Flag) -> `None`**
|
||||
|
||||
*method mean* **::** Добавляет флаг в группу
|
||||
|
||||
---
|
||||
|
||||
#### **.add_flags(flags: list[Flag]) -> `None`**
|
||||
|
||||
*method mean* **::** Добавляет флаги в группу
|
||||
|
||||
---
|
||||
|
||||
#### **.get_flag(name: str) -> `Flag | None`**
|
||||
|
||||
*param* `name: str` **::** Строковый триггер флага без префикса
|
||||
*required* **::** True
|
||||
*example* **::** `'host'`
|
||||
|
||||
*method mean* **::** Возвращает флаг по его триггеру или `None`, если флаг не найден
|
||||
|
||||
---
|
||||
|
||||
## *class* :: `InputFlags`
|
||||
Класс, объединяющий список введённых флагов в один объект, передаётся соответствующему хэндлеру
|
||||
в качестве аргумента
|
||||
|
||||
### Конструктор
|
||||
```python
|
||||
InputFlags(*flags: Flag)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
**Аргументы:**
|
||||
- **name : mean**
|
||||
- `*flags` (`InputFlag`): Неограниченное количество передаваемых флагов
|
||||
|
||||
---
|
||||
|
||||
### ***methods***
|
||||
|
||||
---
|
||||
|
||||
#### **.get_flags() -> `list[Flag]`**
|
||||
|
||||
*method mean* **::** Возвращает введённые флаги
|
||||
|
||||
---
|
||||
|
||||
#### **.get_flag(name: str) -> `InputFlag | None`**
|
||||
|
||||
*param* `name: str` **::** Строковый триггер флага без префикса
|
||||
*required* **::** True
|
||||
*example* **::** `'host'`
|
||||
|
||||
*method mean* **::** Возвращает введённый флаг по его триггеру или `None`, если флаг не найден
|
||||
|
||||
---
|
||||
|
||||
# Тесты
|
||||
|
||||
Запуск тестов:
|
||||
|
||||
```bash
|
||||
python -m unittest discover
|
||||
```
|
||||
or
|
||||
```bash
|
||||
python -m unittest discover -v
|
||||
```
|
||||
|
||||
@@ -1,2 +0,0 @@
|
||||
from .router import *
|
||||
from .app import *
|
||||
@@ -0,0 +1,3 @@
|
||||
__all__ = ["App"]
|
||||
|
||||
from argenta.app.models import App
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
__all__ = ["AutoCompleter"]
|
||||
|
||||
|
||||
from argenta.app.autocompleter.entity import AutoCompleter
|
||||
@@ -0,0 +1,47 @@
|
||||
import os
|
||||
import readline
|
||||
|
||||
|
||||
class AutoCompleter:
|
||||
def __init__(self, history_filename: str = False, autocomplete_button: str = 'tab'):
|
||||
self.history_filename = history_filename
|
||||
self.autocomplete_button = autocomplete_button
|
||||
self.matches = []
|
||||
|
||||
def complete(self, text, state):
|
||||
matches = sorted(cmd for cmd in self.get_history_items() if cmd.startswith(text))
|
||||
if len(matches) > 1:
|
||||
common_prefix = matches[0]
|
||||
for match in matches[1:]:
|
||||
i = 0
|
||||
while i < len(common_prefix) and i < len(match) and common_prefix[i] == match[i]:
|
||||
i += 1
|
||||
common_prefix = common_prefix[:i]
|
||||
if state == 0:
|
||||
readline.insert_text(common_prefix[len(text):])
|
||||
readline.redisplay()
|
||||
return None
|
||||
elif len(matches) == 1:
|
||||
return matches[0] if state == 0 else None
|
||||
else:
|
||||
return None
|
||||
|
||||
def initial_setup(self, all_commands: list[str]):
|
||||
if self.history_filename:
|
||||
if os.path.exists(self.history_filename):
|
||||
readline.read_history_file(self.history_filename)
|
||||
else:
|
||||
for line in all_commands:
|
||||
readline.add_history(line)
|
||||
|
||||
readline.set_completer(self.complete)
|
||||
readline.set_completer_delims(readline.get_completer_delims().replace(' ', ''))
|
||||
readline.parse_and_bind(f'{self.autocomplete_button}: complete')
|
||||
|
||||
def exit_setup(self):
|
||||
if self.history_filename:
|
||||
readline.write_history_file(self.history_filename)
|
||||
|
||||
@staticmethod
|
||||
def get_history_items():
|
||||
return [readline.get_history_item(i) for i in range(1, readline.get_current_history_length() + 1)]
|
||||
@@ -0,0 +1,9 @@
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass
|
||||
class PredeterminedMessages:
|
||||
USAGE = '[b dim]Usage[/b dim]: [i]<command> <[green]flags[/green]>[/i]'
|
||||
HELP = '[b dim]Help[/b dim]: [i]<command>[/i] [b red]--help[/b red]'
|
||||
AUTOCOMPLETE = '[b dim]Autocomplete[/b dim]: [i]<part>[/i] [bold]<tab>'
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
__all__ = ["StaticDividingLine", "DynamicDividingLine"]
|
||||
|
||||
|
||||
from argenta.app.dividing_line.models import StaticDividingLine, DynamicDividingLine
|
||||
@@ -0,0 +1,24 @@
|
||||
class BaseDividingLine:
|
||||
def __init__(self, unit_part: str = '-'):
|
||||
self.unit_part = unit_part
|
||||
|
||||
def get_unit_part(self):
|
||||
if len(self.unit_part) == 0:
|
||||
return ' '
|
||||
else:
|
||||
return self.unit_part[0]
|
||||
|
||||
class StaticDividingLine(BaseDividingLine):
|
||||
def __init__(self, unit_part: str = '-', length: int = 25):
|
||||
super().__init__(unit_part)
|
||||
self.length = length
|
||||
|
||||
def get_full_line(self):
|
||||
return f'\n[dim]{self.length * self.get_unit_part()}[/dim]\n'
|
||||
|
||||
|
||||
class DynamicDividingLine(BaseDividingLine):
|
||||
def get_full_line(self, length: int):
|
||||
return f'\n[dim]{self.get_unit_part() * length}[/dim]\n'
|
||||
|
||||
|
||||
@@ -1,145 +0,0 @@
|
||||
from typing import Callable
|
||||
from ..router.entity import Router
|
||||
from .exceptions import (InvalidRouterInstanceException,
|
||||
InvalidDescriptionMessagePatternException,
|
||||
OnlyOneMainRouterIsAllowedException,
|
||||
MissingMainRouterException,
|
||||
MissingHandlersForUnknownCommandsOnMainRouterException,
|
||||
HandlerForUnknownCommandsCanOnlyBeDeclaredForMainRouterException)
|
||||
|
||||
|
||||
class App:
|
||||
def __init__(self,
|
||||
prompt: str = 'Enter a command',
|
||||
exit_command: str = 'q',
|
||||
ignore_exit_command_register: bool = True,
|
||||
initial_greeting: str = 'Hello',
|
||||
goodbye_message: str = 'GoodBye',
|
||||
line_separate: str = '\n',
|
||||
command_group_description_separate: str = '\n',
|
||||
print_func: Callable[[str], None] = print) -> None:
|
||||
self.prompt = prompt
|
||||
self.print_func = print_func
|
||||
self.exit_command = exit_command
|
||||
self.ignore_exit_command_register = ignore_exit_command_register
|
||||
self.goodbye_message = goodbye_message
|
||||
self.initial_greeting = initial_greeting
|
||||
self.line_separate = line_separate
|
||||
self.command_group_description_separate = command_group_description_separate
|
||||
|
||||
self.routers: list[Router] = []
|
||||
self.registered_commands: list[dict[str, str | list[dict[str, Callable[[], None] | str]] | Router]] = []
|
||||
self.main_app_router: Router | None = None
|
||||
self._description_message_pattern = '[{command}] *=*=* {description}'
|
||||
|
||||
|
||||
def start_polling(self) -> None:
|
||||
self.print_func(self.initial_greeting)
|
||||
self.validate_main_router()
|
||||
|
||||
while True:
|
||||
self.print_command_group_description()
|
||||
self.print_func(self.prompt)
|
||||
|
||||
command: str = input()
|
||||
|
||||
self.checking_command_for_exit_command(command)
|
||||
self.print_func(self.line_separate)
|
||||
|
||||
is_unknown_command: bool = self.check_is_command_unknown(command)
|
||||
|
||||
if is_unknown_command:
|
||||
continue
|
||||
|
||||
for router in self.routers:
|
||||
router.input_command_handler(command)
|
||||
self.print_func(self.line_separate)
|
||||
self.print_func(self.command_group_description_separate)
|
||||
|
||||
|
||||
def set_initial_greeting(self, greeting: str) -> None:
|
||||
self.initial_greeting = greeting
|
||||
|
||||
|
||||
def set_goodbye_message(self, message: str) -> None:
|
||||
self.goodbye_message = message
|
||||
|
||||
|
||||
def set_description_message_pattern(self, pattern: str) -> None:
|
||||
try:
|
||||
pattern.format(command='command',
|
||||
description='description')
|
||||
except KeyError:
|
||||
raise InvalidDescriptionMessagePatternException(pattern)
|
||||
self._description_message_pattern = pattern
|
||||
|
||||
|
||||
def validate_main_router(self):
|
||||
if not self.main_app_router:
|
||||
raise MissingMainRouterException()
|
||||
|
||||
if not self.main_app_router.unknown_command_func:
|
||||
raise MissingHandlersForUnknownCommandsOnMainRouterException()
|
||||
|
||||
for router in self.routers:
|
||||
if router.unknown_command_func and self.main_app_router is not router:
|
||||
raise HandlerForUnknownCommandsCanOnlyBeDeclaredForMainRouterException()
|
||||
|
||||
|
||||
def checking_command_for_exit_command(self, command: str):
|
||||
if command.lower() == self.exit_command.lower():
|
||||
if self.ignore_exit_command_register:
|
||||
self.print_func(self.goodbye_message)
|
||||
exit(0)
|
||||
else:
|
||||
if command == self.exit_command:
|
||||
self.print_func(self.goodbye_message)
|
||||
exit(0)
|
||||
|
||||
|
||||
def check_is_command_unknown(self, command: str):
|
||||
registered_commands = self.registered_commands
|
||||
for router in registered_commands:
|
||||
for command_entity in router['commands']:
|
||||
if command_entity['command'].lower() == command.lower():
|
||||
if router['router'].ignore_command_register:
|
||||
return False
|
||||
else:
|
||||
if command_entity['command'] == command:
|
||||
return False
|
||||
self.main_app_router.unknown_command_handler(command)
|
||||
self.print_func(self.line_separate)
|
||||
self.print_func(self.command_group_description_separate)
|
||||
return True
|
||||
|
||||
|
||||
def print_command_group_description(self):
|
||||
for router in self.registered_commands:
|
||||
self.print_func(router['name'])
|
||||
for command_entity in router['commands']:
|
||||
self.print_func(self._description_message_pattern.format(
|
||||
command=command_entity['command'],
|
||||
description=command_entity['description']
|
||||
)
|
||||
)
|
||||
self.print_func(self.command_group_description_separate)
|
||||
|
||||
|
||||
def include_router(self, router: Router, is_main: bool = False) -> None:
|
||||
if not isinstance(router, Router):
|
||||
raise InvalidRouterInstanceException()
|
||||
|
||||
if is_main:
|
||||
if not self.main_app_router:
|
||||
self.main_app_router = router
|
||||
router.set_router_as_main()
|
||||
else:
|
||||
raise OnlyOneMainRouterIsAllowedException(router)
|
||||
|
||||
self.routers.append(router)
|
||||
|
||||
registered_commands: list[dict[str, Callable[[], None] | str]] = router.get_registered_commands()
|
||||
self.registered_commands.append({'name': router.get_name(),
|
||||
'router': router,
|
||||
'commands': registered_commands})
|
||||
|
||||
@@ -1,39 +1,10 @@
|
||||
class InvalidRouterInstanceException(Exception):
|
||||
class NoRegisteredRoutersException(Exception):
|
||||
def __str__(self):
|
||||
return "Invalid Router Instance"
|
||||
return "No Registered Router Found"
|
||||
|
||||
|
||||
class InvalidDescriptionMessagePatternException(Exception):
|
||||
def __init__(self, pattern: str):
|
||||
self.pattern = pattern
|
||||
class NoRegisteredHandlersException(Exception):
|
||||
def __init__(self, router_name):
|
||||
self.router_name = router_name
|
||||
def __str__(self):
|
||||
return ("Invalid Description Message Pattern\n"
|
||||
"Correct pattern example: [{command}] *=*=* {description}\n"
|
||||
"The pattern must contain two variables: `command` and `description` - description of the command\n"
|
||||
f"Your pattern: {self.pattern}")
|
||||
|
||||
|
||||
class OnlyOneMainRouterIsAllowedException(Exception):
|
||||
def __init__(self, existing_main_router):
|
||||
self.existing_main_router = existing_main_router
|
||||
|
||||
def __str__(self):
|
||||
return ("Only One Main Router Allowed\n"
|
||||
f"Existing main router is: {self.existing_main_router}")
|
||||
|
||||
|
||||
class MissingMainRouterException(Exception):
|
||||
def __str__(self):
|
||||
return ("Missing Main Router\n"
|
||||
"One of the registered routers must be the main one")
|
||||
|
||||
|
||||
class MissingHandlersForUnknownCommandsOnMainRouterException(Exception):
|
||||
def __str__(self):
|
||||
return ("Missing Handlers For Unknown Commands On The Main Router\n"
|
||||
"The main router must have a declared handler for unknown commands")
|
||||
|
||||
|
||||
class HandlerForUnknownCommandsCanOnlyBeDeclaredForMainRouterException(Exception):
|
||||
def __str__(self):
|
||||
return '\nThe handler for unknown commands can only be declared for the main router'
|
||||
return f"No Registered Handlers Found For '{self.router_name}'"
|
||||
|
||||
@@ -0,0 +1,265 @@
|
||||
from typing import Callable
|
||||
from rich.console import Console
|
||||
from rich.markup import escape
|
||||
from art import text2art
|
||||
from contextlib import redirect_stdout
|
||||
import io
|
||||
import re
|
||||
|
||||
from argenta.command.models import Command, InputCommand
|
||||
from argenta.router import Router
|
||||
from argenta.router.defaults import system_router
|
||||
from argenta.app.autocompleter import AutoCompleter
|
||||
from argenta.app.dividing_line.models import StaticDividingLine, DynamicDividingLine
|
||||
from argenta.command.exceptions import (UnprocessedInputFlagException,
|
||||
RepeatedInputFlagsException,
|
||||
EmptyInputCommandException,
|
||||
BaseInputCommandException)
|
||||
from argenta.app.exceptions import (NoRegisteredRoutersException,
|
||||
NoRegisteredHandlersException)
|
||||
from argenta.app.registered_routers.entity import RegisteredRouters
|
||||
|
||||
|
||||
|
||||
class AppInit:
|
||||
def __init__(self,
|
||||
prompt: str = '[italic dim bold]What do you want to do?\n',
|
||||
initial_message: str = '\nArgenta\n',
|
||||
farewell_message: str = '\nSee you\n',
|
||||
exit_command: Command = Command('Q', 'Exit command'),
|
||||
system_points_title: str | None = 'System points:',
|
||||
ignore_command_register: bool = True,
|
||||
dividing_line: StaticDividingLine | DynamicDividingLine = StaticDividingLine(),
|
||||
repeat_command_groups: bool = True,
|
||||
override_system_messages: bool = False,
|
||||
autocompleter: AutoCompleter = AutoCompleter(),
|
||||
print_func: Callable[[str], None] = Console().print) -> None:
|
||||
self._prompt = prompt
|
||||
self._print_func = print_func
|
||||
self._exit_command = exit_command
|
||||
self._system_points_title = system_points_title
|
||||
self._dividing_line = dividing_line
|
||||
self._ignore_command_register = ignore_command_register
|
||||
self._repeat_command_groups_description = repeat_command_groups
|
||||
self._override_system_messages = override_system_messages
|
||||
self._autocompleter = autocompleter
|
||||
|
||||
self._farewell_message = farewell_message
|
||||
self._initial_message = initial_message
|
||||
|
||||
|
||||
self._description_message_gen: Callable[[str, str], str] = lambda command, description: f'[bold red]{escape('['+command+']')}[/bold red] [blue dim]*=*=*[/blue dim] [bold yellow italic]{escape(description)}'
|
||||
self._registered_routers: RegisteredRouters = RegisteredRouters()
|
||||
self._messages_on_startup = []
|
||||
|
||||
self._invalid_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'[red bold]Incorrect flag syntax: {escape(raw_command)}')
|
||||
self._repeated_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'[red bold]Repeated input flags: {escape(raw_command)}')
|
||||
self._empty_input_command_handler: Callable[[], None] = lambda: print_func('[red bold]Empty input command')
|
||||
self._unknown_command_handler: Callable[[InputCommand], None] = lambda command: print_func(f"[red bold]Unknown command: {escape(command.get_trigger())}")
|
||||
self._exit_command_handler: Callable[[], None] = lambda: print_func(self._farewell_message)
|
||||
|
||||
|
||||
class AppSetters(AppInit):
|
||||
def set_description_message_pattern(self, pattern: Callable[[str, str], str]) -> None:
|
||||
self._description_message_gen: Callable[[str, str], str] = pattern
|
||||
|
||||
|
||||
def set_invalid_input_flags_handler(self, handler: Callable[[str], None]) -> None:
|
||||
self._invalid_input_flags_handler = handler
|
||||
|
||||
|
||||
def set_repeated_input_flags_handler(self, handler: Callable[[str], None]) -> None:
|
||||
self._repeated_input_flags_handler = handler
|
||||
|
||||
|
||||
def set_unknown_command_handler(self, handler: Callable[[str], None]) -> None:
|
||||
self._unknown_command_handler = handler
|
||||
|
||||
|
||||
def set_empty_command_handler(self, handler: Callable[[], None]) -> None:
|
||||
self._empty_input_command_handler = handler
|
||||
|
||||
|
||||
def set_exit_command_handler(self, handler: Callable[[], None]) -> None:
|
||||
self._exit_command_handler = handler
|
||||
|
||||
|
||||
class AppPrinters(AppInit):
|
||||
def _print_command_group_description(self):
|
||||
for registered_router in self._registered_routers:
|
||||
if registered_router.get_title():
|
||||
self._print_func(registered_router.get_title())
|
||||
for command_handler in registered_router.get_command_handlers():
|
||||
self._print_func(self._description_message_gen(
|
||||
command_handler.get_handled_command().get_trigger(),
|
||||
command_handler.get_handled_command().get_description()))
|
||||
self._print_func('')
|
||||
|
||||
|
||||
def _print_framed_text_with_dynamic_line(self, text: str):
|
||||
clear_text = re.sub(r'\u001b\[[0-9;]*m', '', text)
|
||||
max_length_line = max([len(line) for line in clear_text.split('\n')])
|
||||
max_length_line = max_length_line if 10 <= max_length_line <= 80 else 80 if max_length_line > 80 else 10
|
||||
self._print_func(self._dividing_line.get_full_line(max_length_line))
|
||||
print(text.strip('\n'))
|
||||
self._print_func(self._dividing_line.get_full_line(max_length_line))
|
||||
|
||||
|
||||
def _print_framed_text(self, text: str):
|
||||
if isinstance(self._dividing_line, StaticDividingLine):
|
||||
self._print_func(self._dividing_line.get_full_line())
|
||||
self._print_func(text)
|
||||
self._print_func(self._dividing_line.get_full_line())
|
||||
elif isinstance(self._dividing_line, DynamicDividingLine):
|
||||
self._print_framed_text_with_dynamic_line(text)
|
||||
|
||||
|
||||
class AppNonStandardHandlers(AppPrinters):
|
||||
def _is_exit_command(self, command: InputCommand):
|
||||
if command.get_trigger().lower() == self._exit_command.get_trigger().lower():
|
||||
if self._ignore_command_register:
|
||||
system_router.input_command_handler(command)
|
||||
return True
|
||||
elif command.get_trigger() == self._exit_command.get_trigger():
|
||||
system_router.input_command_handler(command)
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _is_unknown_command(self, command: InputCommand):
|
||||
for router_entity in self._registered_routers:
|
||||
for command_handler in router_entity.get_command_handlers():
|
||||
handled_command_trigger = command_handler.get_handled_command().get_trigger()
|
||||
handled_command_aliases = command_handler.get_handled_command().get_aliases()
|
||||
if handled_command_trigger.lower() == command.get_trigger().lower() and self._ignore_command_register:
|
||||
return False
|
||||
elif handled_command_trigger == command.get_trigger():
|
||||
return False
|
||||
elif handled_command_aliases:
|
||||
if (command.get_trigger().lower() in [x.lower() for x in handled_command_aliases]) and self._ignore_command_register:
|
||||
return False
|
||||
elif command.get_trigger() in handled_command_trigger:
|
||||
return False
|
||||
if isinstance(self._dividing_line, StaticDividingLine):
|
||||
self._print_func(self._dividing_line.get_full_line())
|
||||
self._unknown_command_handler(command)
|
||||
self._print_func(self._dividing_line.get_full_line())
|
||||
elif isinstance(self._dividing_line, DynamicDividingLine):
|
||||
with redirect_stdout(io.StringIO()) as f:
|
||||
self._unknown_command_handler(command)
|
||||
res: str = f.getvalue()
|
||||
self._print_framed_text_with_dynamic_line(res)
|
||||
return True
|
||||
|
||||
|
||||
def _error_handler(self, error: BaseInputCommandException, raw_command: str) -> None:
|
||||
match error:
|
||||
case UnprocessedInputFlagException():
|
||||
self._invalid_input_flags_handler(raw_command)
|
||||
case RepeatedInputFlagsException():
|
||||
self._repeated_input_flags_handler(raw_command)
|
||||
case EmptyInputCommandException():
|
||||
self._empty_input_command_handler()
|
||||
|
||||
|
||||
class AppValidators(AppInit):
|
||||
def _validate_number_of_routers(self) -> None:
|
||||
if not self._registered_routers:
|
||||
raise NoRegisteredRoutersException()
|
||||
|
||||
|
||||
def _validate_included_routers(self) -> None:
|
||||
for router in self._registered_routers:
|
||||
if not router.get_command_handlers():
|
||||
raise NoRegisteredHandlersException(router.get_name())
|
||||
|
||||
|
||||
class AppSetups(AppValidators, AppPrinters):
|
||||
def _setup_system_router(self):
|
||||
system_router.set_title(self._system_points_title)
|
||||
|
||||
@system_router.command(self._exit_command)
|
||||
def exit_command():
|
||||
self._exit_command_handler()
|
||||
|
||||
if system_router not in self._registered_routers.get_registered_routers():
|
||||
system_router.set_ignore_command_register(self._ignore_command_register)
|
||||
self._registered_routers.add_registered_router(system_router)
|
||||
|
||||
def _setup_default_view(self):
|
||||
if not self._override_system_messages:
|
||||
self._initial_message = f'\n[bold red]{text2art(self._initial_message, font='tarty1')}\n\n'
|
||||
self._farewell_message = (
|
||||
f'[bold red]\n{text2art(f'\n{self._farewell_message}\n', font='chanky')}[/bold red]\n'
|
||||
f'[red i]github.com/koloideal/Argenta[/red i] | [red bold i]made by kolo[/red bold i]\n')
|
||||
|
||||
def _pre_cycle_setup(self):
|
||||
self._setup_default_view()
|
||||
self._setup_system_router()
|
||||
self._validate_number_of_routers()
|
||||
self._validate_included_routers()
|
||||
|
||||
all_triggers: list[str] = []
|
||||
for router_entity in self._registered_routers:
|
||||
all_triggers.extend(router_entity.get_triggers())
|
||||
all_triggers.extend(router_entity.get_aliases())
|
||||
self._autocompleter.initial_setup(all_triggers)
|
||||
|
||||
self._print_func(self._initial_message)
|
||||
|
||||
for message in self._messages_on_startup:
|
||||
self._print_func(message)
|
||||
print('\n\n')
|
||||
|
||||
if not self._repeat_command_groups_description:
|
||||
self._print_command_group_description()
|
||||
|
||||
|
||||
class App(AppSetters, AppNonStandardHandlers, AppSetups):
|
||||
def start_polling(self) -> None:
|
||||
self._pre_cycle_setup()
|
||||
while True:
|
||||
if self._repeat_command_groups_description:
|
||||
self._print_command_group_description()
|
||||
|
||||
raw_command: str = Console().input(self._prompt)
|
||||
|
||||
try:
|
||||
input_command: InputCommand = InputCommand.parse(raw_command=raw_command)
|
||||
except BaseInputCommandException as error:
|
||||
with redirect_stdout(io.StringIO()) as f:
|
||||
self._error_handler(error, raw_command)
|
||||
res: str = f.getvalue()
|
||||
self._print_framed_text(res)
|
||||
continue
|
||||
|
||||
if self._is_exit_command(input_command):
|
||||
self._autocompleter.exit_setup()
|
||||
return
|
||||
|
||||
if self._is_unknown_command(input_command):
|
||||
continue
|
||||
|
||||
with redirect_stdout(io.StringIO()) as f:
|
||||
for registered_router in self._registered_routers:
|
||||
registered_router.input_command_handler(input_command)
|
||||
res: str = f.getvalue()
|
||||
self._print_framed_text(res)
|
||||
|
||||
if not self._repeat_command_groups_description:
|
||||
self._print_func(self._prompt)
|
||||
|
||||
|
||||
def include_router(self, router: Router) -> None:
|
||||
router.set_ignore_command_register(self._ignore_command_register)
|
||||
self._registered_routers.add_registered_router(router)
|
||||
|
||||
|
||||
def include_routers(self, *routers: Router) -> None:
|
||||
for router in routers:
|
||||
self.include_router(router)
|
||||
|
||||
|
||||
def add_message_on_startup(self, message: str) -> None:
|
||||
self._messages_on_startup.append(message)
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
from argenta.router import Router
|
||||
|
||||
|
||||
class RegisteredRouters:
|
||||
def __init__(self, registered_routers: list[Router] = None) -> None:
|
||||
self._registered_routers = registered_routers if registered_routers else []
|
||||
|
||||
def get_registered_routers(self) -> list[Router]:
|
||||
return self._registered_routers
|
||||
|
||||
def add_registered_router(self, router: Router):
|
||||
self._registered_routers.append(router)
|
||||
|
||||
def add_registered_routers(self, *routers: Router):
|
||||
self._registered_routers.extend(routers)
|
||||
|
||||
def __iter__(self):
|
||||
return iter(self._registered_routers)
|
||||
|
||||
def __next__(self):
|
||||
return next(iter(self._registered_routers))
|
||||
@@ -0,0 +1,3 @@
|
||||
__all__ = ["Command"]
|
||||
|
||||
from argenta.command.models import Command
|
||||
@@ -0,0 +1,23 @@
|
||||
from argenta.command.flag.models import InputFlag, Flag
|
||||
|
||||
|
||||
class BaseInputCommandException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class UnprocessedInputFlagException(BaseInputCommandException):
|
||||
def __str__(self):
|
||||
return "Unprocessed Input Flags"
|
||||
|
||||
|
||||
class RepeatedInputFlagsException(BaseInputCommandException):
|
||||
def __init__(self, flag: Flag | InputFlag):
|
||||
self.flag = flag
|
||||
def __str__(self):
|
||||
return ("Repeated Input Flags\n"
|
||||
f"Duplicate flag was detected in the input: '{self.flag.get_string_entity()}'")
|
||||
|
||||
|
||||
class EmptyInputCommandException(BaseInputCommandException):
|
||||
def __str__(self):
|
||||
return "Input Command is empty"
|
||||
@@ -0,0 +1,4 @@
|
||||
__all__ = ('InputFlags', 'InputFlag', 'Flag', 'Flags')
|
||||
|
||||
|
||||
from argenta.command.flag.models import InputFlags, InputFlag, Flags, Flag
|
||||
@@ -0,0 +1,21 @@
|
||||
from dataclasses import dataclass
|
||||
from argenta.command.flag.models import Flag
|
||||
import re
|
||||
|
||||
|
||||
@dataclass
|
||||
class PredeterminedFlags:
|
||||
HELP = Flag(name='help', possible_values=False)
|
||||
SHORT_HELP = Flag(name='h', prefix='-', possible_values=False)
|
||||
|
||||
INFO = Flag(name='info', possible_values=False)
|
||||
SHORT_INFO = Flag(name='i', prefix='-', possible_values=False)
|
||||
|
||||
ALL = Flag(name='all', possible_values=False)
|
||||
SHORT_ALL = Flag(name='a', prefix='-', possible_values=False)
|
||||
|
||||
HOST = Flag(name='host', possible_values=re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$'))
|
||||
SHORT_HOST = Flag(name='h', prefix='-', possible_values=re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$'))
|
||||
|
||||
PORT = Flag(name='port', possible_values=re.compile(r'^\d{1,5}$'))
|
||||
SHORT_PORT = Flag(name='p', prefix='-', possible_values=re.compile(r'^\d{1,5}$'))
|
||||
@@ -0,0 +1,140 @@
|
||||
from typing import Literal, Pattern
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
|
||||
class BaseFlag:
|
||||
def __init__(self, name: str,
|
||||
prefix: Literal['-', '--', '---'] = '--'):
|
||||
self._name = name
|
||||
self._prefix = prefix
|
||||
|
||||
def get_string_entity(self):
|
||||
string_entity: str = self._prefix + self._name
|
||||
return string_entity
|
||||
|
||||
def get_name(self):
|
||||
return self._name
|
||||
|
||||
def get_prefix(self):
|
||||
return self._prefix
|
||||
|
||||
|
||||
|
||||
class InputFlag(BaseFlag):
|
||||
def __init__(self, name: str,
|
||||
prefix: Literal['-', '--', '---'] = '--',
|
||||
value: str = None):
|
||||
super().__init__(name, prefix)
|
||||
self._flag_value = value
|
||||
|
||||
def get_value(self) -> str | None:
|
||||
return self._flag_value
|
||||
|
||||
def set_value(self, value):
|
||||
self._flag_value = value
|
||||
|
||||
|
||||
|
||||
class Flag(BaseFlag):
|
||||
def __init__(self, name: str,
|
||||
prefix: Literal['-', '--', '---'] = '--',
|
||||
possible_values: list[str] | Pattern[str] | False = True):
|
||||
super().__init__(name, prefix)
|
||||
self.possible_values = possible_values
|
||||
|
||||
def validate_input_flag_value(self, input_flag_value: str | None):
|
||||
if self.possible_values is False:
|
||||
if input_flag_value is None:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
elif isinstance(self.possible_values, Pattern):
|
||||
if isinstance(input_flag_value, str):
|
||||
is_valid = bool(self.possible_values.match(input_flag_value))
|
||||
if bool(is_valid):
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
else:
|
||||
return False
|
||||
|
||||
elif isinstance(self.possible_values, list):
|
||||
if input_flag_value in self.possible_values:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
|
||||
|
||||
class BaseFlags(ABC):
|
||||
__slots__ = ('_flags',)
|
||||
|
||||
@abstractmethod
|
||||
def get_flags(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def add_flag(self, flag: Flag | InputFlag):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def add_flags(self, flags: list[Flag] | list[InputFlag]):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_flag(self, name: str):
|
||||
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):
|
||||
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):
|
||||
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
|
||||
|
||||
@@ -0,0 +1,112 @@
|
||||
from argenta.command.flag.models import Flag, InputFlag, Flags, InputFlags
|
||||
from argenta.command.exceptions import (UnprocessedInputFlagException,
|
||||
RepeatedInputFlagsException,
|
||||
EmptyInputCommandException)
|
||||
from typing import Generic, TypeVar, cast, Literal
|
||||
|
||||
|
||||
InputCommandType = TypeVar('InputCommandType')
|
||||
|
||||
|
||||
class BaseCommand:
|
||||
def __init__(self, trigger: str):
|
||||
self._trigger = trigger
|
||||
|
||||
def get_trigger(self) -> str:
|
||||
return self._trigger
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
def __init__(self, trigger: str,
|
||||
description: str = None,
|
||||
flags: Flag | Flags = None,
|
||||
aliases: list[str] = None):
|
||||
super().__init__(trigger)
|
||||
self._registered_flags: Flags = flags if isinstance(flags, Flags) else Flags(flags) if isinstance(flags, Flag) else Flags()
|
||||
self._description = f'Description for "{self._trigger}" command' if not description else description
|
||||
self._aliases = aliases
|
||||
|
||||
def get_registered_flags(self) -> Flags:
|
||||
return self._registered_flags
|
||||
|
||||
def get_aliases(self) -> list[str] | None:
|
||||
return self._aliases
|
||||
|
||||
def validate_input_flag(self, flag: InputFlag):
|
||||
registered_flags: Flags | None = self.get_registered_flags()
|
||||
if registered_flags:
|
||||
if isinstance(registered_flags, Flag):
|
||||
if registered_flags.get_string_entity() == flag.get_string_entity():
|
||||
is_valid = registered_flags.validate_input_flag_value(flag.get_value())
|
||||
if is_valid:
|
||||
return True
|
||||
else:
|
||||
for registered_flag in registered_flags:
|
||||
if registered_flag.get_string_entity() == flag.get_string_entity():
|
||||
is_valid = registered_flag.validate_input_flag_value(flag.get_value())
|
||||
if is_valid:
|
||||
return True
|
||||
return False
|
||||
|
||||
def get_description(self) -> str:
|
||||
return self._description
|
||||
|
||||
|
||||
|
||||
class InputCommand(BaseCommand, Generic[InputCommandType]):
|
||||
def __init__(self, trigger: str,
|
||||
input_flags: InputFlag | InputFlags = None):
|
||||
super().__init__(trigger)
|
||||
self._input_flags: InputFlags = input_flags if isinstance(input_flags, InputFlags) else InputFlags(input_flags) if isinstance(input_flags, InputFlag) else InputFlags()
|
||||
|
||||
def _set_input_flags(self, input_flags: InputFlags):
|
||||
self._input_flags = input_flags
|
||||
|
||||
def get_input_flags(self) -> InputFlags:
|
||||
return self._input_flags
|
||||
|
||||
|
||||
@staticmethod
|
||||
def parse(raw_command: str) -> InputCommandType:
|
||||
if not raw_command:
|
||||
raise EmptyInputCommandException()
|
||||
|
||||
list_of_tokens = raw_command.split()
|
||||
command = list_of_tokens.pop(0)
|
||||
|
||||
input_flags: InputFlags = InputFlags()
|
||||
current_flag_name, current_flag_value = None, None
|
||||
|
||||
for k, _ in enumerate(list_of_tokens):
|
||||
if _.startswith('-'):
|
||||
if current_flag_name or len(_) < 2 or len(_[:_.rfind('-')]) > 3:
|
||||
raise UnprocessedInputFlagException()
|
||||
current_flag_name = _
|
||||
else:
|
||||
if not current_flag_name:
|
||||
raise UnprocessedInputFlagException()
|
||||
current_flag_value = _
|
||||
|
||||
if current_flag_name:
|
||||
if not len(list_of_tokens) == k+1:
|
||||
if not list_of_tokens[k+1].startswith('-'):
|
||||
continue
|
||||
|
||||
input_flag = InputFlag(name=current_flag_name[current_flag_name.rfind('-')+1:],
|
||||
prefix=cast(Literal['-', '--', '---'],
|
||||
current_flag_name[:current_flag_name.rfind('-')+1]),
|
||||
value=current_flag_value)
|
||||
|
||||
all_flags = [flag.get_string_entity() for flag in input_flags.get_flags()]
|
||||
if input_flag.get_string_entity() not in all_flags:
|
||||
input_flags.add_flag(input_flag)
|
||||
else:
|
||||
raise RepeatedInputFlagsException(input_flag)
|
||||
|
||||
current_flag_name, current_flag_value = None, None
|
||||
|
||||
if any([current_flag_name, current_flag_value]):
|
||||
raise UnprocessedInputFlagException()
|
||||
else:
|
||||
return InputCommand(trigger=command, input_flags=input_flags)
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
__all__ = ["Router"]
|
||||
|
||||
|
||||
from argenta.router.entity import Router
|
||||
@@ -0,0 +1,21 @@
|
||||
from typing import Callable
|
||||
from argenta.command import Command
|
||||
from argenta.command.flag.models import InputFlags
|
||||
|
||||
|
||||
class CommandHandler:
|
||||
def __init__(self, handler: Callable[[], None] | Callable[[InputFlags], None], handled_command: Command):
|
||||
self._handler = handler
|
||||
self._handled_command = handled_command
|
||||
|
||||
def handling(self, input_flags: InputFlags = None):
|
||||
if input_flags is not None:
|
||||
self._handler(input_flags)
|
||||
else:
|
||||
self._handler()
|
||||
|
||||
def get_handler(self):
|
||||
return self._handler
|
||||
|
||||
def get_handled_command(self):
|
||||
return self._handled_command
|
||||
@@ -0,0 +1,21 @@
|
||||
from argenta.router.command_handler.entity import CommandHandler
|
||||
|
||||
|
||||
class CommandHandlers:
|
||||
def __init__(self, command_handlers: list[CommandHandler] = None):
|
||||
self.command_handlers = command_handlers if command_handlers else []
|
||||
|
||||
def get_command_handlers(self) -> list[CommandHandler]:
|
||||
return self.command_handlers
|
||||
|
||||
def add_command_handler(self, command_handler: CommandHandler):
|
||||
self.command_handlers.append(command_handler)
|
||||
|
||||
def add_command_handlers(self, *command_handlers: CommandHandler):
|
||||
self.command_handlers.extend(command_handlers)
|
||||
|
||||
def __iter__(self):
|
||||
return iter(self.command_handlers)
|
||||
|
||||
def __next__(self):
|
||||
return next(iter(self.command_handlers))
|
||||
@@ -0,0 +1,5 @@
|
||||
from argenta.router import Router
|
||||
|
||||
|
||||
system_router = Router(title='System points:',
|
||||
name='System')
|
||||
+123
-44
@@ -1,70 +1,149 @@
|
||||
from typing import Callable, Any
|
||||
from ..router.exceptions import (InvalidCommandInstanceException,
|
||||
UnknownCommandHandlerHasAlreadyBeenCreatedException,
|
||||
InvalidDescriptionInstanceException)
|
||||
from inspect import getfullargspec
|
||||
|
||||
from argenta.command import Command
|
||||
from argenta.command.models import InputCommand
|
||||
from argenta.router.command_handlers.entity import CommandHandlers
|
||||
from argenta.router.command_handler.entity import CommandHandler
|
||||
from argenta.command.flag.models import Flag, Flags, InputFlags
|
||||
from argenta.router.exceptions import (RepeatedFlagNameException,
|
||||
TooManyTransferredArgsException,
|
||||
RequiredArgumentNotPassedException,
|
||||
IncorrectNumberOfHandlerArgsException,
|
||||
TriggerCannotContainSpacesException)
|
||||
|
||||
|
||||
class Router:
|
||||
def __init__(self,
|
||||
name: str,
|
||||
ignore_command_register: bool = False):
|
||||
|
||||
self.ignore_command_register: bool = ignore_command_register
|
||||
title: str = None,
|
||||
name: str = 'Default'):
|
||||
self._title = title
|
||||
self._name = name
|
||||
|
||||
self.processed_commands: list[dict[str, Callable[[], None] | str]] = []
|
||||
self.unknown_command_func: Callable[[str], None] | None = None
|
||||
self._is_main_router: bool = False
|
||||
self._command_handlers: CommandHandlers = CommandHandlers()
|
||||
self._ignore_command_register: bool = False
|
||||
self._not_valid_flag_handler: Callable[[Flag], None] = lambda flag: print(f"Undefined or incorrect input flag: {flag.get_string_entity()}{(' '+flag.get_value()) if flag.get_value() else ''}")
|
||||
|
||||
|
||||
def command(self, command: str, description: str) -> Callable[[Any], Any]:
|
||||
if not isinstance(command, str):
|
||||
raise InvalidCommandInstanceException()
|
||||
if not isinstance(description, str):
|
||||
raise InvalidDescriptionInstanceException()
|
||||
else:
|
||||
def command(self, command: Command) -> Callable[[Any], Any]:
|
||||
self._validate_command(command)
|
||||
|
||||
def command_decorator(func):
|
||||
self.processed_commands.append({'func': func,
|
||||
'command': command,
|
||||
'description': description})
|
||||
Router._validate_func_args(command, func)
|
||||
self._command_handlers.add_command_handler(CommandHandler(func, command))
|
||||
|
||||
def wrapper(*args, **kwargs):
|
||||
return func(*args, **kwargs)
|
||||
return wrapper
|
||||
|
||||
return command_decorator
|
||||
|
||||
|
||||
def unknown_command(self, func):
|
||||
if self.unknown_command_func is not None:
|
||||
raise UnknownCommandHandlerHasAlreadyBeenCreatedException()
|
||||
|
||||
self.unknown_command_func = func
|
||||
|
||||
def wrapper(*args, **kwargs):
|
||||
return func(*args, **kwargs)
|
||||
return wrapper
|
||||
|
||||
|
||||
def input_command_handler(self, input_command):
|
||||
for command_entity in self.processed_commands:
|
||||
if input_command.lower() == command_entity['command'].lower():
|
||||
if self.ignore_command_register:
|
||||
return command_entity['func']()
|
||||
def set_invalid_input_flag_handler(self, func):
|
||||
processed_args = getfullargspec(func).args
|
||||
if len(processed_args) != 1:
|
||||
raise IncorrectNumberOfHandlerArgsException()
|
||||
else:
|
||||
if input_command == command_entity['command']:
|
||||
return command_entity['func']()
|
||||
|
||||
def unknown_command_handler(self, unknown_command):
|
||||
self.unknown_command_func(unknown_command)
|
||||
self._not_valid_flag_handler = func
|
||||
|
||||
|
||||
def set_router_as_main(self):
|
||||
self._is_main_router = True
|
||||
def input_command_handler(self, input_command: InputCommand):
|
||||
input_command_name: str = input_command.get_trigger()
|
||||
input_command_flags: InputFlags = input_command.get_input_flags()
|
||||
|
||||
for command_handler in self._command_handlers:
|
||||
handle_command = command_handler.get_handled_command()
|
||||
if input_command_name.lower() == handle_command.get_trigger().lower():
|
||||
self._validate_input_command(input_command_flags, command_handler)
|
||||
elif handle_command.get_aliases():
|
||||
if input_command_name.lower() in handle_command.get_aliases():
|
||||
self._validate_input_command(input_command_flags, command_handler)
|
||||
|
||||
|
||||
def get_registered_commands(self) -> list[dict[str, Callable[[], None] | str]]:
|
||||
return self.processed_commands
|
||||
def _validate_input_command(self, input_command_flags: InputFlags, command_handler: CommandHandler):
|
||||
handle_command = command_handler.get_handled_command()
|
||||
if handle_command.get_registered_flags().get_flags():
|
||||
if input_command_flags.get_flags():
|
||||
if self._validate_input_flags(handle_command, input_command_flags):
|
||||
command_handler.handling(input_command_flags)
|
||||
return
|
||||
else:
|
||||
command_handler.handling(input_command_flags)
|
||||
return
|
||||
else:
|
||||
if input_command_flags.get_flags():
|
||||
self._not_valid_flag_handler(input_command_flags[0])
|
||||
return
|
||||
else:
|
||||
command_handler.handling()
|
||||
return
|
||||
|
||||
|
||||
def _validate_input_flags(self, handle_command: Command, input_flags: InputFlags):
|
||||
for flag in input_flags:
|
||||
is_valid = handle_command.validate_input_flag(flag)
|
||||
if not is_valid:
|
||||
self._not_valid_flag_handler(flag)
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
@staticmethod
|
||||
def _validate_command(command: Command):
|
||||
command_name: str = command.get_trigger()
|
||||
if command_name.find(' ') != -1:
|
||||
raise TriggerCannotContainSpacesException()
|
||||
|
||||
flags: Flags = command.get_registered_flags()
|
||||
if flags:
|
||||
flags_name: list = [x.get_string_entity().lower() for x in flags]
|
||||
if len(set(flags_name)) < len(flags_name):
|
||||
raise RepeatedFlagNameException()
|
||||
|
||||
|
||||
@staticmethod
|
||||
def _validate_func_args(command: Command, func: Callable):
|
||||
registered_args = command.get_registered_flags()
|
||||
transferred_args = getfullargspec(func).args
|
||||
if registered_args.get_flags() and transferred_args:
|
||||
if len(transferred_args) != 1:
|
||||
raise TooManyTransferredArgsException()
|
||||
elif registered_args.get_flags() and not transferred_args:
|
||||
raise RequiredArgumentNotPassedException()
|
||||
elif not registered_args.get_flags() and transferred_args:
|
||||
raise TooManyTransferredArgsException()
|
||||
|
||||
|
||||
def set_ignore_command_register(self, ignore_command_register: bool):
|
||||
self._ignore_command_register = ignore_command_register
|
||||
|
||||
|
||||
def get_triggers(self):
|
||||
all_triggers: list[str] = []
|
||||
for command_handler in self._command_handlers:
|
||||
all_triggers.append(command_handler.get_handled_command().get_trigger())
|
||||
return all_triggers
|
||||
|
||||
|
||||
def get_aliases(self):
|
||||
all_aliases: list[str] = []
|
||||
for command_handler in self._command_handlers:
|
||||
if command_handler.get_handled_command().get_aliases():
|
||||
all_aliases.extend(command_handler.get_handled_command().get_aliases())
|
||||
return all_aliases
|
||||
|
||||
|
||||
def get_command_handlers(self) -> CommandHandlers:
|
||||
return self._command_handlers
|
||||
|
||||
|
||||
def get_name(self) -> str:
|
||||
return self._name
|
||||
|
||||
|
||||
def get_title(self) -> str | None:
|
||||
return self._title
|
||||
|
||||
|
||||
def set_title(self, title: str):
|
||||
self._title = title
|
||||
|
||||
@@ -1,13 +1,23 @@
|
||||
class InvalidCommandInstanceException(Exception):
|
||||
class RepeatedFlagNameException(Exception):
|
||||
def __str__(self):
|
||||
return "Invalid Command Instance"
|
||||
return "Repeated registered_flag name in register command"
|
||||
|
||||
|
||||
class InvalidDescriptionInstanceException(Exception):
|
||||
class TooManyTransferredArgsException(Exception):
|
||||
def __str__(self):
|
||||
return "Invalid Description Instance"
|
||||
return "Too many transferred arguments"
|
||||
|
||||
|
||||
class UnknownCommandHandlerHasAlreadyBeenCreatedException(Exception):
|
||||
class RequiredArgumentNotPassedException(Exception):
|
||||
def __str__(self):
|
||||
return "Only one unknown command handler can be declared"
|
||||
return "Required argument not passed"
|
||||
|
||||
|
||||
class IncorrectNumberOfHandlerArgsException(Exception):
|
||||
def __str__(self):
|
||||
return "Handler has incorrect number of arguments"
|
||||
|
||||
|
||||
class TriggerCannotContainSpacesException(Exception):
|
||||
def __str__(self):
|
||||
return "Command trigger cannot contain spaces"
|
||||
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 60 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 58 KiB |
@@ -0,0 +1,9 @@
|
||||
from argenta.app import App
|
||||
from argenta.app.defaults import PredeterminedMessages
|
||||
|
||||
|
||||
app = App(repeat_command_groups=True)
|
||||
|
||||
app.add_message_on_startup(PredeterminedMessages.USAGE + '\n\n')
|
||||
|
||||
app.start_polling()
|
||||
@@ -0,0 +1,14 @@
|
||||
from argenta.app import App
|
||||
from argenta.command import Command
|
||||
from argenta.router import Router
|
||||
|
||||
router = Router()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
print('test command')
|
||||
|
||||
app = App(ignore_command_register=False)
|
||||
app.include_router(router)
|
||||
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
|
||||
app.start_polling()
|
||||
@@ -0,0 +1,10 @@
|
||||
from rich.console import Console
|
||||
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
def help_command():
|
||||
console.print("[italic bold]The main functionality of the script is to convert an expression from a string "
|
||||
"to a mathematical one and then calculate this expression. "
|
||||
"Project GitHub: https://github.com/koloideal/WordMath[/italic bold]")
|
||||
@@ -0,0 +1,35 @@
|
||||
from rich.console import Console
|
||||
|
||||
from argenta.command import Command
|
||||
from argenta.command.flag import Flags, InputFlags
|
||||
from argenta.command.flag.defaults import PredeterminedFlags
|
||||
from argenta.router import Router
|
||||
from .handlers_implementation.help_command import help_command
|
||||
|
||||
|
||||
work_router: Router = Router(title='Work points:')
|
||||
|
||||
settings_router: Router = Router()
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
@work_router.command(Command('get', 'Get Help', aliases=['help', 'Get_help']))
|
||||
def command_help():
|
||||
help_command()
|
||||
|
||||
|
||||
@work_router.command(Command('start', 'Start Solving',
|
||||
flags=Flags(PredeterminedFlags.HOST, PredeterminedFlags.PORT),
|
||||
aliases=['starting']))
|
||||
def command_start_solving(args: InputFlags):
|
||||
print(args.get_flag('host'))
|
||||
|
||||
|
||||
@settings_router.command(Command('update', 'Update WordMath'))
|
||||
def command_update():
|
||||
print('eeeeeee')
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
from mock.mock_app.handlers.routers import work_router, settings_router
|
||||
|
||||
from argenta.app import App
|
||||
from argenta.app.defaults import PredeterminedMessages
|
||||
from argenta.app.dividing_line import DynamicDividingLine
|
||||
from argenta.app.autocompleter import AutoCompleter
|
||||
|
||||
|
||||
autocompleter = AutoCompleter('./mock/.hist')
|
||||
app: App = App(dividing_line=DynamicDividingLine(),
|
||||
autocompleter=autocompleter)
|
||||
|
||||
|
||||
def main():
|
||||
app.include_routers(work_router, settings_router)
|
||||
|
||||
app.add_message_on_startup(PredeterminedMessages.USAGE)
|
||||
app.add_message_on_startup(PredeterminedMessages.AUTOCOMPLETE)
|
||||
app.add_message_on_startup(PredeterminedMessages.HELP)
|
||||
|
||||
app.start_polling()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Generated
-7
@@ -1,7 +0,0 @@
|
||||
# This file is automatically @generated by Poetry 2.0.1 and should not be changed by hand.
|
||||
package = []
|
||||
|
||||
[metadata]
|
||||
lock-version = "2.1"
|
||||
python-versions = ">=3.11"
|
||||
content-hash = "f5666f5625d676c506924a57dc0520a1f3ed2b2c774baed3dc85353594f8473d"
|
||||
+17
-3
@@ -1,17 +1,31 @@
|
||||
[project]
|
||||
name = "argenta"
|
||||
version = "0.1.1"
|
||||
description = "python library for creating cli apps"
|
||||
version = "0.5.0"
|
||||
description = "Python library for creating TUI"
|
||||
authors = [
|
||||
{name = "kolo", email = "kolo.is.main@gmail.com"}
|
||||
]
|
||||
license = {text = "MIT"}
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.11"
|
||||
dependencies = [
|
||||
dependencies = ["rich (>=14.0.0,<15.0.0)", "art (>=6.4,<7.0)"]
|
||||
|
||||
|
||||
[tool.ruff]
|
||||
exclude = [
|
||||
".idea",
|
||||
"venv",
|
||||
".git",
|
||||
"poetry.lock",
|
||||
".__pycache__",
|
||||
"tests"
|
||||
]
|
||||
|
||||
|
||||
[build-system]
|
||||
requires = ["poetry-core>=2.0.0,<3.0.0"]
|
||||
build-backend = "poetry.core.masonry.api"
|
||||
|
||||
[tool.poetry.group.dev.dependencies]
|
||||
pyreadline3 = "^3.5.4"
|
||||
|
||||
|
||||
@@ -1,23 +0,0 @@
|
||||
from setuptools import setup, find_packages
|
||||
|
||||
with open("README.md", "r", encoding="utf-8") as fh:
|
||||
long_description = fh.read()
|
||||
|
||||
setup(
|
||||
name="argenta",
|
||||
version="0.1.1",
|
||||
author="kolo",
|
||||
author_email="kolo.is.main@gmail.com",
|
||||
description="Python library for creating CLI apps",
|
||||
long_description=long_description,
|
||||
long_description_content_type="text/markdown",
|
||||
packages=find_packages(),
|
||||
install_requires=[
|
||||
"requests",
|
||||
],
|
||||
classifiers=[
|
||||
"Programming Language :: Python :: 3",
|
||||
"Operating System :: OS Independent",
|
||||
],
|
||||
python_requires='>=3.11',
|
||||
)
|
||||
@@ -1,5 +0,0 @@
|
||||
from argenta.app.entity import *
|
||||
|
||||
|
||||
def test():
|
||||
assert App().exit_command == 'q'
|
||||
@@ -0,0 +1,216 @@
|
||||
import _io
|
||||
from unittest.mock import patch, MagicMock
|
||||
import unittest
|
||||
import io
|
||||
import re
|
||||
|
||||
from argenta.app import App
|
||||
from argenta.command import Command
|
||||
from argenta.router import Router
|
||||
from argenta.command.flag.models import Flags, InputFlags
|
||||
from argenta.command.flag.defaults import PredeterminedFlags
|
||||
|
||||
|
||||
|
||||
class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("builtins.input", side_effect=["help", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
print('test command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertIn("\nUnknown command: help\n", output)
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["TeSt", "Q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_incorrect_command2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
print('test command')
|
||||
|
||||
app = App(ignore_command_register=False,
|
||||
override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertIn('\nUnknown command: TeSt\n', output)
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["test --help", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_unregistered_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
print(f'test command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertIn('\nUndefined or incorrect input flag: --help\n', output)
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["test --port 22", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_unregistered_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
print('test command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertIn('\nUndefined or incorrect input flag: --port 22\n', output)
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["test --host 192.168.32.1 --port 132", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_one_correct_flag_an_one_incorrect_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
flags = Flags(PredeterminedFlags.HOST)
|
||||
|
||||
@router.command(Command('test', flags=flags))
|
||||
def test(args: InputFlags):
|
||||
print(f'connecting to host {args.get_flag('host').get_value()}')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertIn('\nUndefined or incorrect input flag: --port 132\n', output)
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["test", "some", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_one_correct_command_and_one_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
print(f'test command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertRegex(output, re.compile(r'\ntest command\n(.|\n)*\nUnknown command: some'))
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["test", "some", "more", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_two_correct_commands_and_one_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
print(f'test command')
|
||||
|
||||
@router.command(Command('more'))
|
||||
def test():
|
||||
print(f'more command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.get_trigger()}'))
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertRegex(output, re.compile(r'\ntest command\n(.|\n)*\nUnknown command: some\n(.|\n)*\nmore command'))
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["test 535 --port", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_incorrect_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
print(f'test command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.set_invalid_input_flags_handler(lambda command: print(f'Incorrect flag syntax: "{command}"'))
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertIn("\nIncorrect flag syntax: \"test 535 --port\"\n", output)
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_empty_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
print(f'test command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.set_empty_command_handler(lambda: print('Empty input command'))
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertIn("\nEmpty input command\n", output)
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["test --port 22 --port 33", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_repeated_flags(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
|
||||
@router.command(Command('test', flags=PredeterminedFlags.PORT))
|
||||
def test(args: InputFlags):
|
||||
print('test command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.set_repeated_input_flags_handler(lambda command: print(f'Repeated input flags: "{command}"'))
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertIn("\nRepeated input flags: \"test --port 22 --port 33\"\n", output)
|
||||
@@ -0,0 +1,222 @@
|
||||
import _io
|
||||
from unittest.mock import patch, MagicMock
|
||||
import unittest
|
||||
import io
|
||||
import re
|
||||
|
||||
from argenta.app import App
|
||||
from argenta.command.models import Command
|
||||
from argenta.router import Router
|
||||
from argenta.command.flag.models import Flag, Flags, InputFlags
|
||||
from argenta.command.flag.defaults import PredeterminedFlags
|
||||
|
||||
|
||||
|
||||
class TestSystemHandlerNormalWork(unittest.TestCase):
|
||||
@patch("builtins.input", side_effect=["test", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
print('test command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertIn('\ntest command\n', output)
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["TeSt", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
print('test command')
|
||||
|
||||
app = App(ignore_command_register=True,
|
||||
override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertIn('\ntest command\n', output)
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["test --help", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_custom_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
flag = Flag('help', '--', False)
|
||||
|
||||
@router.command(Command('test', flags=flag))
|
||||
def test(args: InputFlags):
|
||||
print(f'\nhelp for {args.get_flag('help').get_name()} flag\n')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertIn('\nhelp for help flag\n', output)
|
||||
|
||||
@patch("builtins.input", side_effect=["test --port 22", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_custom_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
flag = Flag('port', '--', re.compile(r'^\d{1,5}$'))
|
||||
|
||||
@router.command(Command('test', flags=flag))
|
||||
def test(args: InputFlags):
|
||||
print(f'flag value for {args.get_flag('port').get_name()} flag : {args.get_flag('port').get_value()}')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertIn('\nflag value for port flag : 22\n', output)
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["test -h", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_default_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
flag = PredeterminedFlags.SHORT_HELP
|
||||
|
||||
@router.command(Command('test', flags=flag))
|
||||
def test(args: InputFlags):
|
||||
print(f'help for {args.get_flag('h').get_name()} flag')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertIn('\nhelp for h flag\n', output)
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["test --info", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_default_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
flag = PredeterminedFlags.INFO
|
||||
|
||||
@router.command(Command('test', flags=flag))
|
||||
def test(args: InputFlags):
|
||||
if args.get_flag('info'):
|
||||
print('info about test command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertIn('\ninfo about test command\n', output)
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["test --host 192.168.0.1", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_default_flag3(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
flag = PredeterminedFlags.HOST
|
||||
|
||||
@router.command(Command('test', flags=flag))
|
||||
def test(args: InputFlags):
|
||||
print(f'connecting to host {args[0].get_value()}')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertIn('\nconnecting to host 192.168.0.1\n', output)
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["test --host 192.168.32.1 --port 132", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_correct_command_with_two_flags(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
flags = Flags(PredeterminedFlags.HOST, PredeterminedFlags.PORT)
|
||||
|
||||
@router.command(Command('test', flags=flags))
|
||||
def test(args: InputFlags):
|
||||
print(f'connecting to host {args[0].get_value()} and port {args[1].get_value()}')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertIn('\nconnecting to host 192.168.32.1 and port 132\n', output)
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["test", "some", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_two_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
print(f'test command')
|
||||
|
||||
@router.command(Command('some'))
|
||||
def test2():
|
||||
print(f'some command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertRegex(output, re.compile(r'\ntest command\n(.|\n)*\nsome command\n'))
|
||||
|
||||
|
||||
@patch("builtins.input", side_effect=["test", "some", "more", "q"])
|
||||
@patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_input_three_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
||||
router = Router()
|
||||
|
||||
@router.command(Command('test'))
|
||||
def test():
|
||||
print(f'test command')
|
||||
|
||||
@router.command(Command('some'))
|
||||
def test():
|
||||
print(f'some command')
|
||||
|
||||
@router.command(Command('more'))
|
||||
def test():
|
||||
print(f'more command')
|
||||
|
||||
app = App(override_system_messages=True,
|
||||
print_func=print)
|
||||
app.include_router(router)
|
||||
app.start_polling()
|
||||
|
||||
output = mock_stdout.getvalue()
|
||||
|
||||
self.assertRegex(output, re.compile(r'\ntest command\n(.|\n)*\nsome command\n(.|\n)*\nmore command'))
|
||||
@@ -0,0 +1,24 @@
|
||||
from argenta.command.models import InputCommand
|
||||
from argenta.command.exceptions import (UnprocessedInputFlagException,
|
||||
RepeatedInputFlagsException,
|
||||
EmptyInputCommandException)
|
||||
|
||||
import unittest
|
||||
|
||||
|
||||
class TestInputCommand(unittest.TestCase):
|
||||
def test_parse_correct_raw_command(self):
|
||||
self.assertEqual(InputCommand.parse('ssh --host 192.168.0.3').get_trigger(), 'ssh')
|
||||
|
||||
def test_parse_raw_command_without_flag_name_with_value(self):
|
||||
with self.assertRaises(UnprocessedInputFlagException):
|
||||
InputCommand.parse('ssh 192.168.0.3')
|
||||
|
||||
def test_parse_raw_command_with_repeated_flag_name(self):
|
||||
with self.assertRaises(RepeatedInputFlagsException):
|
||||
InputCommand.parse('ssh --host 192.168.0.3 --host 172.198.0.43')
|
||||
|
||||
def test_parse_empty_raw_command(self):
|
||||
with self.assertRaises(EmptyInputCommandException):
|
||||
InputCommand.parse('')
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
from argenta.app.dividing_line import DynamicDividingLine, StaticDividingLine
|
||||
|
||||
import unittest
|
||||
|
||||
|
||||
class TestDividingLine(unittest.TestCase):
|
||||
def test_get_static_dividing_line_full_line(self):
|
||||
line = StaticDividingLine('-')
|
||||
self.assertEqual(line.get_full_line().count('-'), 25)
|
||||
|
||||
def test_get_dynamic_dividing_line_full_line(self):
|
||||
line = DynamicDividingLine()
|
||||
self.assertEqual(line.get_full_line(20).count('-'), 20)
|
||||
|
||||
def test_get_dividing_line_unit_part(self):
|
||||
line = StaticDividingLine('')
|
||||
self.assertEqual(line.get_unit_part(), ' ')
|
||||
|
||||
def test_get_dividing_line2_unit_part(self):
|
||||
line = StaticDividingLine('+-0987654321!@#$%^&*()_')
|
||||
self.assertEqual(line.get_unit_part(), '+')
|
||||
@@ -0,0 +1,91 @@
|
||||
from argenta.command.flag.models import Flag, InputFlag
|
||||
|
||||
import unittest
|
||||
import re
|
||||
|
||||
|
||||
class TestFlag(unittest.TestCase):
|
||||
def test_get_string_entity(self):
|
||||
self.assertEqual(Flag(name='test').get_string_entity(),
|
||||
'--test')
|
||||
|
||||
def test_get_string_entity2(self):
|
||||
self.assertEqual(Flag(name='test',
|
||||
prefix='---').get_string_entity(),
|
||||
'---test')
|
||||
|
||||
def test_get_flag_name(self):
|
||||
self.assertEqual(Flag(name='test').get_name(),
|
||||
'test')
|
||||
|
||||
def test_get_flag_prefix(self):
|
||||
self.assertEqual(Flag(name='test').get_prefix(),
|
||||
'--')
|
||||
|
||||
def test_get_flag_prefix2(self):
|
||||
self.assertEqual(Flag(name='test',
|
||||
prefix='--').get_prefix(),
|
||||
'--')
|
||||
|
||||
def test_get_flag_value_without_set(self):
|
||||
self.assertEqual(InputFlag(name='test').get_value(),
|
||||
None)
|
||||
|
||||
def test_get_flag_value_with_set(self):
|
||||
flag = InputFlag(name='test')
|
||||
flag.set_value('example')
|
||||
self.assertEqual(flag.get_value(), 'example')
|
||||
|
||||
def test_validate_incorrect_flag_value_with_list_of_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=['1', '2', '3'])
|
||||
self.assertEqual(flag.validate_input_flag_value('bad value'), False)
|
||||
|
||||
def test_validate_correct_flag_value_with_list_of_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=['1', '2', '3'])
|
||||
self.assertEqual(flag.validate_input_flag_value('1'), True)
|
||||
|
||||
def test_validate_incorrect_flag_value_with_pattern_of_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=re.compile(r'192.168.\d+.\d+'))
|
||||
self.assertEqual(flag.validate_input_flag_value('152.123.9.8'), False)
|
||||
|
||||
def test_validate_correct_flag_value_with_pattern_of_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=re.compile(r'192.168.\d+.\d+'))
|
||||
self.assertEqual(flag.validate_input_flag_value('192.168.9.8'), True)
|
||||
|
||||
def test_validate_correct_empty_flag_value_without_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=False)
|
||||
self.assertEqual(flag.validate_input_flag_value(None), True)
|
||||
|
||||
def test_validate_correct_empty_flag_value_with_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=True)
|
||||
self.assertEqual(flag.validate_input_flag_value(None), True)
|
||||
|
||||
def test_validate_incorrect_random_flag_value_without_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=False)
|
||||
self.assertEqual(flag.validate_input_flag_value('random value'), False)
|
||||
|
||||
def test_validate_correct_random_flag_value_with_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=True)
|
||||
self.assertEqual(flag.validate_input_flag_value('random value'), True)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
from argenta.command.flag.models import Flag, Flags
|
||||
|
||||
import unittest
|
||||
|
||||
|
||||
class TestFlags(unittest.TestCase):
|
||||
def test_get_flags(self):
|
||||
flags = Flags()
|
||||
list_of_flags = [
|
||||
Flag('test1'),
|
||||
Flag('test2'),
|
||||
Flag('test3'),
|
||||
]
|
||||
flags.add_flags(list_of_flags)
|
||||
self.assertEqual(flags.get_flags(),
|
||||
list_of_flags)
|
||||
|
||||
def test_add_flag(self):
|
||||
flags = Flags()
|
||||
flags.add_flag(Flag('test'))
|
||||
self.assertEqual(len(flags.get_flags()), 1)
|
||||
|
||||
def test_add_flags(self):
|
||||
flags = Flags()
|
||||
flags.add_flags([Flag('test'), Flag('test2')])
|
||||
self.assertEqual(len(flags.get_flags()), 2)
|
||||
@@ -0,0 +1,36 @@
|
||||
from argenta.router import Router
|
||||
from argenta.command import Command
|
||||
from argenta.router.exceptions import TriggerCannotContainSpacesException
|
||||
|
||||
import unittest
|
||||
|
||||
|
||||
class TestRouter(unittest.TestCase):
|
||||
def test_get_router_name(self):
|
||||
self.assertEqual(Router(name='test name').get_name(), 'test name')
|
||||
|
||||
def test_get_router_title(self):
|
||||
self.assertEqual(Router(title='test title').get_title(), 'test title')
|
||||
|
||||
def test_register_command_with_spaces_in_trigger(self):
|
||||
router = Router()
|
||||
with self.assertRaises(TriggerCannotContainSpacesException):
|
||||
@router.command(Command(trigger='command with spaces'))
|
||||
def test():
|
||||
return 'correct result'
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user