jyapayne/compose
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #3095 from dnephin/refactor_command_dispatch

Browse source 
Refactor command dispatch and fix api version mismatch error
This commit is contained in:
Aanand Prasad 2016-03-14 16:18:29 +00:00
commit e5cd869c61
8 changed files with 311 additions and 260 deletions
Download patch file Download diff file Expand all files Collapse all files

16
tests/unit/cli/command_test.py
View file

@ -4,28 +4,12 @@ from __future__ import unicode_literals
import os
import pytest
from requests.exceptions import ConnectionError
from compose.cli import errors
from compose.cli.command import friendly_error_message
from compose.cli.command import get_config_path_from_options
from compose.const import IS_WINDOWS_PLATFORM
from tests import mock
class TestFriendlyErrorMessage(object):
def test_dispatch_generic_connection_error(self):
with pytest.raises(errors.ConnectionErrorGeneric):
with mock.patch(
'compose.cli.command.call_silently',
autospec=True,
side_effect=[0, 1]
):
with friendly_error_message():
raise ConnectionError()
class TestGetConfigPathFromOptions(object):
def test_path_from_options(self):

51
tests/unit/cli/errors_test.py Normal file
View file

@ -0,0 +1,51 @@
from __future__ import absolute_import
from __future__ import unicode_literals
import pytest
from docker.errors import APIError
from requests.exceptions import ConnectionError
from compose.cli import errors
from compose.cli.errors import handle_connection_errors
from tests import mock
@pytest.yield_fixture
def mock_logging():
with mock.patch('compose.cli.errors.log', autospec=True) as mock_log:
yield mock_log
def patch_call_silently(side_effect):
return mock.patch(
'compose.cli.errors.call_silently',
autospec=True,
side_effect=side_effect)
class TestHandleConnectionErrors(object):
def test_generic_connection_error(self, mock_logging):
with pytest.raises(errors.ConnectionError):
with patch_call_silently([0, 1]):
with handle_connection_errors(mock.Mock()):
raise ConnectionError()
_, args, _ = mock_logging.error.mock_calls[0]
assert "Couldn't connect to Docker daemon at" in args[0]
def test_api_error_version_mismatch(self, mock_logging):
with pytest.raises(errors.ConnectionError):
with handle_connection_errors(mock.Mock(api_version='1.22')):
raise APIError(None, None, "client is newer than server")
_, args, _ = mock_logging.error.mock_calls[0]
assert "Docker Engine of version 1.10.0 or greater" in args[0]
def test_api_error_version_other(self, mock_logging):
msg = "Something broke!"
with pytest.raises(errors.ConnectionError):
with handle_connection_errors(mock.Mock(api_version='1.22')):
raise APIError(None, None, msg)
mock_logging.error.assert_called_once_with(msg)

67
tests/unit/cli/main_test.py
View file

