From 425342c05985d5454f978694890fe362357c5a30 Mon Sep 17 00:00:00 2001 From: kolo Date: Sun, 7 Dec 2025 01:51:24 +0300 Subject: [PATCH] new tests --- src/argenta/app/models.py | 2 +- src/argenta/command/flag/models.py | 9 +- tests/unit_tests/test_app.py | 290 ++++++++++++++++++++++++++ tests/unit_tests/test_orchestrator.py | 259 +++++++++++++++++++++++ 4 files changed, 554 insertions(+), 6 deletions(-) create mode 100644 tests/unit_tests/test_orchestrator.py diff --git a/src/argenta/app/models.py b/src/argenta/app/models.py index af50cb1..1da6c74 100644 --- a/src/argenta/app/models.py +++ b/src/argenta/app/models.py @@ -340,7 +340,7 @@ class BaseApp: if not self._repeat_command_groups_printing_description: self._print_command_group_description() - def _process_exist_and_valid_command(self, input_command: InputCommand): + def _process_exist_and_valid_command(self, input_command: InputCommand) -> None: processing_router = self._current_matching_triggers_with_routers[input_command.trigger.lower()] if processing_router.disable_redirect_stdout: diff --git a/src/argenta/command/flag/models.py b/src/argenta/command/flag/models.py index 45630c7..2cea82f 100644 --- a/src/argenta/command/flag/models.py +++ b/src/argenta/command/flag/models.py @@ -43,11 +43,10 @@ class Flag: Private. Validates the input flag value :param input_flag_value: The input flag value to validate :return: whether the entered flag is valid as bool - """ - if self.possible_values == PossibleValues.NEITHER: - return input_flag_value == '' - - if self.possible_values == PossibleValues.ALL: + """ + if isinstance(self.possible_values, PossibleValues): + if self.possible_values == PossibleValues.NEITHER: + return input_flag_value == '' return input_flag_value != '' if isinstance(self.possible_values, Pattern): diff --git a/tests/unit_tests/test_app.py b/tests/unit_tests/test_app.py index fd1f7f8..2b99db4 100644 --- a/tests/unit_tests/test_app.py +++ b/tests/unit_tests/test_app.py @@ -6,6 +6,7 @@ from argenta.app.dividing_line import DynamicDividingLine, StaticDividingLine from argenta.app.protocols import DescriptionMessageGenerator, NonStandardBehaviorHandler from argenta.command.models import Command, InputCommand from argenta.response import Response +from argenta.response.status import ResponseStatus from argenta.router import Router @@ -348,3 +349,292 @@ def test_process_command_with_router_with_disabled_stdout_redirect(capsys: Captu stdout = capsys.readouterr() assert 'Hello!' in stdout.out + + +# ============================================================================ +# Tests for handler setters and execution +# ============================================================================ + + +def test_set_unknown_command_handler_stores_handler() -> None: + app = App() + call_tracker = {'called': False} + + def custom_handler(_command: InputCommand) -> None: + call_tracker['called'] = True + + app.set_unknown_command_handler(custom_handler) + app._unknown_command_handler(InputCommand('test')) + + assert call_tracker['called'] + + +def test_set_exit_handler_stores_handler() -> None: + app = App() + call_tracker = {'called': False} + + def custom_handler(_response: Response) -> None: + call_tracker['called'] = True + + app.set_exit_command_handler(custom_handler) + app._exit_command_handler(Response(ResponseStatus.ALL_FLAGS_VALID)) + + assert call_tracker['called'] + + +def test_set_empty_command_handler_stores_handler() -> None: + app = App() + call_tracker = {'called': False} + + def custom_handler() -> None: + call_tracker['called'] = True + + app.set_empty_command_handler(custom_handler) + app._empty_input_command_handler() + + assert call_tracker['called'] + + +def test_set_incorrect_input_syntax_handler_stores_handler() -> None: + app = App() + call_tracker = {'called': False} + + def custom_handler(_command: str) -> None: + call_tracker['called'] = True + + app.set_incorrect_input_syntax_handler(custom_handler) + app._incorrect_input_syntax_handler('test --flag') + + assert call_tracker['called'] + + +def test_set_repeated_input_flags_handler_stores_handler() -> None: + app = App() + call_tracker = {'called': False} + + def custom_handler(_command: str) -> None: + call_tracker['called'] = True + + app.set_repeated_input_flags_handler(custom_handler) + app._repeated_input_flags_handler('test --flag --flag') + + assert call_tracker['called'] + + +# ============================================================================ +# Tests for handler execution with output +# ============================================================================ + + +def test_unknown_command_handler_prints_custom_message(capsys: CaptureFixture[str]) -> None: + app = App(override_system_messages=True) + + def custom_handler(command: InputCommand) -> None: + print(f'Command not found: {command.trigger}') + + app.set_unknown_command_handler(custom_handler) + app._unknown_command_handler(InputCommand('unknown')) + + output = capsys.readouterr() + assert 'Command not found: unknown' in output.out + + +def test_exit_command_handler_prints_custom_message(capsys: CaptureFixture[str]) -> None: + app = App(override_system_messages=True) + + def custom_handler(_response: Response) -> None: + print('Goodbye!') + + app.set_exit_command_handler(custom_handler) + app._exit_command_handler(Response(ResponseStatus.ALL_FLAGS_VALID)) + + output = capsys.readouterr() + assert 'Goodbye!' in output.out + + +def test_empty_command_handler_prints_custom_message(capsys: CaptureFixture[str]) -> None: + app = App(override_system_messages=True) + + def custom_handler() -> None: + print('Please enter a command') + + app.set_empty_command_handler(custom_handler) + app._empty_input_command_handler() + + output = capsys.readouterr() + assert 'Please enter a command' in output.out + + +def test_incorrect_syntax_handler_prints_custom_message(capsys: CaptureFixture[str]) -> None: + app = App(override_system_messages=True) + + def custom_handler(command: str) -> None: + print(f'Syntax error in: {command}') + + app.set_incorrect_input_syntax_handler(custom_handler) + app._incorrect_input_syntax_handler('test --flag') + + output = capsys.readouterr() + assert 'Syntax error in: test --flag' in output.out + + +def test_repeated_flags_handler_prints_custom_message(capsys: CaptureFixture[str]) -> None: + app = App(override_system_messages=True) + + def custom_handler(command: str) -> None: + print(f'Duplicate flags in: {command}') + + app.set_repeated_input_flags_handler(custom_handler) + app._repeated_input_flags_handler('test --flag --flag') + + output = capsys.readouterr() + assert 'Duplicate flags in: test --flag --flag' in output.out + + +# ============================================================================ +# Tests for default handler behavior +# ============================================================================ + + +def test_default_unknown_command_handler_prints_message(capsys: CaptureFixture[str]) -> None: + app = App(override_system_messages=True) + app._unknown_command_handler(InputCommand('unknown')) + + output = capsys.readouterr() + assert 'Unknown command: unknown' in output.out + + +def test_default_empty_command_handler_prints_message(capsys: CaptureFixture[str]) -> None: + app = App(override_system_messages=True) + app._empty_input_command_handler() + + output = capsys.readouterr() + assert 'Empty input command' in output.out + + +def test_default_incorrect_syntax_handler_prints_message(capsys: CaptureFixture[str]) -> None: + app = App(override_system_messages=True) + app._incorrect_input_syntax_handler('test --flag') + + output = capsys.readouterr() + assert 'Incorrect flag syntax: test --flag' in output.out + + +def test_default_repeated_flags_handler_prints_message(capsys: CaptureFixture[str]) -> None: + app = App(override_system_messages=True) + app._repeated_input_flags_handler('test --flag --flag') + + output = capsys.readouterr() + assert 'Repeated input flags: test --flag --flag' in output.out + + +# ============================================================================ +# Tests for handler chaining and multiple calls +# ============================================================================ + + +def test_handler_can_be_replaced_multiple_times() -> None: + app = App() + call_tracker = {'count': 0} + + def handler1(_command: InputCommand) -> None: + call_tracker['count'] += 1 + + def handler2(_command: InputCommand) -> None: + call_tracker['count'] += 10 + + app.set_unknown_command_handler(handler1) + app._unknown_command_handler(InputCommand('test')) + assert call_tracker['count'] == 1 + + app.set_unknown_command_handler(handler2) + app._unknown_command_handler(InputCommand('test')) + assert call_tracker['count'] == 11 + + +def test_handler_receives_correct_parameters() -> None: + app = App() + received_data = {'trigger': None} + + def custom_handler(command: InputCommand) -> None: + received_data['trigger'] = command.trigger + + app.set_unknown_command_handler(custom_handler) + app._unknown_command_handler(InputCommand('mycommand')) + + assert received_data['trigger'] == 'mycommand' + + +def test_exit_handler_receives_response_object() -> None: + app = App() + received_data = {'response': None} + + def custom_handler(response: Response) -> None: + received_data['response'] = response + + app.set_exit_command_handler(custom_handler) + test_response = Response(ResponseStatus.ALL_FLAGS_VALID) + app._exit_command_handler(test_response) + + assert received_data['response'] is test_response + + +# ============================================================================ +# Tests for handler integration with routers +# ============================================================================ + + +def test_app_with_router_and_custom_unknown_handler(capsys: CaptureFixture[str]) -> None: + app = App(override_system_messages=True) + router = Router() + + @router.command(Command('test')) + def handler(_res: Response) -> None: + print('test executed') + + app.include_router(router) + + def custom_unknown_handler(command: InputCommand) -> None: + print(f'Not found: {command.trigger}') + + app.set_unknown_command_handler(custom_unknown_handler) + + # Test that unknown command uses custom handler + assert app._is_unknown_command(InputCommand('unknown')) + app._unknown_command_handler(InputCommand('unknown')) + + output = capsys.readouterr() + assert 'Not found: unknown' in output.out + + +def test_app_handlers_work_with_multiple_routers() -> None: + app = App(override_system_messages=True) + router1 = Router() + router2 = Router() + + @router1.command(Command('cmd1')) + def handler1(_res: Response) -> None: + pass + + @router2.command(Command('cmd2')) + def handler2(_res: Response) -> None: + pass + + app.include_routers(router1, router2) + app._pre_cycle_setup() + + call_tracker = {'called': False} + + def custom_handler(_command: InputCommand) -> None: + call_tracker['called'] = True + + app.set_unknown_command_handler(custom_handler) + + # Both commands should be known + assert not app._is_unknown_command(InputCommand('cmd1')) + assert not app._is_unknown_command(InputCommand('cmd2')) + + # Unknown command should trigger handler + assert app._is_unknown_command(InputCommand('unknown')) + app._unknown_command_handler(InputCommand('unknown')) + assert call_tracker['called'] diff --git a/tests/unit_tests/test_orchestrator.py b/tests/unit_tests/test_orchestrator.py new file mode 100644 index 0000000..9393006 --- /dev/null +++ b/tests/unit_tests/test_orchestrator.py @@ -0,0 +1,259 @@ +import pytest +from dishka import Provider +from pytest_mock import MockerFixture + +from argenta import App, Router +from argenta.command import Command +from argenta.orchestrator import Orchestrator +from argenta.orchestrator.argparser import ArgParser +from argenta.response import Response + + +# ============================================================================ +# Fixtures +# ============================================================================ + + +@pytest.fixture +def mock_argparser(mocker: MockerFixture) -> ArgParser: + """Create a mock ArgParser that doesn't actually parse sys.argv""" + argparser = ArgParser(processed_args=[]) + mocker.patch.object(argparser, '_parse_args') + return argparser + + +@pytest.fixture +def sample_app() -> App: + """Create a sample App for testing""" + return App(override_system_messages=True) + + +@pytest.fixture +def sample_router() -> Router: + """Create a sample Router with a test command""" + router = Router() + + @router.command(Command('test')) + def handler(_res: Response) -> None: + print('test command executed') + + return router + + +# ============================================================================ +# Tests for Orchestrator initialization +# ============================================================================ + + +def test_orchestrator_initializes_with_default_argparser(mocker: MockerFixture) -> None: + """Test Orchestrator initialization with default ArgParser""" + mocker.patch('sys.argv', ['test_program']) + orchestrator = Orchestrator() + assert orchestrator._arg_parser is not None + assert isinstance(orchestrator._arg_parser, ArgParser) + + +def test_orchestrator_initializes_with_custom_argparser(mock_argparser: ArgParser) -> None: + """Test Orchestrator initialization with custom ArgParser""" + orchestrator = Orchestrator(arg_parser=mock_argparser) + assert orchestrator._arg_parser is mock_argparser + + +def test_orchestrator_initializes_with_custom_providers(mocker: MockerFixture) -> None: + """Test Orchestrator initialization with custom providers""" + mocker.patch('sys.argv', ['test_program']) + custom_provider = Provider() + orchestrator = Orchestrator(custom_providers=[custom_provider]) + assert custom_provider in orchestrator._custom_providers + + +def test_orchestrator_initializes_with_auto_inject_disabled(mocker: MockerFixture) -> None: + """Test Orchestrator initialization with auto_inject_handlers disabled""" + mocker.patch('sys.argv', ['test_program']) + orchestrator = Orchestrator(auto_inject_handlers=False) + assert orchestrator._auto_inject_handlers is False + + +def test_orchestrator_initializes_with_auto_inject_enabled(mocker: MockerFixture) -> None: + """Test Orchestrator initialization with auto_inject_handlers enabled (default)""" + mocker.patch('sys.argv', ['test_program']) + orchestrator = Orchestrator() + assert orchestrator._auto_inject_handlers is True + + +def test_orchestrator_parses_args_on_initialization(mocker: MockerFixture, mock_argparser: ArgParser) -> None: + """Test that Orchestrator calls _parse_args on initialization""" + parse_spy = mocker.spy(mock_argparser, '_parse_args') + _orchestrator = Orchestrator(arg_parser=mock_argparser) + parse_spy.assert_called_once() + + +# ============================================================================ +# Tests for start_polling method +# ============================================================================ + + +def test_start_polling_creates_dishka_container( + mocker: MockerFixture, mock_argparser: ArgParser, sample_app: App +) -> None: + """Test that start_polling creates a dishka container""" + mock_make_container = mocker.patch('argenta.orchestrator.entity.make_container') + _mock_setup_dishka = mocker.patch('argenta.orchestrator.entity.setup_dishka') + mocker.patch.object(sample_app, 'run_polling') + + orchestrator = Orchestrator(arg_parser=mock_argparser) + orchestrator.start_polling(sample_app) + + mock_make_container.assert_called_once() + assert mock_make_container.call_args[1]['context'] == {ArgParser: mock_argparser} + + +def test_start_polling_calls_setup_dishka_with_auto_inject_enabled( + mocker: MockerFixture, mock_argparser: ArgParser, sample_app: App +) -> None: + """Test that start_polling calls setup_dishka with auto_inject=True""" + mock_container = mocker.MagicMock() # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] + mocker.patch('argenta.orchestrator.entity.make_container', return_value=mock_container) + mock_setup_dishka = mocker.patch('argenta.orchestrator.entity.setup_dishka') + mocker.patch.object(sample_app, 'run_polling') + + orchestrator = Orchestrator(arg_parser=mock_argparser, auto_inject_handlers=True) + orchestrator.start_polling(sample_app) + + mock_setup_dishka.assert_called_once_with(sample_app, mock_container, auto_inject=True) + + +def test_start_polling_calls_setup_dishka_with_auto_inject_disabled( + mocker: MockerFixture, mock_argparser: ArgParser, sample_app: App +) -> None: + """Test that start_polling calls setup_dishka with auto_inject=False""" + mock_container = mocker.MagicMock() # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] + mocker.patch('argenta.orchestrator.entity.make_container', return_value=mock_container) + mock_setup_dishka = mocker.patch('argenta.orchestrator.entity.setup_dishka') + mocker.patch.object(sample_app, 'run_polling') + + orchestrator = Orchestrator(arg_parser=mock_argparser, auto_inject_handlers=False) + orchestrator.start_polling(sample_app) + + mock_setup_dishka.assert_called_once_with(sample_app, mock_container, auto_inject=False) + + +def test_start_polling_calls_app_run_polling( + mocker: MockerFixture, mock_argparser: ArgParser, sample_app: App +) -> None: + """Test that start_polling calls app.run_polling()""" + mocker.patch('argenta.orchestrator.entity.make_container') + mocker.patch('argenta.orchestrator.entity.setup_dishka') + mock_run_polling = mocker.patch.object(sample_app, 'run_polling') + + orchestrator = Orchestrator(arg_parser=mock_argparser) + orchestrator.start_polling(sample_app) + + mock_run_polling.assert_called_once() + + +def test_start_polling_includes_custom_providers_in_container( + mocker: MockerFixture, mock_argparser: ArgParser, sample_app: App +) -> None: + """Test that start_polling includes custom providers in container""" + custom_provider = Provider() + mock_make_container = mocker.patch('argenta.orchestrator.entity.make_container') + mocker.patch('argenta.orchestrator.entity.setup_dishka') + mocker.patch.object(sample_app, 'run_polling') + + orchestrator = Orchestrator(arg_parser=mock_argparser, custom_providers=[custom_provider]) + orchestrator.start_polling(sample_app) + + # Check that custom_provider was passed to make_container + call_args = mock_make_container.call_args[0] + assert custom_provider in call_args + + +# ============================================================================ +# Tests for integration with App +# ============================================================================ + + +def test_orchestrator_integrates_with_app_with_router( + mocker: MockerFixture, mock_argparser: ArgParser, sample_app: App, sample_router: Router +) -> None: + """Test that Orchestrator properly integrates with App that has routers""" + mocker.patch('argenta.orchestrator.entity.make_container') + mocker.patch('argenta.orchestrator.entity.setup_dishka') + mock_run_polling = mocker.patch.object(sample_app, 'run_polling') + + sample_app.include_router(sample_router) + + orchestrator = Orchestrator(arg_parser=mock_argparser) + orchestrator.start_polling(sample_app) + + mock_run_polling.assert_called_once() + assert len(sample_app.registered_routers.registered_routers) == 1 + + +# ============================================================================ +# Tests for ArgParser integration +# ============================================================================ + + +def test_orchestrator_passes_argparser_to_container_context( + mocker: MockerFixture, mock_argparser: ArgParser, sample_app: App +) -> None: + """Test that Orchestrator passes ArgParser instance to container context""" + mock_make_container = mocker.patch('argenta.orchestrator.entity.make_container') + mocker.patch('argenta.orchestrator.entity.setup_dishka') + mocker.patch.object(sample_app, 'run_polling') + + orchestrator = Orchestrator(arg_parser=mock_argparser) + orchestrator.start_polling(sample_app) + + # Verify that ArgParser was passed in context + call_kwargs = mock_make_container.call_args[1] + assert 'context' in call_kwargs + assert ArgParser in call_kwargs['context'] + assert call_kwargs['context'][ArgParser] is mock_argparser + + +# ============================================================================ +# Tests for error handling +# ============================================================================ + + +def test_orchestrator_handles_app_run_polling_exception( + mocker: MockerFixture, mock_argparser: ArgParser, sample_app: App +) -> None: + """Test that Orchestrator propagates exceptions from app.run_polling()""" + mocker.patch('argenta.orchestrator.entity.make_container') + mocker.patch('argenta.orchestrator.entity.setup_dishka') + mocker.patch.object(sample_app, 'run_polling', side_effect=RuntimeError("Test error")) + + orchestrator = Orchestrator(arg_parser=mock_argparser) + + with pytest.raises(RuntimeError, match="Test error"): + orchestrator.start_polling(sample_app) + + +# ============================================================================ +# Tests for multiple providers +# ============================================================================ + + +def test_orchestrator_accepts_multiple_custom_providers( + mocker: MockerFixture, mock_argparser: ArgParser, sample_app: App +) -> None: + """Test that Orchestrator accepts multiple custom providers""" + provider1 = Provider() + provider2 = Provider() + mock_make_container = mocker.patch('argenta.orchestrator.entity.make_container') + mocker.patch('argenta.orchestrator.entity.setup_dishka') + mocker.patch.object(sample_app, 'run_polling') + + orchestrator = Orchestrator( + arg_parser=mock_argparser, + custom_providers=[provider1, provider2] + ) + orchestrator.start_polling(sample_app) + + call_args = mock_make_container.call_args[0] + assert provider1 in call_args + assert provider2 in call_args