def __init__(self):
        """ Connnects to the docker daemon"""
        # will be used as the tag on the docker image
        self.problem_name = sanitize_name(self.name)
        # use an explicit remote docker daemon per the configuration
        try:
            tls_config = docker.tls.TLSConfig(
                ca_cert=self.docker_ca_cert,
                client_cert=(self.docker_client_cert, self.docker_client_key),
                verify=True)

            self.client = docker.DockerClient(base_url=self.docker_host, tls=tls_config)
            self.api_client = docker.APIClient(base_url=self.docker_host, tls=tls_config)
            logger.debug("Connecting to docker daemon with config")

        # Docker options not set in configuration so use the environment to
        # configure (could be local or remote)
        except AttributeError:
            logger.debug("Connecting to docker daemon with env")
            self.client = docker.from_env()

        # throws an exception if the server returns an error: docker.errors.APIError
        self.client.ping() 

def test_docker_tasker(autoversion, base_url_arg):
    mock_docker()
    base_url = 'unix://var/run/docker.sock'
    kwargs = {}
    if base_url_arg:
        kwargs['base_url'] = base_url
    else:
        os.environ['DOCKER_CONNECTION'] = base_url

    expected_kwargs = {'base_url': base_url, 'timeout': 120}
    if autoversion:
        setattr(docker, 'AutoVersionClient', 'auto')
        expected_kwargs['version'] = 'auto'

    (flexmock(docker.APIClient)
        .should_receive('__init__')
        .with_args(**expected_kwargs)
        .once())

    DockerTasker(**kwargs)

    os.environ.pop('DOCKER_CONNECTION', None)
    if autoversion:
        delattr(docker, 'AutoVersionClient') 

def test_timeout(timeout, expected_timeout):
    if not hasattr(docker, 'APIClient'):
        setattr(docker, 'APIClient', docker.Client)

    expected_kwargs = {
        'timeout': expected_timeout
    }
    if hasattr(docker, 'AutoVersionClient'):
        expected_kwargs['version'] = 'auto'

    (flexmock(docker.APIClient)
        .should_receive('__init__')
        .with_args(**expected_kwargs)
        .once())

    kwargs = {}
    if timeout is not None:
        kwargs['timeout'] = timeout

    DockerTasker(**kwargs) 

def test_rpmqa_plugin_skip(docker_tasker):  # noqa
    """
    Test skipping the plugin if workflow.image_components is already set
    """
    mock_docker()
    workflow = DockerBuildWorkflow(TEST_IMAGE, source=SOURCE)
    workflow.source = StubSource()
    workflow.builder = StubInsideBuilder().for_workflow(workflow)

    image_components = {
        'type': 'rpm',
        'name': 'something'
    }
    setattr(workflow, 'image_components', image_components)

    flexmock(docker.APIClient, logs=mock_logs_raise)
    runner = PostBuildPluginsRunner(docker_tasker, workflow,
                                    [{"name": PostBuildRPMqaPlugin.key,
                                      "args": {'image_id': TEST_IMAGE}}])
    results = runner.run()
    assert results[PostBuildRPMqaPlugin.key] is None
    assert workflow.image_components == image_components 

def test_no_auto_remove(self, types_mock, client_class_mock):

        mock_obj = mock.Mock()

        client_mock = mock.Mock(spec=APIClient)
        client_mock.create_service.return_value = {'ID': 'some_id'}
        client_mock.images.return_value = []
        client_mock.pull.return_value = [b'{"status":"pull log"}']
        client_mock.tasks.return_value = [{'Status': {'State': 'complete'}}]
        types_mock.TaskTemplate.return_value = mock_obj
        types_mock.ContainerSpec.return_value = mock_obj
        types_mock.RestartPolicy.return_value = mock_obj
        types_mock.Resources.return_value = mock_obj

        client_class_mock.return_value = client_mock

        operator = DockerSwarmOperator(image='', auto_remove=False, task_id='unittest', enable_logging=False)
        operator.execute(None)

        self.assertEqual(
            client_mock.remove_service.call_count, 0,
            'Docker service being removed even when `auto_remove` set to `False`'
        ) 

