Python docker 模块，utils() 实例源码

def kwargs_from_client(client, assert_hostname=False):
    """
    More or less stolen from docker-py's kwargs_from_env
    https://github.com/docker/docker-py/blob/c0ec5512ae7ab90f7fac690064e37181186b1928/docker/utils/utils.py
    :type client : docker.Client
    """
    from docker import tls
    if client.base_url == 'http+docker://localunixsocket':
        return {'base_url': 'unix://var/run/docker.sock'}

    params = {'base_url': client.base_url}
    if client.cert:
        # TODO: problem - client.cert is filepaths, and it would be insecure to send those files.
        params['tls'] = tls.TLSConfig(
            client_cert=client.cert,
            ca_cert=client.verify,
            verify=bool(client.verify),
            assert_hostname=assert_hostname)

    return params

def setUp(self):
        self.logger = logging.getLogger(type(self).__name__)
        self.client = docker.Client(**docker.utils.kwargs_from_env(assert_hostname=False))
        options = mock.Mock()
        options.debug = False
        options.config = ['images.test-*.grace_time=1s']
        options.config_path = None
        options.image_gracetime = '1d'
        self.options = options
        self.containers = set()
        self.images = set()

def setUp(self):
        config = docker.utils.kwargs_from_env()
        config['version'] = "1.20"
        self.client = docker.Client(**config)
        self.assertTrue(self.client.images(name=TEST_IMAGE), "docker image %s not found" % TEST_IMAGE)

def docker_machine_client(machine_name='default'):
    import docker.utils

    env = docker_machine_env(machine_name)
    os.environ.update(env)
    client = get_docker_apiclient(**docker.utils.kwargs_from_env(assert_hostname=False))

    return client

def run_image(self):
        client = docker.from_env(version='auto')
        port = self._get_free_port()
        if not self.run_cmd:
            port = str(self._get_free_port())
            run_cmd = ['jupyter', 'notebook', '--ip', '0.0.0.0',
                       '--port', port]
            ports = {'%s/tcp' % port: port}
        else:
            run_cmd = self.run_cmd
            ports = {}
        container_volumes = {}
        if self.volumes:
            api_client = docker.APIClient(
                version='auto',
                **docker.utils.kwargs_from_env()
            )
            image = api_client.inspect_image(self.output_image_spec)
            image_workdir = image['ContainerConfig']['WorkingDir']

            for k, v in self.volumes.items():
                container_volumes[os.path.abspath(k)] = {
                    'bind': v if v.startswith('/') else os.path.join(image_workdir, v),
                    'mode': 'rw'
                }

        container = client.containers.run(
            self.output_image_spec,
            ports=ports,
            detach=True,
            command=run_cmd,
            volumes=container_volumes
        )
        while container.status == 'created':
            time.sleep(0.5)
            container.reload()

        try:
            for line in container.logs(stream=True):
                self.log.info(line.decode('utf-8'),
                              extra=dict(phase='running'))
        finally:
            container.reload()
            if container.status == 'running':
                self.log.info('Stopping container...\n',
                              extra=dict(phase='running'))
                container.kill()
            exit_code = container.attrs['State']['ExitCode']
            container.remove()
            sys.exit(exit_code)