mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 10:05:28 +03:00
refactor tests and add new
This commit is contained in:
+221
-223
@@ -1,249 +1,247 @@
|
||||
from argparse import Namespace
|
||||
import sys
|
||||
from argparse import Namespace
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import call
|
||||
|
||||
import pytest
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from argenta.orchestrator.argparser.arguments.models import (BaseArgument,
|
||||
BooleanArgument,
|
||||
InputArgument,
|
||||
ValueArgument)
|
||||
from argenta.orchestrator.argparser.arguments.models import (
|
||||
BaseArgument,
|
||||
BooleanArgument,
|
||||
InputArgument,
|
||||
ValueArgument,
|
||||
)
|
||||
from argenta.orchestrator.argparser.entity import ArgParser, ArgSpace
|
||||
|
||||
|
||||
class TestArgumentCreation:
|
||||
"""Tests for the creation and attribute validation of argument model classes."""
|
||||
|
||||
def test_value_argument_creation(self):
|
||||
"""Ensures ValueArgument instances are created with correct attributes."""
|
||||
arg = ValueArgument(
|
||||
name="test_arg",
|
||||
prefix="--",
|
||||
help="A test argument.",
|
||||
possible_values=["one", "two"],
|
||||
default="one",
|
||||
is_required=True,
|
||||
is_deprecated=False,
|
||||
)
|
||||
assert arg.name == "test_arg"
|
||||
assert arg.prefix == "--"
|
||||
assert arg.help == "A test argument."
|
||||
assert arg.possible_values == ["one", "two"]
|
||||
assert arg.default == "one"
|
||||
assert arg.is_required is True
|
||||
assert arg.is_deprecated is False
|
||||
assert arg.action == "store"
|
||||
assert arg.string_entity == "--test_arg"
|
||||
|
||||
def test_boolean_argument_creation(self):
|
||||
"""Ensures BooleanArgument instances are created with correct attributes."""
|
||||
arg = BooleanArgument(
|
||||
name="verbose", prefix="-", help="Enable verbose mode.", is_deprecated=True
|
||||
)
|
||||
assert arg.name == "verbose"
|
||||
assert arg.prefix == "-"
|
||||
assert arg.help == "Enable verbose mode."
|
||||
assert arg.is_deprecated is True
|
||||
assert arg.action == "store_true"
|
||||
assert arg.string_entity == "-verbose"
|
||||
|
||||
def test_input_argument_creation(self):
|
||||
"""Ensures InputArgument instances are created with correct attributes."""
|
||||
arg = InputArgument(
|
||||
name="file", value="/path/to/file", founder_class=ValueArgument
|
||||
)
|
||||
assert arg.name == "file"
|
||||
assert arg.value == "/path/to/file"
|
||||
assert arg.founder_class is ValueArgument
|
||||
if TYPE_CHECKING:
|
||||
from pytest_mock.plugin import MockType
|
||||
|
||||
|
||||
class TestArgSpace:
|
||||
"""Tests for the ArgSpace class, which holds parsed argument values."""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_arguments(self) -> list[InputArgument]:
|
||||
"""Provides a list of mock InputArgument objects for testing."""
|
||||
return [
|
||||
InputArgument(name="arg1", value="val1", founder_class=ValueArgument),
|
||||
InputArgument(name="arg2", value=True, founder_class=BooleanArgument),
|
||||
InputArgument(name="arg3", value="val3", founder_class=ValueArgument),
|
||||
]
|
||||
|
||||
@pytest.fixture
|
||||
def arg_space(self, mock_arguments: list[InputArgument]) -> ArgSpace:
|
||||
"""Provides a pre-populated ArgSpace instance."""
|
||||
return ArgSpace(all_arguments=mock_arguments)
|
||||
|
||||
def test_initialization(self, arg_space: ArgSpace, mock_arguments: list[InputArgument]):
|
||||
"""Tests if ArgSpace is initialized correctly with a list of arguments."""
|
||||
assert len(arg_space.all_arguments) == 3
|
||||
assert arg_space.all_arguments == mock_arguments
|
||||
|
||||
def test_get_by_name(self, arg_space: ArgSpace, mock_arguments: list[InputArgument]):
|
||||
"""Tests retrieving an argument by its name."""
|
||||
found_arg = arg_space.get_by_name("arg1")
|
||||
assert found_arg is not None
|
||||
assert found_arg == mock_arguments[0]
|
||||
|
||||
def test_get_by_name_not_found(self, arg_space: ArgSpace):
|
||||
"""Tests that get_by_name returns None for a non-existent argument."""
|
||||
found_arg = arg_space.get_by_name("non_existent_arg")
|
||||
assert found_arg is None
|
||||
|
||||
def test_get_by_type(self, arg_space: ArgSpace, mock_arguments: list[InputArgument]):
|
||||
"""Tests retrieving arguments based on their founder class type."""
|
||||
value_args = arg_space.get_by_type(ValueArgument)
|
||||
assert len(value_args) == 2
|
||||
assert mock_arguments[0] in value_args
|
||||
assert mock_arguments[2] in value_args
|
||||
|
||||
bool_args = arg_space.get_by_type(BooleanArgument)
|
||||
assert len(bool_args) == 1
|
||||
assert mock_arguments[1] in bool_args
|
||||
|
||||
def test_get_by_type_not_found(self, arg_space: ArgSpace):
|
||||
"""Tests that get_by_type returns an empty list for an unused argument type."""
|
||||
class OtherArgument(BaseArgument):
|
||||
pass
|
||||
other_args = arg_space.get_by_type(OtherArgument)
|
||||
assert other_args == []
|
||||
|
||||
def test_from_namespace(self):
|
||||
"""Tests the class method for creating an ArgSpace from an argparse.Namespace."""
|
||||
namespace = Namespace(config="config.json", debug=True, verbose=False)
|
||||
processed_args = [
|
||||
ValueArgument(name="config", prefix="--"),
|
||||
BooleanArgument(name="debug", prefix="-"),
|
||||
BooleanArgument(name="verbose", prefix="-"),
|
||||
]
|
||||
|
||||
arg_space = ArgSpace.from_namespace(namespace, processed_args)
|
||||
assert len(arg_space.all_arguments) == 3
|
||||
|
||||
config_arg = arg_space.get_by_name('config')
|
||||
debug_arg = arg_space.get_by_name('debug')
|
||||
|
||||
assert config_arg is not None
|
||||
assert config_arg.value == "config.json"
|
||||
assert config_arg.founder_class is ValueArgument
|
||||
|
||||
assert debug_arg is not None
|
||||
assert debug_arg.value is True
|
||||
assert debug_arg.founder_class is BooleanArgument
|
||||
def test_value_argument_creation() -> None:
|
||||
arg: ValueArgument = ValueArgument(
|
||||
name="test_arg",
|
||||
prefix="--",
|
||||
help="A test argument.",
|
||||
possible_values=["one", "two"],
|
||||
default="one",
|
||||
is_required=True,
|
||||
is_deprecated=False,
|
||||
)
|
||||
assert arg.name == "test_arg"
|
||||
assert arg.prefix == "--"
|
||||
assert arg.help == "A test argument."
|
||||
assert arg.possible_values == ["one", "two"]
|
||||
assert arg.default == "one"
|
||||
assert arg.is_required is True
|
||||
assert arg.is_deprecated is False
|
||||
assert arg.action == "store"
|
||||
assert arg.string_entity == "--test_arg"
|
||||
|
||||
|
||||
class TestArgParser:
|
||||
"""Tests for the ArgParser class, which orchestrates argument parsing."""
|
||||
def test_boolean_argument_creation() -> None:
|
||||
arg: BooleanArgument = BooleanArgument(
|
||||
name="verbose", prefix="-", help="Enable verbose mode.", is_deprecated=True
|
||||
)
|
||||
assert arg.name == "verbose"
|
||||
assert arg.prefix == "-"
|
||||
assert arg.help == "Enable verbose mode."
|
||||
assert arg.is_deprecated is True
|
||||
assert arg.action == "store_true"
|
||||
assert arg.string_entity == "-verbose"
|
||||
|
||||
@pytest.fixture
|
||||
def value_arg(self) -> ValueArgument:
|
||||
"""Provides a sample ValueArgument."""
|
||||
return ValueArgument(name="config", help="Path to config file", default="dev.json", is_required=False, possible_values=["dev.json", "prod.json"])
|
||||
|
||||
@pytest.fixture
|
||||
def bool_arg(self) -> BooleanArgument:
|
||||
"""Provides a sample BooleanArgument."""
|
||||
return BooleanArgument(name="debug", help="Enable debug mode")
|
||||
def test_input_argument_creation() -> None:
|
||||
arg: InputArgument = InputArgument(
|
||||
name="file", value="/path/to/file", founder_class=ValueArgument
|
||||
)
|
||||
assert arg.name == "file"
|
||||
assert arg.value == "/path/to/file"
|
||||
assert arg.founder_class is ValueArgument
|
||||
|
||||
@pytest.fixture
|
||||
def processed_args(self, value_arg: ValueArgument, bool_arg: BooleanArgument) -> list:
|
||||
"""Provides a list of processed arguments."""
|
||||
return [value_arg, bool_arg]
|
||||
|
||||
def test_initialization(self, processed_args: list):
|
||||
"""Tests that the ArgParser constructor correctly assigns attributes."""
|
||||
parser = ArgParser(
|
||||
processed_args=processed_args,
|
||||
name="TestApp",
|
||||
description="A test application.",
|
||||
epilog="Test epilog.",
|
||||
)
|
||||
assert parser.name == "TestApp"
|
||||
assert parser.description == "A test application."
|
||||
assert parser.epilog == "Test epilog."
|
||||
assert parser.processed_args == processed_args
|
||||
assert isinstance(parser.parsed_argspace, ArgSpace)
|
||||
assert parser.parsed_argspace.all_arguments == []
|
||||
@pytest.fixture
|
||||
def mock_arguments() -> list[InputArgument]:
|
||||
return [
|
||||
InputArgument(name="arg1", value="val1", founder_class=ValueArgument),
|
||||
InputArgument(name="arg2", value=True, founder_class=BooleanArgument),
|
||||
InputArgument(name="arg3", value="val3", founder_class=ValueArgument),
|
||||
]
|
||||
|
||||
@pytest.mark.skipif(sys.version_info < (3, 13), reason="requires python3.13 or higher")
|
||||
def test_register_args(self, mocker, value_arg: ValueArgument, bool_arg: BooleanArgument):
|
||||
"""Tests that arguments are correctly registered with the underlying ArgumentParser."""
|
||||
mock_add_argument = mocker.patch("argparse.ArgumentParser.add_argument")
|
||||
|
||||
parser = ArgParser(processed_args=[value_arg, bool_arg])
|
||||
@pytest.fixture
|
||||
def arg_space(mock_arguments: list[InputArgument]) -> ArgSpace:
|
||||
return ArgSpace(all_arguments=mock_arguments)
|
||||
|
||||
expected_calls = [
|
||||
# Call for the ValueArgument
|
||||
call(
|
||||
value_arg.string_entity,
|
||||
action=value_arg.action,
|
||||
help=value_arg.help,
|
||||
default=value_arg.default,
|
||||
choices=value_arg.possible_values,
|
||||
required=value_arg.is_required,
|
||||
deprecated=value_arg.is_deprecated
|
||||
),
|
||||
# Call for the BooleanArgument
|
||||
call(
|
||||
bool_arg.string_entity,
|
||||
action=bool_arg.action,
|
||||
help=bool_arg.help,
|
||||
deprecated=bool_arg.is_deprecated
|
||||
)
|
||||
]
|
||||
mock_add_argument.assert_has_calls(expected_calls, any_order=True)
|
||||
|
||||
@pytest.mark.skipif(sys.version_info > (3, 12), reason='for more latest python version has been other test')
|
||||
def test_register_args(self, mocker, value_arg: ValueArgument, bool_arg: BooleanArgument):
|
||||
"""Tests that arguments are correctly registered with the underlying ArgumentParser."""
|
||||
mock_add_argument = mocker.patch("argparse.ArgumentParser.add_argument")
|
||||
|
||||
parser = ArgParser(processed_args=[value_arg, bool_arg])
|
||||
def test_argspace_initialization(arg_space: ArgSpace, mock_arguments: list[InputArgument]) -> None:
|
||||
assert len(arg_space.all_arguments) == 3
|
||||
assert arg_space.all_arguments == mock_arguments
|
||||
|
||||
expected_calls = [
|
||||
# Call for the ValueArgument
|
||||
call(
|
||||
value_arg.string_entity,
|
||||
action=value_arg.action,
|
||||
help=value_arg.help,
|
||||
default=value_arg.default,
|
||||
choices=value_arg.possible_values,
|
||||
required=value_arg.is_required
|
||||
),
|
||||
# Call for the BooleanArgument
|
||||
call(
|
||||
bool_arg.string_entity,
|
||||
action=bool_arg.action,
|
||||
help=bool_arg.help
|
||||
)
|
||||
]
|
||||
mock_add_argument.assert_has_calls(expected_calls, any_order=True)
|
||||
|
||||
def test_parse_args_populates_argspace(self, mocker, processed_args: list[ValueArgument | BooleanArgument]):
|
||||
"""Tests that _parse_args correctly calls the parser and populates the ArgSpace."""
|
||||
# 1. Mock the return value of the internal argparse instance
|
||||
mock_namespace = Namespace(config='config.json', debug=True)
|
||||
mocker.patch(
|
||||
'argparse.ArgumentParser.parse_args',
|
||||
return_value=mock_namespace
|
||||
)
|
||||
def test_argspace_get_by_name(arg_space: ArgSpace, mock_arguments: list[InputArgument]) -> None:
|
||||
found_arg: InputArgument | None = arg_space.get_by_name("arg1")
|
||||
assert found_arg is not None
|
||||
assert found_arg == mock_arguments[0]
|
||||
|
||||
# 2. Initialize the parser and call the method under test
|
||||
parser = ArgParser(processed_args=processed_args)
|
||||
parser._parse_args() # Test the private method that contains the logic
|
||||
|
||||
# 3. Assert the results
|
||||
arg_space = parser.parsed_argspace
|
||||
assert isinstance(arg_space, ArgSpace)
|
||||
assert len(arg_space.all_arguments) == 2
|
||||
def test_argspace_get_by_name_not_found(arg_space: ArgSpace) -> None:
|
||||
found_arg: InputArgument | None = arg_space.get_by_name("non_existent_arg")
|
||||
assert found_arg is None
|
||||
|
||||
config_arg = arg_space.get_by_name('config')
|
||||
debug_arg = arg_space.get_by_name('debug')
|
||||
|
||||
assert config_arg is not None
|
||||
assert config_arg.value == 'config.json'
|
||||
assert config_arg.founder_class is ValueArgument
|
||||
def test_argspace_get_by_type(arg_space: ArgSpace, mock_arguments: list[InputArgument]) -> None:
|
||||
value_args: list[InputArgument] = arg_space.get_by_type(ValueArgument)
|
||||
assert len(value_args) == 2
|
||||
assert mock_arguments[0] in value_args
|
||||
assert mock_arguments[2] in value_args
|
||||
|
||||
assert debug_arg is not None
|
||||
assert debug_arg.value is True
|
||||
assert debug_arg.founder_class is BooleanArgument
|
||||
bool_args: list[InputArgument] = arg_space.get_by_type(BooleanArgument)
|
||||
assert len(bool_args) == 1
|
||||
assert mock_arguments[1] in bool_args
|
||||
|
||||
|
||||
def test_argspace_get_by_type_not_found(arg_space: ArgSpace) -> None:
|
||||
class OtherArgument(BaseArgument):
|
||||
pass
|
||||
|
||||
other_args: list[InputArgument] = arg_space.get_by_type(OtherArgument) # pyright: ignore[reportAssignmentType]
|
||||
assert other_args == []
|
||||
|
||||
|
||||
def test_argspace_from_namespace() -> None:
|
||||
namespace: Namespace = Namespace(config="config.json", debug=True, verbose=False)
|
||||
processed_args: list[ValueArgument | BooleanArgument] = [
|
||||
ValueArgument(name="config", prefix="--"),
|
||||
BooleanArgument(name="debug", prefix="-"),
|
||||
BooleanArgument(name="verbose", prefix="-"),
|
||||
]
|
||||
|
||||
arg_space: ArgSpace = ArgSpace.from_namespace(namespace, processed_args)
|
||||
assert len(arg_space.all_arguments) == 3
|
||||
|
||||
config_arg: InputArgument | None = arg_space.get_by_name('config')
|
||||
debug_arg: InputArgument | None = arg_space.get_by_name('debug')
|
||||
|
||||
assert config_arg is not None
|
||||
assert config_arg.value == "config.json"
|
||||
assert config_arg.founder_class is ValueArgument
|
||||
|
||||
assert debug_arg is not None
|
||||
assert debug_arg.value is True
|
||||
assert debug_arg.founder_class is BooleanArgument
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def value_arg() -> ValueArgument:
|
||||
return ValueArgument(
|
||||
name="config",
|
||||
help="Path to config file",
|
||||
default="dev.json",
|
||||
is_required=False,
|
||||
possible_values=["dev.json", "prod.json"],
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def bool_arg() -> BooleanArgument:
|
||||
return BooleanArgument(name="debug", help="Enable debug mode")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def processed_args(value_arg: ValueArgument, bool_arg: BooleanArgument) -> list[ValueArgument | BooleanArgument]:
|
||||
return [value_arg, bool_arg]
|
||||
|
||||
|
||||
def test_argparser_initialization(processed_args: list[ValueArgument | BooleanArgument]) -> None:
|
||||
parser: ArgParser = ArgParser(
|
||||
processed_args=processed_args,
|
||||
name="TestApp",
|
||||
description="A test application.",
|
||||
epilog="Test epilog.",
|
||||
)
|
||||
assert parser.name == "TestApp"
|
||||
assert parser.description == "A test application."
|
||||
assert parser.epilog == "Test epilog."
|
||||
assert parser.processed_args == processed_args
|
||||
assert isinstance(parser.parsed_argspace, ArgSpace)
|
||||
assert parser.parsed_argspace.all_arguments == []
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.version_info < (3, 13), reason="requires python3.13 or higher")
|
||||
def test_argparser_register_args_py313(
|
||||
mocker: MockerFixture, value_arg: ValueArgument, bool_arg: BooleanArgument
|
||||
) -> None:
|
||||
mock_add_argument: MockType = mocker.patch("argparse.ArgumentParser.add_argument")
|
||||
|
||||
parser: ArgParser = ArgParser(processed_args=[value_arg, bool_arg]) # pyright: ignore[reportUnusedVariable]
|
||||
|
||||
expected_calls: list[call] = [
|
||||
call(
|
||||
value_arg.string_entity,
|
||||
action=value_arg.action,
|
||||
help=value_arg.help,
|
||||
default=value_arg.default,
|
||||
choices=value_arg.possible_values,
|
||||
required=value_arg.is_required,
|
||||
deprecated=value_arg.is_deprecated,
|
||||
),
|
||||
call(
|
||||
bool_arg.string_entity,
|
||||
action=bool_arg.action,
|
||||
help=bool_arg.help,
|
||||
deprecated=bool_arg.is_deprecated,
|
||||
),
|
||||
]
|
||||
mock_add_argument.assert_has_calls(expected_calls, any_order=True)
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.version_info > (3, 12), reason="for more latest python version has been other test")
|
||||
def test_argparser_register_args_py312(
|
||||
mocker: MockerFixture, value_arg: ValueArgument, bool_arg: BooleanArgument
|
||||
) -> None:
|
||||
mock_add_argument: MockType = mocker.patch("argparse.ArgumentParser.add_argument")
|
||||
|
||||
parser: ArgParser = ArgParser(processed_args=[value_arg, bool_arg])
|
||||
|
||||
expected_calls: list[call] = [
|
||||
call(
|
||||
value_arg.string_entity,
|
||||
action=value_arg.action,
|
||||
help=value_arg.help,
|
||||
default=value_arg.default,
|
||||
choices=value_arg.possible_values,
|
||||
required=value_arg.is_required,
|
||||
),
|
||||
call(
|
||||
bool_arg.string_entity,
|
||||
action=bool_arg.action,
|
||||
help=bool_arg.help,
|
||||
),
|
||||
]
|
||||
mock_add_argument.assert_has_calls(expected_calls, any_order=True)
|
||||
|
||||
|
||||
def test_argparser_parse_args_populates_argspace(
|
||||
mocker: MockerFixture, processed_args: list[ValueArgument | BooleanArgument]
|
||||
) -> None:
|
||||
mock_namespace: Namespace = Namespace(config='config.json', debug=True)
|
||||
mocker.patch('argparse.ArgumentParser.parse_args', return_value=mock_namespace)
|
||||
|
||||
parser: ArgParser = ArgParser(processed_args=processed_args)
|
||||
parser._parse_args()
|
||||
|
||||
arg_space: ArgSpace = parser.parsed_argspace
|
||||
assert isinstance(arg_space, ArgSpace)
|
||||
assert len(arg_space.all_arguments) == 2
|
||||
|
||||
config_arg: InputArgument | None = arg_space.get_by_name('config')
|
||||
debug_arg: InputArgument | None = arg_space.get_by_name('debug')
|
||||
|
||||
assert config_arg is not None
|
||||
assert config_arg.value == 'config.json'
|
||||
assert config_arg.founder_class is ValueArgument
|
||||
|
||||
assert debug_arg is not None
|
||||
assert debug_arg.value is True
|
||||
assert debug_arg.founder_class is BooleanArgument
|
||||
|
||||
@@ -1,198 +1,173 @@
|
||||
import os
|
||||
from unittest.mock import MagicMock, call, patch
|
||||
|
||||
import pytest
|
||||
from pyfakefs.fake_filesystem import FakeFilesystem
|
||||
from pytest_mock import MockerFixture
|
||||
from pytest_mock.plugin import MockType
|
||||
|
||||
from argenta.app.autocompleter.entity import (
|
||||
AutoCompleter,
|
||||
_get_history_items,
|
||||
_is_command_exist,
|
||||
)
|
||||
|
||||
# Since readline is not available on all platforms (e.g., Windows) for testing,
|
||||
# it is mocked for all tests.
|
||||
readline_mock = MagicMock()
|
||||
|
||||
# We patch the module where it's imported, not where it's defined.
|
||||
@pytest.fixture
|
||||
def mock_readline():
|
||||
"""Fixture to provide a mock of the `readline` module."""
|
||||
with patch('argenta.app.autocompleter.entity.readline', readline_mock) as mock:
|
||||
# This nested state simulates readline's internal history list.
|
||||
_history = []
|
||||
def mock_readline(mocker: MockerFixture) -> MockType:
|
||||
_history: list[str] = []
|
||||
|
||||
def add_history(item: str) -> None:
|
||||
_history.append(item)
|
||||
def add_history(item: str) -> None:
|
||||
_history.append(item)
|
||||
|
||||
def get_history_item(index: int) -> str | None:
|
||||
# readline history is 1-based.
|
||||
if 1 <= index <= len(_history):
|
||||
return _history[index - 1]
|
||||
return None
|
||||
def get_history_item(index: int) -> str | None:
|
||||
if 1 <= index <= len(_history):
|
||||
return _history[index - 1]
|
||||
return None
|
||||
|
||||
def get_current_history_length() -> int:
|
||||
return len(_history)
|
||||
def get_current_history_length() -> int:
|
||||
return len(_history)
|
||||
|
||||
def clear_history() -> None:
|
||||
_history.clear()
|
||||
def clear_history() -> None:
|
||||
_history.clear()
|
||||
|
||||
# Reset all mocks and the internal history before each test.
|
||||
mock.reset_mock()
|
||||
clear_history()
|
||||
mock: MockType = mocker.MagicMock()
|
||||
mocker.patch('argenta.app.autocompleter.entity.readline', mock)
|
||||
|
||||
# Apply side effects to mock functions to simulate real behavior.
|
||||
mock.add_history.side_effect = add_history
|
||||
mock.get_history_item.side_effect = get_history_item
|
||||
mock.get_current_history_length.side_effect = get_current_history_length
|
||||
mock.reset_mock()
|
||||
clear_history()
|
||||
|
||||
# Provide a default return value for functions that are read from.
|
||||
mock.get_completer_delims.return_value = " "
|
||||
mock.add_history.side_effect = add_history
|
||||
mock.get_history_item.side_effect = get_history_item
|
||||
mock.get_current_history_length.side_effect = get_current_history_length
|
||||
mock.get_completer_delims.return_value = " "
|
||||
|
||||
yield mock
|
||||
|
||||
# We import the class under test after setting up the patch context if needed,
|
||||
# or ensure patches target the correct import location.
|
||||
from argenta.app.autocompleter.entity import (AutoCompleter,
|
||||
_get_history_items,
|
||||
_is_command_exist)
|
||||
return mock
|
||||
|
||||
|
||||
class TestAutoCompleter:
|
||||
"""Test suite for the AutoCompleter class."""
|
||||
HISTORY_FILE = "test_history.txt"
|
||||
COMMANDS = ["start", "stop", "status"]
|
||||
|
||||
def test_initialization(self):
|
||||
"""Tests that the constructor correctly assigns attributes."""
|
||||
completer = AutoCompleter(history_filename=self.HISTORY_FILE, autocomplete_button="tab")
|
||||
assert completer.history_filename == self.HISTORY_FILE
|
||||
assert completer.autocomplete_button == "tab"
|
||||
|
||||
def test_initial_setup_if_history_file_does_not_exist(self, fs, mock_readline):
|
||||
"""Tests initial setup creates history from commands when the history file is absent."""
|
||||
# Ensure the file does not exist in the fake filesystem.
|
||||
if os.path.exists(self.HISTORY_FILE):
|
||||
os.remove(self.HISTORY_FILE)
|
||||
|
||||
completer = AutoCompleter(history_filename=self.HISTORY_FILE)
|
||||
completer.initial_setup(self.COMMANDS)
|
||||
|
||||
mock_readline.read_history_file.assert_not_called()
|
||||
expected_calls = [call(cmd) for cmd in self.COMMANDS]
|
||||
mock_readline.add_history.assert_has_calls(expected_calls, any_order=True)
|
||||
assert mock_readline.add_history.call_count == len(self.COMMANDS)
|
||||
|
||||
mock_readline.set_completer.assert_called_with(completer._complete)
|
||||
mock_readline.parse_and_bind.assert_called_with("tab: complete")
|
||||
|
||||
def test_initial_setup_if_history_file_exists(self, fs, mock_readline):
|
||||
"""Tests initial setup reads from an existing history file."""
|
||||
fs.create_file(self.HISTORY_FILE, contents="previous_command\n")
|
||||
|
||||
completer = AutoCompleter(history_filename=self.HISTORY_FILE)
|
||||
completer.initial_setup(self.COMMANDS)
|
||||
|
||||
mock_readline.read_history_file.assert_called_once_with(self.HISTORY_FILE)
|
||||
mock_readline.add_history.assert_not_called()
|
||||
mock_readline.set_completer.assert_called_once()
|
||||
mock_readline.parse_and_bind.assert_called_once()
|
||||
|
||||
def test_initial_setup_with_no_history_filename(self, mock_readline):
|
||||
"""Tests initial setup when no history filename is provided."""
|
||||
completer = AutoCompleter(history_filename=None)
|
||||
completer.initial_setup(self.COMMANDS)
|
||||
|
||||
mock_readline.read_history_file.assert_not_called()
|
||||
expected_calls = [call(cmd) for cmd in self.COMMANDS]
|
||||
mock_readline.add_history.assert_has_calls(expected_calls, any_order=True)
|
||||
|
||||
def test_exit_setup_writes_and_filters_history(self, fs, mock_readline):
|
||||
"""Tests that exit_setup writes a filtered and unique history to the file."""
|
||||
# 1. Populate the mock readline history.
|
||||
mock_readline.add_history.side_effect(None) # Temporarily disable side effect to just record calls
|
||||
mock_readline.add_history("start server")
|
||||
mock_readline.add_history("stop client")
|
||||
mock_readline.add_history("invalid command")
|
||||
mock_readline.add_history("start server") # Add a duplicate.
|
||||
|
||||
# 2. Simulate the state of the history file after readline.write_history_file would have run.
|
||||
raw_history_content = "\n".join(["start server", "stop client", "invalid command", "start server"])
|
||||
fs.create_file(self.HISTORY_FILE, contents=raw_history_content)
|
||||
|
||||
# 3. Call the method under test.
|
||||
completer = AutoCompleter(history_filename=self.HISTORY_FILE)
|
||||
completer.exit_setup(all_commands=["start", "stop"], ignore_command_register=False)
|
||||
|
||||
# 4. Assert that readline's write function was called.
|
||||
mock_readline.write_history_file.assert_called_once_with(self.HISTORY_FILE)
|
||||
|
||||
# 5. Assert the file was correctly re-written with filtered and unique content.
|
||||
with open(self.HISTORY_FILE, "r") as f:
|
||||
content = f.read()
|
||||
lines = sorted(content.strip().split("\n"))
|
||||
assert lines == ["start server", "stop client"]
|
||||
|
||||
def test_exit_setup_with_no_history_filename(self, mock_readline):
|
||||
"""Tests that exit_setup does nothing if no filename is provided."""
|
||||
completer = AutoCompleter(history_filename=None)
|
||||
completer.exit_setup(all_commands=self.COMMANDS, ignore_command_register=False)
|
||||
mock_readline.write_history_file.assert_not_called()
|
||||
|
||||
def test_complete_with_no_matches(self, mock_readline):
|
||||
"""Tests the _complete method when there are no matching history items."""
|
||||
for cmd in ["start", "stop"]:
|
||||
mock_readline.add_history(cmd)
|
||||
|
||||
completer = AutoCompleter()
|
||||
assert completer._complete("run", 0) is None
|
||||
assert completer._complete("run", 1) is None
|
||||
|
||||
def test_complete_with_one_match(self, mock_readline):
|
||||
"""Tests the _complete method when there is exactly one match."""
|
||||
mock_readline.add_history("start server")
|
||||
mock_readline.add_history("stop server")
|
||||
|
||||
completer = AutoCompleter()
|
||||
assert completer._complete("start", 0) == "start server"
|
||||
assert completer._complete("start", 1) is None # Subsequent states yield no matches
|
||||
|
||||
def test_complete_with_multiple_matches(self, mock_readline):
|
||||
"""Tests _complete with multiple matches that share a common prefix."""
|
||||
mock_readline.add_history("status client")
|
||||
mock_readline.add_history("status server")
|
||||
mock_readline.add_history("stop")
|
||||
|
||||
completer = AutoCompleter()
|
||||
|
||||
# On state 0, it should insert the common prefix via readline and return None.
|
||||
result = completer._complete("stat", 0)
|
||||
assert result is None
|
||||
mock_readline.insert_text.assert_called_once_with("us ") # Completes "stat" to "status "
|
||||
mock_readline.redisplay.assert_called_once()
|
||||
|
||||
# On subsequent states, it should do nothing.
|
||||
mock_readline.reset_mock()
|
||||
result_state_1 = completer._complete("stat", 1)
|
||||
assert result_state_1 is None
|
||||
mock_readline.insert_text.assert_not_called()
|
||||
HISTORY_FILE: str = "test_history.txt"
|
||||
COMMANDS: list[str] = ["start", "stop", "status"]
|
||||
|
||||
|
||||
class TestHelperFunctions:
|
||||
"""Test suite for helper functions in the autocompleter module."""
|
||||
def test_initialization() -> None:
|
||||
completer: AutoCompleter = AutoCompleter(history_filename=HISTORY_FILE, autocomplete_button="tab")
|
||||
assert completer.history_filename == HISTORY_FILE
|
||||
assert completer.autocomplete_button == "tab"
|
||||
|
||||
def test_is_command_exist(self):
|
||||
"""Tests the _is_command_exist helper function."""
|
||||
existing = ["start", "stop", "status"]
|
||||
|
||||
# Case-sensitive check
|
||||
assert _is_command_exist("start", existing, ignore_command_register=False) is True
|
||||
assert _is_command_exist("START", existing, ignore_command_register=False) is False
|
||||
assert _is_command_exist("unknown", existing, ignore_command_register=False) is False
|
||||
def test_initial_setup_if_history_file_does_not_exist(fs: FakeFilesystem, mock_readline: MockType) -> None:
|
||||
if os.path.exists(HISTORY_FILE):
|
||||
os.remove(HISTORY_FILE)
|
||||
|
||||
# Case-insensitive check
|
||||
assert _is_command_exist("start", existing, ignore_command_register=True) is True
|
||||
assert _is_command_exist("START", existing, ignore_command_register=True) is True
|
||||
assert _is_command_exist("unknown", existing, ignore_command_register=True) is False
|
||||
completer: AutoCompleter = AutoCompleter(history_filename=HISTORY_FILE)
|
||||
completer.initial_setup(COMMANDS)
|
||||
|
||||
def test_get_history_items(self, mock_readline):
|
||||
"""Tests the _get_history_items helper function."""
|
||||
assert _get_history_items() == []
|
||||
mock_readline.read_history_file.assert_not_called()
|
||||
assert mock_readline.add_history.call_count == len(COMMANDS)
|
||||
|
||||
mock_readline.add_history("first item")
|
||||
mock_readline.add_history("second item")
|
||||
mock_readline.set_completer.assert_called_with(completer._complete)
|
||||
mock_readline.parse_and_bind.assert_called_with("tab: complete")
|
||||
|
||||
assert _get_history_items() == ["first item", "second item"]
|
||||
|
||||
def test_initial_setup_if_history_file_exists(fs: FakeFilesystem, mock_readline: MockType) -> None:
|
||||
fs.create_file(HISTORY_FILE, contents="previous_command\n")
|
||||
|
||||
completer: AutoCompleter = AutoCompleter(history_filename=HISTORY_FILE)
|
||||
completer.initial_setup(COMMANDS)
|
||||
|
||||
mock_readline.read_history_file.assert_called_once_with(HISTORY_FILE)
|
||||
mock_readline.add_history.assert_not_called()
|
||||
mock_readline.set_completer.assert_called_once()
|
||||
mock_readline.parse_and_bind.assert_called_once()
|
||||
|
||||
|
||||
def test_initial_setup_with_no_history_filename(mock_readline: MockType) -> None:
|
||||
completer: AutoCompleter = AutoCompleter(history_filename=None)
|
||||
completer.initial_setup(COMMANDS)
|
||||
|
||||
mock_readline.read_history_file.assert_not_called()
|
||||
assert mock_readline.add_history.call_count == len(COMMANDS)
|
||||
|
||||
|
||||
def test_exit_setup_writes_and_filters_history(fs: FakeFilesystem, mock_readline: MockType) -> None:
|
||||
mock_readline.add_history.side_effect = None
|
||||
mock_readline.add_history("start server")
|
||||
mock_readline.add_history("stop client")
|
||||
mock_readline.add_history("invalid command")
|
||||
mock_readline.add_history("start server")
|
||||
|
||||
raw_history_content: str = "\n".join(["start server", "stop client", "invalid command", "start server"])
|
||||
fs.create_file(HISTORY_FILE, contents=raw_history_content)
|
||||
|
||||
completer: AutoCompleter = AutoCompleter(history_filename=HISTORY_FILE)
|
||||
completer.exit_setup(all_commands=["start", "stop"], ignore_command_register=False)
|
||||
|
||||
mock_readline.write_history_file.assert_called_once_with(HISTORY_FILE)
|
||||
|
||||
with open(HISTORY_FILE) as f:
|
||||
content: str = f.read()
|
||||
lines: list[str] = sorted(content.strip().split("\n"))
|
||||
assert lines == ["start server", "stop client"]
|
||||
|
||||
|
||||
def test_exit_setup_with_no_history_filename(mock_readline: MockType) -> None:
|
||||
completer: AutoCompleter = AutoCompleter(history_filename=None)
|
||||
completer.exit_setup(all_commands=COMMANDS, ignore_command_register=False)
|
||||
mock_readline.write_history_file.assert_not_called()
|
||||
|
||||
|
||||
def test_complete_with_no_matches(mock_readline: MockType) -> None:
|
||||
cmd: str
|
||||
for cmd in ["start", "stop"]:
|
||||
mock_readline.add_history(cmd)
|
||||
|
||||
completer: AutoCompleter = AutoCompleter()
|
||||
assert completer._complete("run", 0) is None
|
||||
assert completer._complete("run", 1) is None
|
||||
|
||||
|
||||
def test_complete_with_one_match(mock_readline: MockType) -> None:
|
||||
mock_readline.add_history("start server")
|
||||
mock_readline.add_history("stop server")
|
||||
|
||||
completer: AutoCompleter = AutoCompleter()
|
||||
assert completer._complete("start", 0) == "start server"
|
||||
assert completer._complete("start", 1) is None
|
||||
|
||||
|
||||
def test_complete_with_multiple_matches(mock_readline: MockType) -> None:
|
||||
mock_readline.add_history("status client")
|
||||
mock_readline.add_history("status server")
|
||||
mock_readline.add_history("stop")
|
||||
|
||||
completer: AutoCompleter = AutoCompleter()
|
||||
|
||||
result: str | None = completer._complete("stat", 0)
|
||||
assert result is None
|
||||
mock_readline.insert_text.assert_called_once_with("us ")
|
||||
mock_readline.redisplay.assert_called_once()
|
||||
|
||||
mock_readline.reset_mock()
|
||||
result_state_1: str | None = completer._complete("stat", 1)
|
||||
assert result_state_1 is None
|
||||
mock_readline.insert_text.assert_not_called()
|
||||
|
||||
|
||||
def test_is_command_exist() -> None:
|
||||
existing: list[str] = ["start", "stop", "status"]
|
||||
|
||||
assert _is_command_exist("start", existing, ignore_command_register=False) is True
|
||||
assert _is_command_exist("START", existing, ignore_command_register=False) is False
|
||||
assert _is_command_exist("unknown", existing, ignore_command_register=False) is False
|
||||
|
||||
assert _is_command_exist("start", existing, ignore_command_register=True) is True
|
||||
assert _is_command_exist("START", existing, ignore_command_register=True) is True
|
||||
assert _is_command_exist("unknown", existing, ignore_command_register=True) is False
|
||||
|
||||
|
||||
def test_get_history_items(mock_readline: MockType) -> None:
|
||||
assert _get_history_items() == []
|
||||
|
||||
mock_readline.add_history("first item")
|
||||
mock_readline.add_history("second item")
|
||||
|
||||
assert _get_history_items() == ["first item", "second item"]
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import re
|
||||
import unittest
|
||||
|
||||
import pytest
|
||||
|
||||
from argenta.command.exceptions import (EmptyInputCommandException,
|
||||
RepeatedInputFlagsException,
|
||||
@@ -10,54 +11,65 @@ from argenta.command.flag.models import PossibleValues, ValidationStatus
|
||||
from argenta.command.models import Command, InputCommand
|
||||
|
||||
|
||||
class TestInputCommand(unittest.TestCase):
|
||||
def test_parse_correct_raw_command(self):
|
||||
self.assertEqual(InputCommand.parse('ssh --host 192.168.0.3').trigger, 'ssh')
|
||||
def test_parse_correct_raw_command():
|
||||
assert InputCommand.parse('ssh --host 192.168.0.3').trigger == 'ssh'
|
||||
|
||||
def test_parse_raw_command_without_flag_name_with_value(self):
|
||||
with self.assertRaises(UnprocessedInputFlagException):
|
||||
InputCommand.parse('ssh 192.168.0.3')
|
||||
|
||||
def test_parse_raw_command_with_repeated_flag_name(self):
|
||||
with self.assertRaises(RepeatedInputFlagsException):
|
||||
InputCommand.parse('ssh --host 192.168.0.3 --host 172.198.0.43')
|
||||
def test_parse_raw_command_without_flag_name_with_value():
|
||||
with pytest.raises(UnprocessedInputFlagException):
|
||||
InputCommand.parse('ssh 192.168.0.3')
|
||||
|
||||
def test_parse_empty_raw_command(self):
|
||||
with self.assertRaises(EmptyInputCommandException):
|
||||
InputCommand.parse('')
|
||||
|
||||
def test_validate_invalid_input_flag1(self):
|
||||
command = Command('some', flags=Flag('test'))
|
||||
self.assertEqual(command.validate_input_flag(InputFlag('test', input_value='', status=None)), ValidationStatus.INVALID)
|
||||
def test_parse_raw_command_with_repeated_flag_name():
|
||||
with pytest.raises(RepeatedInputFlagsException):
|
||||
InputCommand.parse('ssh --host 192.168.0.3 --host 172.198.0.43')
|
||||
|
||||
def test_validate_valid_input_flag2(self):
|
||||
command = Command('some', flags=Flags([Flag('test'), Flag('more')]))
|
||||
self.assertEqual(command.validate_input_flag(InputFlag('more', input_value='random-value', status=None)), ValidationStatus.VALID)
|
||||
|
||||
def test_validate_undefined_input_flag1(self):
|
||||
command = Command('some', flags=Flag('test'))
|
||||
self.assertEqual(command.validate_input_flag(InputFlag('more', input_value='', status=None)), ValidationStatus.UNDEFINED)
|
||||
def test_parse_empty_raw_command():
|
||||
with pytest.raises(EmptyInputCommandException):
|
||||
InputCommand.parse('')
|
||||
|
||||
def test_validate_undefined_input_flag2(self):
|
||||
command = Command('some', flags=Flags([Flag('test'), Flag('more')]))
|
||||
self.assertEqual(command.validate_input_flag(InputFlag('case', input_value='', status=None)), ValidationStatus.UNDEFINED)
|
||||
|
||||
def test_validate_undefined_input_flag3(self):
|
||||
command = Command('some')
|
||||
self.assertEqual(command.validate_input_flag(InputFlag('case', input_value='', status=None)), ValidationStatus.UNDEFINED)
|
||||
def test_validate_invalid_input_flag1():
|
||||
command = Command('some', flags=Flag('test'))
|
||||
assert command.validate_input_flag(InputFlag('test', input_value='', status=None)) == ValidationStatus.INVALID
|
||||
|
||||
def test_invalid_input_flag1(self):
|
||||
command = Command('some', flags=Flag('test', possible_values=PossibleValues.NEITHER))
|
||||
self.assertEqual(command.validate_input_flag(InputFlag('test', input_value='example', status=None)), ValidationStatus.INVALID)
|
||||
|
||||
def test_invalid_input_flag2(self):
|
||||
command = Command('some', flags=Flag('test', possible_values=['some', 'case']))
|
||||
self.assertEqual(command.validate_input_flag(InputFlag('test', input_value='slay', status=None)), ValidationStatus.INVALID)
|
||||
def test_validate_valid_input_flag2():
|
||||
command = Command('some', flags=Flags([Flag('test'), Flag('more')]))
|
||||
assert command.validate_input_flag(InputFlag('more', input_value='random-value', status=None)) == ValidationStatus.VALID
|
||||
|
||||
def test_invalid_input_flag3(self):
|
||||
command = Command('some', flags=Flag('test', possible_values=re.compile(r'^ex\d{, 2}op$')))
|
||||
self.assertEqual(command.validate_input_flag(InputFlag('test', input_value='example', status=None)), ValidationStatus.INVALID)
|
||||
|
||||
def test_isinstance_parse_correct_raw_command(self):
|
||||
cmd = InputCommand.parse('ssh --host 192.168.0.3')
|
||||
self.assertIsInstance(cmd, InputCommand)
|
||||
def test_validate_undefined_input_flag1():
|
||||
command = Command('some', flags=Flag('test'))
|
||||
assert command.validate_input_flag(InputFlag('more', input_value='', status=None)) == ValidationStatus.UNDEFINED
|
||||
|
||||
|
||||
def test_validate_undefined_input_flag2():
|
||||
command = Command('some', flags=Flags([Flag('test'), Flag('more')]))
|
||||
assert command.validate_input_flag(InputFlag('case', input_value='', status=None)) == ValidationStatus.UNDEFINED
|
||||
|
||||
|
||||
def test_validate_undefined_input_flag3():
|
||||
command = Command('some')
|
||||
assert command.validate_input_flag(InputFlag('case', input_value='', status=None)) == ValidationStatus.UNDEFINED
|
||||
|
||||
|
||||
def test_invalid_input_flag1():
|
||||
command = Command('some', flags=Flag('test', possible_values=PossibleValues.NEITHER))
|
||||
assert command.validate_input_flag(InputFlag('test', input_value='example', status=None)) == ValidationStatus.INVALID
|
||||
|
||||
|
||||
def test_invalid_input_flag2():
|
||||
command = Command('some', flags=Flag('test', possible_values=['some', 'case']))
|
||||
assert command.validate_input_flag(InputFlag('test', input_value='slay', status=None)) == ValidationStatus.INVALID
|
||||
|
||||
|
||||
def test_invalid_input_flag3():
|
||||
command = Command('some', flags=Flag('test', possible_values=re.compile(r'^ex\d{, 2}op$')))
|
||||
assert command.validate_input_flag(InputFlag('test', input_value='example', status=None)) == ValidationStatus.INVALID
|
||||
|
||||
|
||||
def test_isinstance_parse_correct_raw_command():
|
||||
cmd = InputCommand.parse('ssh --host 192.168.0.3')
|
||||
assert isinstance(cmd, InputCommand)
|
||||
|
||||
+88
-101
@@ -1,129 +1,116 @@
|
||||
import re
|
||||
import unittest
|
||||
|
||||
from argenta.command.flag import Flag, InputFlag, PossibleValues
|
||||
from argenta.command.flag.flags import Flags, InputFlags
|
||||
|
||||
|
||||
class TestFlag(unittest.TestCase):
|
||||
def test_get_string_entity(self):
|
||||
self.assertEqual(Flag(name='test').string_entity,
|
||||
'--test')
|
||||
|
||||
def test_get_string_entity2(self):
|
||||
self.assertEqual(Flag(name='test',
|
||||
prefix='---').string_entity,
|
||||
'---test')
|
||||
|
||||
def test_get_flag_name(self):
|
||||
self.assertEqual(Flag(name='test').name,
|
||||
'test')
|
||||
|
||||
def test_get_flag_prefix(self):
|
||||
self.assertEqual(Flag(name='test').prefix,
|
||||
'--')
|
||||
|
||||
def test_get_flag_prefix2(self):
|
||||
self.assertEqual(Flag(name='test',
|
||||
prefix='--').prefix,
|
||||
'--')
|
||||
|
||||
def test_get_flag_value_without_set(self):
|
||||
self.assertEqual(InputFlag(name='test', input_value='', status=None).input_value,
|
||||
'')
|
||||
|
||||
def test_get_flag_value_with_set(self):
|
||||
flag = InputFlag(name='test', input_value='example', status=None)
|
||||
self.assertEqual(flag.input_value, 'example')
|
||||
|
||||
def test_validate_incorrect_flag_value_with_list_of_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=['1', '2', '3'])
|
||||
self.assertEqual(flag.validate_input_flag_value('bad value'), False)
|
||||
|
||||
def test_validate_correct_flag_value_with_list_of_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=['1', '2', '3'])
|
||||
self.assertEqual(flag.validate_input_flag_value('1'), True)
|
||||
|
||||
def test_validate_incorrect_flag_value_with_pattern_of_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=re.compile(r'192.168.\d+.\d+'))
|
||||
self.assertEqual(flag.validate_input_flag_value('152.123.9.8'), False)
|
||||
|
||||
def test_validate_correct_flag_value_with_pattern_of_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=re.compile(r'192.168.\d+.\d+'))
|
||||
self.assertEqual(flag.validate_input_flag_value('192.168.9.8'), True)
|
||||
|
||||
def test_validate_correct_empty_flag_value_without_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=PossibleValues.NEITHER)
|
||||
self.assertEqual(flag.validate_input_flag_value(''), True)
|
||||
|
||||
def test_validate_correct_empty_flag_value_with_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=PossibleValues.NEITHER)
|
||||
self.assertEqual(flag.validate_input_flag_value(''), True)
|
||||
|
||||
def test_validate_incorrect_random_flag_value_without_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=PossibleValues.NEITHER)
|
||||
self.assertEqual(flag.validate_input_flag_value('random value'), False)
|
||||
|
||||
def test_validate_correct_random_flag_value_with_possible_flag_values(self):
|
||||
flag = Flag(name='test', possible_values=PossibleValues.ALL)
|
||||
self.assertEqual(flag.validate_input_flag_value('random value'), True)
|
||||
|
||||
def test_get_input_flag1(self):
|
||||
flag = InputFlag(name='test', input_value='', status=None)
|
||||
input_flags = InputFlags([flag])
|
||||
self.assertEqual(input_flags.get_flag_by_name('test'), flag)
|
||||
|
||||
def test_get_input_flag2(self):
|
||||
flag = InputFlag(name='test', input_value='', status=None)
|
||||
flag2 = InputFlag(name='some', input_value='', status=None)
|
||||
input_flags = InputFlags([flag, flag2])
|
||||
self.assertEqual(input_flags.get_flag_by_name('some'), flag2)
|
||||
|
||||
def test_get_undefined_input_flag(self):
|
||||
flag = InputFlag(name='test', input_value='', status=None)
|
||||
flag2 = InputFlag(name='some', input_value='', status=None)
|
||||
input_flags = InputFlags([flag, flag2])
|
||||
self.assertEqual(input_flags.get_flag_by_name('case'), None)
|
||||
|
||||
def test_get_flags(self):
|
||||
flags = Flags()
|
||||
list_of_flags = [
|
||||
Flag('test1'),
|
||||
Flag('test2'),
|
||||
Flag('test3'),
|
||||
]
|
||||
flags.add_flags(list_of_flags)
|
||||
self.assertEqual(flags.flags,
|
||||
list_of_flags)
|
||||
|
||||
def test_add_flag(self):
|
||||
flags = Flags()
|
||||
flags.add_flag(Flag('test'))
|
||||
self.assertEqual(len(flags.flags), 1)
|
||||
|
||||
def test_add_flags(self):
|
||||
flags = Flags()
|
||||
flags.add_flags([Flag('test'), Flag('test2')])
|
||||
self.assertEqual(len(flags.flags), 2)
|
||||
def test_get_string_entity():
|
||||
assert Flag(name='test').string_entity == '--test'
|
||||
|
||||
|
||||
def test_get_string_entity2():
|
||||
assert Flag(name='test', prefix='---').string_entity == '---test'
|
||||
|
||||
|
||||
def test_get_flag_name():
|
||||
assert Flag(name='test').name == 'test'
|
||||
|
||||
|
||||
def test_get_flag_prefix():
|
||||
assert Flag(name='test').prefix == '--'
|
||||
|
||||
|
||||
def test_get_flag_prefix2():
|
||||
assert Flag(name='test', prefix='--').prefix == '--'
|
||||
|
||||
|
||||
def test_get_flag_value_without_set():
|
||||
assert InputFlag(name='test', input_value='', status=None).input_value == ''
|
||||
|
||||
|
||||
def test_get_flag_value_with_set():
|
||||
flag = InputFlag(name='test', input_value='example', status=None)
|
||||
assert flag.input_value == 'example'
|
||||
|
||||
|
||||
def test_validate_incorrect_flag_value_with_list_of_possible_flag_values():
|
||||
flag = Flag(name='test', possible_values=['1', '2', '3'])
|
||||
assert flag.validate_input_flag_value('bad value') is False
|
||||
|
||||
|
||||
def test_validate_correct_flag_value_with_list_of_possible_flag_values():
|
||||
flag = Flag(name='test', possible_values=['1', '2', '3'])
|
||||
assert flag.validate_input_flag_value('1') is True
|
||||
|
||||
|
||||
def test_validate_incorrect_flag_value_with_pattern_of_possible_flag_values():
|
||||
flag = Flag(name='test', possible_values=re.compile(r'192.168.\d+.\d+'))
|
||||
assert flag.validate_input_flag_value('152.123.9.8') is False
|
||||
|
||||
|
||||
def test_validate_correct_flag_value_with_pattern_of_possible_flag_values():
|
||||
flag = Flag(name='test', possible_values=re.compile(r'192.168.\d+.\d+'))
|
||||
assert flag.validate_input_flag_value('192.168.9.8') is True
|
||||
|
||||
|
||||
def test_validate_correct_empty_flag_value_without_possible_flag_values():
|
||||
flag = Flag(name='test', possible_values=PossibleValues.NEITHER)
|
||||
assert flag.validate_input_flag_value('') is True
|
||||
|
||||
|
||||
def test_validate_correct_empty_flag_value_with_possible_flag_values():
|
||||
flag = Flag(name='test', possible_values=PossibleValues.NEITHER)
|
||||
assert flag.validate_input_flag_value('') is True
|
||||
|
||||
|
||||
def test_validate_incorrect_random_flag_value_without_possible_flag_values():
|
||||
flag = Flag(name='test', possible_values=PossibleValues.NEITHER)
|
||||
assert flag.validate_input_flag_value('random value') is False
|
||||
|
||||
|
||||
def test_validate_correct_random_flag_value_with_possible_flag_values():
|
||||
flag = Flag(name='test', possible_values=PossibleValues.ALL)
|
||||
assert flag.validate_input_flag_value('random value') is True
|
||||
|
||||
|
||||
def test_get_input_flag1():
|
||||
flag = InputFlag(name='test', input_value='', status=None)
|
||||
input_flags = InputFlags([flag])
|
||||
assert input_flags.get_flag_by_name('test') == flag
|
||||
|
||||
|
||||
def test_get_input_flag2():
|
||||
flag = InputFlag(name='test', input_value='', status=None)
|
||||
flag2 = InputFlag(name='some', input_value='', status=None)
|
||||
input_flags = InputFlags([flag, flag2])
|
||||
assert input_flags.get_flag_by_name('some') == flag2
|
||||
|
||||
|
||||
def test_get_undefined_input_flag():
|
||||
flag = InputFlag(name='test', input_value='', status=None)
|
||||
flag2 = InputFlag(name='some', input_value='', status=None)
|
||||
input_flags = InputFlags([flag, flag2])
|
||||
assert input_flags.get_flag_by_name('case') is None
|
||||
|
||||
|
||||
def test_get_flags():
|
||||
flags = Flags()
|
||||
list_of_flags = [
|
||||
Flag('test1'),
|
||||
Flag('test2'),
|
||||
Flag('test3'),
|
||||
]
|
||||
flags.add_flags(list_of_flags)
|
||||
assert flags.flags == list_of_flags
|
||||
|
||||
|
||||
def test_add_flag():
|
||||
flags = Flags()
|
||||
flags.add_flag(Flag('test'))
|
||||
assert len(flags.flags) == 1
|
||||
|
||||
|
||||
def test_add_flags():
|
||||
flags = Flags()
|
||||
flags.add_flags([Flag('test'), Flag('test2')])
|
||||
assert len(flags.flags) == 2
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import unittest
|
||||
from datetime import date, datetime
|
||||
|
||||
import pytest
|
||||
|
||||
from argenta.data_bridge import DataBridge
|
||||
from argenta.command.flag.models import InputFlag
|
||||
from argenta.command.flag.flags.models import InputFlags
|
||||
@@ -8,96 +9,100 @@ from argenta.response.entity import EMPTY_INPUT_FLAGS, Response
|
||||
from argenta.response.status import ResponseStatus
|
||||
|
||||
|
||||
class TestDataBridge(unittest.TestCase):
|
||||
def setUp(self):
|
||||
"""Create a new DataBridge instance for each test"""
|
||||
self.data_bridge = DataBridge()
|
||||
|
||||
def test_update_data_basic(self):
|
||||
"""Test basic data update functionality"""
|
||||
test_data = {"key1": "value1", "key2": "value2"}
|
||||
self.data_bridge.update(test_data)
|
||||
self.assertEqual(self.data_bridge.get_all(), test_data)
|
||||
|
||||
def test_update_data_with_datetime(self):
|
||||
"""Test updating data with datetime objects"""
|
||||
test_datetime = datetime(2024, 1, 15, 10, 30, 45)
|
||||
test_data = {"created_at": test_datetime, "name": "test"}
|
||||
self.data_bridge.update(test_data)
|
||||
|
||||
result = self.data_bridge.get_all()
|
||||
self.assertEqual(result["created_at"], test_datetime)
|
||||
self.assertEqual(result["name"], "test")
|
||||
|
||||
def test_update_data_multiple_calls(self):
|
||||
"""Test multiple update calls merge data"""
|
||||
first_data = {"key1": "value1"}
|
||||
second_data = {"key2": "value2"}
|
||||
self.data_bridge.update(first_data)
|
||||
self.data_bridge.update(second_data)
|
||||
self.assertEqual(len(self.data_bridge.get_all()), 2)
|
||||
|
||||
def test_get_data_empty(self):
|
||||
"""Test get_all returns empty dict when no data"""
|
||||
self.assertEqual(self.data_bridge.get_all(), {})
|
||||
|
||||
def test_clear_data(self):
|
||||
"""Test clear_all removes all data"""
|
||||
self.data_bridge.update({"key": "value"})
|
||||
self.assertNotEqual(self.data_bridge.get_all(), {})
|
||||
self.data_bridge.clear_all()
|
||||
self.assertEqual(self.data_bridge.get_all(), {})
|
||||
|
||||
def test_delete_from_data(self):
|
||||
"""Test delete_by_key removes specific key"""
|
||||
test_data = {"key1": "value1", "key2": "value2"}
|
||||
self.data_bridge.update(test_data)
|
||||
self.data_bridge.delete_by_key("key1")
|
||||
result = self.data_bridge.get_all()
|
||||
self.assertNotIn("key1", result)
|
||||
self.assertIn("key2", result)
|
||||
|
||||
def test_delete_from_data_nonexistent_key(self):
|
||||
"""Test delete_by_key with nonexistent key raises KeyError"""
|
||||
with self.assertRaises(KeyError):
|
||||
self.data_bridge.delete_by_key("nonexistent_key")
|
||||
|
||||
def test_get_by_key(self):
|
||||
"""Test get_by_key retrieves correct value"""
|
||||
test_data = {"key1": "value1", "key2": date(2024, 1, 1)}
|
||||
self.data_bridge.update(test_data)
|
||||
self.assertEqual(self.data_bridge.get_by_key("key1"), "value1")
|
||||
self.assertEqual(self.data_bridge.get_by_key("key2"), date(2024, 1, 1))
|
||||
self.assertIsNone(self.data_bridge.get_by_key("nonexistent"))
|
||||
@pytest.fixture
|
||||
def data_bridge():
|
||||
"""Create a new DataBridge instance for each test"""
|
||||
return DataBridge()
|
||||
|
||||
|
||||
class TestResponse(unittest.TestCase):
|
||||
def test_response_initialization_basic(self):
|
||||
"""Test basic Response initialization"""
|
||||
response = Response(ResponseStatus.ALL_FLAGS_VALID)
|
||||
self.assertEqual(response.status, ResponseStatus.ALL_FLAGS_VALID)
|
||||
self.assertEqual(response.input_flags, EMPTY_INPUT_FLAGS)
|
||||
|
||||
def test_response_initialization_with_flags(self):
|
||||
"""Test Response initialization with input flags"""
|
||||
input_flags = InputFlags([InputFlag('test', input_value='value', status=None)])
|
||||
response = Response(ResponseStatus.INVALID_VALUE_FLAGS, input_flags)
|
||||
self.assertEqual(response.status, ResponseStatus.INVALID_VALUE_FLAGS)
|
||||
self.assertEqual(response.input_flags, input_flags)
|
||||
|
||||
def test_response_status_types(self):
|
||||
"""Test Response with different status types"""
|
||||
statuses = [
|
||||
ResponseStatus.ALL_FLAGS_VALID,
|
||||
ResponseStatus.UNDEFINED_FLAGS,
|
||||
ResponseStatus.INVALID_VALUE_FLAGS,
|
||||
ResponseStatus.UNDEFINED_AND_INVALID_FLAGS
|
||||
]
|
||||
for status in statuses:
|
||||
with self.subTest(status=status):
|
||||
response = Response(status)
|
||||
self.assertEqual(response.status, status)
|
||||
def test_update_data_basic(data_bridge: DataBridge):
|
||||
"""Test basic data update functionality"""
|
||||
test_data = {"key1": "value1", "key2": "value2"}
|
||||
data_bridge.update(test_data)
|
||||
assert data_bridge.get_all() == test_data
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
def test_update_data_with_datetime(data_bridge: DataBridge):
|
||||
"""Test updating data with datetime objects"""
|
||||
test_datetime = datetime(2024, 1, 15, 10, 30, 45)
|
||||
test_data = {"created_at": test_datetime, "name": "test"}
|
||||
data_bridge.update(test_data)
|
||||
|
||||
result = data_bridge.get_all()
|
||||
assert result["created_at"] == test_datetime
|
||||
assert result["name"] == "test"
|
||||
|
||||
|
||||
def test_update_data_multiple_calls(data_bridge: DataBridge):
|
||||
"""Test multiple update calls merge data"""
|
||||
first_data = {"key1": "value1"}
|
||||
second_data = {"key2": "value2"}
|
||||
data_bridge.update(first_data)
|
||||
data_bridge.update(second_data)
|
||||
assert len(data_bridge.get_all()) == 2
|
||||
|
||||
|
||||
def test_get_data_empty(data_bridge: DataBridge):
|
||||
"""Test get_all returns empty dict when no data"""
|
||||
assert data_bridge.get_all() == {}
|
||||
|
||||
|
||||
def test_clear_data(data_bridge: DataBridge):
|
||||
"""Test clear_all removes all data"""
|
||||
data_bridge.update({"key": "value"})
|
||||
assert data_bridge.get_all() != {}
|
||||
data_bridge.clear_all()
|
||||
assert data_bridge.get_all() == {}
|
||||
|
||||
|
||||
def test_delete_from_data(data_bridge: DataBridge):
|
||||
"""Test delete_by_key removes specific key"""
|
||||
test_data = {"key1": "value1", "key2": "value2"}
|
||||
data_bridge.update(test_data)
|
||||
data_bridge.delete_by_key("key1")
|
||||
result = data_bridge.get_all()
|
||||
assert "key1" not in result
|
||||
assert "key2" in result
|
||||
|
||||
|
||||
def test_delete_from_data_nonexistent_key(data_bridge: DataBridge):
|
||||
"""Test delete_by_key with nonexistent key raises KeyError"""
|
||||
with pytest.raises(KeyError):
|
||||
data_bridge.delete_by_key("nonexistent_key")
|
||||
|
||||
|
||||
def test_get_by_key(data_bridge: DataBridge):
|
||||
"""Test get_by_key retrieves correct value"""
|
||||
test_data = {"key1": "value1", "key2": date(2024, 1, 1)}
|
||||
data_bridge.update(test_data)
|
||||
assert data_bridge.get_by_key("key1") == "value1"
|
||||
assert data_bridge.get_by_key("key2") == date(2024, 1, 1)
|
||||
assert data_bridge.get_by_key("nonexistent") is None
|
||||
|
||||
|
||||
def test_response_initialization_basic():
|
||||
"""Test basic Response initialization"""
|
||||
response = Response(ResponseStatus.ALL_FLAGS_VALID)
|
||||
assert response.status == ResponseStatus.ALL_FLAGS_VALID
|
||||
assert response.input_flags == EMPTY_INPUT_FLAGS
|
||||
|
||||
|
||||
def test_response_initialization_with_flags():
|
||||
"""Test Response initialization with input flags"""
|
||||
input_flags = InputFlags([InputFlag('test', input_value='value', status=None)])
|
||||
response = Response(ResponseStatus.INVALID_VALUE_FLAGS, input_flags)
|
||||
assert response.status == ResponseStatus.INVALID_VALUE_FLAGS
|
||||
assert response.input_flags == input_flags
|
||||
|
||||
|
||||
def test_response_status_types():
|
||||
"""Test Response with different status types"""
|
||||
statuses = [
|
||||
ResponseStatus.ALL_FLAGS_VALID,
|
||||
ResponseStatus.UNDEFINED_FLAGS,
|
||||
ResponseStatus.INVALID_VALUE_FLAGS,
|
||||
ResponseStatus.UNDEFINED_AND_INVALID_FLAGS
|
||||
]
|
||||
for status in statuses:
|
||||
response = Response(status)
|
||||
assert response.status == status
|
||||
|
||||
Reference in New Issue
Block a user