diff --git a/.gitignore b/.gitignore index 897b850..a71cd19 100644 --- a/.gitignore +++ b/.gitignore @@ -318,3 +318,5 @@ http-client.private.env.json # Apifox Helper cache .idea/.cache/.Apifox_Helper .idea/ApifoxUploaderProjectSetting.xml + +.zed \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index c742baa..eb066f7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -59,7 +59,7 @@ reportUnusedFunction = false branch = true omit = [ "src/argenta/app/protocols.py", - "src/argenta/command/exceptions.py", + "src/argenta/*/exceptions.py", "src/argenta/metrics/*" ] diff --git a/src/argenta/router/entity.py b/src/argenta/router/entity.py index fa9a5ea..b267a4f 100644 --- a/src/argenta/router/entity.py +++ b/src/argenta/router/entity.py @@ -129,17 +129,6 @@ class Router: 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: """ Private. Validates flags of input command diff --git a/tests/system_tests/test_system_handling_non_standard_behavior.py b/tests/system_tests/test_system_handling_non_standard_behavior.py index 47375f1..e952df0 100644 --- a/tests/system_tests/test_system_handling_non_standard_behavior.py +++ b/tests/system_tests/test_system_handling_non_standard_behavior.py @@ -1,10 +1,8 @@ -import io import re import sys -from unittest import TestCase -from unittest.mock import MagicMock, patch +from collections.abc import Iterator -import _io +import pytest from argenta import App, Orchestrator, Router from argenta.command import Command, PredefinedFlags @@ -13,252 +11,255 @@ from argenta.command.flag.models import ValidationStatus from argenta.response import Response -class PatchedArgvTestCase(TestCase): - def setUp(self): - super().setUp() - self.patcher = patch.object(sys, 'argv', ['program.py']) - self.mock_argv = self.patcher.start() - self.addCleanup(self.patcher.stop) +@pytest.fixture(autouse=True) +def patch_argv(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(sys, 'argv', ['program.py']) -class TestSystemHandlerNormalWork(PatchedArgvTestCase): - @patch("builtins.input", side_effect=["help", "q"]) - @patch("sys.stdout", new_callable=io.StringIO) - def test_input_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): - router = Router() - orchestrator = Orchestrator() +def _mock_input(inputs: Iterator[str]) -> str: + return next(inputs) - @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) - app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}')) - orchestrator.start_polling(app) +# ============================================================================ +# Tests for empty input handling +# ============================================================================ - 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"]) - @patch("sys.stdout", new_callable=io.StringIO) - def test_input_incorrect_command2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): - router = Router() - orchestrator = Orchestrator() + app = App(override_system_messages=True, print_func=print) + app.include_router(router) + app.set_empty_command_handler(lambda: print('Empty input command')) + orchestrator.start_polling(app) - @router.command(Command('test')) - def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction] - print('test command') + output = capsys.readouterr().out - 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.trigger}')) - orchestrator.start_polling(app) + assert "\nEmpty input command\n" in output - output = mock_stdout.getvalue() - self.assertIn('\nUnknown command: TeSt\n', output) +# ============================================================================ +# Tests for unknown command handling +# ============================================================================ - @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() - orchestrator = Orchestrator() +def test_unknown_command_triggers_unknown_command_handler(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + inputs = iter(["help", "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] - 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}') + @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) + 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) - 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"]) - @patch("sys.stdout", new_callable=io.StringIO) - def test_input_correct_command_with_unregistered_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): - router = Router() - orchestrator = Orchestrator() +def test_case_sensitive_command_triggers_unknown_command_handler(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + inputs = iter(["TeSt", "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] - 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 + @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) + 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.trigger}')) + 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"]) - @patch("sys.stdout", new_callable=io.StringIO) - def test_input_correct_command_with_one_correct_flag_an_one_incorrect_flag(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): - router = Router() - orchestrator = Orchestrator() - flags = Flags([PredefinedFlags.HOST]) +def test_mixed_valid_and_unknown_commands_handled_correctly(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + inputs = iter(["test", "some", "q"]) + monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs)) + + router = Router() + orchestrator = Orchestrator() - @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}') + @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) + 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) - 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"]) - @patch("sys.stdout", new_callable=io.StringIO) - def test_input_one_correct_command_and_one_incorrect_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): - router = Router() - orchestrator = Orchestrator() +def test_multiple_commands_with_unknown_command_in_between(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + inputs = iter(["test", "some", "more", "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(f'test command') + @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) - app.set_unknown_command_handler(lambda command: print(f'Unknown command: {command.trigger}')) - orchestrator.start_polling(app) + @router.command(Command('more')) + def test1(_response: Response) -> None: # pyright: ignore[reportUnusedFunction] + print('more command') - 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] - print(f'test command') +# ============================================================================ +# Tests for unregistered flag handling +# ============================================================================ - @router.command(Command('more')) - def test1(response: Response) -> None: # pyright: ignore[reportUnusedFunction] - 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.trigger}')) - orchestrator.start_polling(app) +def test_unregistered_flag_without_value_is_accessible(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + inputs = iter(["test --help", "q"]) + monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs)) + + 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"]) - @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() + assert '\ntest command with undefined flag: --help\n' in output - @router.command(Command('test')) - def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction] - print(f'test command') - app = App(override_system_messages=True, - print_func=print) - app.include_router(router) - app.set_incorrect_input_syntax_handler(lambda command: print(f'Incorrect flag syntax: "{command}"')) - orchestrator.start_polling(app) +def test_unregistered_flag_with_value_is_accessible(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + inputs = iter(["test --port 22", "q"]) + monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs)) + + 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"]) - @patch("sys.stdout", new_callable=io.StringIO) - def test_input_empty_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): - router = Router() - orchestrator = Orchestrator() + assert '\ntest command with undefined flag with value: --port 22\n' in output - @router.command(Command('test')) - def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction] - 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')) - orchestrator.start_polling(app) +def test_registered_and_unregistered_flags_coexist(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]) - 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"]) - @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() + assert '\nconnecting to host with flag: --port 132\n' in output - @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) +# ============================================================================ +# Tests for incorrect flag syntax handling +# ============================================================================ - 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"]) - @patch("sys.stdout", new_callable=io.StringIO) - def test_input_correct_command_with_unregistered_flag3(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') - @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}') + app = App(override_system_messages=True, print_func=print) + app.include_router(router) + app.set_incorrect_input_syntax_handler(lambda command: print(f'Incorrect flag syntax: "{command}"')) + orchestrator.start_polling(app) - app = App(override_system_messages=True, - print_func=print) - app.include_router(router) - orchestrator.start_polling(app) + output = capsys.readouterr().out - 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 diff --git a/tests/system_tests/test_system_handling_normal_behavior.py b/tests/system_tests/test_system_handling_normal_behavior.py index 92ade73..77e88c6 100644 --- a/tests/system_tests/test_system_handling_normal_behavior.py +++ b/tests/system_tests/test_system_handling_normal_behavior.py @@ -1,10 +1,8 @@ -import io import re import sys -from unittest import TestCase -from unittest.mock import MagicMock, patch +from collections.abc import Iterator -import _io +import pytest from argenta import App, Orchestrator, Router from argenta.command import Command, PredefinedFlags @@ -14,244 +12,261 @@ from argenta.command.flag.models import PossibleValues, ValidationStatus from argenta.response import Response -class PatchedArgvTestCase(TestCase): - def setUp(self): - super().setUp() - self.patcher = patch.object(sys, 'argv', ['program.py']) - self.mock_argv = self.patcher.start() - self.addCleanup(self.patcher.stop) +@pytest.fixture(autouse=True) +def patch_argv(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(sys, 'argv', ['program.py']) -class TestSystemHandlerNormalWork(PatchedArgvTestCase): - @patch("builtins.input", side_effect=["test", "q"]) - @patch("sys.stdout", new_callable=io.StringIO) - def test_input_correct_command(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): - router = Router() - orchestrator = Orchestrator() - - @router.command(Command('test')) - def test(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) +def _mock_input(inputs: Iterator[str]) -> str: + return next(inputs) - @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() - 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) +# ============================================================================ +# Tests for basic command execution +# ============================================================================ - @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() - orchestrator = Orchestrator() - flag = Flag('help', prefix='--', possible_values=PossibleValues.NEITHER) +def test_simple_command_executes_successfully(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + inputs = iter(["test", "q"]) + monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs)) + + router = Router() + orchestrator = Orchestrator() - @router.command(Command('test', flags=flag)) - def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction] - 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('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) + app = App(override_system_messages=True, print_func=print) + app.include_router(router) + orchestrator.start_polling(app) - output = mock_stdout.getvalue() + output = capsys.readouterr().out - 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() - 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) + assert '\ntest command\n' in 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() - orchestrator = Orchestrator() - flag = PredefinedFlags.SHORT_HELP +def test_case_insensitive_command_executes_when_enabled(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + inputs = iter(["TeSt", "q"]) + monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs)) + + router = Router() + orchestrator = Orchestrator() - @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') + @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) + 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() + 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"]) - @patch("sys.stdout", new_callable=io.StringIO) - def test_input_correct_command_with_default_flag2(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): - router = Router() - orchestrator = Orchestrator() - flag = PredefinedFlags.INFO +def test_two_commands_execute_sequentially(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + inputs = iter(["test", "some", "q"]) + monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs)) + + router = Router() + orchestrator = Orchestrator() - @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') + @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) + @router.command(Command('some')) + def test2(_response: Response) -> None: # pyright: ignore[reportUnusedFunction] + print('some command') - 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"]) - @patch("sys.stdout", new_callable=io.StringIO) - def test_input_correct_command_with_default_flag3(self, mock_stdout: _io.StringIO, magick_mock: MagicMock): - router = Router() - orchestrator = Orchestrator() - flag = PredefinedFlags.HOST +def test_three_commands_execute_sequentially(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + inputs = iter(["test", "some", "more", "q"]) + monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs)) + + router = Router() + orchestrator = Orchestrator() - @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}') + @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) + @router.command(Command('some')) + def test1(_response: Response) -> None: # pyright: ignore[reportUnusedFunction] + print('some command') - 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) - 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) +# ============================================================================ +# Tests for custom flag handling +# ============================================================================ - @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() - orchestrator = Orchestrator() +def test_custom_flag_without_value_is_recognized(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + inputs = iter(["test --help", "q"]) + monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs)) + + router = Router() + orchestrator = Orchestrator() + flag = Flag('help', prefix='--', possible_values=PossibleValues.NEITHER) - @router.command(Command('test')) - def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction] - print(f'test command') + @router.command(Command('test', flags=flag)) + def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction] + 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')) - def test2(response: Response) -> None: # pyright: ignore[reportUnusedFunction] - print(f'some command') + app = App(override_system_messages=True, print_func=print) + app.include_router(router) + orchestrator.start_polling(app) - app = App(override_system_messages=True, - print_func=print) - app.include_router(router) - orchestrator.start_polling(app) + output = capsys.readouterr().out - output = mock_stdout.getvalue() - - self.assertRegex(output, re.compile(r'\ntest command\n(.|\n)*\nsome command\n')) + assert '\nhelp for help flag\n' in output - @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() - orchestrator = Orchestrator() +def test_custom_flag_with_regex_validation_accepts_valid_value(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + inputs = iter(["test --port 22", "q"]) + monkeypatch.setattr('builtins.input', lambda _prompt="": _mock_input(inputs)) + + router = Router() + orchestrator = Orchestrator() + flag = Flag('port', prefix='--', possible_values=re.compile(r'^\d{1,5}$')) - @router.command(Command('test')) - def test(response: Response) -> None: # pyright: ignore[reportUnusedFunction] - print(f'test command') + @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}') - @router.command(Command('some')) - def test1(response: Response) -> None: # pyright: ignore[reportUnusedFunction] - print(f'some command') + app = App(override_system_messages=True, repeat_command_groups_printing=True, print_func=print) + app.include_router(router) + orchestrator.start_polling(app) - @router.command(Command('more')) - def test2(response: Response) -> None: # pyright: ignore[reportUnusedFunction] - print(f'more command') + output = capsys.readouterr().out - app = App(override_system_messages=True, - print_func=print) - app.include_router(router) - orchestrator.start_polling(app) + assert '\nflag value for port flag : 22\n' in output - 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 diff --git a/tests/unit_tests/test_flag.py b/tests/unit_tests/test_flag.py index 2e077d4..deb092e 100644 --- a/tests/unit_tests/test_flag.py +++ b/tests/unit_tests/test_flag.py @@ -1,5 +1,4 @@ import re -from sys import flags from argenta.command.flag import Flag, InputFlag, PossibleValues from argenta.command.flag.flags import Flags, InputFlags diff --git a/tests/unit_tests/test_router.py b/tests/unit_tests/test_router.py index 506baf4..125cb4c 100644 --- a/tests/unit_tests/test_router.py +++ b/tests/unit_tests/test_router.py @@ -1,5 +1,6 @@ import re import pytest +from pytest import CaptureFixture from argenta.command import Command, InputCommand from argenta.command.flag import Flag, InputFlag @@ -89,64 +90,59 @@ def test_get_router_aliases(): def test_get_router_aliases2(): router = Router() @router.command(Command('some', aliases={'test', 'case'})) - def handler(response: Response): + def handler(response: Response): pass @router.command(Command('ext', aliases={'more', 'foo'})) - def handler2(response: Response): + def handler2(response: Response): pass assert router.aliases == {'test', 'case', 'more', 'foo'} def test_get_router_aliases3(): router = Router() @router.command(Command('some')) - def handler(response: Response): + def handler(response: Response): pass assert router.aliases == set() - + def test_find_appropiate_handler(capsys: pytest.CaptureFixture[str]): router = Router() - + @router.command(Command('hello', aliases={'hi'})) def handler(res: Response): print("Hello World!") - + router.finds_appropriate_handler(InputCommand('hi')) - + output = capsys.readouterr() - + assert "Hello World!" in output.out - + def test_find_appropiate_handler2(capsys: CaptureFixture[str]): router = Router() - + @router.command(Command('hello', flags=Flag('flag'), aliases={'hi'})) def handler(res: Response): print("Hello World!") - + router.finds_appropriate_handler(InputCommand('hi')) - + output = capsys.readouterr() - + assert "Hello World!" in output.out - + def test_wrong_typehint(capsys: pytest.CaptureFixture[str]): class NotResponse: pass - + def func(response: NotResponse): pass - + _validate_func_args(func) - + output = capsys.readouterr() - + assert "WARNING" in output.out - + def test_missing_typehint(capsys: pytest.CaptureFixture[str]): def func(response): pass # pyright: ignore[reportMissingParameterType, reportUnknownParameterType] - _validate_func_args(func) # pyright: ignore[reportUnknownArgumentType] - output = capsys.readouterr() - assert output.out == '' - -