test_wdocker.py 文件源码

def test__wait_for_port_with_unsupported_errors(self, time_sleep_mock, socket_mock):
        # GIVEN
        docker_container = DockerContainer(
            image='alpine',
            wait_for_port=1234,
        )

        docker_container._client = mock.MagicMock()
        docker_container._client.containers.get.return_value.attrs = dict(
            NetworkSettings=dict(IPAddress='172.10.0.2')
        )
        docker_container._container = mock.MagicMock(id='c5f0cad13259')
        docker_container._container.logs.return_value = b'\n'.join([
            b'some container error ...',
            b'container failed to start'
        ])

        socket_mock.return_value.connect_ex.side_effect = [1, 1, errno.EHOSTUNREACH]

        # WHEN
        with self.assertRaises(DockerContainerError) as cm:
            docker_container._wait_for_port()

        # THEN
        expected_message = '\n'.join([
            '[alpine] Host 172.10.0.2 cannot be reach. The container may exit abnormally. Container logs :',
            'some container error ...',
            'container failed to start'
        ])
        self.assertEquals(str(cm.exception), expected_message, 'Message should be explicit: %s' % expected_message)
        time_sleep_mock.assert_called()
        socket_mock.assert_called_with(socket.AF_INET, socket.SOCK_STREAM)
        connect_ex_mock = socket_mock.return_value.connect_ex
        connect_ex_mock.assert_called_with(('172.10.0.2', 1234))
        docker_container._container.logs.assert_called_once_with(stream=False)
        self.assertEqual(connect_ex_mock.call_count, 3, 'Should have been called 3 times instead of {call_nb}'.format(
            call_nb=connect_ex_mock.call_count
        ))