refactor tests and add new

This commit is contained in:
2025-12-06 11:55:50 +03:00
parent a2ef2652ed
commit 1d2ab6f6bb
17 changed files with 294 additions and 31 deletions
+17 -3
View File
@@ -1,5 +1,19 @@
from argenta import App
from argenta import App, DataBridge, Response, Router
from argenta.di import FromDishka
from argenta.di.integration import setup_dishka, _auto_inject_handlers
from argenta.di.providers import SystemProvider
from dishka import make_container
container = make_container()
Response.patch_by_container(container)
app = App()
app._setup_default_view()
app._empty_input_command_handler()
router = Router()
@router.command('command')
def handler(res: Response, data_bridge: FromDishka[DataBridge]):
print(data_bridge)
_auto_inject_handlers(app)
_auto_inject_handlers(app)
+2 -1
View File
@@ -59,7 +59,8 @@ reportUnusedFunction = false
branch = true
omit = [
"src/argenta/app/protocols.py",
"src/argenta/command/exceptions.py"
"src/argenta/command/exceptions.py",
"src/argenta/metrics/*"
]
[tool.mypy]
+2 -5
View File
@@ -39,9 +39,6 @@ class BaseFlags(Generic[FlagType]):
def __iter__(self) -> Iterator[FlagType]:
return iter(self.flags)
def __next__(self) -> FlagType:
return next(iter(self))
def __getitem__(self, flag_index: int) -> FlagType:
return self.flags[flag_index]
@@ -61,7 +58,7 @@ class Flags(BaseFlags[Flag]):
@override
def __eq__(self, other: object) -> bool:
if not isinstance(other, Flags):
return NotImplemented
return False
if len(self.flags) != len(other.flags):
return False
@@ -91,7 +88,7 @@ class InputFlags(BaseFlags[InputFlag]):
@override
def __eq__(self, other: object) -> bool:
if not isinstance(other, InputFlags):
raise NotImplementedError
return False
if len(self.flags) != len(other.flags):
return False
+2 -5
View File
@@ -53,11 +53,8 @@ class Flag:
if isinstance(self.possible_values, Pattern):
return bool(self.possible_values.match(input_flag_value))
if isinstance(self.possible_values, list):
return input_flag_value in self.possible_values
return False
@property
def string_entity(self) -> str:
"""
@@ -88,9 +85,9 @@ class InputFlag:
self,
name: str,
*,
prefix: PREFIX_TYPE = "--",
input_value: str,
status: ValidationStatus | None,
prefix: PREFIX_TYPE = "--",
status: ValidationStatus | None = None,
):
"""
Public. The entity of the flag of the entered command
-3
View File
@@ -104,9 +104,6 @@ class InputCommand:
else:
raise UnprocessedInputFlagException
if not name:
raise UnprocessedInputFlagException
if i + 1 < len(tokens) and not tokens[i + 1].startswith("-"):
input_value = tokens[i + 1]
i += 2
+3 -3
View File
@@ -20,16 +20,16 @@ def inject(func: Callable[..., T]) -> Callable[..., T]:
def setup_dishka(app: App, container: Container, *, auto_inject: bool = False) -> None:
Response.patch_by_container(container)
if auto_inject:
_auto_inject_handlers(app)
Response.patch_by_container(container)
def _get_container_from_response(args: tuple[Any, ...], kwargs: dict[str, Any]) -> Container:
for arg in args:
if isinstance(arg, Response):
if hasattr(arg, "_dishka_container"):
return arg._dishka_container # pyright: ignore[reportPrivateUsage]
if hasattr(arg, "__dishka_container__"):
return arg.__dishka_container__ # pyright: ignore[reportPrivateUsage]
break
raise RuntimeError("dishka container not found in Response")
+1 -1
View File
@@ -94,7 +94,7 @@ class ArgParser:
namespace=self._core.parse_args(), processed_args=self.processed_args
)
def _register_args(self, processed_args: list[ValueArgument | BooleanArgument]) -> None:
def _register_args(self, processed_args: list[ValueArgument | BooleanArgument]) -> None: # pragma: no cover
if sys.version_info >= (3, 13):
for arg in processed_args:
if isinstance(arg, BooleanArgument):
+1 -1
View File
@@ -26,7 +26,7 @@ class Orchestrator:
self._custom_providers: list[Provider] = custom_providers
self._auto_inject_handlers: bool = auto_inject_handlers
self._arg_parser._parse_args()
self._arg_parser._parse_args() # pyright: ignore[reportPrivateUsage]
def start_polling(self, app: App) -> None:
"""
+2 -2
View File
@@ -9,7 +9,7 @@ EMPTY_INPUT_FLAGS: InputFlags = InputFlags()
class Response:
_dishka_container: Container
__dishka_container__: Container
def __init__(
self,
@@ -26,4 +26,4 @@ class Response:
@classmethod
def patch_by_container(cls, container: Container) -> None:
cls._dishka_container = container
cls.__dishka_container__ = container
@@ -44,6 +44,3 @@ class CommandHandlers:
def __iter__(self) -> Iterator[CommandHandler]:
return iter(self.command_handlers)
def __next__(self) -> CommandHandler:
return next(iter(self.command_handlers))
+1 -1
View File
@@ -20,7 +20,7 @@ class RequiredArgumentNotPassedException(Exception):
@override
def __str__(self) -> str:
return "Required argument not passed"
return "Required argument with type Response not passed"
class TriggerContainSpacesException(Exception):
+8
View File
@@ -245,3 +245,11 @@ def test_argparser_parse_args_populates_argspace(
assert debug_arg is not None
assert debug_arg.value is True
assert debug_arg.founder_class is BooleanArgument
def test_str_input_argument():
arg = InputArgument('host', value='192.168.0.0', founder_class=ValueArgument)
assert str(arg) == 'InputArgument(host=192.168.0.0)'
def test_repr_input_argument():
arg = InputArgument('host', value='192.168.0.0', founder_class=ValueArgument)
assert repr(arg) == "InputArgument<name=host, value=192.168.0.0, founder_class=ValueArgument>"
+12
View File
@@ -25,6 +25,18 @@ def test_parse_raw_command_with_repeated_flag_name():
InputCommand.parse('ssh --host 192.168.0.3 --host 172.198.0.43')
def test_parse_raw_command_with_triple_prefix():
assert InputCommand.parse(
'ssh ---host 192.168.0.0'
).input_flags.get_flag_by_name('host') == \
InputFlag('host', input_value='192.168.0.0', prefix='---')
def test_parse_raw_command_with_unprocessed_entity():
with pytest.raises(UnprocessedInputFlagException):
InputCommand.parse('ssh --host 192.168.0.3 9977')
def test_parse_empty_raw_command():
with pytest.raises(EmptyInputCommandException):
InputCommand.parse('')
+81
View File
@@ -0,0 +1,81 @@
from typing import Generator
from argenta import App, DataBridge, Router
from argenta.di.providers import SystemProvider
from argenta.orchestrator.argparser import ArgParser, ArgSpace
from argenta.response import ResponseStatus
from dishka import Container, make_container
import pytest
from argenta.response.entity import Response
from argenta.di.integration import FromDishka, _get_container_from_response, setup_dishka, _auto_inject_handlers
@pytest.fixture
def argparser() -> ArgParser:
return ArgParser(processed_args=[])
@pytest.fixture
def container(argparser: ArgParser) -> Generator[Container]:
container = make_container(SystemProvider(), context={ArgParser: argparser})
yield container
container.close()
def test_get_container_from_response(container: Container):
Response.patch_by_container(container)
response = Response(ResponseStatus.ALL_FLAGS_VALID)
assert _get_container_from_response((response,), {}) == container
def test_get_container_from_response4(container: Container):
Response.patch_by_container(container)
response = Response(ResponseStatus.ALL_FLAGS_VALID)
assert _get_container_from_response((object(), response,), {}) == container
def test_get_container_from_response2(container: Container):
delattr(Response, '__dishka_container__')
response = Response(ResponseStatus.ALL_FLAGS_VALID)
with pytest.raises(RuntimeError):
_get_container_from_response((response,), {})
def test_get_container_from_response3(container: Container):
Response.patch_by_container(container)
with pytest.raises(RuntimeError):
assert _get_container_from_response((), {}) == container
def test_setup_dishka(container: Container):
app = App()
router = Router()
@router.command('command')
def handler(res: Response, data_bridge: FromDishka[DataBridge]):
print(data_bridge)
app.include_router(router)
assert setup_dishka(app, container, auto_inject=True) is None
def test_setup_dishka2(container: Container):
app = App()
assert setup_dishka(app, container, auto_inject=False) is None
def test_auto_inject_handlers(container: Container):
Response.patch_by_container(container)
app = App()
router = Router()
@router.command('command')
def handler(res: Response, data_bridge: FromDishka[DataBridge]):
print(data_bridge)
app.include_router(router)
_auto_inject_handlers(app)
_auto_inject_handlers(app) # check idempotency
def test_get_from_container(container: Container):
assert isinstance(container.get(ArgSpace), ArgSpace)
def test_get_from_container2(container: Container):
assert isinstance(container.get(DataBridge), DataBridge)
+111
View File
@@ -1,7 +1,9 @@
import re
from sys import flags
from argenta.command.flag import Flag, InputFlag, PossibleValues
from argenta.command.flag.flags import Flags, InputFlags
import pytest
def test_get_string_entity():
@@ -114,3 +116,112 @@ def test_add_flags():
flags = Flags()
flags.add_flags([Flag('test'), Flag('test2')])
assert len(flags.flags) == 2
def test_eq_flags():
flags = Flags([Flag('some')])
flags2 = Flags([Flag('some')])
assert flags == flags2
def test_contains_flags():
flags = Flags([Flag('some')])
flag = Flag('some')
assert flag in flags
def test_eq_flags2():
flags = Flags([Flag('some')])
flags2 = Flags([Flag('other')])
assert flags != flags2
def test_eq_flags3():
flags = Flags([Flag('some')])
flags2 = Flags([Flag('some'), Flag('other')])
assert flags != flags2
def test_eq_flags4():
flags = Flags([Flag('some')])
not_flags = object()
assert flags != not_flags
def test_contains_flags2():
flags = Flags([Flag('some')])
flag = Flag('nonexists')
assert flag not in flags
def test_contains_flags3():
flags = Flags([Flag('some')])
not_flag = object
with pytest.raises(TypeError):
not_flag in flags # pyright: ignore[reportUnusedExpression]
def test_get_flag_by_name():
flags = Flags([Flag('some')])
assert flags.get_flag_by_name('some') == Flag('some')
def test_eq_input_flags3():
flags = InputFlags([InputFlag('some', input_value='')])
flags2 = InputFlags([
InputFlag('some', input_value=''),
InputFlag('some2', input_value='')
])
assert flags != flags2
def test_eq_input_flags4():
flags = InputFlags([InputFlag('some', input_value='')])
not_flags = object()
assert flags != not_flags
def test_contains_input_flags2():
flags = InputFlags([InputFlag('some', input_value='')])
flag = InputFlag('nonexists', input_value='')
assert flag not in flags
def test_contains_input_flags3():
flags = InputFlags([InputFlag('some', input_value='')])
not_flag = object
with pytest.raises(TypeError):
not_flag in flags # pyright: ignore[reportUnusedExpression]
def test_len_flags():
flags = Flags([Flag('one'), Flag('two')])
assert len(flags) == 2
def test_bool_flags():
flags = Flags([Flag('one'), Flag('two')])
assert bool(flags)
def test_bool_flags2():
flags = Flags([])
assert not bool(flags)
def test_getitem_flags():
flags = Flags([Flag('one'), Flag('two')])
assert flags[1] == Flag('two')
def test_str_flag():
flag = Flag('two')
assert str(flag) == '--two'
def test_repr_flag():
flag = Flag('two')
assert repr(flag) == 'Flag<name=two, prefix=-->'
def test_eq_flag():
flag = Flag('two')
not_flag = object()
with pytest.raises(NotImplementedError):
flag == not_flag # pyright: ignore[reportUnusedExpression]
def test_str_input_flag():
flag = InputFlag('two', input_value='value')
assert str(flag) == '--two value'
def test_repr_input_flag():
flag = InputFlag('two', input_value='some_value')
assert repr(flag) == 'InputFlag<name=two, prefix=--, value=some_value, status=None>'
def test_eq_input_flag():
flag = InputFlag('two', input_value='')
not_flag = object()
with pytest.raises(NotImplementedError):
flag == not_flag # pyright: ignore[reportUnusedExpression]
+49 -1
View File
@@ -1,7 +1,7 @@
import re
import pytest
from argenta.command import Command
from argenta.command import Command, InputCommand
from argenta.command.flag import Flag, InputFlag
from argenta.command.flag.flags import Flags, InputFlags
from argenta.command.flag.models import PossibleValues, ValidationStatus
@@ -102,3 +102,51 @@ def test_get_router_aliases3():
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 == ''