mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 10:05:28 +03:00
refactor system tests
This commit is contained in:
@@ -318,3 +318,5 @@ http-client.private.env.json
|
|||||||
# Apifox Helper cache
|
# Apifox Helper cache
|
||||||
.idea/.cache/.Apifox_Helper
|
.idea/.cache/.Apifox_Helper
|
||||||
.idea/ApifoxUploaderProjectSetting.xml
|
.idea/ApifoxUploaderProjectSetting.xml
|
||||||
|
|
||||||
|
.zed
|
||||||
+1
-1
@@ -59,7 +59,7 @@ reportUnusedFunction = false
|
|||||||
branch = true
|
branch = true
|
||||||
omit = [
|
omit = [
|
||||||
"src/argenta/app/protocols.py",
|
"src/argenta/app/protocols.py",
|
||||||
"src/argenta/command/exceptions.py",
|
"src/argenta/*/exceptions.py",
|
||||||
"src/argenta/metrics/*"
|
"src/argenta/metrics/*"
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|||||||
@@ -129,17 +129,6 @@ class Router:
|
|||||||
command_handler.handling(response)
|
command_handler.handling(response)
|
||||||
|
|
||||||
|
|
||||||
class CommandDecorator:
|
|
||||||
def __init__(self, router_instance: Router, command: Command):
|
|
||||||
self.router: Router = router_instance
|
|
||||||
self.command: Command = command
|
|
||||||
|
|
||||||
def __call__(self, handler_func: Callable[..., None]) -> Callable[..., None]:
|
|
||||||
_validate_func_args(handler_func)
|
|
||||||
self.router.command_handlers.add_handler(CommandHandler(handler_func, self.command))
|
|
||||||
return handler_func
|
|
||||||
|
|
||||||
|
|
||||||
def _structuring_input_flags(handled_command: Command, input_flags: InputFlags) -> Response:
|
def _structuring_input_flags(handled_command: Command, input_flags: InputFlags) -> Response:
|
||||||
"""
|
"""
|
||||||
Private. Validates flags of input command
|
Private. Validates flags of input command
|
||||||
|
|||||||
@@ -1,10 +1,8 @@
|
|||||||
import io
|
|
||||||
import re
|
import re
|
||||||
import sys
|
import sys
|
||||||
from unittest import TestCase
|
from collections.abc import Iterator
|
||||||
from unittest.mock import MagicMock, patch
|
|
||||||
|
|
||||||
import _io
|
import pytest
|
||||||
|
|
||||||
from argenta import App, Orchestrator, Router
|
from argenta import App, Orchestrator, Router
|
||||||
from argenta.command import Command, PredefinedFlags
|
from argenta.command import Command, PredefinedFlags
|
||||||
@@ -13,252 +11,255 @@ from argenta.command.flag.models import ValidationStatus
|
|||||||
from argenta.response import Response
|
from argenta.response import Response
|
||||||
|
|
||||||
|
|
||||||
class PatchedArgvTestCase(TestCase):
|
@pytest.fixture(autouse=True)
|
||||||
def setUp(self):
|
def patch_argv(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||||
super().setUp()
|
monkeypatch.setattr(sys, 'argv', ['program.py'])
|
||||||
self.patcher = patch.object(sys, 'argv', ['program.py'])
|
|
||||||
self.mock_argv = self.patcher.start()
|
|
||||||
self.addCleanup(self.patcher.stop)
|
|
||||||
|
|
||||||
|
|
||||||
class TestSystemHandlerNormalWork(PatchedArgvTestCase):
|
def _mock_input(inputs: Iterator[str]) -> str:
|
||||||
@patch("builtins.input", side_effect=["help", "q"])
|
return next(inputs)
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
|
||||||
def test_input_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
|
||||||
|
|
||||||
@router.command(Command('test'))
|
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
|
||||||
print('test command')
|
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
# ============================================================================
|
||||||
print_func=print)
|
# Tests for empty input handling
|
||||||
app.include_router(router)
|
# ============================================================================
|
||||||
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}'))
|
|
||||||
orchestrator.start_polling(app)
|
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
|
||||||
|
|
||||||
self.assertIn("\nUnknown command: help\n", output)
|
def test_empty_input_triggers_empty_command_handler(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
|
inputs = iter(["", "q"])
|
||||||
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
|
|
||||||
|
router = Router()
|
||||||
|
orchestrator = Orchestrator()
|
||||||
|
|
||||||
|
@router.command(Command('test'))
|
||||||
|
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
|
print('test command')
|
||||||
|
|
||||||
@patch("builtins.input", side_effect=["TeSt", "Q"])
|
app = App(override_system_messages=True, print_func=print)
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
app.include_router(router)
|
||||||
def test_input_incorrect_command2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
app.set_empty_command_handler(lambda: print('Empty input command'))
|
||||||
router = Router()
|
orchestrator.start_polling(app)
|
||||||
orchestrator = Orchestrator()
|
|
||||||
|
|
||||||
@router.command(Command('test'))
|
output = capsys.readouterr().out
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
|
||||||
print('test command')
|
|
||||||
|
|
||||||
app = App(ignore_command_register=False,
|
assert "\nEmpty input command\n" in output
|
||||||
override_system_messages=True,
|
|
||||||
print_func=print)
|
|
||||||
app.include_router(router)
|
|
||||||
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}'))
|
|
||||||
orchestrator.start_polling(app)
|
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
|
||||||
|
|
||||||
self.assertIn('\nUnknown command: TeSt\n', output)
|
# ============================================================================
|
||||||
|
# Tests for unknown command handling
|
||||||
|
# ============================================================================
|
||||||
|
|
||||||
|
|
||||||
@patch("builtins.input", side_effect=["test --help", "q"])
|
def test_unknown_command_triggers_unknown_command_handler(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
inputs = iter(["help", "q"])
|
||||||
def test_input_correct_command_with_unregistered_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
router = Router()
|
||||||
|
orchestrator = Orchestrator()
|
||||||
|
|
||||||
@router.command(Command('test'))
|
@router.command(Command('test'))
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
undefined_flag = response.input_flags.get_flag_by_name('help')
|
print('test command')
|
||||||
if undefined_flag and undefined_flag.status == ValidationStatus.UNDEFINED:
|
|
||||||
print(f'test command with undefined flag: {undefined_flag.string_entity}')
|
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
app = App(override_system_messages=True, print_func=print)
|
||||||
print_func=print)
|
app.include_router(router)
|
||||||
app.include_router(router)
|
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}'))
|
||||||
orchestrator.start_polling(app)
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
output = capsys.readouterr().out
|
||||||
|
|
||||||
self.assertIn('\ntest command with undefined flag: --help\n', output)
|
assert "\nUnknown command: help\n" in output
|
||||||
|
|
||||||
|
|
||||||
@patch("builtins.input", side_effect=["test --port 22", "q"])
|
def test_case_sensitive_command_triggers_unknown_command_handler(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
inputs = iter(["TeSt", "Q"])
|
||||||
def test_input_correct_command_with_unregistered_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
router = Router()
|
||||||
|
orchestrator = Orchestrator()
|
||||||
|
|
||||||
@router.command(Command('test'))
|
@router.command(Command('test'))
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
undefined_flag = response.input_flags.get_flag_by_name("port")
|
print('test command')
|
||||||
if undefined_flag and undefined_flag.status == ValidationStatus.UNDEFINED:
|
|
||||||
print(f'test command with undefined flag with value: {undefined_flag.string_entity} {undefined_flag.input_value}')
|
|
||||||
else:
|
|
||||||
raise
|
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
app = App(ignore_command_register=False, override_system_messages=True, print_func=print)
|
||||||
print_func=print)
|
app.include_router(router)
|
||||||
app.include_router(router)
|
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}'))
|
||||||
orchestrator.start_polling(app)
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
output = capsys.readouterr().out
|
||||||
|
|
||||||
self.assertIn('\ntest command with undefined flag with value: --port 22\n', output)
|
assert '\nUnknown command: TeSt\n' in output
|
||||||
|
|
||||||
|
|
||||||
@patch("builtins.input", side_effect=["test --host 192.168.32.1 --port 132", "q"])
|
def test_mixed_valid_and_unknown_commands_handled_correctly(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
inputs = iter(["test", "some", "q"])
|
||||||
def test_input_correct_command_with_one_correct_flag_an_one_incorrect_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
router = Router()
|
||||||
flags = Flags([PredefinedFlags.HOST])
|
orchestrator = Orchestrator()
|
||||||
|
|
||||||
@router.command(Command('test', flags=flags))
|
@router.command(Command('test'))
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
undefined_flag = response.input_flags.get_flag_by_name("port")
|
print('test command')
|
||||||
if undefined_flag and undefined_flag.status == ValidationStatus.UNDEFINED:
|
|
||||||
print(f'connecting to host with flag: {undefined_flag.string_entity} {undefined_flag.input_value}')
|
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
app = App(override_system_messages=True, print_func=print)
|
||||||
print_func=print)
|
app.include_router(router)
|
||||||
app.include_router(router)
|
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}'))
|
||||||
orchestrator.start_polling(app)
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
output = capsys.readouterr().out
|
||||||
|
|
||||||
self.assertIn('\nconnecting to host with flag: --port 132\n', output)
|
assert re.search(r'\ntest command\n(.|\n)*\nUnknown command: some', output)
|
||||||
|
|
||||||
|
|
||||||
@patch("builtins.input", side_effect=["test", "some", "q"])
|
def test_multiple_commands_with_unknown_command_in_between(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
inputs = iter(["test", "some", "more", "q"])
|
||||||
def test_input_one_correct_command_and_one_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
router = Router()
|
||||||
|
orchestrator = Orchestrator()
|
||||||
|
|
||||||
@router.command(Command('test'))
|
@router.command(Command('test'))
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
print(f'test command')
|
print('test command')
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
@router.command(Command('more'))
|
||||||
print_func=print)
|
def test1(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
app.include_router(router)
|
print('more command')
|
||||||
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}'))
|
|
||||||
orchestrator.start_polling(app)
|
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
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.trigger}'))
|
||||||
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
self.assertRegex(output, re.compile(r'\ntest command\n(.|\n)*\nUnknown command: some'))
|
output = capsys.readouterr().out
|
||||||
|
|
||||||
|
assert re.search(r'\ntest command\n(.|\n)*\nUnknown command: some\n(.|\n)*\nmore command', output)
|
||||||
|
|
||||||
@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()
|
|
||||||
orchestrator = Orchestrator()
|
|
||||||
|
|
||||||
@router.command(Command('test'))
|
# ============================================================================
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
# Tests for unregistered flag handling
|
||||||
print(f'test command')
|
# ============================================================================
|
||||||
|
|
||||||
@router.command(Command('more'))
|
|
||||||
def test1(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
|
||||||
print(f'more command')
|
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
def test_unregistered_flag_without_value_is_accessible(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
print_func=print)
|
inputs = iter(["test --help", "q"])
|
||||||
app.include_router(router)
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}'))
|
|
||||||
orchestrator.start_polling(app)
|
router = Router()
|
||||||
|
orchestrator = Orchestrator()
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
@router.command(Command('test'))
|
||||||
|
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
|
undefined_flag = response.input_flags.get_flag_by_name('help')
|
||||||
|
if undefined_flag and undefined_flag.status == ValidationStatus.UNDEFINED:
|
||||||
|
print(f'test command with undefined flag: {undefined_flag.string_entity}')
|
||||||
|
|
||||||
self.assertRegex(output, re.compile(r'\ntest command\n(.|\n)*\nUnknown command: some\n(.|\n)*\nmore command'))
|
app = App(override_system_messages=True, print_func=print)
|
||||||
|
app.include_router(router)
|
||||||
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
|
output = capsys.readouterr().out
|
||||||
|
|
||||||
@patch("builtins.input", side_effect=["test 535 --port", "q"])
|
assert '\ntest command with undefined flag: --help\n' in output
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
|
||||||
def test_input_correct_command_with_incorrect_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
|
||||||
|
|
||||||
@router.command(Command('test'))
|
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
|
||||||
print(f'test command')
|
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
def test_unregistered_flag_with_value_is_accessible(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
print_func=print)
|
inputs = iter(["test --port 22", "q"])
|
||||||
app.include_router(router)
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
app.set_incorrect_input_syntax_handler(lambda command: print(f'Incorrect flag syntax: "{command}"'))
|
|
||||||
orchestrator.start_polling(app)
|
router = Router()
|
||||||
|
orchestrator = Orchestrator()
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
@router.command(Command('test'))
|
||||||
|
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
|
undefined_flag = response.input_flags.get_flag_by_name("port")
|
||||||
|
if undefined_flag and undefined_flag.status == ValidationStatus.UNDEFINED:
|
||||||
|
print(f'test command with undefined flag with value: {undefined_flag.string_entity} {undefined_flag.input_value}')
|
||||||
|
else:
|
||||||
|
raise
|
||||||
|
|
||||||
self.assertIn("\nIncorrect flag syntax: \"test 535 --port\"\n", output)
|
app = App(override_system_messages=True, print_func=print)
|
||||||
|
app.include_router(router)
|
||||||
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
|
output = capsys.readouterr().out
|
||||||
|
|
||||||
@patch("builtins.input", side_effect=["", "q"])
|
assert '\ntest command with undefined flag with value: --port 22\n' in output
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
|
||||||
def test_input_empty_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
|
||||||
|
|
||||||
@router.command(Command('test'))
|
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
|
||||||
print(f'test command')
|
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
def test_registered_and_unregistered_flags_coexist(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
print_func=print)
|
inputs = iter(["test --host 192.168.32.1 --port 132", "q"])
|
||||||
app.include_router(router)
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
app.set_empty_command_handler(lambda: print('Empty input command'))
|
|
||||||
orchestrator.start_polling(app)
|
router = Router()
|
||||||
|
orchestrator = Orchestrator()
|
||||||
|
flags = Flags([PredefinedFlags.HOST])
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
@router.command(Command('test', flags=flags))
|
||||||
|
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
|
undefined_flag = response.input_flags.get_flag_by_name("port")
|
||||||
|
if undefined_flag and undefined_flag.status == ValidationStatus.UNDEFINED:
|
||||||
|
print(f'connecting to host with flag: {undefined_flag.string_entity} {undefined_flag.input_value}')
|
||||||
|
|
||||||
self.assertIn("\nEmpty input command\n", output)
|
app = App(override_system_messages=True, print_func=print)
|
||||||
|
app.include_router(router)
|
||||||
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
|
output = capsys.readouterr().out
|
||||||
|
|
||||||
@patch("builtins.input", side_effect=["test --port 22 --port 33", "q"])
|
assert '\nconnecting to host with flag: --port 132\n' in output
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
|
||||||
def test_input_correct_command_with_repeated_flags(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
|
||||||
|
|
||||||
@router.command(Command('test', flags=PredefinedFlags.PORT))
|
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
|
||||||
print('test command')
|
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
# ============================================================================
|
||||||
print_func=print)
|
# Tests for incorrect flag syntax handling
|
||||||
app.include_router(router)
|
# ============================================================================
|
||||||
app.set_repeated_input_flags_handler(lambda command: print(f'Repeated input flags: "{command}"'))
|
|
||||||
orchestrator.start_polling(app)
|
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
|
||||||
|
|
||||||
self.assertIn('Repeated input flags: "test --port 22 --port 33"', output)
|
def test_flag_without_value_triggers_incorrect_syntax_handler(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
|
inputs = iter(["test 535 --port", "q"])
|
||||||
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
|
|
||||||
|
router = Router()
|
||||||
|
orchestrator = Orchestrator()
|
||||||
|
|
||||||
@patch("builtins.input", side_effect=["test --help", "q"])
|
@router.command(Command('test'))
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
def test_input_correct_command_with_unregistered_flag3(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
print('test command')
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
|
||||||
|
|
||||||
@router.command(Command('test'))
|
app = App(override_system_messages=True, print_func=print)
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
app.include_router(router)
|
||||||
undefined_flag = response.input_flags.get_flag_by_name('help')
|
app.set_incorrect_input_syntax_handler(lambda command: print(f'Incorrect flag syntax: "{command}"'))
|
||||||
if undefined_flag and undefined_flag.status == ValidationStatus.UNDEFINED:
|
orchestrator.start_polling(app)
|
||||||
print(f'test command with undefined flag: {undefined_flag.string_entity}')
|
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
output = capsys.readouterr().out
|
||||||
print_func=print)
|
|
||||||
app.include_router(router)
|
|
||||||
orchestrator.start_polling(app)
|
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
assert "\nIncorrect flag syntax: \"test 535 --port\"\n" in output
|
||||||
|
|
||||||
self.assertIn('\ntest command with undefined flag: --help\n', output)
|
|
||||||
|
# ============================================================================
|
||||||
|
# Tests for repeated flag handling
|
||||||
|
# ============================================================================
|
||||||
|
|
||||||
|
|
||||||
|
def test_repeated_flags_trigger_repeated_flags_handler(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
|
inputs = iter(["test --port 22 --port 33", "q"])
|
||||||
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
|
|
||||||
|
router = Router()
|
||||||
|
orchestrator = Orchestrator()
|
||||||
|
|
||||||
|
@router.command(Command('test', flags=PredefinedFlags.PORT))
|
||||||
|
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
|
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}"'))
|
||||||
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
|
output = capsys.readouterr().out
|
||||||
|
|
||||||
|
assert 'Repeated input flags: "test --port 22 --port 33"' in output
|
||||||
|
|||||||
@@ -1,10 +1,8 @@
|
|||||||
import io
|
|
||||||
import re
|
import re
|
||||||
import sys
|
import sys
|
||||||
from unittest import TestCase
|
from collections.abc import Iterator
|
||||||
from unittest.mock import MagicMock, patch
|
|
||||||
|
|
||||||
import _io
|
import pytest
|
||||||
|
|
||||||
from argenta import App, Orchestrator, Router
|
from argenta import App, Orchestrator, Router
|
||||||
from argenta.command import Command, PredefinedFlags
|
from argenta.command import Command, PredefinedFlags
|
||||||
@@ -14,244 +12,261 @@ from argenta.command.flag.models import PossibleValues, ValidationStatus
|
|||||||
from argenta.response import Response
|
from argenta.response import Response
|
||||||
|
|
||||||
|
|
||||||
class PatchedArgvTestCase(TestCase):
|
@pytest.fixture(autouse=True)
|
||||||
def setUp(self):
|
def patch_argv(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||||
super().setUp()
|
monkeypatch.setattr(sys, 'argv', ['program.py'])
|
||||||
self.patcher = patch.object(sys, 'argv', ['program.py'])
|
|
||||||
self.mock_argv = self.patcher.start()
|
|
||||||
self.addCleanup(self.patcher.stop)
|
|
||||||
|
|
||||||
|
|
||||||
class TestSystemHandlerNormalWork(PatchedArgvTestCase):
|
def _mock_input(inputs: Iterator[str]) -> str:
|
||||||
@patch("builtins.input", side_effect=["test", "q"])
|
return next(inputs)
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
|
||||||
def test_input_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
|
||||||
|
|
||||||
@router.command(Command('test'))
|
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
|
||||||
print('test command')
|
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
|
||||||
print_func=print)
|
|
||||||
app.include_router(router)
|
|
||||||
orchestrator.start_polling(app)
|
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
|
||||||
|
|
||||||
self.assertIn('\ntest command\n', output)
|
|
||||||
|
|
||||||
|
|
||||||
@patch("builtins.input", side_effect=["TeSt", "q"])
|
# ============================================================================
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
# Tests for basic command execution
|
||||||
def test_input_correct_command2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
# ============================================================================
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
|
||||||
|
|
||||||
@router.command(Command('test'))
|
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
|
||||||
print('test command')
|
|
||||||
|
|
||||||
app = App(ignore_command_register=True,
|
|
||||||
override_system_messages=True,
|
|
||||||
print_func=print)
|
|
||||||
app.include_router(router)
|
|
||||||
orchestrator.start_polling(app)
|
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
|
||||||
|
|
||||||
self.assertIn('\ntest command\n', output)
|
|
||||||
|
|
||||||
|
|
||||||
@patch("builtins.input", side_effect=["test --help", "q"])
|
def test_simple_command_executes_successfully(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
inputs = iter(["test", "q"])
|
||||||
def test_input_correct_command_with_custom_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
router = Router()
|
||||||
flag = Flag('help', prefix='--', possible_values=PossibleValues.NEITHER)
|
orchestrator = Orchestrator()
|
||||||
|
|
||||||
@router.command(Command('test', flags=flag))
|
@router.command(Command('test'))
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
valid_flag = response.input_flags.get_flag_by_name('help')
|
print('test command')
|
||||||
if valid_flag and valid_flag.status == ValidationStatus.VALID:
|
|
||||||
print(f'\nhelp for {valid_flag.name} flag\n')
|
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
app = App(override_system_messages=True, print_func=print)
|
||||||
print_func=print)
|
app.include_router(router)
|
||||||
app.include_router(router)
|
orchestrator.start_polling(app)
|
||||||
orchestrator.start_polling(app)
|
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
output = capsys.readouterr().out
|
||||||
|
|
||||||
self.assertIn('\nhelp for help flag\n', output)
|
assert '\ntest command\n' in 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()
|
|
||||||
orchestrator = Orchestrator()
|
|
||||||
flag = Flag('port', prefix='--', possible_values=re.compile(r'^\d{1,5}$'))
|
|
||||||
|
|
||||||
@router.command(Command('test', flags=flag))
|
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
|
||||||
valid_flag = response.input_flags.get_flag_by_name('port')
|
|
||||||
if valid_flag and valid_flag.status == ValidationStatus.VALID:
|
|
||||||
print(f'flag value for {valid_flag.name} flag : {valid_flag.input_value}')
|
|
||||||
|
|
||||||
app = App(
|
|
||||||
override_system_messages=True,
|
|
||||||
repeat_command_groups_printing=True,
|
|
||||||
print_func=print
|
|
||||||
)
|
|
||||||
app.include_router(router)
|
|
||||||
orchestrator.start_polling(app)
|
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
|
||||||
|
|
||||||
self.assertIn('\nflag value for port flag : 22\n', output)
|
|
||||||
|
|
||||||
|
|
||||||
@patch("builtins.input", side_effect=["test -H", "q"])
|
def test_case_insensitive_command_executes_when_enabled(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
inputs = iter(["TeSt", "q"])
|
||||||
def test_input_correct_command_with_default_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
router = Router()
|
||||||
flag = PredefinedFlags.SHORT_HELP
|
orchestrator = Orchestrator()
|
||||||
|
|
||||||
@router.command(Command('test', flags=flag))
|
@router.command(Command('test'))
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
valid_flag = response.input_flags.get_flag_by_name('H')
|
print('test command')
|
||||||
if valid_flag and valid_flag.status == ValidationStatus.VALID:
|
|
||||||
print(f'help for {valid_flag.name} flag')
|
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
app = App(ignore_command_register=True, override_system_messages=True, print_func=print)
|
||||||
print_func=print)
|
app.include_router(router)
|
||||||
app.include_router(router)
|
orchestrator.start_polling(app)
|
||||||
orchestrator.start_polling(app)
|
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
output = capsys.readouterr().out
|
||||||
|
|
||||||
self.assertIn('\nhelp for H flag\n', output)
|
assert '\ntest command\n' in output
|
||||||
|
|
||||||
|
|
||||||
@patch("builtins.input", side_effect=["test --info", "q"])
|
def test_two_commands_execute_sequentially(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
inputs = iter(["test", "some", "q"])
|
||||||
def test_input_correct_command_with_default_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
router = Router()
|
||||||
flag = PredefinedFlags.INFO
|
orchestrator = Orchestrator()
|
||||||
|
|
||||||
@router.command(Command('test', flags=flag))
|
@router.command(Command('test'))
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
valid_flag = response.input_flags.get_flag_by_name('info')
|
print('test command')
|
||||||
if valid_flag and valid_flag.status == ValidationStatus.VALID:
|
|
||||||
print('info about test command')
|
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
@router.command(Command('some'))
|
||||||
print_func=print)
|
def test2(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
app.include_router(router)
|
print('some command')
|
||||||
orchestrator.start_polling(app)
|
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
app = App(override_system_messages=True, print_func=print)
|
||||||
|
app.include_router(router)
|
||||||
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
self.assertIn('\ninfo about test command\n', output)
|
output = capsys.readouterr().out
|
||||||
|
|
||||||
|
assert re.search(r'\ntest command\n(.|\n)*\nsome command\n', output)
|
||||||
|
|
||||||
|
|
||||||
@patch("builtins.input", side_effect=["test --host 192.168.0.1", "q"])
|
def test_three_commands_execute_sequentially(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
inputs = iter(["test", "some", "more", "q"])
|
||||||
def test_input_correct_command_with_default_flag3(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
router = Router()
|
||||||
flag = PredefinedFlags.HOST
|
orchestrator = Orchestrator()
|
||||||
|
|
||||||
@router.command(Command('test', flags=flag))
|
@router.command(Command('test'))
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
def test(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
valid_flag = response.input_flags.get_flag_by_name('host')
|
print('test command')
|
||||||
if valid_flag and valid_flag.status == ValidationStatus.VALID:
|
|
||||||
print(f'connecting to host {valid_flag.input_value}')
|
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
@router.command(Command('some'))
|
||||||
print_func=print)
|
def test1(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
app.include_router(router)
|
print('some command')
|
||||||
orchestrator.start_polling(app)
|
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
@router.command(Command('more'))
|
||||||
|
def test2(_response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
|
print('more command')
|
||||||
|
|
||||||
self.assertIn('\nconnecting to host 192.168.0.1\n', output)
|
app = App(override_system_messages=True, print_func=print)
|
||||||
|
app.include_router(router)
|
||||||
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
|
output = capsys.readouterr().out
|
||||||
|
|
||||||
|
assert re.search(r'\ntest command\n(.|\n)*\nsome command\n(.|\n)*\nmore command', output)
|
||||||
|
|
||||||
|
|
||||||
@patch("builtins.input", side_effect=["test --host 192.168.32.1 --port 132", "q"])
|
# ============================================================================
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
# Tests for custom flag handling
|
||||||
def test_input_correct_command_with_two_flags(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
# ============================================================================
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
|
||||||
flags = Flags([PredefinedFlags.HOST, PredefinedFlags.PORT])
|
|
||||||
|
|
||||||
@router.command(Command('test', flags=flags))
|
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
|
||||||
host_flag = response.input_flags.get_flag_by_name('host')
|
|
||||||
port_flag = response.input_flags.get_flag_by_name('port')
|
|
||||||
if (host_flag and host_flag.status == ValidationStatus.VALID) and (port_flag and port_flag.status == ValidationStatus.VALID):
|
|
||||||
print(f'connecting to host {host_flag.input_value} and port {port_flag.input_value}')
|
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
|
||||||
print_func=print)
|
|
||||||
app.include_router(router)
|
|
||||||
orchestrator.start_polling(app)
|
|
||||||
|
|
||||||
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"])
|
def test_custom_flag_without_value_is_recognized(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
inputs = iter(["test --help", "q"])
|
||||||
def test_input_two_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
router = Router()
|
||||||
|
orchestrator = Orchestrator()
|
||||||
|
flag = Flag('help', prefix='--', possible_values=PossibleValues.NEITHER)
|
||||||
|
|
||||||
@router.command(Command('test'))
|
@router.command(Command('test', flags=flag))
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
print(f'test command')
|
valid_flag = response.input_flags.get_flag_by_name('help')
|
||||||
|
if valid_flag and valid_flag.status == ValidationStatus.VALID:
|
||||||
|
print(f'\nhelp for {valid_flag.name} flag\n')
|
||||||
|
|
||||||
@router.command(Command('some'))
|
app = App(override_system_messages=True, print_func=print)
|
||||||
def test2(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
app.include_router(router)
|
||||||
print(f'some command')
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
output = capsys.readouterr().out
|
||||||
print_func=print)
|
|
||||||
app.include_router(router)
|
|
||||||
orchestrator.start_polling(app)
|
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
assert '\nhelp for help flag\n' in output
|
||||||
|
|
||||||
self.assertRegex(output, re.compile(r'\ntest command\n(.|\n)*\nsome command\n'))
|
|
||||||
|
|
||||||
|
|
||||||
@patch("builtins.input", side_effect=["test", "some", "more", "q"])
|
def test_custom_flag_with_regex_validation_accepts_valid_value(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
@patch("sys.stdout", new_callable=io.StringIO)
|
inputs = iter(["test --port 22", "q"])
|
||||||
def test_input_three_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock):
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
router = Router()
|
|
||||||
orchestrator = Orchestrator()
|
router = Router()
|
||||||
|
orchestrator = Orchestrator()
|
||||||
|
flag = Flag('port', prefix='--', possible_values=re.compile(r'^\d{1,5}$'))
|
||||||
|
|
||||||
@router.command(Command('test'))
|
@router.command(Command('test', flags=flag))
|
||||||
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
print(f'test command')
|
valid_flag = response.input_flags.get_flag_by_name('port')
|
||||||
|
if valid_flag and valid_flag.status == ValidationStatus.VALID:
|
||||||
|
print(f'flag value for {valid_flag.name} flag : {valid_flag.input_value}')
|
||||||
|
|
||||||
@router.command(Command('some'))
|
app = App(override_system_messages=True, repeat_command_groups_printing=True, print_func=print)
|
||||||
def test1(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
app.include_router(router)
|
||||||
print(f'some command')
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
@router.command(Command('more'))
|
output = capsys.readouterr().out
|
||||||
def test2(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
|
||||||
print(f'more command')
|
|
||||||
|
|
||||||
app = App(override_system_messages=True,
|
assert '\nflag value for port flag : 22\n' in output
|
||||||
print_func=print)
|
|
||||||
app.include_router(router)
|
|
||||||
orchestrator.start_polling(app)
|
|
||||||
|
|
||||||
output = mock_stdout.getvalue()
|
|
||||||
|
|
||||||
self.assertRegex(output, re.compile(r'\ntest command\n(.|\n)*\nsome command\n(.|\n)*\nmore command'))
|
# ============================================================================
|
||||||
|
# Tests for predefined flag handling
|
||||||
|
# ============================================================================
|
||||||
|
|
||||||
|
|
||||||
|
def test_predefined_short_help_flag_is_recognized(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
|
inputs = iter(["test -H", "q"])
|
||||||
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
|
|
||||||
|
router = Router()
|
||||||
|
orchestrator = Orchestrator()
|
||||||
|
flag = PredefinedFlags.SHORT_HELP
|
||||||
|
|
||||||
|
@router.command(Command('test', flags=flag))
|
||||||
|
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
|
valid_flag = response.input_flags.get_flag_by_name('H')
|
||||||
|
if valid_flag and valid_flag.status == ValidationStatus.VALID:
|
||||||
|
print(f'help for {valid_flag.name} flag')
|
||||||
|
|
||||||
|
app = App(override_system_messages=True, print_func=print)
|
||||||
|
app.include_router(router)
|
||||||
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
|
output = capsys.readouterr().out
|
||||||
|
|
||||||
|
assert '\nhelp for H flag\n' in output
|
||||||
|
|
||||||
|
|
||||||
|
def test_predefined_info_flag_is_recognized(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
|
inputs = iter(["test --info", "q"])
|
||||||
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
|
|
||||||
|
router = Router()
|
||||||
|
orchestrator = Orchestrator()
|
||||||
|
flag = PredefinedFlags.INFO
|
||||||
|
|
||||||
|
@router.command(Command('test', flags=flag))
|
||||||
|
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
|
valid_flag = response.input_flags.get_flag_by_name('info')
|
||||||
|
if valid_flag and valid_flag.status == ValidationStatus.VALID:
|
||||||
|
print('info about test command')
|
||||||
|
|
||||||
|
app = App(override_system_messages=True, print_func=print)
|
||||||
|
app.include_router(router)
|
||||||
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
|
output = capsys.readouterr().out
|
||||||
|
|
||||||
|
assert '\ninfo about test command\n' in output
|
||||||
|
|
||||||
|
|
||||||
|
def test_predefined_host_flag_with_value_is_recognized(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
|
inputs = iter(["test --host 192.168.0.1", "q"])
|
||||||
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
|
|
||||||
|
router = Router()
|
||||||
|
orchestrator = Orchestrator()
|
||||||
|
flag = PredefinedFlags.HOST
|
||||||
|
|
||||||
|
@router.command(Command('test', flags=flag))
|
||||||
|
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
|
valid_flag = response.input_flags.get_flag_by_name('host')
|
||||||
|
if valid_flag and valid_flag.status == ValidationStatus.VALID:
|
||||||
|
print(f'connecting to host {valid_flag.input_value}')
|
||||||
|
|
||||||
|
app = App(override_system_messages=True, print_func=print)
|
||||||
|
app.include_router(router)
|
||||||
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
|
output = capsys.readouterr().out
|
||||||
|
|
||||||
|
assert '\nconnecting to host 192.168.0.1\n' in output
|
||||||
|
|
||||||
|
|
||||||
|
# ============================================================================
|
||||||
|
# Tests for multiple flag handling
|
||||||
|
# ============================================================================
|
||||||
|
|
||||||
|
|
||||||
|
def test_two_predefined_flags_are_recognized_together(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
|
inputs = iter(["test --host 192.168.32.1 --port 132", "q"])
|
||||||
|
monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs))
|
||||||
|
|
||||||
|
router = Router()
|
||||||
|
orchestrator = Orchestrator()
|
||||||
|
flags = Flags([PredefinedFlags.HOST, PredefinedFlags.PORT])
|
||||||
|
|
||||||
|
@router.command(Command('test', flags=flags))
|
||||||
|
def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction]
|
||||||
|
host_flag = response.input_flags.get_flag_by_name('host')
|
||||||
|
port_flag = response.input_flags.get_flag_by_name('port')
|
||||||
|
if (host_flag and host_flag.status == ValidationStatus.VALID) and (port_flag and port_flag.status == ValidationStatus.VALID):
|
||||||
|
print(f'connecting to host {host_flag.input_value} and port {port_flag.input_value}')
|
||||||
|
|
||||||
|
app = App(override_system_messages=True, print_func=print)
|
||||||
|
app.include_router(router)
|
||||||
|
orchestrator.start_polling(app)
|
||||||
|
|
||||||
|
output = capsys.readouterr().out
|
||||||
|
|
||||||
|
assert '\nconnecting to host 192.168.32.1 and port 132\n' in output
|
||||||
|
|||||||
@@ -1,5 +1,4 @@
|
|||||||
import re
|
import re
|
||||||
from sys import flags
|
|
||||||
|
|
||||||
from argenta.command.flag import Flag, InputFlag, PossibleValues
|
from argenta.command.flag import Flag, InputFlag, PossibleValues
|
||||||
from argenta.command.flag.flags import Flags, InputFlags
|
from argenta.command.flag.flags import Flags, InputFlags
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
import re
|
import re
|
||||||
import pytest
|
import pytest
|
||||||
|
from pytest import CaptureFixture
|
||||||
|
|
||||||
from argenta.command import Command, InputCommand
|
from argenta.command import Command, InputCommand
|
||||||
from argenta.command.flag import Flag, InputFlag
|
from argenta.command.flag import Flag, InputFlag
|
||||||
@@ -89,64 +90,59 @@ def test_get_router_aliases():
|
|||||||
def test_get_router_aliases2():
|
def test_get_router_aliases2():
|
||||||
router = Router()
|
router = Router()
|
||||||
@router.command(Command('some', aliases={'test', 'case'}))
|
@router.command(Command('some', aliases={'test', 'case'}))
|
||||||
def handler(response: Response):
|
def handler(response: Response):
|
||||||
pass
|
pass
|
||||||
@router.command(Command('ext', aliases={'more', 'foo'}))
|
@router.command(Command('ext', aliases={'more', 'foo'}))
|
||||||
def handler2(response: Response):
|
def handler2(response: Response):
|
||||||
pass
|
pass
|
||||||
assert router.aliases == {'test', 'case', 'more', 'foo'}
|
assert router.aliases == {'test', 'case', 'more', 'foo'}
|
||||||
|
|
||||||
def test_get_router_aliases3():
|
def test_get_router_aliases3():
|
||||||
router = Router()
|
router = Router()
|
||||||
@router.command(Command('some'))
|
@router.command(Command('some'))
|
||||||
def handler(response: Response):
|
def handler(response: Response):
|
||||||
pass
|
pass
|
||||||
assert router.aliases == set()
|
assert router.aliases == set()
|
||||||
|
|
||||||
def test_find_appropiate_handler(capsys: pytest.CaptureFixture[str]):
|
def test_find_appropiate_handler(capsys: pytest.CaptureFixture[str]):
|
||||||
router = Router()
|
router = Router()
|
||||||
|
|
||||||
@router.command(Command('hello', aliases={'hi'}))
|
@router.command(Command('hello', aliases={'hi'}))
|
||||||
def handler(res: Response):
|
def handler(res: Response):
|
||||||
print("Hello World!")
|
print("Hello World!")
|
||||||
|
|
||||||
router.finds_appropriate_handler(InputCommand('hi'))
|
router.finds_appropriate_handler(InputCommand('hi'))
|
||||||
|
|
||||||
output = capsys.readouterr()
|
output = capsys.readouterr()
|
||||||
|
|
||||||
assert "Hello World!" in output.out
|
assert "Hello World!" in output.out
|
||||||
|
|
||||||
def test_find_appropiate_handler2(capsys: CaptureFixture[str]):
|
def test_find_appropiate_handler2(capsys: CaptureFixture[str]):
|
||||||
router = Router()
|
router = Router()
|
||||||
|
|
||||||
@router.command(Command('hello', flags=Flag('flag'), aliases={'hi'}))
|
@router.command(Command('hello', flags=Flag('flag'), aliases={'hi'}))
|
||||||
def handler(res: Response):
|
def handler(res: Response):
|
||||||
print("Hello World!")
|
print("Hello World!")
|
||||||
|
|
||||||
router.finds_appropriate_handler(InputCommand('hi'))
|
router.finds_appropriate_handler(InputCommand('hi'))
|
||||||
|
|
||||||
output = capsys.readouterr()
|
output = capsys.readouterr()
|
||||||
|
|
||||||
assert "Hello World!" in output.out
|
assert "Hello World!" in output.out
|
||||||
|
|
||||||
def test_wrong_typehint(capsys: pytest.CaptureFixture[str]):
|
def test_wrong_typehint(capsys: pytest.CaptureFixture[str]):
|
||||||
class NotResponse: pass
|
class NotResponse: pass
|
||||||
|
|
||||||
def func(response: NotResponse): pass
|
def func(response: NotResponse): pass
|
||||||
|
|
||||||
_validate_func_args(func)
|
_validate_func_args(func)
|
||||||
|
|
||||||
output = capsys.readouterr()
|
output = capsys.readouterr()
|
||||||
|
|
||||||
assert "WARNING" in output.out
|
assert "WARNING" in output.out
|
||||||
|
|
||||||
def test_missing_typehint(capsys: pytest.CaptureFixture[str]):
|
def test_missing_typehint(capsys: pytest.CaptureFixture[str]):
|
||||||
def func(response): pass # pyright: ignore[reportMissingParameterType, reportUnknownParameterType]
|
def func(response): pass # pyright: ignore[reportMissingParameterType, reportUnknownParameterType]
|
||||||
|
|
||||||
_validate_func_args(func) # pyright: ignore[reportUnknownArgumentType]
|
_validate_func_args(func) # pyright: ignore[reportUnknownArgumentType]
|
||||||
|
|
||||||
output = capsys.readouterr()
|
output = capsys.readouterr()
|
||||||
|
|
||||||
assert output.out == ''
|
assert output.out == ''
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user