Skip to content

Commit

Permalink
add unittests for cli consume
Browse files Browse the repository at this point in the history 
Signed-off-by: Sebastian Wojciechowski <s.wojciechowski89@gmail.com>
[Jeremy Cline: Slight adjustment to a single test to check log output]
Signed-off-by: Jeremy Cline <jcline@redhat.com>
  • Loading branch information
@sebwoj @mergify
sebwoj authored and mergify[bot] committed Aug 28, 2018
1 parent 010380f commit 54d704e
Show file tree
Hide file tree
Showing 2 changed files with 266 additions and 10 deletions.
2 changes: 1 addition & 1 deletion fedora_messaging/cli.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ def consume(amqp_url, exchange, queue_name, routing_key, callback, app_name):
raise click.ClickException(
"Unable to parse the callback path ({}); the "
'expected format is "my_package.module:'
'callable_object"'.format(str(e))
'callable_object"'.format(callback_path)
)
try:
module = importlib.import_module(module)
Expand Down
274 changes: 265 additions & 9 deletions fedora_messaging/tests/unit/test_cli.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
from click.testing import CliRunner
import mock

from fedora_messaging import cli
from fedora_messaging import cli, exceptions

FIXTURES_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "../fixtures/"))
GOOD_CONF = os.path.join(FIXTURES_DIR, "good_conf.toml")
Expand All @@ -47,11 +47,13 @@ def test_no_conf(self):
class ConsumeCliTests(unittest.TestCase):
"""Unit tests for the 'consume' command of the CLI."""

def setUp(self):
self.runner = CliRunner()

