| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667 | from __future__ import absolute_importfrom __future__ import unicode_literalsimport pytestfrom docker.errors import APIErrorfrom requests.exceptions import ConnectionErrorfrom compose.cli import errorsfrom compose.cli.errors import handle_connection_errorsfrom tests import mock@pytest.yield_fixturedef mock_logging():    with mock.patch('compose.cli.errors.log', autospec=True) as mock_log:        yield mock_logdef patch_find_executable(side_effect):    return mock.patch(        'compose.cli.errors.find_executable',        autospec=True,        side_effect=side_effect)class TestHandleConnectionErrors(object):    def test_generic_connection_error(self, mock_logging):        with pytest.raises(errors.ConnectionError):            with patch_find_executable(['/bin/docker', None]):                with handle_connection_errors(mock.Mock()):                    raise ConnectionError()        _, args, _ = mock_logging.error.mock_calls[0]        assert "Couldn't connect to Docker daemon" 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, b"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_mismatch_unicode_explanation(self, mock_logging):        with pytest.raises(errors.ConnectionError):            with handle_connection_errors(mock.Mock(api_version='1.22')):                raise APIError(None, None, u"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 = b"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.decode('utf-8'))    def test_api_error_version_other_unicode_explanation(self, mock_logging):        msg = u"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)
 |