generated from daniil-berg/boilerplate-py
big rework of the session-parser-interaction;
dynamically adding pool methods/properties as parser commands; dynamically executing selected pool method/property; greatly simplified `ControlSession` class; removed the need for hard-coded command names; adjusted unittests accordingly
This commit is contained in:
@ -87,14 +87,6 @@ class HelpersTestCase(IsolatedAsyncioTestCase):
|
||||
self.assertIsNone(await helpers.join_queue(mock_queue))
|
||||
mock_join.assert_awaited_once_with()
|
||||
|
||||
def test_task_str(self):
|
||||
self.assertEqual("task", helpers.tasks_str(1))
|
||||
self.assertEqual("tasks", helpers.tasks_str(0))
|
||||
self.assertEqual("tasks", helpers.tasks_str(-1))
|
||||
self.assertEqual("tasks", helpers.tasks_str(2))
|
||||
self.assertEqual("tasks", helpers.tasks_str(-10))
|
||||
self.assertEqual("tasks", helpers.tasks_str(42))
|
||||
|
||||
def test_get_first_doc_line(self):
|
||||
expected_output = 'foo bar baz'
|
||||
mock_obj = MagicMock(__doc__=f"""{expected_output}
|
||||
|
||||
259
tests/test_parser.py
Normal file
259
tests/test_parser.py
Normal file
@ -0,0 +1,259 @@
|
||||
__author__ = "Daniil Fajnberg"
|
||||
__copyright__ = "Copyright © 2022 Daniil Fajnberg"
|
||||
__license__ = """GNU LGPLv3.0
|
||||
|
||||
This file is part of asyncio-taskpool.
|
||||
|
||||
asyncio-taskpool is free software: you can redistribute it and/or modify it under the terms of
|
||||
version 3.0 of the GNU Lesser General Public License as published by the Free Software Foundation.
|
||||
|
||||
asyncio-taskpool is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
|
||||
without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
|
||||
See the GNU Lesser General Public License for more details.
|
||||
|
||||
You should have received a copy of the GNU Lesser General Public License along with asyncio-taskpool.
|
||||
If not, see <https://www.gnu.org/licenses/>."""
|
||||
|
||||
__doc__ = """
|
||||
Unittests for the `asyncio_taskpool.control.parser` module.
|
||||
"""
|
||||
|
||||
|
||||
from argparse import ArgumentParser, HelpFormatter, ArgumentDefaultsHelpFormatter, RawTextHelpFormatter, SUPPRESS
|
||||
from inspect import signature
|
||||
from unittest import TestCase
|
||||
from unittest.mock import MagicMock, call, patch
|
||||
|
||||
from asyncio_taskpool import parser
|
||||
from asyncio_taskpool.exceptions import HelpRequested
|
||||
|
||||
|
||||
FOO, BAR = 'foo', 'bar'
|
||||
|
||||
|
||||
class ControlServerTestCase(TestCase):
|
||||
|
||||
def setUp(self) -> None:
|
||||
self.help_formatter_factory_patcher = patch.object(parser.ControlParser, 'help_formatter_factory')
|
||||
self.mock_help_formatter_factory = self.help_formatter_factory_patcher.start()
|
||||
self.mock_help_formatter_factory.return_value = RawTextHelpFormatter
|
||||
self.stream_writer, self.terminal_width = MagicMock(), 420
|
||||
self.kwargs = {
|
||||
'stream_writer': self.stream_writer,
|
||||
'terminal_width': self.terminal_width,
|
||||
parser.FORMATTER_CLASS: FOO
|
||||
}
|
||||
self.parser = parser.ControlParser(**self.kwargs)
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.help_formatter_factory_patcher.stop()
|
||||
|
||||
def test_help_formatter_factory(self):
|
||||
self.help_formatter_factory_patcher.stop()
|
||||
|
||||
class MockBaseClass(HelpFormatter):
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
terminal_width = 123456789
|
||||
cls = parser.ControlParser.help_formatter_factory(terminal_width, MockBaseClass)
|
||||
self.assertTrue(issubclass(cls, MockBaseClass))
|
||||
instance = cls('prog')
|
||||
self.assertEqual(terminal_width, getattr(instance, '_width'))
|
||||
|
||||
cls = parser.ControlParser.help_formatter_factory(terminal_width)
|
||||
self.assertTrue(issubclass(cls, ArgumentDefaultsHelpFormatter))
|
||||
instance = cls('prog')
|
||||
self.assertEqual(terminal_width, getattr(instance, '_width'))
|
||||
|
||||
def test_init(self):
|
||||
self.assertIsInstance(self.parser, ArgumentParser)
|
||||
self.assertEqual(self.stream_writer, self.parser._stream_writer)
|
||||
self.assertEqual(self.terminal_width, self.parser._terminal_width)
|
||||
self.mock_help_formatter_factory.assert_called_once_with(self.terminal_width, FOO)
|
||||
self.assertFalse(getattr(self.parser, 'exit_on_error'))
|
||||
self.assertEqual(RawTextHelpFormatter, getattr(self.parser, 'formatter_class'))
|
||||
self.assertSetEqual(set(), self.parser._flags)
|
||||
self.assertIsNone(self.parser._commands)
|
||||
|
||||
@patch.object(parser, 'get_first_doc_line')
|
||||
def test_add_function_command(self, mock_get_first_doc_line: MagicMock):
|
||||
def foo_bar(): pass
|
||||
mock_subparser = MagicMock()
|
||||
mock_add_parser = MagicMock(return_value=mock_subparser)
|
||||
self.parser._commands = MagicMock(add_parser=mock_add_parser)
|
||||
mock_get_first_doc_line.return_value = mock_help = 'help 123'
|
||||
kwargs = {FOO: 1, BAR: 2, parser.DESCRIPTION: FOO + BAR}
|
||||
expected_name = 'foo-bar'
|
||||
expected_kwargs = {parser.NAME: expected_name, parser.PROG: expected_name, parser.HELP: mock_help} | kwargs
|
||||
to_omit = ['abc', 'xyz']
|
||||
output = self.parser.add_function_command(foo_bar, omit_params=to_omit, **kwargs)
|
||||
self.assertEqual(mock_subparser, output)
|
||||
mock_add_parser.assert_called_once_with(**expected_kwargs)
|
||||
mock_subparser.add_function_args.assert_called_once_with(foo_bar, to_omit)
|
||||
|
||||
@patch.object(parser, 'get_first_doc_line')
|
||||
def test_add_property_command(self, mock_get_first_doc_line: MagicMock):
|
||||
def get_prop(_self): pass
|
||||
def set_prop(_self, _value): pass
|
||||
prop = property(get_prop)
|
||||
mock_subparser = MagicMock()
|
||||
mock_add_parser = MagicMock(return_value=mock_subparser)
|
||||
self.parser._commands = MagicMock(add_parser=mock_add_parser)
|
||||
mock_get_first_doc_line.return_value = mock_help = 'help 123'
|
||||
kwargs = {FOO: 1, BAR: 2, parser.DESCRIPTION: FOO + BAR}
|
||||
expected_name = 'get-prop'
|
||||
expected_kwargs = {parser.NAME: expected_name, parser.PROG: expected_name, parser.HELP: mock_help} | kwargs
|
||||
output = self.parser.add_property_command(prop, **kwargs)
|
||||
self.assertEqual(mock_subparser, output)
|
||||
mock_get_first_doc_line.assert_called_once_with(get_prop)
|
||||
mock_add_parser.assert_called_once_with(**expected_kwargs)
|
||||
mock_subparser.add_function_arg.assert_not_called()
|
||||
|
||||
mock_get_first_doc_line.reset_mock()
|
||||
mock_add_parser.reset_mock()
|
||||
|
||||
prop = property(get_prop, set_prop)
|
||||
expected_help = f"Get/set the `.{expected_name}` property"
|
||||
expected_kwargs = {parser.NAME: expected_name, parser.PROG: expected_name, parser.HELP: expected_help} | kwargs
|
||||
output = self.parser.add_property_command(prop, **kwargs)
|
||||
self.assertEqual(mock_subparser, output)
|
||||
mock_get_first_doc_line.assert_has_calls([call(get_prop), call(set_prop)])
|
||||
mock_add_parser.assert_called_once_with(**expected_kwargs)
|
||||
mock_subparser.add_function_arg.assert_called_once_with(
|
||||
tuple(signature(set_prop).parameters.values())[1],
|
||||
nargs='?',
|
||||
default=SUPPRESS,
|
||||
help=f"If provided: {mock_help} If omitted: {mock_help}"
|
||||
)
|
||||
|
||||
@patch.object(parser.ControlParser, 'add_property_command')
|
||||
@patch.object(parser.ControlParser, 'add_function_command')
|
||||
def test_add_class_commands(self, mock_add_function_command: MagicMock, mock_add_property_command: MagicMock):
|
||||
class FooBar:
|
||||
some_attribute = None
|
||||
|
||||
def _protected(self, _): pass
|
||||
|
||||
def __private(self, _): pass
|
||||
|
||||
def to_omit(self, _): pass
|
||||
|
||||
def method(self, _): pass
|
||||
|
||||
@property
|
||||
def prop(self): return None
|
||||
|
||||
mock_set_defaults = MagicMock()
|
||||
mock_subparser = MagicMock(set_defaults=mock_set_defaults)
|
||||
mock_add_function_command.return_value = mock_add_property_command.return_value = mock_subparser
|
||||
x = 'x'
|
||||
common_kwargs = {parser.STREAM_WRITER: self.parser._stream_writer,
|
||||
parser.CLIENT_INFO.TERMINAL_WIDTH: self.parser._terminal_width}
|
||||
expected_output = {'method': mock_subparser, 'prop': mock_subparser}
|
||||
output = self.parser.add_class_commands(FooBar, public_only=True, omit_members=['to_omit'], member_arg_name=x)
|
||||
self.assertDictEqual(expected_output, output)
|
||||
mock_add_function_command.assert_called_once_with(FooBar.method, **common_kwargs)
|
||||
mock_add_property_command.assert_called_once_with(FooBar.prop, FooBar.__name__, **common_kwargs)
|
||||
mock_set_defaults.assert_has_calls([call(**{x: FooBar.method}), call(**{x: FooBar.prop})])
|
||||
|
||||
def test__print_message(self):
|
||||
self.stream_writer.write = MagicMock()
|
||||
self.assertIsNone(self.parser._print_message(''))
|
||||
self.stream_writer.write.assert_not_called()
|
||||
msg = 'foo bar baz'
|
||||
self.assertIsNone(self.parser._print_message(msg))
|
||||
self.stream_writer.write.assert_called_once_with(msg.encode())
|
||||
|
||||
@patch.object(parser.ControlParser, '_print_message')
|
||||
def test_exit(self, mock__print_message: MagicMock):
|
||||
self.assertIsNone(self.parser.exit(123, ''))
|
||||
mock__print_message.assert_not_called()
|
||||
msg = 'foo bar baz'
|
||||
self.assertIsNone(self.parser.exit(123, msg))
|
||||
mock__print_message.assert_called_once_with(msg)
|
||||
|
||||
@patch.object(parser.ArgumentParser, 'error')
|
||||
def test_error(self, mock_supercls_error: MagicMock):
|
||||
with self.assertRaises(HelpRequested):
|
||||
self.parser.error(FOO + BAR)
|
||||
mock_supercls_error.assert_called_once_with(message=FOO + BAR)
|
||||
|
||||
@patch.object(parser.ArgumentParser, 'print_help')
|
||||
def test_print_help(self, mock_print_help: MagicMock):
|
||||
arg = MagicMock()
|
||||
with self.assertRaises(HelpRequested):
|
||||
self.parser.print_help(arg)
|
||||
mock_print_help.assert_called_once_with(arg)
|
||||
|
||||
@patch.object(parser, 'get_arg_type_wrapper')
|
||||
@patch.object(parser.ArgumentParser, 'add_argument')
|
||||
def test_add_function_arg(self, mock_add_argument: MagicMock, mock_get_arg_type_wrapper: MagicMock):
|
||||
mock_add_argument.return_value = expected_output = 'action'
|
||||
mock_get_arg_type_wrapper.return_value = mock_type = 'fake'
|
||||
|
||||
foo_type, args_type, bar_type, baz_type, boo_type = tuple, str, int, float, complex
|
||||
bar_default, baz_default, boo_default = 1, 0.1, 1j
|
||||
|
||||
def func(foo: foo_type, *args: args_type, bar: bar_type = bar_default, baz: baz_type = baz_default,
|
||||
boo: boo_type = boo_default, flag: bool = False):
|
||||
return foo, args, bar, baz, boo, flag
|
||||
|
||||
param_foo, param_args, param_bar, param_baz, param_boo, param_flag = signature(func).parameters.values()
|
||||
kwargs = {FOO + BAR: 'xyz'}
|
||||
self.assertEqual(expected_output, self.parser.add_function_arg(param_foo, **kwargs))
|
||||
mock_add_argument.assert_called_once_with('foo', type=mock_type, **kwargs)
|
||||
mock_get_arg_type_wrapper.assert_called_once_with(foo_type)
|
||||
|
||||
mock_add_argument.reset_mock()
|
||||
mock_get_arg_type_wrapper.reset_mock()
|
||||
|
||||
self.assertEqual(expected_output, self.parser.add_function_arg(param_args, **kwargs))
|
||||
mock_add_argument.assert_called_once_with('args', nargs='*', type=mock_type, **kwargs)
|
||||
mock_get_arg_type_wrapper.assert_called_once_with(args_type)
|
||||
|
||||
mock_add_argument.reset_mock()
|
||||
mock_get_arg_type_wrapper.reset_mock()
|
||||
|
||||
self.assertEqual(expected_output, self.parser.add_function_arg(param_bar, **kwargs))
|
||||
mock_add_argument.assert_called_once_with('-b', '--bar', default=bar_default, type=mock_type, **kwargs)
|
||||
mock_get_arg_type_wrapper.assert_called_once_with(bar_type)
|
||||
|
||||
mock_add_argument.reset_mock()
|
||||
mock_get_arg_type_wrapper.reset_mock()
|
||||
|
||||
self.assertEqual(expected_output, self.parser.add_function_arg(param_baz, **kwargs))
|
||||
mock_add_argument.assert_called_once_with('-B', '--baz', default=baz_default, type=mock_type, **kwargs)
|
||||
mock_get_arg_type_wrapper.assert_called_once_with(baz_type)
|
||||
|
||||
mock_add_argument.reset_mock()
|
||||
mock_get_arg_type_wrapper.reset_mock()
|
||||
|
||||
self.assertEqual(expected_output, self.parser.add_function_arg(param_boo, **kwargs))
|
||||
mock_add_argument.assert_called_once_with('--boo', default=boo_default, type=mock_type, **kwargs)
|
||||
mock_get_arg_type_wrapper.assert_called_once_with(boo_type)
|
||||
|
||||
mock_add_argument.reset_mock()
|
||||
mock_get_arg_type_wrapper.reset_mock()
|
||||
|
||||
self.assertEqual(expected_output, self.parser.add_function_arg(param_flag, **kwargs))
|
||||
mock_add_argument.assert_called_once_with('-f', '--flag', action='store_true', **kwargs)
|
||||
mock_get_arg_type_wrapper.assert_not_called()
|
||||
|
||||
@patch.object(parser.ControlParser, 'add_function_arg')
|
||||
def test_add_function_args(self, mock_add_function_arg: MagicMock):
|
||||
def func(foo: str, *args: int, bar: float = 0.1):
|
||||
return foo, args, bar
|
||||
_, param_args, param_bar = signature(func).parameters.values()
|
||||
self.assertIsNone(self.parser.add_function_args(func, omit=['foo']))
|
||||
mock_add_function_arg.assert_has_calls([
|
||||
call(param_args, help=repr(param_args.annotation)),
|
||||
call(param_bar, help=repr(param_bar.annotation)),
|
||||
])
|
||||
|
||||
|
||||
class RestTestCase(TestCase):
|
||||
def test_get_arg_type_wrapper(self):
|
||||
type_wrap = parser.get_arg_type_wrapper(int)
|
||||
self.assertEqual(SUPPRESS, type_wrap(SUPPRESS))
|
||||
self.assertEqual(13, type_wrap('13'))
|
||||
@ -25,9 +25,9 @@ from unittest import IsolatedAsyncioTestCase
|
||||
from unittest.mock import AsyncMock, MagicMock, patch, call
|
||||
|
||||
from asyncio_taskpool import session
|
||||
from asyncio_taskpool.constants import CLIENT_INFO, CMD, SESSION_MSG_BYTES, SESSION_WRITER
|
||||
from asyncio_taskpool.exceptions import HelpRequested, NotATaskPool, UnknownTaskPoolClass
|
||||
from asyncio_taskpool.pool import BaseTaskPool, TaskPool, SimpleTaskPool
|
||||
from asyncio_taskpool.constants import CLIENT_INFO, CMD, SESSION_MSG_BYTES, STREAM_WRITER
|
||||
from asyncio_taskpool.exceptions import HelpRequested
|
||||
from asyncio_taskpool.pool import SimpleTaskPool
|
||||
|
||||
|
||||
FOO, BAR = 'foo', 'bar'
|
||||
@ -61,236 +61,105 @@ class ControlServerTestCase(IsolatedAsyncioTestCase):
|
||||
self.assertEqual(self.mock_reader, self.session._reader)
|
||||
self.assertEqual(self.mock_writer, self.session._writer)
|
||||
self.assertIsNone(self.session._parser)
|
||||
self.assertIsNone(self.session._subparsers)
|
||||
|
||||
def test__add_command(self):
|
||||
expected_output = 123456
|
||||
mock_add_parser = MagicMock(return_value=expected_output)
|
||||
self.session._subparsers = MagicMock(add_parser=mock_add_parser)
|
||||
self.session._parser = MagicMock()
|
||||
name, prog, short_help, long_help = 'abc', None, 'short123', None
|
||||
kwargs = {'x': 1, 'y': 2}
|
||||
output = self.session._add_command(name, prog, short_help, long_help, **kwargs)
|
||||
self.assertEqual(expected_output, output)
|
||||
mock_add_parser.assert_called_once_with(name, prog=name, help=short_help, description=short_help,
|
||||
parent=self.session._parser, **kwargs)
|
||||
@patch.object(session, 'return_or_exception')
|
||||
async def test__exec_method_and_respond(self, mock_return_or_exception: AsyncMock):
|
||||
def method(self, arg1, arg2, *var_args, **rest): pass
|
||||
test_arg1, test_arg2, test_var_args, test_rest = 123, 'xyz', [0.1, 0.2, 0.3], {'aaa': 1, 'bbb': 11}
|
||||
kwargs = {'arg1': test_arg1, 'arg2': test_arg2, 'var_args': test_var_args} | test_rest
|
||||
mock_return_or_exception.return_value = None
|
||||
self.assertIsNone(await self.session._exec_method_and_respond(method, **kwargs))
|
||||
mock_return_or_exception.assert_awaited_once_with(
|
||||
method, self.mock_pool, test_arg1, test_arg2, *test_var_args, **test_rest
|
||||
)
|
||||
self.mock_writer.write.assert_called_once_with(session.CMD_OK)
|
||||
|
||||
mock_add_parser.reset_mock()
|
||||
@patch.object(session, 'return_or_exception')
|
||||
async def test__exec_property_and_respond(self, mock_return_or_exception: AsyncMock):
|
||||
def prop_get(_): pass
|
||||
def prop_set(_): pass
|
||||
prop = property(prop_get, prop_set)
|
||||
kwargs = {'value': 'something'}
|
||||
mock_return_or_exception.return_value = None
|
||||
self.assertIsNone(await self.session._exec_property_and_respond(prop, **kwargs))
|
||||
mock_return_or_exception.assert_awaited_once_with(prop_set, self.mock_pool, **kwargs)
|
||||
self.mock_writer.write.assert_called_once_with(session.CMD_OK)
|
||||
|
||||
prog, long_help = 'ffffff', 'so long, wow'
|
||||
output = self.session._add_command(name, prog, short_help, long_help, **kwargs)
|
||||
self.assertEqual(expected_output, output)
|
||||
mock_add_parser.assert_called_once_with(name, prog=prog, help=short_help, description=long_help,
|
||||
parent=self.session._parser, **kwargs)
|
||||
mock_return_or_exception.reset_mock()
|
||||
self.mock_writer.write.reset_mock()
|
||||
|
||||
mock_add_parser.reset_mock()
|
||||
mock_return_or_exception.return_value = val = 420.69
|
||||
self.assertIsNone(await self.session._exec_property_and_respond(prop))
|
||||
mock_return_or_exception.assert_awaited_once_with(prop_get, self.mock_pool)
|
||||
self.mock_writer.write.assert_called_once_with(str(val).encode())
|
||||
|
||||
short_help = None
|
||||
output = self.session._add_command(name, prog, short_help, long_help, **kwargs)
|
||||
self.assertEqual(expected_output, output)
|
||||
mock_add_parser.assert_called_once_with(name, prog=prog, help=long_help, description=long_help,
|
||||
parent=self.session._parser, **kwargs)
|
||||
|
||||
@patch.object(session, 'get_first_doc_line')
|
||||
@patch.object(session.ControlSession, '_add_command')
|
||||
def test__adding_commands(self, mock__add_command: MagicMock, mock_get_first_doc_line: MagicMock):
|
||||
self.assertIsNone(self.session._add_base_commands())
|
||||
mock__add_command.assert_called()
|
||||
mock_get_first_doc_line.assert_called()
|
||||
|
||||
mock__add_command.reset_mock()
|
||||
mock_get_first_doc_line.reset_mock()
|
||||
|
||||
self.assertIsNone(self.session._add_simple_commands())
|
||||
mock__add_command.assert_called()
|
||||
mock_get_first_doc_line.assert_called()
|
||||
|
||||
with self.assertRaises(NotImplementedError):
|
||||
self.session._add_advanced_commands()
|
||||
|
||||
@patch.object(session.ControlSession, '_add_simple_commands')
|
||||
@patch.object(session.ControlSession, '_add_advanced_commands')
|
||||
@patch.object(session.ControlSession, '_add_base_commands')
|
||||
@patch.object(session, 'CommandParser')
|
||||
def test__init_parser(self, mock_command_parser_cls: MagicMock, mock__add_base_commands: MagicMock,
|
||||
mock__add_advanced_commands: MagicMock, mock__add_simple_commands: MagicMock):
|
||||
mock_command_parser_cls.return_value = mock_parser = MagicMock()
|
||||
self.session._pool = TaskPool()
|
||||
width = 1234
|
||||
expected_parser_kwargs = {
|
||||
'prog': '',
|
||||
SESSION_WRITER: self.mock_writer,
|
||||
CLIENT_INFO.TERMINAL_WIDTH: width,
|
||||
}
|
||||
self.assertIsNone(self.session._init_parser(width))
|
||||
mock_command_parser_cls.assert_called_once_with(**expected_parser_kwargs)
|
||||
mock_parser.add_subparsers.assert_called_once_with(title="Commands", dest=CMD.CMD)
|
||||
mock__add_base_commands.assert_called_once_with()
|
||||
mock__add_advanced_commands.assert_called_once_with()
|
||||
mock__add_simple_commands.assert_not_called()
|
||||
|
||||
mock_command_parser_cls.reset_mock()
|
||||
mock_parser.add_subparsers.reset_mock()
|
||||
mock__add_base_commands.reset_mock()
|
||||
mock__add_advanced_commands.reset_mock()
|
||||
mock__add_simple_commands.reset_mock()
|
||||
|
||||
async def fake_coroutine(): pass
|
||||
|
||||
self.session._pool = SimpleTaskPool(fake_coroutine)
|
||||
self.assertIsNone(self.session._init_parser(width))
|
||||
mock_command_parser_cls.assert_called_once_with(**expected_parser_kwargs)
|
||||
mock_parser.add_subparsers.assert_called_once_with(title="Commands", dest=CMD.CMD)
|
||||
mock__add_base_commands.assert_called_once_with()
|
||||
mock__add_advanced_commands.assert_not_called()
|
||||
mock__add_simple_commands.assert_called_once_with()
|
||||
|
||||
mock_command_parser_cls.reset_mock()
|
||||
mock_parser.add_subparsers.reset_mock()
|
||||
mock__add_base_commands.reset_mock()
|
||||
mock__add_advanced_commands.reset_mock()
|
||||
mock__add_simple_commands.reset_mock()
|
||||
|
||||
class FakeTaskPool(BaseTaskPool):
|
||||
pass
|
||||
|
||||
self.session._pool = FakeTaskPool()
|
||||
with self.assertRaises(UnknownTaskPoolClass):
|
||||
self.session._init_parser(width)
|
||||
mock_command_parser_cls.assert_called_once_with(**expected_parser_kwargs)
|
||||
mock_parser.add_subparsers.assert_called_once_with(title="Commands", dest=CMD.CMD)
|
||||
mock__add_base_commands.assert_called_once_with()
|
||||
mock__add_advanced_commands.assert_not_called()
|
||||
mock__add_simple_commands.assert_not_called()
|
||||
|
||||
mock_command_parser_cls.reset_mock()
|
||||
mock_parser.add_subparsers.reset_mock()
|
||||
mock__add_base_commands.reset_mock()
|
||||
mock__add_advanced_commands.reset_mock()
|
||||
mock__add_simple_commands.reset_mock()
|
||||
|
||||
self.session._pool = MagicMock()
|
||||
with self.assertRaises(NotATaskPool):
|
||||
self.session._init_parser(width)
|
||||
mock_command_parser_cls.assert_called_once_with(**expected_parser_kwargs)
|
||||
mock_parser.add_subparsers.assert_called_once_with(title="Commands", dest=CMD.CMD)
|
||||
mock__add_base_commands.assert_called_once_with()
|
||||
mock__add_advanced_commands.assert_not_called()
|
||||
mock__add_simple_commands.assert_not_called()
|
||||
|
||||
@patch.object(session.ControlSession, '_init_parser')
|
||||
async def test_client_handshake(self, mock__init_parser: MagicMock):
|
||||
@patch.object(session, 'ControlParser')
|
||||
async def test_client_handshake(self, mock_parser_cls: MagicMock):
|
||||
mock_add_subparsers, mock_add_class_commands = MagicMock(), MagicMock()
|
||||
mock_parser = MagicMock(add_subparsers=mock_add_subparsers, add_class_commands=mock_add_class_commands)
|
||||
mock_parser_cls.return_value = mock_parser
|
||||
width = 5678
|
||||
msg = ' ' + json.dumps({CLIENT_INFO.TERMINAL_WIDTH: width, FOO: BAR}) + ' '
|
||||
mock_read = AsyncMock(return_value=msg.encode())
|
||||
self.mock_reader.read = mock_read
|
||||
self.mock_writer.drain = AsyncMock()
|
||||
expected_parser_kwargs = {
|
||||
STREAM_WRITER: self.mock_writer,
|
||||
CLIENT_INFO.TERMINAL_WIDTH: width,
|
||||
'prog': '',
|
||||
'usage': f'%(prog)s [-h] [{CMD}] ...'
|
||||
}
|
||||
expected_subparsers_kwargs = {
|
||||
'title': "Commands",
|
||||
'metavar': "(A command followed by '-h' or '--help' will show command-specific help.)"
|
||||
}
|
||||
self.assertIsNone(await self.session.client_handshake())
|
||||
self.assertEqual(mock_parser, self.session._parser)
|
||||
mock_read.assert_awaited_once_with(SESSION_MSG_BYTES)
|
||||
mock__init_parser.assert_called_once_with(width)
|
||||
mock_parser_cls.assert_called_once_with(**expected_parser_kwargs)
|
||||
mock_add_subparsers.assert_called_once_with(**expected_subparsers_kwargs)
|
||||
mock_add_class_commands.assert_called_once_with(self.mock_pool.__class__)
|
||||
self.mock_writer.write.assert_called_once_with(str(self.mock_pool).encode())
|
||||
self.mock_writer.drain.assert_awaited_once_with()
|
||||
|
||||
@patch.object(session, 'return_or_exception')
|
||||
async def test__write_function_output(self, mock_return_or_exception: MagicMock):
|
||||
self.mock_writer.write = MagicMock()
|
||||
mock_return_or_exception.return_value = None
|
||||
func, args, kwargs = MagicMock(), (1, 2, 3), {'a': 'A', 'b': 'B'}
|
||||
self.assertIsNone(await self.session._write_function_output(func, *args, **kwargs))
|
||||
mock_return_or_exception.assert_called_once_with(func, *args, **kwargs)
|
||||
self.mock_writer.write.assert_called_once_with(b"ok")
|
||||
|
||||
mock_return_or_exception.reset_mock()
|
||||
self.mock_writer.write.reset_mock()
|
||||
|
||||
mock_return_or_exception.return_value = output = MagicMock()
|
||||
self.assertIsNone(await self.session._write_function_output(func, *args, **kwargs))
|
||||
mock_return_or_exception.assert_called_once_with(func, *args, **kwargs)
|
||||
self.mock_writer.write.assert_called_once_with(str(output).encode())
|
||||
|
||||
@patch.object(session.ControlSession, '_write_function_output')
|
||||
async def test__cmd_name(self, mock__write_function_output: AsyncMock):
|
||||
self.assertIsNone(await self.session._cmd_name())
|
||||
mock__write_function_output.assert_awaited_once_with(self.mock_pool.__class__.__str__, self.session._pool)
|
||||
|
||||
@patch.object(session.ControlSession, '_write_function_output')
|
||||
async def test__cmd_pool_size(self, mock__write_function_output: AsyncMock):
|
||||
num = 12345
|
||||
kwargs = {session.NUM: num, FOO: BAR}
|
||||
self.assertIsNone(await self.session._cmd_pool_size(**kwargs))
|
||||
mock__write_function_output.assert_awaited_once_with(
|
||||
self.mock_pool.__class__.pool_size.fset, self.session._pool, num
|
||||
)
|
||||
|
||||
mock__write_function_output.reset_mock()
|
||||
|
||||
kwargs.pop(session.NUM)
|
||||
self.assertIsNone(await self.session._cmd_pool_size(**kwargs))
|
||||
mock__write_function_output.assert_awaited_once_with(
|
||||
self.mock_pool.__class__.pool_size.fget, self.session._pool
|
||||
)
|
||||
|
||||
@patch.object(session.ControlSession, '_write_function_output')
|
||||
async def test__cmd_num_running(self, mock__write_function_output: AsyncMock):
|
||||
self.assertIsNone(await self.session._cmd_num_running())
|
||||
mock__write_function_output.assert_awaited_once_with(
|
||||
self.mock_pool.__class__.num_running.fget, self.session._pool
|
||||
)
|
||||
|
||||
@patch.object(session.ControlSession, '_write_function_output')
|
||||
async def test__cmd_start(self, mock__write_function_output: AsyncMock):
|
||||
num = 12345
|
||||
kwargs = {session.NUM: num, FOO: BAR}
|
||||
self.assertIsNone(await self.session._cmd_start(**kwargs))
|
||||
mock__write_function_output.assert_awaited_once_with(self.mock_pool.start, num)
|
||||
|
||||
@patch.object(session.ControlSession, '_write_function_output')
|
||||
async def test__cmd_stop(self, mock__write_function_output: AsyncMock):
|
||||
num = 12345
|
||||
kwargs = {session.NUM: num, FOO: BAR}
|
||||
self.assertIsNone(await self.session._cmd_stop(**kwargs))
|
||||
mock__write_function_output.assert_awaited_once_with(self.mock_pool.stop, num)
|
||||
|
||||
@patch.object(session.ControlSession, '_write_function_output')
|
||||
async def test__cmd_stop_all(self, mock__write_function_output: AsyncMock):
|
||||
self.assertIsNone(await self.session._cmd_stop_all())
|
||||
mock__write_function_output.assert_awaited_once_with(self.mock_pool.stop_all)
|
||||
|
||||
@patch.object(session.ControlSession, '_write_function_output')
|
||||
async def test__cmd_func_name(self, mock__write_function_output: AsyncMock):
|
||||
self.assertIsNone(await self.session._cmd_func_name())
|
||||
mock__write_function_output.assert_awaited_once_with(
|
||||
self.mock_pool.__class__.func_name.fget, self.session._pool
|
||||
)
|
||||
|
||||
async def test__execute_command(self):
|
||||
mock_method = AsyncMock()
|
||||
cmd = 'this-is-a-test'
|
||||
setattr(self.session, '_cmd_' + cmd.replace('-', '_'), mock_method)
|
||||
kwargs = {FOO: BAR, 'hello': 'python'}
|
||||
self.assertIsNone(await self.session._execute_command(**{CMD.CMD: cmd}, **kwargs))
|
||||
mock_method.assert_awaited_once_with(**kwargs)
|
||||
|
||||
@patch.object(session.ControlSession, '_execute_command')
|
||||
async def test__parse_command(self, mock__execute_command: AsyncMock):
|
||||
@patch.object(session.ControlSession, '_exec_property_and_respond')
|
||||
@patch.object(session.ControlSession, '_exec_method_and_respond')
|
||||
async def test__parse_command(self, mock__exec_method_and_respond: AsyncMock,
|
||||
mock__exec_property_and_respond: AsyncMock):
|
||||
def method(_): pass
|
||||
prop = property(method)
|
||||
msg = 'asdf asd as a'
|
||||
kwargs = {FOO: BAR, 'hello': 'python'}
|
||||
mock_parse_args = MagicMock(return_value=Namespace(**kwargs))
|
||||
mock_parse_args = MagicMock(return_value=Namespace(**{CMD: method}, **kwargs))
|
||||
self.session._parser = MagicMock(parse_args=mock_parse_args)
|
||||
self.mock_writer.write = MagicMock()
|
||||
self.assertIsNone(await self.session._parse_command(msg))
|
||||
mock_parse_args.assert_called_once_with(msg.split(' '))
|
||||
self.mock_writer.write.assert_not_called()
|
||||
mock__execute_command.assert_awaited_once_with(**kwargs)
|
||||
mock__exec_method_and_respond.assert_awaited_once_with(method, **kwargs)
|
||||
mock__exec_property_and_respond.assert_not_called()
|
||||
|
||||
mock__execute_command.reset_mock()
|
||||
mock__exec_method_and_respond.reset_mock()
|
||||
mock_parse_args.reset_mock()
|
||||
|
||||
mock_parse_args = MagicMock(return_value=Namespace(**{CMD: prop}, **kwargs))
|
||||
self.session._parser = MagicMock(parse_args=mock_parse_args)
|
||||
self.mock_writer.write = MagicMock()
|
||||
self.assertIsNone(await self.session._parse_command(msg))
|
||||
mock_parse_args.assert_called_once_with(msg.split(' '))
|
||||
self.mock_writer.write.assert_not_called()
|
||||
mock__exec_method_and_respond.assert_not_called()
|
||||
mock__exec_property_and_respond.assert_awaited_once_with(prop, **kwargs)
|
||||
|
||||
mock__exec_property_and_respond.reset_mock()
|
||||
mock_parse_args.reset_mock()
|
||||
|
||||
mock_parse_args.side_effect = exc = ArgumentError(MagicMock(), "oops")
|
||||
self.assertIsNone(await self.session._parse_command(msg))
|
||||
mock_parse_args.assert_called_once_with(msg.split(' '))
|
||||
self.mock_writer.write.assert_called_once_with(str(exc).encode())
|
||||
mock__execute_command.assert_not_awaited()
|
||||
mock__exec_method_and_respond.assert_not_awaited()
|
||||
mock__exec_property_and_respond.assert_not_awaited()
|
||||
|
||||
self.mock_writer.write.reset_mock()
|
||||
mock_parse_args.reset_mock()
|
||||
@ -299,7 +168,8 @@ class ControlServerTestCase(IsolatedAsyncioTestCase):
|
||||
self.assertIsNone(await self.session._parse_command(msg))
|
||||
mock_parse_args.assert_called_once_with(msg.split(' '))
|
||||
self.mock_writer.write.assert_not_called()
|
||||
mock__execute_command.assert_not_awaited()
|
||||
mock__exec_method_and_respond.assert_not_awaited()
|
||||
mock__exec_property_and_respond.assert_not_awaited()
|
||||
|
||||
@patch.object(session.ControlSession, '_parse_command')
|
||||
async def test_listen(self, mock__parse_command: AsyncMock):
|
||||
|
||||
@ -1,134 +0,0 @@
|
||||
__author__ = "Daniil Fajnberg"
|
||||
__copyright__ = "Copyright © 2022 Daniil Fajnberg"
|
||||
__license__ = """GNU LGPLv3.0
|
||||
|
||||
This file is part of asyncio-taskpool.
|
||||
|
||||
asyncio-taskpool is free software: you can redistribute it and/or modify it under the terms of
|
||||
version 3.0 of the GNU Lesser General Public License as published by the Free Software Foundation.
|
||||
|
||||
asyncio-taskpool is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
|
||||
without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
|
||||
See the GNU Lesser General Public License for more details.
|
||||
|
||||
You should have received a copy of the GNU Lesser General Public License along with asyncio-taskpool.
|
||||
If not, see <https://www.gnu.org/licenses/>."""
|
||||
|
||||
__doc__ = """
|
||||
Unittests for the `asyncio_taskpool.session_parser` module.
|
||||
"""
|
||||
|
||||
|
||||
from argparse import Action, ArgumentParser, HelpFormatter, ArgumentDefaultsHelpFormatter, RawTextHelpFormatter
|
||||
from unittest import IsolatedAsyncioTestCase
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from asyncio_taskpool import session_parser
|
||||
from asyncio_taskpool.constants import SESSION_WRITER, CLIENT_INFO
|
||||
from asyncio_taskpool.exceptions import HelpRequested
|
||||
|
||||
|
||||
FOO = 'foo'
|
||||
|
||||
|
||||
class ControlServerTestCase(IsolatedAsyncioTestCase):
|
||||
|
||||
def setUp(self) -> None:
|
||||
self.help_formatter_factory_patcher = patch.object(session_parser.CommandParser, 'help_formatter_factory')
|
||||
self.mock_help_formatter_factory = self.help_formatter_factory_patcher.start()
|
||||
self.mock_help_formatter_factory.return_value = RawTextHelpFormatter
|
||||
self.session_writer, self.terminal_width = MagicMock(), 420
|
||||
self.kwargs = {
|
||||
SESSION_WRITER: self.session_writer,
|
||||
CLIENT_INFO.TERMINAL_WIDTH: self.terminal_width,
|
||||
session_parser.FORMATTER_CLASS: FOO
|
||||
}
|
||||
self.parser = session_parser.CommandParser(**self.kwargs)
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.help_formatter_factory_patcher.stop()
|
||||
|
||||
def test_help_formatter_factory(self):
|
||||
self.help_formatter_factory_patcher.stop()
|
||||
|
||||
class MockBaseClass(HelpFormatter):
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
terminal_width = 123456789
|
||||
cls = session_parser.CommandParser.help_formatter_factory(terminal_width, MockBaseClass)
|
||||
self.assertTrue(issubclass(cls, MockBaseClass))
|
||||
instance = cls('prog')
|
||||
self.assertEqual(terminal_width, getattr(instance, '_width'))
|
||||
|
||||
cls = session_parser.CommandParser.help_formatter_factory(terminal_width)
|
||||
self.assertTrue(issubclass(cls, ArgumentDefaultsHelpFormatter))
|
||||
instance = cls('prog')
|
||||
self.assertEqual(terminal_width, getattr(instance, '_width'))
|
||||
|
||||
def test_init(self):
|
||||
self.assertIsInstance(self.parser, ArgumentParser)
|
||||
self.assertEqual(self.session_writer, self.parser._session_writer)
|
||||
self.assertEqual(self.terminal_width, self.parser._terminal_width)
|
||||
self.mock_help_formatter_factory.assert_called_once_with(self.terminal_width, FOO)
|
||||
self.assertFalse(getattr(self.parser, 'exit_on_error'))
|
||||
self.assertEqual(RawTextHelpFormatter, getattr(self.parser, 'formatter_class'))
|
||||
|
||||
def test_session_writer(self):
|
||||
self.assertEqual(self.session_writer, self.parser.session_writer)
|
||||
|
||||
def test_terminal_width(self):
|
||||
self.assertEqual(self.terminal_width, self.parser.terminal_width)
|
||||
|
||||
def test__print_message(self):
|
||||
self.session_writer.write = MagicMock()
|
||||
self.assertIsNone(self.parser._print_message(''))
|
||||
self.session_writer.write.assert_not_called()
|
||||
msg = 'foo bar baz'
|
||||
self.assertIsNone(self.parser._print_message(msg))
|
||||
self.session_writer.write.assert_called_once_with(msg.encode())
|
||||
|
||||
@patch.object(session_parser.CommandParser, '_print_message')
|
||||
def test_exit(self, mock__print_message: MagicMock):
|
||||
self.assertIsNone(self.parser.exit(123, ''))
|
||||
mock__print_message.assert_not_called()
|
||||
msg = 'foo bar baz'
|
||||
self.assertIsNone(self.parser.exit(123, msg))
|
||||
mock__print_message.assert_called_once_with(msg)
|
||||
|
||||
@patch.object(session_parser.ArgumentParser, 'print_help')
|
||||
def test_print_help(self, mock_print_help: MagicMock):
|
||||
arg = MagicMock()
|
||||
with self.assertRaises(HelpRequested):
|
||||
self.parser.print_help(arg)
|
||||
mock_print_help.assert_called_once_with(arg)
|
||||
|
||||
def test_add_optional_num_argument(self):
|
||||
metavar = 'FOOBAR'
|
||||
action = self.parser.add_optional_num_argument(metavar=metavar)
|
||||
self.assertIsInstance(action, Action)
|
||||
self.assertEqual('?', action.nargs)
|
||||
self.assertEqual(1, action.default)
|
||||
self.assertEqual(int, action.type)
|
||||
self.assertEqual(metavar, action.metavar)
|
||||
num = 111
|
||||
kwargs = vars(self.parser.parse_args([f'{num}']))
|
||||
self.assertDictEqual({session_parser.NUM: num}, kwargs)
|
||||
|
||||
name = f'--{FOO}'
|
||||
nargs = '+'
|
||||
default = 1
|
||||
_type = float
|
||||
required = True
|
||||
dest = 'foo_bar'
|
||||
action = self.parser.add_optional_num_argument(name, nargs=nargs, default=default, type=_type,
|
||||
required=required, metavar=metavar, dest=dest)
|
||||
self.assertIsInstance(action, Action)
|
||||
self.assertEqual(nargs, action.nargs)
|
||||
self.assertEqual(default, action.default)
|
||||
self.assertEqual(_type, action.type)
|
||||
self.assertEqual(required, action.required)
|
||||
self.assertEqual(metavar, action.metavar)
|
||||
self.assertEqual(dest, action.dest)
|
||||
kwargs = vars(self.parser.parse_args([f'{num}', name, '1', '1.5']))
|
||||
self.assertDictEqual({session_parser.NUM: num, dest: [1.0, 1.5]}, kwargs)
|
||||
Reference in New Issue
Block a user