@mock.patch("fedora_messaging.cli.api.consume")
def test_good_conf(self, mock_consume):
"""Assert providing a configuration file via the CLI works."""
runner = CliRunner()
result = runner.invoke(cli.cli, ["--conf=" + GOOD_CONF, "consume"])
result = self.runner.invoke(cli.cli, ["--conf=" + GOOD_CONF, "consume"])
mock_consume.assert_called_with(
echo, [{"exchange": "e", "queue_name": "q", "routing_key": "#"}]
)
Expand All @@ -60,8 +62,7 @@ def test_good_conf(self, mock_consume):
@mock.patch("fedora_messaging.cli.api.consume")
def test_conf_env_support(self, mock_consume):
"""Assert FEDORA_MESSAGING_CONF environment variable is supported."""
runner = CliRunner()
result = runner.invoke(
result = self.runner.invoke(
cli.cli, ["consume"], env={"FEDORA_MESSAGING_CONF": GOOD_CONF}
)
mock_consume.assert_called_with(
Expand All @@ -76,17 +77,272 @@ def test_bad_conf(self, mock_consume):
"Error: Invalid value: Configuration error: Failed to parse"
" {}: error at line 1, column 1".format(BAD_CONF)
)
runner = CliRunner()
result = runner.invoke(cli.cli, ["--conf=" + BAD_CONF, "consume"])
result = self.runner.invoke(cli.cli, ["--conf=" + BAD_CONF, "consume"])
self.assertEqual(2, result.exit_code)
self.assertIn(expected_err, result.output)

@mock.patch("fedora_messaging.cli.api.consume")
def test_missing_conf(self, mock_consume):
"""Assert a missing configuration file is reported."""
runner = CliRunner()
result = runner.invoke(cli.cli, ["--conf=thispathdoesnotexist", "consume"])
result = self.runner.invoke(cli.cli, ["--conf=thispathdoesnotexist", "consume"])
self.assertEqual(2, result.exit_code)
self.assertIn(
"Error: Invalid value: thispathdoesnotexist is not a file", result.output
)

@mock.patch.dict(
"fedora_messaging.config.conf",
{"amqp_url": "a", "bindings": None, "callback": "mod:callable"},
)
@mock.patch("fedora_messaging.cli.importlib")
@mock.patch("fedora_messaging.cli.api.consume")
def test_good_cli_bindings(self, mock_consume, mock_importlib):
"""Assert providing a bindings via the CLI works."""
cli_options = {"exchange": "e", "queue-name": "qn", "routing-key": "rk"}
mock_mod_with_callable = mock.Mock(spec=["callable"])
mock_importlib.import_module.return_value = mock_mod_with_callable
result = self.runner.invoke(
cli.cli,
[
"consume",
"--exchange=" + cli_options["exchange"],
"--queue-name=" + cli_options["queue-name"],
"--routing-key=" + cli_options["routing-key"],
],
)
mock_importlib.import_module.called_once_with("mod")
mock_consume.assert_called_once_with(
mock_mod_with_callable.callable,
[
{
"exchange": cli_options["exchange"],
"queue_name": cli_options["queue-name"],
"routing_key": cli_options["routing-key"],
}
],
)
self.assertEqual(0, result.exit_code)

@mock.patch.dict(
"fedora_messaging.config.conf",
{"amqp_url": "a", "bindings": "b", "callback": "mod:callable"},
)
@mock.patch("fedora_messaging.cli.importlib")
@mock.patch("fedora_messaging.cli.api.consume")
def test_no_cli_bindings(self, mock_consume, mock_importlib):
"""Assert providing a bindings via configuration works."""
mock_mod_with_callable = mock.Mock(spec=["callable"])
mock_importlib.import_module.return_value = mock_mod_with_callable
result = self.runner.invoke(cli.cli, ["consume"])
mock_importlib.import_module.called_once_with("mod")
mock_consume.assert_called_once_with(mock_mod_with_callable.callable, "b")
self.assertEqual(0, result.exit_code)

@mock.patch.dict(
"fedora_messaging.config.conf",
{"amqp_url": "a", "bindings": None, "callback": "mod:callable"},
)
@mock.patch("fedora_messaging.cli.importlib")
@mock.patch("fedora_messaging.cli.api.consume")
def test_wrong_cli_bindings(self, mock_consume, mock_importlib):
"""Assert providing improper bindings is reported."""
cli_options = {"queue-name": "qn", "routing-key": "rk"}
mock_mod_with_callable = mock.Mock(spec=["callable"])
mock_importlib.import_module.return_value = mock_mod_with_callable
result = self.runner.invoke(
cli.cli,
[
"consume",
"--queue-name=" + cli_options["queue-name"],
"--routing-key=" + cli_options["routing-key"],
],
)
mock_importlib.import_module.called_once_with("mod")
mock_consume.assert_not_called()
self.assertIn(
"You must define all three of exchange, queue_name and"
" routing_key, or none of them to use the configuration",
result.output,
)
self.assertEqual(1, result.exit_code)

@mock.patch.dict(
"fedora_messaging.config.conf",
{"amqp_url": "a", "bindings": None, "callback": "mod:callable"},
)
@mock.patch("fedora_messaging.cli.importlib")
@mock.patch("fedora_messaging.cli.api.consume")
def test_missing_cli_and_conf_bindings(self, mock_consume, mock_importlib):
"""Assert missing bindings via cli and in conf is reported."""
mock_mod_with_callable = mock.Mock(spec=["callable"])
mock_importlib.import_module.return_value = mock_mod_with_callable
result = self.runner.invoke(cli.cli, ["consume"])
mock_importlib.import_module.not_called()
mock_consume.assert_not_called()
self.assertIn(
"No bindings are defined in the configuration file"
" and none were provided as arguments!",
result.output,
)
self.assertEqual(1, result.exit_code)

@mock.patch.dict("fedora_messaging.config.conf", {"amqp_url": "a", "bindings": "b"})
@mock.patch("fedora_messaging.cli.importlib")
@mock.patch("fedora_messaging.cli.api.consume")
def test_good_cli_callable(self, mock_consume, mock_importlib):
"""Assert providing a callable via the CLI works."""
cli_options = {"callback": "mod:callable"}
mock_mod_with_callable = mock.Mock(spec=["callable"])
mock_importlib.import_module.return_value = mock_mod_with_callable
result = self.runner.invoke(
cli.cli, ["consume", "--callback=" + cli_options["callback"]]
)
mock_importlib.import_module.called_once_with("mod")
mock_consume.assert_called_once_with(mock_mod_with_callable.callable, "b")
self.assertEqual(0, result.exit_code)

@mock.patch.dict(
"fedora_messaging.config.conf",
{"amqp_url": "a", "bindings": "b", "callback": None},
)
@mock.patch("fedora_messaging.cli.importlib")
@mock.patch("fedora_messaging.cli.api.consume")
def test_missing_cli_and_conf_callable(self, mock_consume, mock_importlib):
"""Assert missing callable via cli and in conf is reported."""
mock_mod_with_callable = mock.Mock(spec=["callable"])
mock_importlib.import_module.return_value = mock_mod_with_callable
result = self.runner.invoke(cli.cli, ["consume"])
mock_importlib.import_module.not_called()
mock_consume.assert_not_called()
self.assertIn(
'"callback" must be the Python path to a' " callable to consume",
result.output,
)
self.assertEqual(1, result.exit_code)

@mock.patch.dict("fedora_messaging.config.conf", {"amqp_url": "a", "bindings": "b"})
@mock.patch("fedora_messaging.cli.importlib")
@mock.patch("fedora_messaging.cli.api.consume")
def test_cli_callable_wrong_format(self, mock_consume, mock_importlib):
"""Assert a wrong callable format is reported."""
cli_options = {"callback": "modcallable"}
mock_mod_with_callable = mock.Mock(spec=["callable"])
mock_importlib.import_module.return_value = mock_mod_with_callable
result = self.runner.invoke(
cli.cli, ["consume", "--callback=" + cli_options["callback"]]
)
mock_importlib.import_module.not_called()
mock_consume.assert_not_called()
self.assertIn(
"Unable to parse the callback path ({}); the "
'expected format is "my_package.module:'
'callable_object"'.format(cli_options["callback"]),
result.output,
)
self.assertEqual(1, result.exit_code)

@mock.patch.dict("fedora_messaging.config.conf", {"amqp_url": "a", "bindings": "b"})
@mock.patch("fedora_messaging.cli.importlib")
@mock.patch("fedora_messaging.cli.api.consume")
def test_cli_callable_import_failure(self, mock_consume, mock_importlib):
"""Assert module with callable import failure is reported."""
cli_options = {"callback": "mod:callable"}
error_message = "No module named 'mod'"
mock_importlib.import_module.side_effect = ImportError(error_message)
result = self.runner.invoke(
cli.cli, ["consume", "--callback=" + cli_options["callback"]]
)
mock_importlib.import_module.called_once_with("mod")
mock_consume.assert_not_called()
self.assertIn(
"Failed to import the callback module ({})".format(error_message),
result.output,
)
self.assertEqual(1, result.exit_code)

@mock.patch("fedora_messaging.cli.getattr")
@mock.patch.dict("fedora_messaging.config.conf", {"amqp_url": "a", "bindings": "b"})
@mock.patch("fedora_messaging.cli.importlib")
@mock.patch("fedora_messaging.cli.api.consume")
def test_callable_getattr_failure(self, mock_consume, mock_importlib, mock_getattr):
"""Assert finding callable in module failure is reported."""
cli_options = {"callback": "mod:callable"}
error_message = "module 'mod' has no attribute 'callable'"
mock_mod_with_callable = mock.Mock(spec=["callable"])
mock_importlib.import_module.return_value = mock_mod_with_callable
mock_getattr.side_effect = AttributeError(error_message)
result = self.runner.invoke(
cli.cli, ["consume", "--callback=" + cli_options["callback"]]
)
mock_importlib.import_module.called_once_with("mod")
mock_consume.assert_not_called()
self.assertIn(
"Unable to import {} ({}); is the package installed? The python path should "
'be in the format "my_package.module:callable_object"'.format(
cli_options["callback"], error_message
),
result.output,
)
self.assertEqual(1, result.exit_code)

@mock.patch.dict("fedora_messaging.config.conf", {"amqp_url": "a", "bindings": "b"})
@mock.patch("fedora_messaging.cli.importlib")
@mock.patch("fedora_messaging.cli.api.consume")
def test_consume_improper_callback_object(self, mock_consume, mock_importlib):
"""Assert improper callback object type failure is reported."""
cli_options = {"callback": "mod:callable"}
error_message = (
"Callback must be a class that implements __call__ or a function."
)
mock_mod_with_callable = mock.Mock(spec=["callable"])
mock_importlib.import_module.return_value = mock_mod_with_callable
mock_consume.side_effect = ValueError(error_message)
result = self.runner.invoke(
cli.cli, ["consume", "--callback=" + cli_options["callback"]]
)
mock_importlib.import_module.called_once_with("mod")
mock_consume.assert_called_once_with(mock_mod_with_callable.callable, "b")
self.assertIn(error_message, result.output)
self.assertEqual(2, result.exit_code)

@mock.patch.dict("fedora_messaging.config.conf", {"amqp_url": "a", "bindings": "b"})
@mock.patch("fedora_messaging.cli.importlib")
@mock.patch("fedora_messaging.cli.api.consume")
def test_consume_halt_with_exitcode(self, mock_consume, mock_importlib):
"""Assert user execution halt with reason and exit_code is reported."""
cli_options = {"callback": "mod:callable"}
halt_message = "User halted execution"
halt_exit_code = 5
mock_mod_with_callable = mock.Mock(spec=["callable"])
mock_importlib.import_module.return_value = mock_mod_with_callable
mock_consume.side_effect = exceptions.HaltConsumer(
exit_code=halt_exit_code, reason=halt_message
)
with mock.patch("fedora_messaging.cli._log") as mock_log:
result = self.runner.invoke(
cli.cli, ["consume", "--callback=" + cli_options["callback"]]
)
mock_importlib.import_module.called_once_with("mod")
mock_consume.assert_called_once_with(mock_mod_with_callable.callable, "b")
mock_log.error.assert_called_once_with(
"Consumer halted with non-zero exit code (%d): %s",
5,
"User halted execution",
)
self.assertEqual(halt_exit_code, result.exit_code)

@mock.patch.dict("fedora_messaging.config.conf", {"amqp_url": "a", "bindings": "b"})
@mock.patch("fedora_messaging.cli.importlib")
@mock.patch("fedora_messaging.cli.api.consume")
def test_consume_halt_without_exitcode(self, mock_consume, mock_importlib):
"""Assert user execution halt is reported."""
cli_options = {"callback": "mod:callable"}
mock_mod_with_callable = mock.Mock(spec=["callable"])
mock_importlib.import_module.return_value = mock_mod_with_callable
mock_consume.side_effect = exceptions.HaltConsumer(exit_code=0)
result = self.runner.invoke(
cli.cli, ["consume", "--callback=" + cli_options["callback"]]
)
mock_importlib.import_module.called_once_with("mod")
mock_consume.assert_called_once_with(mock_mod_with_callable.callable, "b")
self.assertEqual(0, result.exit_code)

0 comments on commit 54d704e

Please sign in to comment.