def test_failed_service_raises_error(self, types_mock, client_class_mock):

        mock_obj = mock.Mock()

        client_mock = mock.Mock(spec=APIClient)
        client_mock.create_service.return_value = {'ID': 'some_id'}
        client_mock.images.return_value = []
        client_mock.pull.return_value = [b'{"status":"pull log"}']
        client_mock.tasks.return_value = [{'Status': {'State': 'failed'}}]
        types_mock.TaskTemplate.return_value = mock_obj
        types_mock.ContainerSpec.return_value = mock_obj
        types_mock.RestartPolicy.return_value = mock_obj
        types_mock.Resources.return_value = mock_obj

        client_class_mock.return_value = client_mock

        operator = DockerSwarmOperator(image='', auto_remove=False, task_id='unittest', enable_logging=False)
        msg = "Service failed: {'ID': 'some_id'}"
        with self.assertRaises(AirflowException) as error:
            operator.execute(None)
        self.assertEqual(str(error.exception), msg) 

def test_execute_tls(self, client_class_mock, tls_class_mock):
        client_mock = mock.Mock(spec=APIClient)
        client_mock.create_container.return_value = {'Id': 'some_id'}
        client_mock.create_host_config.return_value = mock.Mock()
        client_mock.images.return_value = []
        client_mock.attach.return_value = []
        client_mock.pull.return_value = []
        client_mock.wait.return_value = {"StatusCode": 0}

        client_class_mock.return_value = client_mock
        tls_mock = mock.Mock()
        tls_class_mock.return_value = tls_mock

        operator = DockerOperator(docker_url='tcp://127.0.0.1:2376', image='ubuntu',
                                  owner='unittest', task_id='unittest', tls_client_cert='cert.pem',
                                  tls_ca_cert='ca.pem', tls_client_key='key.pem')
        operator.execute(None)

        tls_class_mock.assert_called_once_with(assert_hostname=None, ca_cert='ca.pem',
                                               client_cert=('cert.pem', 'key.pem'),
                                               ssl_version=None, verify=True)

        client_class_mock.assert_called_once_with(base_url='https://127.0.0.1:2376',
                                                  tls=tls_mock, version=None) 

def __init__(self):
        self._containers = None
        self._images = None  # displayed images
        self._all_images = None  # docker images -a
        self._df = None

        kwargs = {"version": "auto"}
        kwargs.update(docker.utils.kwargs_from_env(assert_hostname=False))

        try:
            APIClientClass = docker.Client  # 1.x
        except AttributeError:
            APIClientClass = docker.APIClient  # 2.x

        try:
            self.client = APIClientClass(**kwargs)
        except docker.errors.DockerException as ex:
            raise TerminateApplication("can't establish connection to docker daemon: {0}".format(str(ex)))

        self.scratch_image = RootImage(self)

    # backend queries 

def __init__(self, image=DEFAULT_IMAGE, container_timeout=3600,
                 container_expire=3000, container_prefix=None,
                 docker_client=None):
        self._image = image
        self._container_timeout = container_timeout
        self._container_expire = container_expire

        if container_prefix:
            self._container_prefix = container_prefix
        elif os.environ.get(CONTAINER_PREFIX_ENV):
            self._container_prefix = os.environ[CONTAINER_PREFIX_ENV]
        else:
            self._container_prefix = DEFAULT_CONTAINER_PREFIX

        self._docker_client = docker_client or docker.APIClient(
            **docker.utils.kwargs_from_env(assert_hostname=False)
        )
        self._lock = threading.RLock()
        self._reset_container() 

def prepare_image(update, service):
    """
    Prepare a Docker image for execution.

    This is usually the longest operation during a chute installation, so
    instead of running this step in the update thread, we spin off a worker
    thread and return a Deferred. This will suspend processing of the current
    update until the worker thread finishes.
    """
    client = docker.APIClient(base_url="unix://var/run/docker.sock", version='auto')

    image_name = service.get_image_name()

    def call(func, *args, **kwargs):
        return func(*args, **kwargs)

    if settings.CONCURRENT_BUILDS:
        wrapper = deferToThread
    else:
        wrapper = call

    if service.type == "image":
        return wrapper(_pull_image, update, client, image_name)

    elif service.type == "inline":
        return wrapper(_build_image, update, service, client, True,
                rm=True, tag=image_name, fileobj=service.dockerfile)

    else:
        return wrapper(_build_image, update, service, client, False,
                rm=True, tag=image_name, path=update.workdir) 

def remove_image(update, service):
    """
    Remove a Docker image.
    """
    client = docker.APIClient(base_url="unix://var/run/docker.sock", version='auto')

    image_name = service.get_image_name()
    out.info("Removing image {}\n".format(image_name))

    try:
        client = docker.DockerClient(base_url="unix://var/run/docker.sock",
                version='auto')
        client.images.remove(image=image_name)
    except Exception as error:
        out.warn("Error removing image: {}".format(error)) 