@ -3,6 +3,8 @@ from __future__ import unicode_literals
import logging
import pytest
from compose import container
from compose.cli.errors import UserError
from compose.cli.formatter import ConsoleWarningFormatter
@ -11,7 +13,6 @@ from compose.cli.main import convergence_strategy_from_opts
from compose.cli.main import setup_console_handler
from compose.service import ConvergenceStrategy
from tests import mock
from tests import unittest
def mock_container(service, number):
@ -22,7 +23,14 @@ def mock_container(service, number):
name_without_project='{0}_{1}'.format(service, number))
class CLIMainTestCase(unittest.TestCase):
@pytest.fixture
def logging_handler():
stream = mock.Mock()
stream.isatty.return_value = True
return logging.StreamHandler(stream=stream)
class TestCLIMainTestCase(object):
def test_build_log_printer(self):
containers = [
@ -34,7 +42,7 @@ class CLIMainTestCase(unittest.TestCase):
]
service_names = ['web', 'db']
log_printer = build_log_printer(containers, service_names, True, False, {'follow': True})
self.assertEqual(log_printer.containers, containers[:3])
assert log_printer.containers == containers[:3]
def test_build_log_printer_all_services(self):
containers = [
@ -44,58 +52,53 @@ class CLIMainTestCase(unittest.TestCase):
]
service_names = []
log_printer = build_log_printer(containers, service_names, True, False, {'follow': True})
self.assertEqual(log_printer.containers, containers)
assert log_printer.containers == containers
class SetupConsoleHandlerTestCase(unittest.TestCase):
class TestSetupConsoleHandlerTestCase(object):
def setUp(self):
self.stream = mock.Mock()
self.stream.isatty.return_value = True
self.handler = logging.StreamHandler(stream=self.stream)
def test_with_tty_verbose(self, logging_handler):
setup_console_handler(logging_handler, True)
assert type(logging_handler.formatter) == ConsoleWarningFormatter
assert '%(name)s' in logging_handler.formatter._fmt
assert '%(funcName)s' in logging_handler.formatter._fmt
def test_with_tty_verbose(self):
setup_console_handler(self.handler, True)
assert type(self.handler.formatter) == ConsoleWarningFormatter
assert '%(name)s' in self.handler.formatter._fmt
assert '%(funcName)s' in self.handler.formatter._fmt
def test_with_tty_not_verbose(self, logging_handler):
setup_console_handler(logging_handler, False)
assert type(logging_handler.formatter) == ConsoleWarningFormatter
assert '%(name)s' not in logging_handler.formatter._fmt
assert '%(funcName)s' not in logging_handler.formatter._fmt
def test_with_tty_not_verbose(self):
setup_console_handler(self.handler, False)
assert type(self.handler.formatter) == ConsoleWarningFormatter
assert '%(name)s' not in self.handler.formatter._fmt
assert '%(funcName)s' not in self.handler.formatter._fmt
def test_with_not_a_tty(self):
self.stream.isatty.return_value = False
setup_console_handler(self.handler, False)
assert type(self.handler.formatter) == logging.Formatter
def test_with_not_a_tty(self, logging_handler):
logging_handler.stream.isatty.return_value = False
setup_console_handler(logging_handler, False)
assert type(logging_handler.formatter) == logging.Formatter
class ConvergeStrategyFromOptsTestCase(unittest.TestCase):
class TestConvergeStrategyFromOptsTestCase(object):
def test_invalid_opts(self):
options = {'--force-recreate': True, '--no-recreate': True}
with self.assertRaises(UserError):
with pytest.raises(UserError):
convergence_strategy_from_opts(options)
def test_always(self):
options = {'--force-recreate': True, '--no-recreate': False}
self.assertEqual(
convergence_strategy_from_opts(options),
assert (
convergence_strategy_from_opts(options) ==
ConvergenceStrategy.always
)
def test_never(self):
options = {'--force-recreate': False, '--no-recreate': True}
self.assertEqual(
convergence_strategy_from_opts(options),
assert (
convergence_strategy_from_opts(options) ==
ConvergenceStrategy.never
)
def test_changed(self):
options = {'--force-recreate': False, '--no-recreate': False}
self.assertEqual(
convergence_strategy_from_opts(options),
assert (
convergence_strategy_from_opts(options) ==
ConvergenceStrategy.changed
)

31
tests/unit/cli_test.py
View file

@ -64,26 +64,20 @@ class CLITestCase(unittest.TestCase):
self.assertTrue(project.client)
self.assertTrue(project.services)
def test_help(self):
command = TopLevelCommand()
with self.assertRaises(SystemExit):
command.dispatch(['-h'])
def test_command_help(self):
with self.assertRaises(SystemExit) as ctx:
TopLevelCommand().dispatch(['help', 'up'])
with pytest.raises(SystemExit) as exc:
TopLevelCommand.help({'COMMAND': 'up'})
self.assertIn('Usage: up', str(ctx.exception))
assert 'Usage: up' in exc.exconly()
def test_command_help_nonexistent(self):
with self.assertRaises(NoSuchCommand):
TopLevelCommand().dispatch(['help', 'nonexistent'])
with pytest.raises(NoSuchCommand):
TopLevelCommand.help({'COMMAND': 'nonexistent'})
@pytest.mark.xfail(IS_WINDOWS_PLATFORM, reason="requires dockerpty")
@mock.patch('compose.cli.main.RunOperation', autospec=True)
@mock.patch('compose.cli.main.PseudoTerminal', autospec=True)
def test_run_interactive_passes_logs_false(self, mock_pseudo_terminal, mock_run_operation):
command = TopLevelCommand()
mock_client = mock.create_autospec(docker.Client)
project = Project.from_config(
name='composetest',
@ -92,9 +86,10 @@ class CLITestCase(unittest.TestCase):
'service': {'image': 'busybox'}
}),
)
command = TopLevelCommand(project)
with pytest.raises(SystemExit):
command.run(project, {
command.run({
'SERVICE': 'service',
'COMMAND': None,
'-e': [],
@ -126,8 +121,8 @@ class CLITestCase(unittest.TestCase):
}),
)
command = TopLevelCommand()
command.run(project, {
command = TopLevelCommand(project)
command.run({
'SERVICE': 'service',
'COMMAND': None,
'-e': [],
@ -147,8 +142,8 @@ class CLITestCase(unittest.TestCase):
'always'
)
command = TopLevelCommand()
command.run(project, {
command = TopLevelCommand(project)
command.run({
'SERVICE': 'service',
'COMMAND': None,
'-e': [],
@ -168,7 +163,6 @@ class CLITestCase(unittest.TestCase):
)
def test_command_manula_and_service_ports_together(self):
command = TopLevelCommand()
project = Project.from_config(
name='composetest',
client=None,
@ -176,9 +170,10 @@ class CLITestCase(unittest.TestCase):
'service': {'image': 'busybox'},
}),
)
command = TopLevelCommand(project)
with self.assertRaises(UserError):
command.run(project, {
command.run({
'SERVICE': 'service',
'COMMAND': None,
'-e': [],