def inspect(self):
        """
        Return the full container status from Docker.
        """
        client = docker.APIClient(base_url=self.docker_url, version='auto')
        try:
            info = client.inspect_container(self.name)
            return info
        except docker.errors.NotFound:
            raise ChuteNotFound("The chute could not be found.") 

def _new_exec_client(self, container, token, uuid, exec_id):
        exec_instance = None
        for e in container.exec_instances:
            if token == e.token and exec_id == e.exec_id:
                exec_instance = e

        if not exec_instance:
            raise exception.InvalidWebsocketToken(token)

        access_url = '%s?token=%s&uuid=%s' % (CONF.websocket_proxy.base_url,
                                              token, uuid)

        self._verify_origin(access_url)

        client = docker.APIClient(base_url=exec_instance.url)
        tsock = client.exec_start(exec_id, socket=True, tty=True)
        if hasattr(tsock, "_sock"):
            # NOTE(hongbin): dockerpy returns different socket class depending
            # on python version and base_url (see _get_raw_response_socket) so
            # we need to handle it in here.
            tsock = tsock._sock

        try:
            self.do_proxy(tsock)
        finally:
            if tsock:
                tsock.shutdown(socket.SHUT_RDWR)
                tsock.close()
                self.vmsg(_("%s: Closed target") % exec_instance.url) 

def setUp(self):
        super(BaseFullStackTestCase, self).setUp()

        self.docker = docker.APIClient(base_url='tcp://0.0.0.0:2375')
        try:
            self.zun = utils.get_zun_client_from_env()
        except Exception as e:
            # We may missing or didn't source configured openrc file.
            message = ("Missing environment variable %s in your local."
                       "Please add it and also check other missing "
                       "environment variables. After that please source "
                       "the openrc file. "
                       "Trying credentials from DevStack cloud.yaml ...")
            LOG.warning(message, e.args[0])
            self.zun = utils.get_zun_client_from_creds() 

def get_clients():
    """
    Get a high level and a low level docker client connection,

    Ensures that only one global docker client exists per thread. If the client
    does not exist a new one is created and returned.
    """

    global __client, __api_client
    if not __client or not __api_client:
        try:
            conf = current_app.config

            # use an explicit remote docker daemon per the configuration
            opts = ["DOCKER_HOST", "DOCKER_CA", "DOCKER_CLIENT", "DOCKER_KEY"]
            if all([o in conf for o in opts]):
                host, ca, client, key = [conf[o] for o in opts]

                log.debug("Connecting to docker daemon with config")
                tls_config = docker.tls.TLSConfig(ca_cert=ca, client_cert=(client, key), verify=True)
                __api_client = docker.APIClient(base_url=host, tls=tls_config)
                __client = docker.DockerClient(base_url=host, tls=tls_config)

            # Docker options not set in configuration so attempt to use unix socket
            else:
                log.debug("Connecting to docker daemon on local unix socket")
                __api_client = docker.APIClient(base_url="unix:///var/run/docker.sock")
                __client = docker.DockerClient(base_url="unix:///var/run/docker.sock")

            # ensure a responsive connection
            __client.ping()
        except docker.errors.APIError as e:
            log.debug("Could not connect to docker daemon:" + e)
            raise PicoException(
                "On Demand backend unavailible. Please contact an admin."
            )

    return __client, __api_client 

def __init__(self, job_backend, cpu_cores=1, gpu_devices=None, docker_container=None):
        Thread.__init__(self)

        self.job_backend = job_backend
        self.gpu_devices = gpu_devices
        self.docker_container = docker_container
        self.max_minutes = 0
        self.cpu_cores = cpu_cores

        job = self.job_backend.job
        if 'maxTime' in job['config'] and isinstance(job['config']['maxTime'], int) and job['config']['maxTime'] > 0:
            self.max_minutes = job['config']['maxTime']

        self.hardware_stream = self.job_backend.git.stream_file('aetros/job/monitoring.csv')

        header = ["second", "cpu", "memory"]
        try:
            if self.gpu_devices:
                for gpu_id, gpu in enumerate(aetros.cuda_gpu.get_ordered_devices()):
                    if gpu_id in gpu_devices:
                        header.append("memory_gpu" + str(gpu['id']))
        except aetros.cuda_gpu.CudaNotImplementedException: pass

        if job_backend.get_job_model().has_dpu():
            header += ['dpu0']

        self.hardware_stream.write(simplejson.dumps(header)[1:-1] + "\n")
        self.running = True
        self.early_stopped = False
        self.handle_max_time = True
        self.client = docker.from_env()
        self.docker_api = docker.APIClient(**docker.utils.kwargs_from_env())
        self.stat_stream = None

        self.docker_last_last_reponse = None
        self.docker_last_stream_data = 0
        self.docker_last_mem = None
        self.docker_last_cpu = None 

def docker():
    docker = APIClient(version='auto')
    return docker 

def __init__(self, **kwargs):
        self.retry_times = kwargs.pop('retry', None)

        try:
            # docker-py 2.x
            self.wrapped = docker.APIClient(**kwargs)
        except AttributeError:
            # docker-py 1.x
            self.wrapped = docker.Client(**kwargs) 

def test_wait_for_command():
    if MOCK:
        mock_docker()

    d = docker.APIClient()
    logs_gen = d.pull(INPUT_IMAGE, decode=True, stream=True)
    assert wait_for_command(logs_gen) is not None 

def test_tag_image_same_name(temp_image_name, docker_tasker):  # noqa
    if MOCK:
        mock_docker()

    temp_image_name.registry = "somewhere.example.com"
    temp_image_name.tag = "1"

    flexmock(docker.APIClient).should_receive('tag').never()
    docker_tasker.tag_image(temp_image_name, temp_image_name.copy()) 

def test_login(tmpdir, docker_tasker, dockerconfig_contents, should_raise):
    if MOCK:
        mock_docker()
        fake_api = flexmock(docker.APIClient, login=lambda username, registry,
                            dockercfg_path: {'Status': 'Login Succeeded'})

    tmpdir_path = str(tmpdir.realpath())
    file_name = '.dockercfg'
    dockercfg_path = os.path.join(tmpdir_path, file_name)
    with open(dockercfg_path, "w+") as dockerconfig:
        dockerconfig.write(json.dumps(dockerconfig_contents))
        dockerconfig.flush()
    if should_raise:
        if 'auth' in dockerconfig_contents[LOCALHOST_REGISTRY]:
            with pytest.raises(ValueError) as exc:
                docker_tasker.login(LOCALHOST_REGISTRY, tmpdir_path)
                assert "Failed to parse 'auth'" in str(exc.value)
        else:
            with pytest.raises(RuntimeError) as exc:
                docker_tasker.login(LOCALHOST_REGISTRY, tmpdir_path)
                assert "Failed to extract a username" in str(exc.value)
    else:
        if MOCK:
            (fake_api
             .should_receive('login')
             .with_args(username='user', registry=LOCALHOST_REGISTRY, dockercfg_path=dockercfg_path)
             .once().and_return({'Status': 'Login Succeeded'}))
        docker_tasker.login(LOCALHOST_REGISTRY, tmpdir_path) 

def test_rpmqa_plugin(caplog, docker_tasker, base_from_scratch, remove_container_error,
                      ignore_autogenerated):
    should_raise_error = {}
    if remove_container_error:
        should_raise_error['remove_container'] = None
    mock_docker(should_raise_error=should_raise_error)

    workflow = DockerBuildWorkflow(TEST_IMAGE, source=SOURCE)
    workflow.source = StubSource()
    workflow.builder = StubInsideBuilder().for_workflow(workflow)
    workflow.builder.set_base_from_scratch(base_from_scratch)

    flexmock(docker.APIClient, logs=mock_logs)
    runner = PostBuildPluginsRunner(
        docker_tasker,
        workflow,
        [{"name": PostBuildRPMqaPlugin.key,
          "args": {
              'image_id': TEST_IMAGE,
              "ignore_autogenerated_gpg_keys": ignore_autogenerated["ignore"]}}
         ])
    results = runner.run()
    if base_from_scratch:
        log_msg = "from scratch can't run rpmqa"
        assert log_msg in caplog.text
        assert results[PostBuildRPMqaPlugin.key] is None
        assert workflow.image_components is None
    else:
        assert results[PostBuildRPMqaPlugin.key] == ignore_autogenerated["package_list"]
        assert workflow.image_components == parse_rpm_output(ignore_autogenerated["package_list"]) 

def test_rpmqa_plugin_exception(docker_tasker):  # noqa
    mock_docker()
    workflow = DockerBuildWorkflow(TEST_IMAGE, source=SOURCE)
    workflow.source = StubSource()
    workflow.builder = StubInsideBuilder().for_workflow(workflow)

    flexmock(docker.APIClient, logs=mock_logs_raise)
    runner = PostBuildPluginsRunner(docker_tasker, workflow,
                                    [{"name": PostBuildRPMqaPlugin.key,
                                      "args": {'image_id': TEST_IMAGE}}])
    with pytest.raises(PluginFailedException):
        runner.run() 

def test_empty_logs_retry(docker_tasker):  # noqa
    mock_docker()
    workflow = DockerBuildWorkflow(TEST_IMAGE, source=SOURCE)
    workflow.source = StubSource()
    workflow.builder = StubInsideBuilder().for_workflow(workflow)
    workflow.builder.set_base_from_scratch(False)

    mock_logs_retry = setup_mock_logs_retry()
    flexmock(docker.APIClient, logs=mock_logs_retry)
    runner = PostBuildPluginsRunner(docker_tasker, workflow,
                                    [{"name": PostBuildRPMqaPlugin.key,
                                      "args": {'image_id': TEST_IMAGE}}])
    results = runner.run()
    assert results[PostBuildRPMqaPlugin.key] == PACKAGE_LIST
    assert workflow.image_components == parse_rpm_output(PACKAGE_LIST) 

def client(self):
        """single global client instance"""
        cls = self.__class__

        if cls._client is None:
            kwargs = {}
            if self.tls_config:
                kwargs['tls'] = docker.tls.TLSConfig(**self.tls_config)
            kwargs.update(kwargs_from_env())
            client = docker.APIClient(version='auto', **kwargs)

            cls._client = client
        return cls._client 

def docker() -> libdocker.APIClient:
    with libdocker.APIClient(version="auto") as client:
        yield client 

def postgres_server(docker: libdocker.APIClient) -> None:
    warnings.filterwarnings("ignore", category=DeprecationWarning)

    if not USE_LOCAL_DB:  # pragma: no cover
        pull_image(docker, POSTGRES_DOCKER_IMAGE)

        container = docker.create_container(
            image=POSTGRES_DOCKER_IMAGE,
            name="test-postgres-{}".format(uuid.uuid4()),
            detach=True,
        )
        docker.start(container=container["Id"])
        inspection = docker.inspect_container(container["Id"])
        host = inspection["NetworkSettings"]["IPAddress"]

        dsn = f"postgres://postgres:postgres@{host}/postgres"

        try:
            ping_postgres(dsn)
            environ["DB_CONNECTION"] = dsn

            yield container
        finally:
            docker.kill(container["Id"])
            docker.remove_container(container["Id"])
    else:  # pragma: no cover
        yield
        return 

def pull_image(client: APIClient, image: str) -> None:  # pragma: no cover
    client.pull(image) 

def get_api_client():
    """Return the API client or initialize it."""
    if 'api_client' not in __st__:
        from docker import APIClient, utils
        params = utils.kwargs_from_env()
        base_url = None if 'base_url' not in params else params['base_url']
        tls = None if 'tls' not in params else params['tls']

        __st__['api_client'] = APIClient(base_url=base_url, tls=tls)

    return __st__['api_client'] 

def test_logging_with_requests_timeout(self, types_mock, client_class_mock):

        mock_obj = mock.Mock()

        def _client_tasks_side_effect():
            for _ in range(2):
                yield [{'Status': {'State': 'pending'}}]
            while True:
                yield [{'Status': {'State': 'complete'}}]

        def _client_service_logs_effect():
            yield b'Testing is awesome.'
            raise requests.exceptions.ConnectionError('')

        client_mock = mock.Mock(spec=APIClient)
        client_mock.create_service.return_value = {'ID': 'some_id'}
        client_mock.service_logs.return_value = _client_service_logs_effect()
        client_mock.images.return_value = []
        client_mock.pull.return_value = [b'{"status":"pull log"}']
        client_mock.tasks.side_effect = _client_tasks_side_effect()
        types_mock.TaskTemplate.return_value = mock_obj
        types_mock.ContainerSpec.return_value = mock_obj
        types_mock.RestartPolicy.return_value = mock_obj
        types_mock.Resources.return_value = mock_obj

        client_class_mock.return_value = client_mock

        operator = DockerSwarmOperator(
            api_version='1.19', command='env', environment={'UNIT': 'TEST'}, image='ubuntu:latest',
            mem_limit='128m', user='unittest', task_id='unittest', auto_remove=True, tty=True,
            enable_logging=True
        )
        operator.execute(None)

        client_mock.service_logs.assert_called_once_with(
            'some_id', follow=True, stdout=True, stderr=True, is_tty=True
        )