Repository: exasol/integration-test-docker-environment (language: python)

Issue:
Make jobid a parameter of the Tasks
# Background

- Currently, we inject the jobid as config into the tasks
- However, you can't change the jobid after you run the first task with luigi
- So, we can't run multiple jobs in the same process

# Acceptance Criteria

- Make jobid a parameter of the BaseTask
- Make sure the jobid is forwarded from parent to child task


Discussion / hints:

Fix patch:
diff --git a/doc/changes/changes_0.6.0.md b/doc/changes/changes_0.6.0.md
index fb6e88ed4..1631a9e22 100644
--- a/doc/changes/changes_0.6.0.md
+++ b/doc/changes/changes_0.6.0.md
@@ -30,7 +30,7 @@ If you need further versions, please open an issue.
 
 ## Refactoring:
 
-n/a
+ - #122: make jobid a parameter of the task
 
 ## Security:
  -#126: Update urllib3 package
\ No newline at end of file


Test patch:
diff --git a/exasol_integration_test_docker_environment/cli/commands/build_test_container.py b/exasol_integration_test_docker_environment/cli/commands/build_test_container.py
index 9e16b5c45..6396ff77e 100644
--- a/exasol_integration_test_docker_environment/cli/commands/build_test_container.py
+++ b/exasol_integration_test_docker_environment/cli/commands/build_test_container.py
@@ -2,7 +2,7 @@
 
 from exasol_integration_test_docker_environment.cli.cli import cli
 from exasol_integration_test_docker_environment.cli.common import add_options, set_build_config, \
-    set_docker_repository_config, set_job_id, run_task
+    set_docker_repository_config, run_task, generate_root_task
 from exasol_integration_test_docker_environment.cli.options.build_options import build_options
 from exasol_integration_test_docker_environment.cli.options.docker_repository_options import docker_repository_options
 from exasol_integration_test_docker_environment.cli.options.system_options import system_options
@@ -53,8 +53,7 @@ def build_test_container(
                                  source_docker_tag_prefix, "source")
     set_docker_repository_config(target_docker_password, target_docker_repository_name, target_docker_username,
                                  target_docker_tag_prefix, "target")
-    set_job_id(DockerTestContainerBuild.__name__)
-    task_creator = lambda: DockerTestContainerBuild()
+    task_creator = lambda: generate_root_task(task_class=DockerTestContainerBuild)
     success, task = run_task(task_creator, workers, task_dependencies_dot_file)
     if not success:
         exit(1)
diff --git a/exasol_integration_test_docker_environment/cli/commands/push_test_container.py b/exasol_integration_test_docker_environment/cli/commands/push_test_container.py
index 96a251b13..ff78b6409 100644
--- a/exasol_integration_test_docker_environment/cli/commands/push_test_container.py
+++ b/exasol_integration_test_docker_environment/cli/commands/push_test_container.py
@@ -1,8 +1,8 @@
 from typing import Tuple
 
 from exasol_integration_test_docker_environment.cli.cli import cli
-from exasol_integration_test_docker_environment.cli.common import add_options, set_docker_repository_config, set_job_id, \
-    run_task, set_build_config
+from exasol_integration_test_docker_environment.cli.common import add_options, set_docker_repository_config, \
+    run_task, set_build_config, generate_root_task
 from exasol_integration_test_docker_environment.cli.options.build_options import build_options
 from exasol_integration_test_docker_environment.cli.options.docker_repository_options import docker_repository_options
 from exasol_integration_test_docker_environment.cli.options.push_options import push_options
@@ -56,8 +56,8 @@ def push_test_container(
                                  source_docker_tag_prefix, "source")
     set_docker_repository_config(target_docker_password, target_docker_repository_name, target_docker_username,
                                  target_docker_tag_prefix, "target")
-    set_job_id(DockerTestContainerPush.__name__)
-    task_creator = lambda: DockerTestContainerPush(force_push=force_push,
+    task_creator = lambda: generate_root_task(task_class=DockerTestContainerPush,
+                                                   force_push=force_push,
                                                    push_all=push_all)
     success, task = run_task(task_creator, workers, task_dependencies_dot_file)
     if not success:
diff --git a/exasol_integration_test_docker_environment/cli/commands/spawn_test_environment.py b/exasol_integration_test_docker_environment/cli/commands/spawn_test_environment.py
index f53b50b31..509a4dcd5 100644
--- a/exasol_integration_test_docker_environment/cli/commands/spawn_test_environment.py
+++ b/exasol_integration_test_docker_environment/cli/commands/spawn_test_environment.py
@@ -3,8 +3,8 @@
 import humanfriendly
 
 from exasol_integration_test_docker_environment.cli.cli import cli
-from exasol_integration_test_docker_environment.cli.common import add_options, set_build_config, \
-    set_job_id, run_task
+from exasol_integration_test_docker_environment.cli.common import add_options, set_build_config, run_task, \
+    generate_root_task
 from exasol_integration_test_docker_environment.cli.options.system_options import system_options, \
     output_directory_option, tempory_base_directory_option
 from exasol_integration_test_docker_environment.cli.options.test_environment_options import docker_db_options
@@ -68,7 +68,7 @@ def spawn_test_environment(
                      temporary_base_directory,
                      None,
                      None)
-    task_creator = lambda: SpawnTestEnvironmentWithDockerDB(
+    task_creator = lambda: generate_root_task(task_class=SpawnTestEnvironmentWithDockerDB,
         environment_name=environment_name,
         database_port_forward=str(database_port_forward) if database_port_forward is not None else None,
         bucketfs_port_forward=str(bucketfs_port_forward) if bucketfs_port_forward is not None else None,
@@ -87,7 +87,6 @@ def spawn_test_environment(
         no_database_cleanup_after_failure=False,
         is_setup_database_activated=not deactivate_database_setup
     )
-    set_job_id(SpawnTestEnvironmentWithDockerDB.__name__)
     success, task = run_task(task_creator, workers, task_dependencies_dot_file)
     if not success:
         exit(1)
diff --git a/exasol_integration_test_docker_environment/cli/common.py b/exasol_integration_test_docker_environment/cli/common.py
index 4d74cf6dc..b8aebcf51 100644
--- a/exasol_integration_test_docker_environment/cli/common.py
+++ b/exasol_integration_test_docker_environment/cli/common.py
@@ -77,11 +77,11 @@ def import_build_steps(flavor_path: Tuple[str, ...]):
         spec.loader.exec_module(module)
 
 
-def set_job_id(name):
+def generate_root_task(task_class, *args, **kwargs):
     strftime = datetime.now().strftime('%Y_%m_%d_%H_%M_%S')
-    job_id = f"{strftime}_{name}"
-    config = luigi.configuration.get_config()
-    config.set('job_config', 'job_id', job_id)
+    params = {"job_id": f"{strftime}_{task_class.__name__}"}
+    params.update(kwargs)
+    return task_class(**params)
 
 
 def run_task(task_creator: Callable[[], DependencyLoggerBaseTask],
diff --git a/exasol_integration_test_docker_environment/lib/base/base_task.py b/exasol_integration_test_docker_environment/lib/base/base_task.py
index 073143af5..f3bcd11de 100644
--- a/exasol_integration_test_docker_environment/lib/base/base_task.py
+++ b/exasol_integration_test_docker_environment/lib/base/base_task.py
@@ -14,7 +14,6 @@
 from exasol_integration_test_docker_environment.abstract_method_exception import AbstractMethodException
 from exasol_integration_test_docker_environment.lib.base.abstract_task_future import AbstractTaskFuture, \
     DEFAULT_RETURN_OBJECT_NAME
-from exasol_integration_test_docker_environment.lib.base.job_config import job_config
 from exasol_integration_test_docker_environment.lib.base.pickle_target import PickleTarget
 from exasol_integration_test_docker_environment.lib.base.task_logger_wrapper import TaskLoggerWrapper
 from exasol_integration_test_docker_environment.lib.base.task_state import TaskState
@@ -71,6 +70,7 @@ def _get_outputs_dict(self) -> Dict[str, PickleTarget]:
 
 class BaseTask(Task):
     caller_output_path = luigi.ListParameter([], significant=False, visibility=ParameterVisibility.HIDDEN)
+    job_id = luigi.Parameter()
 
     def __init__(self, *args, **kwargs):
         self._registered_tasks = []
@@ -113,7 +113,7 @@ def task_id_str(self, task_family, params):
         # task_id is a concatenation of task family, the first values of the first 3 parameters
         # sorted by parameter name and a md5hash of the family/parameters as a cananocalised json.
         param_str = json.dumps(params, separators=(',', ':'), sort_keys=True)
-        hash_input = job_config().job_id + param_str
+        hash_input = param_str
         param_hash = hashlib.sha3_256(hash_input.encode('utf-8')).hexdigest()
         return '{}_{}'.format(task_family, param_hash[:TASK_ID_TRUNCATE_HASH])
 
@@ -137,7 +137,7 @@ def get_output_path(self) -> Path:
 
     def _get_output_path_for_job(self) -> Path:
         return Path(build_config().output_directory,
-                    "jobs", job_config().job_id)
+                    "jobs", self.job_id)
 
     def _extend_output_path(self):
         extension = self.extend_output_path()
@@ -306,12 +306,14 @@ def __repr__(self):
     def create_child_task_with_common_params(self, task_class, **kwargs):
         params = util.common_params(self, task_class)
         params["caller_output_path"] = self._extend_output_path()
+        params["job_id"] = self.job_id
         params.update(kwargs)
         return task_class(**params)
 
     def create_child_task(self, task_class, **kwargs):
         params = {}
         params["caller_output_path"] = self._extend_output_path()
+        params["job_id"] = self.job_id
         params.update(kwargs)
         return task_class(**params)
 
diff --git a/exasol_integration_test_docker_environment/lib/base/docker_base_task.py b/exasol_integration_test_docker_environment/lib/base/docker_base_task.py
index d0ef5bb06..f18fc42f1 100644
--- a/exasol_integration_test_docker_environment/lib/base/docker_base_task.py
+++ b/exasol_integration_test_docker_environment/lib/base/docker_base_task.py
@@ -1,29 +1,9 @@
 from exasol_integration_test_docker_environment.lib.base.dependency_logger_base_task import DependencyLoggerBaseTask
-from exasol_integration_test_docker_environment.lib.config.docker_config import docker_client_config
+from exasol_integration_test_docker_environment.lib.base.docker_parameter import DockerParameter
+from exasol_integration_test_docker_environment.lib.docker import ContextDockerClient
 
 
-class DockerBaseTask(DependencyLoggerBaseTask):
+class DockerBaseTask(DependencyLoggerBaseTask, DockerParameter):
 
-    def __init__(self, *args, **kwargs):
-        self._init()
-        super().__init__(*args, **kwargs)
-
-    def _init(self):
-        self._client = docker_client_config().get_client()
-        self._low_level_client = docker_client_config().get_low_level_client()
-
-    def __del__(self):
-        if "_client" in self.__dict__:
-            self._client.close()
-        if "_low_level_client" in self.__dict__:
-            self._low_level_client.close()
-
-    def __getstate__(self):
-        new_dict = super().__getstate__()
-        del new_dict["_client"]
-        del new_dict["_low_level_client"]
-        return new_dict
-
-    def __setstate__(self, new_dict):
-        super().__setstate__(new_dict)
-        self._init()
+    def _get_docker_client(self):
+        return ContextDockerClient(timeout=self.timeout)
diff --git a/exasol_integration_test_docker_environment/lib/base/docker_parameter.py b/exasol_integration_test_docker_environment/lib/base/docker_parameter.py
new file mode 100644
index 000000000..25a9b9308
--- /dev/null
+++ b/exasol_integration_test_docker_environment/lib/base/docker_parameter.py
@@ -0,0 +1,12 @@
+import luigi
+from luigi import Config
+from luigi.parameter import ParameterVisibility
+
+
+class DockerParameter(Config):
+    """
+    Docker parameters used for Tasks accessing Docker client.
+    """
+
+    timeout = luigi.IntParameter(100000, significant=False, visibility=ParameterVisibility.PRIVATE)
+    no_cache = luigi.BoolParameter(False)
diff --git a/exasol_integration_test_docker_environment/lib/base/job_config.py b/exasol_integration_test_docker_environment/lib/base/job_config.py
deleted file mode 100644
index da1f5a902..000000000
--- a/exasol_integration_test_docker_environment/lib/base/job_config.py
+++ /dev/null
@@ -1,8 +0,0 @@
-import luigi
-from luigi import Config
-
-
-class job_config(Config):
-    job_id = luigi.Parameter()
-    temporary_base_directory = luigi.OptionalParameter("/tmp")
-    output_directory = luigi.Parameter(".build_output")
diff --git a/exasol_integration_test_docker_environment/lib/config/docker_config.py b/exasol_integration_test_docker_environment/lib/config/docker_config.py
index e17fb6772..3c4fd517c 100644
--- a/exasol_integration_test_docker_environment/lib/config/docker_config.py
+++ b/exasol_integration_test_docker_environment/lib/config/docker_config.py
@@ -1,19 +1,7 @@
-import docker
 import luigi
 from luigi.parameter import ParameterVisibility
 
 
-class docker_client_config(luigi.Config):
-    timeout = luigi.IntParameter(100000, significant=False, visibility=ParameterVisibility.PRIVATE)
-    no_cache = luigi.BoolParameter(False)
-
-    def get_client(self):
-        return docker.from_env(timeout=self.timeout, assert_hostname=False)
-
-    def get_low_level_client(self):
-        return self.get_client().api
-
-
 class source_docker_repository_config(luigi.Config):
     repository_name = luigi.Parameter("exasol/script-language-container")
     tag_prefix = luigi.Parameter("")
diff --git a/exasol_integration_test_docker_environment/lib/docker/__init__.py b/exasol_integration_test_docker_environment/lib/docker/__init__.py
index e69de29bb..e8f893978 100644
--- a/exasol_integration_test_docker_environment/lib/docker/__init__.py
+++ b/exasol_integration_test_docker_environment/lib/docker/__init__.py
@@ -0,0 +1,16 @@
+import docker
+
+
+class ContextDockerClient(object):
+    def __init__(self, **kwargs):
+        self.kwargs = kwargs
+        self._client = None
+
+    def __enter__(self):
+        self._client = docker.from_env(**self.kwargs)
+        return self._client
+
+    def __exit__(self, type_, value, traceback):
+        if self._client is not None:
+            self._client.close()
+        self._client = None
diff --git a/exasol_integration_test_docker_environment/lib/docker/images/clean/clean_images.py b/exasol_integration_test_docker_environment/lib/docker/images/clean/clean_images.py
index 8452f218c..be12e136b 100644
--- a/exasol_integration_test_docker_environment/lib/docker/images/clean/clean_images.py
+++ b/exasol_integration_test_docker_environment/lib/docker/images/clean/clean_images.py
@@ -1,11 +1,9 @@
 import luigi
 from exasol_integration_test_docker_environment.lib.base.docker_base_task import DockerBaseTask
-from exasol_integration_test_docker_environment.lib.config.docker_config import target_docker_repository_config
 from exasol_integration_test_docker_environment.lib.docker.images.utils import find_images_by_tag
 
 
 class CleanImageTask(DockerBaseTask):
-    jobid = luigi.Parameter("")
     image_id = luigi.Parameter()
 
     def __init__(self, *args, **kwargs):
@@ -17,22 +15,25 @@ def run_task(self):
         for i in range(3):
             try:
                 self.logger.info("Try to remove image %s" % self.image_id)
-                self._client.images.remove(image=self.image_id, force=True)
+                with self._get_docker_client() as docker_client:
+                    docker_client.images.remove(image=self.image_id, force=True)
                 self.logger.info("Removed image %s" % self.image_id)
                 break
             except Exception as e:
                 self.logger.info("Could not removed image %s got exception %s" % (self.image_id, e))
 
     def get_clean_image_tasks_for_dependent_images(self):
-        image_ids = [str(possible_child).replace("sha256:", "") for possible_child
-                     in self._low_level_client.images(all=True, quiet=True)
-                     if self.is_child_image(possible_child)]
-        return [self.create_child_task(CleanImageTask, image_id=image_id)
-                for image_id in image_ids]
+        with self._get_docker_client() as docker_client:
+            image_ids = [str(possible_child).replace("sha256:", "") for possible_child
+                         in docker_client.api.images(all=True, quiet=True)
+                         if self.is_child_image(possible_child)]
+            return [self.create_child_task(CleanImageTask, image_id=image_id)
+                    for image_id in image_ids]
 
     def is_child_image(self, possible_child):
         try:
-            inspect = self._low_level_client.inspect_image(image=str(possible_child).replace("sha256:", ""))
+            with self._get_docker_client() as docker_client:
+                inspect = docker_client.api.inspect_image(image=str(possible_child).replace("sha256:", ""))
             return str(inspect["Parent"]).replace("sha256:", "") == self.image_id
         except:
             return False
@@ -40,7 +41,6 @@ def is_child_image(self, possible_child):
 
 class CleanImagesStartingWith(DockerBaseTask):
 
-    jobid = luigi.Parameter("")
     starts_with_pattern = luigi.Parameter()
 
     def register_required(self):
@@ -51,10 +51,11 @@ def register_required(self):
 
     def find_images_to_clean(self):
         self.logger.info("Going to remove all images starting with %s" % self.starts_with_pattern)
-        filter_images = find_images_by_tag(self._client, lambda tag: tag.startswith(self.starts_with_pattern))
-        for i in filter_images:
-            self.logger.info("Going to remove following image: %s" % i.tags)
-        return filter_images
+        with self._get_docker_client() as docker_client:
+            filter_images = find_images_by_tag(docker_client, lambda tag: tag.startswith(self.starts_with_pattern))
+            for i in filter_images:
+                self.logger.info("Going to remove following image: %s" % i.tags)
+            return filter_images
 
     def run_task(self):
         pass
diff --git a/exasol_integration_test_docker_environment/lib/docker/images/create/docker_image_build_task.py b/exasol_integration_test_docker_environment/lib/docker/images/create/docker_image_build_task.py
index be3aa40ae..7acd6638e 100644
--- a/exasol_integration_test_docker_environment/lib/docker/images/create/docker_image_build_task.py
+++ b/exasol_integration_test_docker_environment/lib/docker/images/create/docker_image_build_task.py
@@ -4,8 +4,7 @@
 
 from exasol_integration_test_docker_environment.lib.base.still_running_logger import StillRunningLogger
 from exasol_integration_test_docker_environment.lib.config.build_config import build_config
-from exasol_integration_test_docker_environment.lib.config.docker_config import docker_client_config, \
-    docker_build_arguments
+from exasol_integration_test_docker_environment.lib.config.docker_config import docker_build_arguments
 from exasol_integration_test_docker_environment.lib.docker.images.create.docker_image_creator_base_task import \
     DockerImageCreatorBaseTask
 from exasol_integration_test_docker_environment.lib.docker.images.create.utils.build_context_creator import \
@@ -27,15 +26,16 @@ def run_task(self):
                                                         self.image_info,
                                                         self.get_log_path())
             build_context_creator.prepare_build_context_to_temp_dir()
-            output_generator = \
-                self._low_level_client.build(path=temp_directory,
-                                             nocache=docker_client_config.no_cache,
-                                             tag=self.image_info.get_target_complete_name(),
-                                             rm=True,
-                                             buildargs=dict(**image_description.transparent_build_arguments,
-                                                            **image_description.image_changing_build_arguments,
-                                                            **docker_build_arguments().secret))
-            self._handle_output(output_generator, self.image_info)
+            with self._get_docker_client() as docker_client:
+                output_generator = \
+                    docker_client.api.build(path=temp_directory,
+                                            nocache=self.no_cache,
+                                            tag=self.image_info.get_target_complete_name(),
+                                            rm=True,
+                                            buildargs=dict(**image_description.transparent_build_arguments,
+                                                           **image_description.image_changing_build_arguments,
+                                                           **docker_build_arguments().secret))
+                self._handle_output(output_generator, self.image_info)
         finally:
             shutil.rmtree(temp_directory)
 
diff --git a/exasol_integration_test_docker_environment/lib/docker/images/create/docker_image_create_task.py b/exasol_integration_test_docker_environment/lib/docker/images/create/docker_image_create_task.py
index fc96bb0a4..55fe41af2 100644
--- a/exasol_integration_test_docker_environment/lib/docker/images/create/docker_image_create_task.py
+++ b/exasol_integration_test_docker_environment/lib/docker/images/create/docker_image_create_task.py
@@ -61,10 +61,11 @@ def build(self, image_info: ImageInfo):
                             self.task_id, image_info.image_state, image_info.get_target_complete_name())
 
     def rename_source_image_to_target_image(self, image_info):
-        self._client.images.get(image_info.get_source_complete_name()).tag(
-            repository=image_info.target_repository_name,
-            tag=image_info.get_target_complete_tag()
-        )
+        with self._get_docker_client() as docker_client:
+            docker_client.images.get(image_info.get_source_complete_name()).tag(
+                repository=image_info.target_repository_name,
+                tag=image_info.get_target_complete_tag()
+            )
 
 
 class DockerCreateImageTaskWithDeps(DockerCreateImageTask):
diff --git a/exasol_integration_test_docker_environment/lib/docker/images/create/docker_image_load_task.py b/exasol_integration_test_docker_environment/lib/docker/images/create/docker_image_load_task.py
index cb81cd556..2d7d602b7 100644
--- a/exasol_integration_test_docker_environment/lib/docker/images/create/docker_image_load_task.py
+++ b/exasol_integration_test_docker_environment/lib/docker/images/create/docker_image_load_task.py
@@ -12,9 +12,10 @@ def run_task(self):
             .joinpath(self.image_info.get_source_complete_name() + ".tar")
         self.logger.info("Try to load docker image %s from %s",
                          self.image_info.get_source_complete_name(), image_archive_path)
-        with image_archive_path.open("rb") as f:
-            self._client.images.load(f)
-        self._client.images.get(self.image_info.get_source_complete_name()).tag(
-            repository=self.image_info.target_repository_name,
-            tag=self.image_info.get_target_complete_tag()
-        )
+        with self._get_docker_client() as docker_client:
+            with image_archive_path.open("rb") as f:
+                docker_client.images.load(f)
+            docker_client.images.get(self.image_info.get_source_complete_name()).tag(
+                repository=self.image_info.target_repository_name,
+                tag=self.image_info.get_target_complete_tag()
+            )
diff --git a/exasol_integration_test_docker_environment/lib/docker/images/create/docker_image_pull_task.py b/exasol_integration_test_docker_environment/lib/docker/images/create/docker_image_pull_task.py
index dd681b3f3..586e66524 100644
--- a/exasol_integration_test_docker_environment/lib/docker/images/create/docker_image_pull_task.py
+++ b/exasol_integration_test_docker_environment/lib/docker/images/create/docker_image_pull_task.py
@@ -35,18 +35,20 @@ def run_task(self):
                 else:
                     raise e
 
-        self._client.images.get(self.image_info.get_source_complete_name()).tag(
-            repository=self.image_info.target_repository_name,
-            tag=self.image_info.get_target_complete_tag()
-        )
+        with self._get_docker_client() as docker_client:
+            docker_client.images.get(self.image_info.get_source_complete_name()).tag(
+                repository=self.image_info.target_repository_name,
+                tag=self.image_info.get_target_complete_tag()
+            )
 
     def pull(self, image_target, auth_config):
         self.logger.info("Try to pull docker image %s", image_target.get_complete_name())
-        output_generator = self._low_level_client.pull(
-            repository=image_target.image_name, tag=image_target.image_tag,
-            auth_config=auth_config,
-            stream=True)
-        self._handle_output(output_generator, self.image_info)
+        with self._get_docker_client() as docker_client:
+            output_generator = docker_client.api.pull(
+                repository=image_target.image_name, tag=image_target.image_tag,
+                auth_config=auth_config,
+                stream=True)
+            self._handle_output(output_generator, self.image_info)
 
     def _handle_output(self, output_generator, image_info):
         log_file_path = self.get_log_path().joinpath("pull_docker_db_image.log")
diff --git a/exasol_integration_test_docker_environment/lib/docker/images/create/utils/docker_image_target.py b/exasol_integration_test_docker_environment/lib/docker/images/create/utils/docker_image_target.py
index 1c01db576..07f2c89ac 100644
--- a/exasol_integration_test_docker_environment/lib/docker/images/create/utils/docker_image_target.py
+++ b/exasol_integration_test_docker_environment/lib/docker/images/create/utils/docker_image_target.py
@@ -1,7 +1,7 @@
 import docker
 import luigi
 
-from exasol_integration_test_docker_environment.lib.config.docker_config import docker_client_config
+from exasol_integration_test_docker_environment.lib.docker import ContextDockerClient
 
 
 class DockerImageTarget(luigi.Target):
@@ -13,11 +13,9 @@ def get_complete_name(self):
         return f"{self.image_name}:{self.image_tag}"
 
     def exists(self) -> bool:
-        client = docker_client_config().get_client()
-        try:
-            image = client.images.get(self.get_complete_name())
-            return True
-        except docker.errors.DockerException as e:
-            return False
-        finally:
-            client.close()
+        with ContextDockerClient() as docker_client:
+            try:
+                image = docker_client.images.get(self.get_complete_name())
+                return True
+            except docker.errors.DockerException as e:
+                return False
diff --git a/exasol_integration_test_docker_environment/lib/docker/images/create/utils/docker_registry_image_checker.py b/exasol_integration_test_docker_environment/lib/docker/images/create/utils/docker_registry_image_checker.py
index f8b221942..0f2da542e 100644
--- a/exasol_integration_test_docker_environment/lib/docker/images/create/utils/docker_registry_image_checker.py
+++ b/exasol_integration_test_docker_environment/lib/docker/images/create/utils/docker_registry_image_checker.py
@@ -1,8 +1,8 @@
 import json
 import multiprocessing as mp
 
-import docker
 
+from exasol_integration_test_docker_environment.lib.docker import ContextDockerClient
 from exasol_integration_test_docker_environment.lib.logging.abstract_log_handler import AbstractLogHandler
 
 
@@ -23,19 +23,15 @@ def handle_log_line(self, log_line, error: bool = False):
 
 class DockerRegistryImageChecker:
     def map(self, image: str, queue: mp.Queue):
-        client = docker.from_env()  # TODO replace by docker_config
-        try:
-            generator = client.api.pull(
-                repository=image,
-                stream=True,
-            )
-            for log_line in generator:
-                queue.put(log_line)
-            queue.put(None)
-        except Exception as e:
-            queue.put(e)
-        finally:
-            client.close()
+
+        with ContextDockerClient() as docker_client:
+            try:
+                generator = docker_client.api.pull(repository=image, stream=True)
+                for log_line in generator:
+                    queue.put(log_line)
+                queue.put(None)
+            except Exception as e:
+                queue.put(e)
 
     def check(self, image: str):
         log_handler = DockerRegistryImageCheckerPullLogHandler()
diff --git a/exasol_integration_test_docker_environment/lib/docker/images/push/docker_image_push_base_task.py b/exasol_integration_test_docker_environment/lib/docker/images/push/docker_image_push_base_task.py
index 82040f05f..5f8bfdeb1 100644
--- a/exasol_integration_test_docker_environment/lib/docker/images/push/docker_image_push_base_task.py
+++ b/exasol_integration_test_docker_environment/lib/docker/images/push/docker_image_push_base_task.py
@@ -30,11 +30,12 @@ def run_task(self):
                 "username": target_docker_repository_config().username,
                 "password": target_docker_repository_config().password
             }
-            generator = self._client.images.push(repository=image_info.get_target_complete_name(),
-                                                 tag=image_info.get_target_complete_tag(),
-                                                 auth_config=auth_config,
-                                                 stream=True)
-            self._handle_output(generator, image_info)
+            with self._get_docker_client() as docker_client:
+                generator = docker_client.images.push(repository=image_info.get_target_complete_name(),
+                                               tag=image_info.get_target_complete_tag(),
+                                               auth_config=auth_config,
+                                               stream=True)
+                self._handle_output(generator, image_info)
         self.return_object(image_info)
 
     def _handle_output(self, output_generator, image_info: ImageInfo):
diff --git a/exasol_integration_test_docker_environment/lib/docker/images/save/docker_image_save_base_task.py b/exasol_integration_test_docker_environment/lib/docker/images/save/docker_image_save_base_task.py
index 3e5fca7ed..d12a15baa 100644
--- a/exasol_integration_test_docker_environment/lib/docker/images/save/docker_image_save_base_task.py
+++ b/exasol_integration_test_docker_environment/lib/docker/images/save/docker_image_save_base_task.py
@@ -41,9 +41,10 @@ def get_tag_for_save(self, image_info: ImageInfo):
 
     def save_image(self, image_info: ImageInfo, tag_for_save: str, save_file_path: pathlib.Path):
         self.remove_save_file_if_necassary(save_file_path)
-        image = self._client.images.get(image_info.get_target_complete_name())
-        generator = image.save(named=tag_for_save)
-        self.write_image_to_file(save_file_path, image_info, generator)
+        with self._get_docker_client() as docker_client:
+            image = docker_client.images.get(image_info.get_target_complete_name())
+            generator = image.save(named=tag_for_save)
+            self.write_image_to_file(save_file_path, image_info, generator)
 
     def remove_save_file_if_necassary(self, save_file_path: pathlib.Path):
         save_file_path.parent.mkdir(exist_ok=True, parents=True)
diff --git a/exasol_integration_test_docker_environment/lib/docker/images/utils.py b/exasol_integration_test_docker_environment/lib/docker/images/utils.py
index e414b72a9..c9f14763b 100644
--- a/exasol_integration_test_docker_environment/lib/docker/images/utils.py
+++ b/exasol_integration_test_docker_environment/lib/docker/images/utils.py
@@ -1,8 +1,8 @@
 from typing import Callable, List
 
 
-def find_images_by_tag(client, condition: Callable[[str], bool]) -> List:
-    images = client.images.list()
+def find_images_by_tag(docker_client, condition: Callable[[str], bool]) -> List:
+    images = docker_client.images.list()
     filter_images = [image for image in images
                      if image.tags is not None and len(image.tags) > 0 and
                      any([condition(tag) for tag in image.tags])]
diff --git a/exasol_integration_test_docker_environment/lib/test_environment/abstract_spawn_test_environment.py b/exasol_integration_test_docker_environment/lib/test_environment/abstract_spawn_test_environment.py
index 6e443a3f4..2111357c4 100644
--- a/exasol_integration_test_docker_environment/lib/test_environment/abstract_spawn_test_environment.py
+++ b/exasol_integration_test_docker_environment/lib/test_environment/abstract_spawn_test_environment.py
@@ -67,40 +67,41 @@ def _attempt_database_start(self):
     def create_test_environment_info_in_test_container_and_on_host(
             self, test_environment_info: EnvironmentInfo):
         test_container_name = test_environment_info.test_container_info.container_name
-        test_container = self._client.containers.get(test_container_name)
-        test_environment_info_base_host_path = Path(self.get_cache_path(),
-                                                    f"environments/{self.environment_name}")
-        test_environment_info_base_host_path.mkdir(exist_ok=True, parents=True)
-        self.logger.info(
-            f"Create test environment info in test container '{test_container_name}' at '/' "
-            f"and on the host at '{test_environment_info_base_host_path}'")
-
-        copy = DockerContainerCopy(test_container)
-        json = test_environment_info.to_json()
-        copy.add_string_to_file("environment_info.json", json)
-        cache_environment_info_json_path = Path(test_environment_info_base_host_path,
-                                                "environment_info.json")
-        with cache_environment_info_json_path.open("w") as f:
-            f.write(json)
-
-        environment_variables = \
-            self.collect_environment_info_variables(test_container_name,
-                                                    test_environment_info)
-        copy.add_string_to_file("environment_info.conf", environment_variables)
-        cache_environment_info_conf_path = Path(test_environment_info_base_host_path,
-                                                "environment_info.conf")
-        with cache_environment_info_conf_path.open("w") as f:
-            f.write(environment_variables)
-
-        environment_variables_with_export = ""
-        for line in environment_variables.splitlines():
-            environment_variables_with_export += f"export {line}\n"
-        copy.add_string_to_file("environment_info.sh", environment_variables_with_export)
-        cache_environment_info_sh_path = Path(test_environment_info_base_host_path, "environment_info.sh")
-        with cache_environment_info_sh_path.open("w") as f:
-            f.write(environment_variables_with_export)
-
-        copy.copy("/")
+        with self._get_docker_client() as docker_client:
+            test_container = docker_client.containers.get(test_container_name)
+            test_environment_info_base_host_path = Path(self.get_cache_path(),
+                                                        f"environments/{self.environment_name}")
+            test_environment_info_base_host_path.mkdir(exist_ok=True, parents=True)
+            self.logger.info(
+                f"Create test environment info in test container '{test_container_name}' at '/' "
+                f"and on the host at '{test_environment_info_base_host_path}'")
+
+            copy = DockerContainerCopy(test_container)
+            json = test_environment_info.to_json()
+            copy.add_string_to_file("environment_info.json", json)
+            cache_environment_info_json_path = Path(test_environment_info_base_host_path,
+                                                    "environment_info.json")
+            with cache_environment_info_json_path.open("w") as f:
+                f.write(json)
+
+            environment_variables = \
+                self.collect_environment_info_variables(test_container_name,
+                                                        test_environment_info)
+            copy.add_string_to_file("environment_info.conf", environment_variables)
+            cache_environment_info_conf_path = Path(test_environment_info_base_host_path,
+                                                    "environment_info.conf")
+            with cache_environment_info_conf_path.open("w") as f:
+                f.write(environment_variables)
+
+            environment_variables_with_export = ""
+            for line in environment_variables.splitlines():
+                environment_variables_with_export += f"export {line}\n"
+            copy.add_string_to_file("environment_info.sh", environment_variables_with_export)
+            cache_environment_info_sh_path = Path(test_environment_info_base_host_path, "environment_info.sh")
+            with cache_environment_info_sh_path.open("w") as f:
+                f.write(environment_variables_with_export)
+
+            copy.copy("/")
 
     def collect_environment_info_variables(self, test_container_name, test_environment_info):
         environment_variables = ""
@@ -116,10 +117,11 @@ def collect_environment_info_variables(self, test_container_name, test_environme
             environment_variables += f"""ENVIRONMENT_DATABASE_CONTAINER_NETWORK_ALIASES="{database_container_network_aliases}"\n"""
             environment_variables += f"""ENVIRONMENT_DATABASE_CONTAINER_IP_ADDRESS={test_environment_info.database_info.container_info.ip_address}\n"""
             environment_variables += f"""ENVIRONMENT_DATABASE_CONTAINER_VOLUMNE_NAME={test_environment_info.database_info.container_info.volume_name}\n"""
-            db_container = self._client.containers.get(
-                test_environment_info.database_info.container_info.container_name)
-            db_container.reload()
-            default_bridge_ip_address = db_container.attrs["NetworkSettings"]["Networks"]["bridge"]["IPAddress"]
+            with self._get_docker_client() as docker_client:
+                db_container = docker_client.containers.get(
+                    test_environment_info.database_info.container_info.container_name)
+                db_container.reload()
+                default_bridge_ip_address = db_container.attrs["NetworkSettings"]["Networks"]["bridge"]["IPAddress"]
             environment_variables += f"""ENVIRONMENT_DATABASE_CONTAINER_DEFAULT_BRIDGE_IP_ADDRESS={default_bridge_ip_address}\n"""
         if test_environment_info.test_container_info is not None:
             environment_variables += f"""ENVIRONMENT_TEST_CONTAINER_NAME={test_container_name}\n"""
diff --git a/exasol_integration_test_docker_environment/lib/test_environment/analyze_test_container.py b/exasol_integration_test_docker_environment/lib/test_environment/analyze_test_container.py
index ffc1e09a8..43f3ee9e5 100644
--- a/exasol_integration_test_docker_environment/lib/test_environment/analyze_test_container.py
+++ b/exasol_integration_test_docker_environment/lib/test_environment/analyze_test_container.py
@@ -43,7 +43,7 @@ def is_rebuild_requested(self) -> bool:
 class DockerTestContainerBuildBase(DockerBuildBase):
 
     def get_goal_class_map(self) -> Dict[str, DockerAnalyzeImageTask]:
-        goal_class_map = {"test-container": AnalyzeTestContainer()}
+        goal_class_map = {"test-container": self.create_child_task(task_class=AnalyzeTestContainer)}
         return goal_class_map
 
     def get_default_goals(self) -> Set[str]:
diff --git a/exasol_integration_test_docker_environment/lib/test_environment/database_setup/populate_data.py b/exasol_integration_test_docker_environment/lib/test_environment/database_setup/populate_data.py
index 526a7d53a..03a45d951 100644
--- a/exasol_integration_test_docker_environment/lib/test_environment/database_setup/populate_data.py
+++ b/exasol_integration_test_docker_environment/lib/test_environment/database_setup/populate_data.py
@@ -33,10 +33,11 @@ def populate_data(self):
         self.logger.warning("Uploading data")
         username = self.db_user
         password = self.db_password
-        test_container = self._client.containers.get(self._test_container_info.container_name)
-        cmd = f"""cd /tests/test/enginedb_small; $EXAPLUS -c '{self._database_info.host}:{self._database_info.db_port}' -u '{username}' -p '{password}' -f import.sql"""
-        bash_cmd = f"""bash -c "{cmd}" """
-        exit_code, output = test_container.exec_run(cmd=bash_cmd)
+        with self._get_docker_client() as docker_client:
+            test_container = docker_client.containers.get(self._test_container_info.container_name)
+            cmd = f"""cd /tests/test/enginedb_small; $EXAPLUS -c '{self._database_info.host}:{self._database_info.db_port}' -u '{username}' -p '{password}' -f import.sql"""
+            bash_cmd = f"""bash -c "{cmd}" """
+            exit_code, output = test_container.exec_run(cmd=bash_cmd)
         self.write_logs(output.decode("utf-8"))
         if exit_code != 0:
             raise Exception("Failed to populate the database with data.\nLog: %s" % cmd + "\n" + output.decode("utf-8"))
diff --git a/exasol_integration_test_docker_environment/lib/test_environment/database_setup/upload_file_to_db.py b/exasol_integration_test_docker_environment/lib/test_environment/database_setup/upload_file_to_db.py
index c49d9f835..f5e579805 100644
--- a/exasol_integration_test_docker_environment/lib/test_environment/database_setup/upload_file_to_db.py
+++ b/exasol_integration_test_docker_environment/lib/test_environment/database_setup/upload_file_to_db.py
@@ -36,22 +36,23 @@ def run_task(self):
         log_file = self.get_log_file()
         sync_time_estimation = self.get_sync_time_estimation()
 
-        if self._database_info.container_info is not None:
-            database_container = self._client.containers.get(
-                self._database_info.container_info.container_name)
-        else:
-            database_container = None
-        if not self.should_be_reused(upload_target):
-            self.upload_and_wait(database_container,
-                                 file_to_upload,
-                                 upload_target,
-                                 log_file,
-                                 pattern_to_wait_for,
-                                 sync_time_estimation)
-        else:
-            self.logger.warning("Reusing uploaded target %s instead of file %s",
-                                upload_target, file_to_upload)
-            self.write_logs("Reusing")
+        with self._get_docker_client() as docker_client:
+            if self._database_info.container_info is not None:
+                database_container = docker_client.containers.get(
+                    self._database_info.container_info.container_name)
+            else:
+                database_container = None
+            if not self.should_be_reused(upload_target):
+                self.upload_and_wait(database_container,
+                                     file_to_upload,
+                                     upload_target,
+                                     log_file,
+                                     pattern_to_wait_for,
+                                     sync_time_estimation)
+            else:
+                self.logger.warning("Reusing uploaded target %s instead of file %s",
+                                    upload_target, file_to_upload)
+                self.write_logs("Reusing")
 
     def upload_and_wait(self, database_container,
                         file_to_upload: str, upload_target: str,
@@ -132,12 +133,13 @@ def generate_upload_command(self, file_to_upload, upload_target):
         return cmd
 
     def run_command(self, command_type, cmd):
-        test_container = self._client.containers.get(self._test_container_info.container_name)
-        self.logger.info("start %s command %s", command_type, cmd)
-        exit_code, output = test_container.exec_run(cmd=cmd)
-        self.logger.info("finish %s command %s", command_type, cmd)
-        log_output = cmd + "\n\n" + output.decode("utf-8")
-        return exit_code, log_output
+        with self._get_docker_client() as docker_client:
+            test_container = docker_client.containers.get(self._test_container_info.container_name)
+            self.logger.info("start %s command %s", command_type, cmd)
+            exit_code, output = test_container.exec_run(cmd=cmd)
+            self.logger.info("finish %s command %s", command_type, cmd)
+            log_output = cmd + "\n\n" + output.decode("utf-8")
+            return exit_code, log_output
 
     def write_logs(self, output):
         log_file = Path(self.get_log_path(), "log")
diff --git a/exasol_integration_test_docker_environment/lib/test_environment/database_waiters/wait_for_external_database.py b/exasol_integration_test_docker_environment/lib/test_environment/database_waiters/wait_for_external_database.py
index df8aa7c6e..43e352fa1 100644
--- a/exasol_integration_test_docker_environment/lib/test_environment/database_waiters/wait_for_external_database.py
+++ b/exasol_integration_test_docker_environment/lib/test_environment/database_waiters/wait_for_external_database.py
@@ -22,9 +22,10 @@ class WaitForTestExternalDatabase(DockerBaseTask,
     attempt = luigi.IntParameter(1)
 
     def run_task(self):
-        test_container = self._client.containers.get(self.test_container_info.container_name)
-        is_database_ready = self.wait_for_database_startup(test_container)
-        self.return_object(is_database_ready)
+        with self._get_docker_client() as docker_client:
+            test_container = docker_client.containers.get(self.test_container_info.container_name)
+            is_database_ready = self.wait_for_database_startup(test_container)
+            self.return_object(is_database_ready)
 
     def wait_for_database_startup(self, test_container: Container):
         is_database_ready_thread = self.start_wait_threads(test_container)
diff --git a/exasol_integration_test_docker_environment/lib/test_environment/database_waiters/wait_for_test_docker_database.py b/exasol_integration_test_docker_environment/lib/test_environment/database_waiters/wait_for_test_docker_database.py
index 5020cca27..81b47ab5d 100644
--- a/exasol_integration_test_docker_environment/lib/test_environment/database_waiters/wait_for_test_docker_database.py
+++ b/exasol_integration_test_docker_environment/lib/test_environment/database_waiters/wait_for_test_docker_database.py
@@ -25,14 +25,15 @@ class WaitForTestDockerDatabase(DockerBaseTask, DatabaseCredentialsParameter):
     attempt = luigi.IntParameter(1)
 
     def run_task(self):
-        test_container = self._client.containers.get(self.test_container_info.container_name)
-        db_container_name = self.database_info.container_info.container_name
-        db_container = self._client.containers.get(db_container_name)
-        is_database_ready = \
-            self.wait_for_database_startup(test_container, db_container)
-        after_startup_db_log_file = self.get_log_path().joinpath("after_startup_db_log.tar.gz")
-        self.save_db_log_files_as_gzip_tar(after_startup_db_log_file, db_container)
-        self.return_object(is_database_ready)
+        with self._get_docker_client() as docker_client:
+            test_container = docker_client.containers.get(self.test_container_info.container_name)
+            db_container_name = self.database_info.container_info.container_name
+            db_container = docker_client.containers.get(db_container_name)
+            is_database_ready = \
+                self.wait_for_database_startup(test_container, db_container)
+            after_startup_db_log_file = self.get_log_path().joinpath("after_startup_db_log.tar.gz")
+            self.save_db_log_files_as_gzip_tar(after_startup_db_log_file, db_container)
+            self.return_object(is_database_ready)
 
     def wait_for_database_startup(self,
                                   test_container: Container,
diff --git a/exasol_integration_test_docker_environment/lib/test_environment/prepare_network_for_test_environment.py b/exasol_integration_test_docker_environment/lib/test_environment/prepare_network_for_test_environment.py
index 0383ea308..06e25f8f1 100644
--- a/exasol_integration_test_docker_environment/lib/test_environment/prepare_network_for_test_environment.py
+++ b/exasol_integration_test_docker_environment/lib/test_environment/prepare_network_for_test_environment.py
@@ -29,50 +29,54 @@ def run_task(self):
         self.return_object(self.network_info)
 
     def get_network_info(self, reused: bool):
-        network_properties = self._low_level_client.inspect_network(self.network_name)
-        network_config = network_properties["IPAM"]["Config"][0]
-        return DockerNetworkInfo(network_name=self.network_name, subnet=network_config["Subnet"],
-                                 gateway=network_config["Gateway"], reused=reused)
+        with self._get_docker_client() as docker_client:
+            network_properties = docker_client.api.inspect_network(self.network_name)
+            network_config = network_properties["IPAM"]["Config"][0]
+            return DockerNetworkInfo(network_name=self.network_name, subnet=network_config["Subnet"],
+                                     gateway=network_config["Gateway"], reused=reused)
 
     def create_docker_network(self) -> DockerNetworkInfo:
         self.remove_container(self.test_container_name)
         if self.db_container_name is not None:
             self.remove_container(self.db_container_name)
         self.remove_network(self.network_name)
-        network = self._client.networks.create(
-            name=self.network_name,
-            driver="bridge",
-        )
-        network_info = self.get_network_info(reused=False)
-        subnet = network_info.subnet
-        gateway = network_info.gateway
-        ipam_pool = docker.types.IPAMPool(
-            subnet=subnet,
-            gateway=gateway
-        )
-        ipam_config = docker.types.IPAMConfig(
-            pool_configs=[ipam_pool]
-        )
-        self.remove_network(self.network_name)  # TODO race condition possible, add retry
-        network = self._client.networks.create(
-            name=self.network_name,
-            driver="bridge",
-            ipam=ipam_config
-        )
+        with self._get_docker_client() as docker_client:
+            network = docker_client.networks.create(
+                name=self.network_name,
+                driver="bridge",
+            )
+            network_info = self.get_network_info(reused=False)
+            subnet = network_info.subnet
+            gateway = network_info.gateway
+            ipam_pool = docker.types.IPAMPool(
+                subnet=subnet,
+                gateway=gateway
+            )
+            ipam_config = docker.types.IPAMConfig(
+                pool_configs=[ipam_pool]
+            )
+            self.remove_network(self.network_name)  # TODO race condition possible, add retry
+            network = docker_client.networks.create(
+                name=self.network_name,
+                driver="bridge",
+                ipam=ipam_config
+            )
         return network_info
 
     def remove_network(self, network_name):
         try:
-            self._client.networks.get(network_name).remove()
-            self.logger.info("Removed network %s", network_name)
+            with self._get_docker_client() as docker_client:
+                docker_client.networks.get(network_name).remove()
+                self.logger.info("Removed network %s", network_name)
         except docker.errors.NotFound:
             pass
 
     def remove_container(self, container_name: str):
         try:
-            container = self._client.containers.get(container_name)
-            container.remove(force=True)
-            self.logger.info("Removed container %s", container_name)
+            with self._get_docker_client() as docker_client:
+                container = docker_client.containers.get(container_name)
+                container.remove(force=True)
+                self.logger.info("Removed container %s", container_name)
         except docker.errors.NotFound:
             pass
 
diff --git a/exasol_integration_test_docker_environment/lib/test_environment/spawn_test_container.py b/exasol_integration_test_docker_environment/lib/test_environment/spawn_test_container.py
index f321ce133..549e06aa7 100644
--- a/exasol_integration_test_docker_environment/lib/test_environment/spawn_test_container.py
+++ b/exasol_integration_test_docker_environment/lib/test_environment/spawn_test_container.py
@@ -36,22 +36,24 @@ def __init__(self, *args, **kwargs):
                 % self.ip_address_index_in_subnet)
 
     def register_required(self):
-        self.test_container_image_future = self.register_dependency(DockerTestContainerBuild())
-        self.export_directory_future = self.register_dependency(CreateExportDirectory())
+        self.test_container_image_future = \
+            self.register_dependency(self.create_child_task(task_class=DockerTestContainerBuild))
+        self.export_directory_future = \
+            self.register_dependency(self.create_child_task(task_class=CreateExportDirectory))
 
     def is_reuse_possible(self) -> bool:
         test_container_image_info = \
             self.get_values_from_futures(self.test_container_image_future)["test-container"]  # type: ImageInfo
         test_container = None
-        try:
-            test_container = \
-                self._client.containers.get(self.test_container_name)
-        except Exception as e:
-            pass
-        ret_val = self.network_info.reused and self.reuse_test_container and \
-               test_container is not None and \
-               test_container_image_info.get_target_complete_name() in test_container.image.tags and \
-               test_container_image_info.image_state == ImageState.USED_LOCAL.name
+        with self._get_docker_client() as docker_client:
+            try:
+                test_container = docker_client.containers.get(self.test_container_name)
+            except Exception as e:
+                pass
+            ret_val = self.network_info.reused and self.reuse_test_container and \
+                      test_container is not None and \
+                      test_container_image_info.get_target_complete_name() in test_container.image.tags and \
+                      test_container_image_info.image_state == ImageState.USED_LOCAL.name
 
         return ret_val
 
@@ -64,20 +66,20 @@ def run_task(self):
             container_info = self._try_to_reuse_test_container(ip_address, self.network_info)
         if container_info is None:
             container_info = self._create_test_container(ip_address, self.network_info)
-        test_container = \
-            self._client.containers.get(self.test_container_name)
+        with self._get_docker_client() as docker_client:
+            docker_client.containers.get(self.test_container_name)
         self._copy_tests()
         self.return_object(container_info)
 
     def _copy_tests(self):
         self.logger.warning("Copy tests in test container %s.", self.test_container_name)
-        test_container = \
-            self._client.containers.get(self.test_container_name)
-        try:
-            test_container.exec_run(cmd="rm -r /tests")
-        except:
-            pass
-        test_container.exec_run(cmd="cp -r /tests_src /tests")
+        with self._get_docker_client() as docker_client:
+            test_container = docker_client.containers.get(self.test_container_name)
+            try:
+                test_container.exec_run(cmd="rm -r /tests")
+            except:
+                pass
+            test_container.exec_run(cmd="cp -r /tests_src /tests")
 
     def _try_to_reuse_test_container(self, ip_address: str,
                                      network_info: DockerNetworkInfo) -> ContainerInfo:
@@ -114,30 +116,32 @@ def _create_test_container(self, ip_address,
                 "mode": "rw"
             }
         }
-        docker_unix_sockets = [i for i in self._client.api.adapters.values() if isinstance(i, unixconn.UnixHTTPAdapter)]
-        if len(docker_unix_sockets) > 0:
-            host_docker_socker_path = docker_unix_sockets[0].socket_path
-            volumes[host_docker_socker_path] = {
-                "bind": "/var/run/docker.sock",
-                "mode": "rw"
-            }
-        test_container = \
-            self._client.containers.create(
-                image=test_container_image_info.get_target_complete_name(),
-                name=self.test_container_name,
-                network_mode=None,
-                command="sleep infinity",
-                detach=True,
-                volumes=volumes,
-                labels={"test_environment_name": self.environment_name, "container_type": "test_container"},
-                runtime=self.docker_runtime
-            )
-        docker_network = self._client.networks.get(network_info.network_name)
-        network_aliases = self._get_network_aliases()
-        docker_network.connect(test_container, ipv4_address=ip_address, aliases=network_aliases)
-        test_container.start()
-        container_info = self.create_container_info(ip_address, network_aliases, network_info)
-        return container_info
+        with self._get_docker_client() as docker_client:
+            docker_unix_sockets = [i for i in docker_client.api.adapters.values()
+                                   if isinstance(i, unixconn.UnixHTTPAdapter)]
+            if len(docker_unix_sockets) > 0:
+                host_docker_socker_path = docker_unix_sockets[0].socket_path
+                volumes[host_docker_socker_path] = {
+                    "bind": "/var/run/docker.sock",
+                    "mode": "rw"
+                }
+            test_container = \
+                docker_client.containers.create(
+                    image=test_container_image_info.get_target_complete_name(),
+                    name=self.test_container_name,
+                    network_mode=None,
+                    command="sleep infinity",
+                    detach=True,
+                    volumes=volumes,
+                    labels={"test_environment_name": self.environment_name, "container_type": "test_container"},
+                    runtime=self.docker_runtime
+                )
+            docker_network = docker_client.networks.get(network_info.network_name)
+            network_aliases = self._get_network_aliases()
+            docker_network.connect(test_container, ipv4_address=ip_address, aliases=network_aliases)
+            test_container.start()
+            container_info = self.create_container_info(ip_address, network_aliases, network_info)
+            return container_info
 
     def _get_network_aliases(self):
         network_aliases = ["test_container", self.test_container_name]
@@ -145,13 +149,14 @@ def _get_network_aliases(self):
 
     def create_container_info(self, ip_address: str, network_aliases: List[str],
                               network_info: DockerNetworkInfo) -> ContainerInfo:
-        test_container = self._client.containers.get(self.test_container_name)
-        if test_container.status != "running":
-            raise Exception(f"Container {self.test_container_name} not running")
-        container_info = ContainerInfo(container_name=self.test_container_name,
-                                       ip_address=ip_address,
-                                       network_aliases=network_aliases,
-                                       network_info=network_info)
+        with self._get_docker_client() as docker_client:
+            test_container = docker_client.containers.get(self.test_container_name)
+            if test_container.status != "running":
+                raise Exception(f"Container {self.test_container_name} not running")
+            container_info = ContainerInfo(container_name=self.test_container_name,
+                                           ip_address=ip_address,
+                                           network_aliases=network_aliases,
+                                           network_info=network_info)
         return container_info
 
     def _get_export_directory(self):
@@ -159,18 +164,17 @@ def _get_export_directory(self):
 
     def _remove_container(self, container_name: str):
         try:
-            container = self._client.containers.get(container_name)
-            container.remove(force=True)
-            self.logger.info(f"Removed container: name: '{container_name}', id: '{container.short_id}'")
+            with self._get_docker_client() as docker_client:
+                container = docker_client.containers.get(container_name)
+                container.remove(force=True)
+                self.logger.info(f"Removed container: name: '{container_name}', id: '{container.short_id}'")
         except Exception as e:
             pass
 
-    def cleanup_task(self, success:bool):
-        self.logger.info(f"___________cleanup_task")
+    def cleanup_task(self, success: bool):
         if (success and not self.no_test_container_cleanup_after_success) or \
                 (not success and not self.no_test_container_cleanup_after_failure):
             try:
-                print(f"Cleaning up container %s", self.test_container_name)
                 self.logger.info(f"Cleaning up container %s", self.test_container_name)
                 self._remove_container(self.test_container_name)
             except Exception as e:
diff --git a/exasol_integration_test_docker_environment/lib/test_environment/spawn_test_database.py b/exasol_integration_test_docker_environment/lib/test_environment/spawn_test_database.py
index 2bc41b2a7..0106658de 100644
--- a/exasol_integration_test_docker_environment/lib/test_environment/spawn_test_database.py
+++ b/exasol_integration_test_docker_environment/lib/test_environment/spawn_test_database.py
@@ -79,54 +79,56 @@ def _handle_output(self, output_generator, image_info: ImageInfo):
 
     def _create_database_container(self, db_ip_address: str, db_private_network: str):
         self.logger.info("Starting database container %s", self.db_container_name)
-        try:
-            self._client.containers.get(self.db_container_name).remove(force=True, v=True)
-        except:
-            pass
-        docker_db_image_info = self._pull_docker_db_images_if_necassary()
-        db_volume = self._prepare_db_volume(db_private_network, docker_db_image_info)
-        ports = {}
-        if self.database_port_forward is not None:
-            ports[f"{DB_PORT}/tcp"] = ('0.0.0.0', int(self.database_port_forward))
-        if self.bucketfs_port_forward is not None:
-            ports[f"{BUCKETFS_PORT}/tcp"] = ('0.0.0.0', int(self.bucketfs_port_forward))
-        db_container = \
-            self._client.containers.create(
-                image="%s" % (docker_db_image_info.get_source_complete_name()),
-                name=self.db_container_name,
-                detach=True,
-                privileged=True,
-                volumes={db_volume.name: {"bind": "/exa", "mode": "rw"}},
-                network_mode=None,
-                ports=ports,
-                runtime=self.docker_runtime
-            )
-        docker_network = self._client.networks.get(self.network_info.network_name)
-        network_aliases = self._get_network_aliases()
-        docker_network.connect(db_container, ipv4_address=db_ip_address, aliases=network_aliases)
-        db_container.start()
-        database_info = self._create_database_info(db_ip_address=db_ip_address, reused=False)
-        return database_info
+        with self._get_docker_client() as docker_client:
+            try:
+                docker_client.containers.get(self.db_container_name).remove(force=True, v=True)
+            except:
+                pass
+            docker_db_image_info = self._pull_docker_db_images_if_necassary()
+            db_volume = self._prepare_db_volume(docker_client, db_private_network, docker_db_image_info)
+            ports = {}
+            if self.database_port_forward is not None:
+                ports[f"{DB_PORT}/tcp"] = ('0.0.0.0', int(self.database_port_forward))
+            if self.bucketfs_port_forward is not None:
+                ports[f"{BUCKETFS_PORT}/tcp"] = ('0.0.0.0', int(self.bucketfs_port_forward))
+            db_container = \
+                docker_client.containers.create(
+                    image="%s" % (docker_db_image_info.get_source_complete_name()),
+                    name=self.db_container_name,
+                    detach=True,
+                    privileged=True,
+                    volumes={db_volume.name: {"bind": "/exa", "mode": "rw"}},
+                    network_mode=None,
+                    ports=ports,
+                    runtime=self.docker_runtime
+                )
+            docker_network = docker_client.networks.get(self.network_info.network_name)
+            network_aliases = self._get_network_aliases()
+            docker_network.connect(db_container, ipv4_address=db_ip_address, aliases=network_aliases)
+            db_container.start()
+            database_info = self._create_database_info(db_ip_address=db_ip_address, reused=False)
+            return database_info
 
     def _get_network_aliases(self):
         network_aliases = ["exasol_test_database", self.db_container_name]
         return network_aliases
 
     def _create_database_info(self, db_ip_address: str, reused: bool) -> DatabaseInfo:
-        db_container = self._client.containers.get(self.db_container_name)
-        if db_container.status != "running":
-            raise Exception(f"Container {self.db_container_name} not running")
-        network_aliases = self._get_network_aliases()
-        container_info = \
-            ContainerInfo(container_name=self.db_container_name,
-                          ip_address=db_ip_address,
-                          network_aliases=network_aliases,
-                          network_info=self.network_info,
-                          volume_name=self._get_db_volume_name())
-        database_info = \
-            DatabaseInfo(host=db_ip_address, db_port=DB_PORT, bucketfs_port=BUCKETFS_PORT,
-                         reused=reused, container_info=container_info)
-        return database_info
+        with self._get_docker_client() as docker_client:
+            db_container = docker_client.containers.get(self.db_container_name)
+            if db_container.status != "running":
+                raise Exception(f"Container {self.db_container_name} not running")
+            network_aliases = self._get_network_aliases()
+            container_info = \
+                ContainerInfo(container_name=self.db_container_name,
+                              ip_address=db_ip_address,
+                              network_aliases=network_aliases,
+                              network_info=self.network_info,
+                              volume_name=self._get_db_volume_name())
+            database_info = \
+                DatabaseInfo(host=db_ip_address, db_port=DB_PORT, bucketfs_port=BUCKETFS_PORT,
+                             reused=reused, container_info=container_info)
+            return database_info
 
     def _pull_docker_db_images_if_necassary(self):
         image_name = "exasol/docker-db"
@@ -137,19 +139,20 @@ def _pull_docker_db_images_if_necassary(self):
             target_tag=self.docker_db_image_version,
             hash_value="", commit="",
             image_description=None)
-        try:
-            self._client.images.get(docker_db_image_info.get_source_complete_name())
-        except docker.errors.ImageNotFound as e:
-            self.logger.info("Pulling docker-db image %s",
-                             docker_db_image_info.get_source_complete_name())
-            output_generator = self._low_level_client.pull(
-                docker_db_image_info.source_repository_name,
-                tag=docker_db_image_info.source_tag,
-                stream=True)
-            self._handle_output(output_generator, docker_db_image_info)
+        with self._get_docker_client() as docker_client:
+            try:
+                docker_client.images.get(docker_db_image_info.get_source_complete_name())
+            except docker.errors.ImageNotFound as e:
+                self.logger.info("Pulling docker-db image %s",
+                                 docker_db_image_info.get_source_complete_name())
+                output_generator = docker_client.api.pull(
+                    docker_db_image_info.source_repository_name,
+                    tag=docker_db_image_info.source_tag,
+                    stream=True)
+                self._handle_output(output_generator, docker_db_image_info)
         return docker_db_image_info
 
-    def _prepare_db_volume(self, db_private_network: str,
+    def _prepare_db_volume(self, docker_client, db_private_network: str,
                            docker_db_image_info: ImageInfo) -> Volume:
         db_volume_name = self._get_db_volume_name()
         db_volume_preperation_container_name = self._get_db_volume_preperation_container_name()
@@ -157,7 +160,7 @@ def _prepare_db_volume(self, db_private_network: str,
         self._remove_volume(db_volume_name)
         db_volume, volume_preparation_container = \
             volume_preparation_container, volume_preparation_container = \
-            self._create_volume_and_container(db_volume_name,
+            self._create_volume_and_container(docker_client, db_volume_name,
                                               db_volume_preperation_container_name)
         try:
             self._upload_init_db_files(volume_preparation_container,
@@ -177,23 +180,25 @@ def _get_db_volume_name(self):
 
     def _remove_container(self, db_volume_preperation_container_name):
         try:
-            self._client.containers.get(db_volume_preperation_container_name).remove(force=True)
-            self.logger.info("Removed container %s", db_volume_preperation_container_name)
+            with self._get_docker_client() as docker_client:
+                docker_client.containers.get(db_volume_preperation_container_name).remove(force=True)
+                self.logger.info("Removed container %s", db_volume_preperation_container_name)
         except docker.errors.NotFound:
             pass
 
     def _remove_volume(self, db_volume_name):
         try:
-            self._client.volumes.get(db_volume_name).remove(force=True)
-            self.logger.info("Removed volume %s", db_volume_name)
+            with self._get_docker_client() as docker_client:
+                docker_client.volumes.get(db_volume_name).remove(force=True)
+                self.logger.info("Removed volume %s", db_volume_name)
         except docker.errors.NotFound:
             pass
 
-    def _create_volume_and_container(self, db_volume_name, db_volume_preperation_container_name) \
+    def _create_volume_and_container(self, docker_client, db_volume_name, db_volume_preperation_container_name) \
             -> Tuple[Volume, Container]:
-        db_volume = self._client.volumes.create(db_volume_name)
+        db_volume = docker_client.volumes.create(db_volume_name)
         volume_preparation_container = \
-            self._client.containers.run(
+            docker_client.containers.run(
                 image="ubuntu:18.04",
                 name=db_volume_preperation_container_name,
                 auto_remove=True,
diff --git a/exasol_integration_test_docker_environment/test/test_base_task.py b/exasol_integration_test_docker_environment/test/test_base_task.py
index 0472095f5..58f932905 100644
--- a/exasol_integration_test_docker_environment/test/test_base_task.py
+++ b/exasol_integration_test_docker_environment/test/test_base_task.py
@@ -1,11 +1,11 @@
 import shutil
 import time
 import unittest
-from datetime import datetime
 
 import luigi
 from luigi import Parameter, Config
 
+from exasol_integration_test_docker_environment.cli.common import generate_root_task
 from exasol_integration_test_docker_environment.lib.base.dependency_logger_base_task import DependencyLoggerBaseTask
 from exasol_integration_test_docker_environment.lib.base.json_pickle_parameter import JsonPickleParameter
 
@@ -14,7 +14,7 @@
 
 class TestTask1(TestBaseTask):
     def register_required(self):
-        self.task2 = self.register_dependency(TestTask2())
+        self.task2 = self.register_dependency(self.create_child_task(task_class=TestTask2))
 
     def run_task(self):
         self.logger.info("RUN")
@@ -70,7 +70,7 @@ class TestParameter(Config):
 class TestTask7(TestBaseTask, TestParameter):
 
     def register_required(self):
-        task8 = self.create_child_task_with_common_params(TestTask8, new_parameter="new")
+        task8 = self.create_child_task_with_common_params(task_class=TestTask8, new_parameter="new")
         self.task8_future = self.register_dependency(task8)
 
     def run_task(self):
@@ -104,11 +104,11 @@ def __init__(self, *args, **kwargs):
 
     def register_required(self):
         inputs = [Data(i, f"{i}") for i in range(2)]
-        tasks = [TestTask10(parameter_1=input) for input in inputs]
+        tasks = [self.create_child_task(task_class=TestTask10, parameter_1=input) for input in inputs]
         self.register_dependencies(tasks)
 
     def run_task(self):
-        yield from self.run_dependencies(TestTask10(parameter_1=Data(3,"3")))
+        yield from self.run_dependencies(self.create_child_task(task_class=TestTask10, parameter_1=Data(3, "3")))
 
 
 class TestTask10(TestBaseTask):
@@ -118,6 +118,7 @@ def run_task(self):
         time.sleep(1)
         print(self.parameter_1)
 
+
 class TestTask9_Fail(TestBaseTask):
 
     def __init__(self, *args, **kwargs):
@@ -127,12 +128,13 @@ def register_required(self):
         pass
 
     def run_task(self):
-        yield from self.run_dependencies(TestTask10(parameter_1=Data1()))
+        yield from self.run_dependencies(self.create_child_task(task_class=TestTask10, parameter_1=Data1()))
+
 
 class TestTask11(TestBaseTask):
 
     def run_task(self):
-        tasks = [TestTask12(p=f"{i}") for i in range(10)]
+        tasks = [self.create_child_task(task_class=TestTask12, p=f"{i}") for i in range(10)]
         self.logger.info(tasks)
         yield from self.run_dependencies(tasks)
 
@@ -151,7 +153,7 @@ def run_task(self):
 class TestTask13(TestBaseTask):
 
     def register_required(self):
-        tasks = [TestTask14(p=f"{i}") for i in range(10)]
+        tasks = [self.create_child_task(task_class=TestTask14, p=f"{i}") for i in range(10)]
         self.register_dependencies(tasks)
 
     def run_task(self):
@@ -162,11 +164,12 @@ class TestTask14(TestBaseTask):
     p = Parameter()
 
     def register_required(self):
-        self.register_dependency(TestTask15())
+        self.register_dependency(self.create_child_task(task_class=TestTask15))
 
     def run_task(self):
         pass
 
+
 class TestTask15(TestBaseTask):
     def run_task(self):
         raise Exception("%s" % self.task_id)
@@ -174,37 +177,26 @@ def run_task(self):
 
 class BaseTaskTest(unittest.TestCase):
 
-    def set_job_id(self, task_cls):
-        strftime = datetime.now().strftime('%Y_%m_%d_%H_%M_%S')
-        job_id = f"{strftime}_{task_cls.__name__}"
-        config = luigi.configuration.get_config()
-        config.set('job_config', 'job_id', job_id)
-        # config.reload()
-
     def test_dependency_creation(self):
-        self.set_job_id(TestTask1)
-        task = TestTask1()
+        task = generate_root_task(task_class=TestTask1)
         luigi.build([task], workers=1, local_scheduler=True, log_level="INFO")
         if task._get_tmp_path_for_job().exists():
             shutil.rmtree(str(task._get_tmp_path_for_job()))
 
     def test_failing_task(self):
-        self.set_job_id(TestTask4)
-        task = TestTask4()
+        task = generate_root_task(task_class=TestTask4)
         luigi.build([task], workers=1, local_scheduler=True, log_level="INFO")
         if task._get_tmp_path_for_job().exists():
             shutil.rmtree(str(task._get_tmp_path_for_job()))
 
     def test_common_parameter(self):
-        self.set_job_id(TestTask7)
-        task = TestTask7(test_parameter="input")
+        task = generate_root_task(task_class=TestTask7, test_parameter="input")
         luigi.build([task], workers=1, local_scheduler=True, log_level="INFO")
         if task._get_tmp_path_for_job().exists():
             shutil.rmtree(str(task._get_tmp_path_for_job()))
 
     def test_json_pickle_parameter_success(self):
-        self.set_job_id(TestTask9_Success)
-        task = TestTask9_Success()
+        task = generate_root_task(task_class=TestTask9_Success)
         try:
             luigi.build([task], workers=3, local_scheduler=True, log_level="INFO")
         finally:
@@ -212,8 +204,7 @@ def test_json_pickle_parameter_success(self):
                 shutil.rmtree(str(task._get_tmp_path_for_job()))
     
     def test_json_pickle_parameter_fail(self):
-        self.set_job_id(TestTask9_Fail)
-        task = TestTask9_Fail()
+        task = generate_root_task(task_class=TestTask9_Fail)
         try:
             with self.assertRaises(Exception) as context:
                 luigi.build([task], workers=3, local_scheduler=True, log_level="INFO")
@@ -223,8 +214,7 @@ def test_json_pickle_parameter_fail(self):
                 shutil.rmtree(str(task._get_tmp_path_for_job()))
 
     def test_collect_failures_diffrent_task_fail(self):
-        self.set_job_id(TestTask11)
-        task = TestTask11()
+        task = generate_root_task(task_class=TestTask11)
         try:
             luigi.build([task], workers=5, local_scheduler=True, log_level="INFO")
             failures = task.collect_failures()
@@ -240,8 +230,7 @@ def test_collect_failures_diffrent_task_fail(self):
                 shutil.rmtree(str(task._get_tmp_path_for_job()))
 
     def test_collect_failures_one_task_fails_but_is_dependency_of_multiple(self):
-        self.set_job_id(TestTask13)
-        task = TestTask13()
+        task = generate_root_task(task_class=TestTask13)
         try:
             luigi.build([task], workers=3, local_scheduler=True, log_level="INFO")
             failures = task.collect_failures()
@@ -256,5 +245,6 @@ def test_collect_failures_one_task_fails_but_is_dependency_of_multiple(self):
             if task._get_tmp_path_for_job().exists():
                 shutil.rmtree(str(task._get_tmp_path_for_job()))
 
+
 if __name__ == '__main__':
     unittest.main()
diff --git a/exasol_integration_test_docker_environment/test/test_docker_build_base.py b/exasol_integration_test_docker_environment/test/test_docker_build_base.py
index 445aa24f7..71ac0a885 100644
--- a/exasol_integration_test_docker_environment/test/test_docker_build_base.py
+++ b/exasol_integration_test_docker_environment/test/test_docker_build_base.py
@@ -1,31 +1,23 @@
 import shutil
-import time
 import unittest
 from pathlib import Path
-from datetime import datetime
 
 import docker
 import luigi
-from luigi import Parameter, Config
-
-from exasol_integration_test_docker_environment.lib.base.dependency_logger_base_task import DependencyLoggerBaseTask
-from exasol_integration_test_docker_environment.lib.base.json_pickle_parameter import JsonPickleParameter
+from luigi import Parameter
 
 from typing import Set, Dict
 
-from exasol_integration_test_docker_environment.lib.config.docker_config import target_docker_repository_config, \
-    source_docker_repository_config
+from exasol_integration_test_docker_environment.cli.common import generate_root_task
+from exasol_integration_test_docker_environment.lib.docker import ContextDockerClient
 from exasol_integration_test_docker_environment.lib.docker.images.create.docker_build_base import DockerBuildBase
 from exasol_integration_test_docker_environment.lib.docker.images.create.docker_image_analyze_task import \
     DockerAnalyzeImageTask
-from exasol_integration_test_docker_environment.lib.docker.images.push.docker_push_parameter import DockerPushParameter
-from exasol_integration_test_docker_environment.lib.docker.images.push.push_task_creator_for_build_tasks import \
-    PushTaskCreatorFromBuildTasks
 from exasol_integration_test_docker_environment.lib.docker.images.clean.clean_images import CleanImagesStartingWith
 from exasol_integration_test_docker_environment.lib.docker.images.utils import find_images_by_tag
 
-class TestDockerBuildBaseTestAnalyzeImage(DockerAnalyzeImageTask):
 
+class TestDockerBuildBaseTestAnalyzeImage(DockerAnalyzeImageTask):
     task_name = Parameter()
 
     def get_target_repository_name(self) -> str:
@@ -57,9 +49,13 @@ class TestDockerBuildBase(DockerBuildBase):
 
     def get_goal_class_map(self) -> Dict[str, DockerAnalyzeImageTask]:
         goal_class_map = {
-                "test-analyze-image-1": TestDockerBuildBaseTestAnalyzeImage(task_name="test-analyze-image-1"),
-                "test-analyze-image-2": TestDockerBuildBaseTestAnalyzeImage(task_name="test-analyze-image-2")
-                }
+            "test-analyze-image-1":
+                self.create_child_task(task_class=TestDockerBuildBaseTestAnalyzeImage,
+                                       task_name="test-analyze-image-1"),
+            "test-analyze-image-2":
+                self.create_child_task(TestDockerBuildBaseTestAnalyzeImage,
+                                       task_name="test-analyze-image-2")
+        }
         return goal_class_map
 
     def get_default_goals(self) -> Set[str]:
@@ -75,37 +71,29 @@ def run_task(self):
         image_infos = self.get_values_from_futures(image_infos_futures)
         self.return_object(image_infos)
 
-class DockerBuildBaseTest(unittest.TestCase):
 
-    def set_job_id(self, task_cls):
-        strftime = datetime.now().strftime('%Y_%m_%d_%H_%M_%S')
-        job_id = f"{strftime}_{task_cls.__name__}"
-        config = luigi.configuration.get_config()
-        config.set('job_config', 'job_id', job_id)
-        return job_id
+class DockerBuildBaseTest(unittest.TestCase):
 
     def clean(self):
-        jobid = self.set_job_id(CleanImagesStartingWith)
-        task = CleanImagesStartingWith(starts_with_pattern="exasol-test-docker-build-base", jobid=jobid)
+        task = generate_root_task(task_class=CleanImagesStartingWith,
+                                  starts_with_pattern="exasol-test-docker-build-base")
         luigi.build([task], workers=1, local_scheduler=True, log_level="INFO")
         if task._get_tmp_path_for_job().exists():
             shutil.rmtree(str(task._get_tmp_path_for_job()))
 
     def setUp(self):
-        self.client = docker.from_env()
         self.clean()
 
     def tearDown(self):
         self.clean()
-        self.client.close()
 
     def assert_image_exists(self, prefix):
-        image_list = find_images_by_tag(self.client, lambda x: x.startswith(prefix))
-        self.assertEquals(len(image_list),1,f"Image with prefix {prefix} not found")
+        with ContextDockerClient() as docker_client:
+            image_list = find_images_by_tag(docker_client, lambda x: x.startswith(prefix))
+            self.assertEquals(len(image_list), 1, f"Image with prefix {prefix} not found")
 
     def test_default_parameter(self):
-        self.set_job_id(TestDockerBuildBase)
-        task = TestDockerBuildBase()
+        task = generate_root_task(task_class=TestDockerBuildBase)
         try:
             luigi.build([task], workers=1, local_scheduler=True, log_level="INFO")
             self.assert_image_exists("exasol-test-docker-build-base:test-analyze-image-1")
@@ -113,10 +101,8 @@ def test_default_parameter(self):
             if task._get_tmp_path_for_job().exists():
                 shutil.rmtree(str(task._get_tmp_path_for_job()))
 
-
     def test_valid_non_default_goal(self):
-        self.set_job_id(TestDockerBuildBase)
-        task = TestDockerBuildBase(goals=["test-analyze-image-2"])
+        task = generate_root_task(task_class=TestDockerBuildBase, goals=["test-analyze-image-2"])
         try:
             luigi.build([task], workers=1, local_scheduler=True, log_level="INFO")
             self.assert_image_exists("exasol-test-docker-build-base:test-analyze-image-2")
@@ -125,10 +111,9 @@ def test_valid_non_default_goal(self):
                 shutil.rmtree(str(task._get_tmp_path_for_job()))
 
     def test_non_valid_non_default_goal(self):
-        self.set_job_id(TestDockerBuildBase)
         with self.assertRaises(Exception) as contex:
-            task = TestDockerBuildBase(goals=["test-analyze-image-3"])
-        self.assertIn("Unknown goal(s)",str(contex.exception))
+            task = generate_root_task(task_class=TestDockerBuildBase, goals=["test-analyze-image-3"])
+        self.assertIn("Unknown goal(s)", str(contex.exception))
 
 
 if __name__ == '__main__':
diff --git a/exasol_integration_test_docker_environment/test/test_test_container_reuse.py b/exasol_integration_test_docker_environment/test/test_test_container_reuse.py
index 7652acf4c..34b099848 100644
--- a/exasol_integration_test_docker_environment/test/test_test_container_reuse.py
+++ b/exasol_integration_test_docker_environment/test/test_test_container_reuse.py
@@ -4,17 +4,17 @@
 import unittest
 from pathlib import Path
 
-import docker
 import luigi
 
-from exasol_integration_test_docker_environment.test.utils import process_spawn_utils
-from exasol_integration_test_docker_environment.cli.common import set_build_config, set_docker_repository_config
+from exasol_integration_test_docker_environment.cli.common import set_build_config, set_docker_repository_config, \
+    generate_root_task
 from exasol_integration_test_docker_environment.lib.base.docker_base_task import DockerBaseTask
 from exasol_integration_test_docker_environment.lib.data.container_info import ContainerInfo
+from exasol_integration_test_docker_environment.lib.docker import ContextDockerClient
 from exasol_integration_test_docker_environment.lib.test_environment.prepare_network_for_test_environment import \
     PrepareDockerNetworkForTestEnvironment
 from exasol_integration_test_docker_environment.lib.test_environment.spawn_test_container import SpawnTestContainer
-from exasol_integration_test_docker_environment.test.utils import luigi_utils
+from exasol_integration_test_docker_environment.testing import luigi_utils
 
 
 class TestTask(DockerBaseTask):
@@ -22,58 +22,57 @@ class TestTask(DockerBaseTask):
     attempt = luigi.IntParameter()
 
     def run_task(self):
-        docker_network_task_1 = PrepareDockerNetworkForTestEnvironment(
-            environment_name="test_environment_TestContainerReuseTest",
-            network_name="docker_network_TestContainerReuseTest",
-            test_container_name="test_container_TestContainerReuseTest",
-            db_container_name="db_container_TestContainerReuseTest",
-            reuse=self.reuse,
-            no_cleanup_after_success=True,
-            no_cleanup_after_failure=False,
-            attempt=self.attempt
-        )
+        docker_network_task_1 = self.create_child_task(task_class=PrepareDockerNetworkForTestEnvironment,
+                                                       environment_name="test_environment_TestContainerReuseTest",
+                                                       network_name="docker_network_TestContainerReuseTest",
+                                                       test_container_name="test_container_TestContainerReuseTest",
+                                                       db_container_name="db_container_TestContainerReuseTest",
+                                                       reuse=self.reuse,
+                                                       no_cleanup_after_success=True,
+                                                       no_cleanup_after_failure=False,
+                                                       attempt=self.attempt
+                                                       )
         self.docker_network_future_1 = yield from self.run_dependencies(docker_network_task_1)
 
         test_container_task_1 = \
-            SpawnTestContainer(
-                environment_name="test_environment_TestContainerReuseTest",
-                test_container_name="test_container_TestContainerReuseTest",
-                network_info=self.docker_network_future_1.get_output(),
-                ip_address_index_in_subnet=2,
-                attempt=self.attempt,
-                reuse_test_container=self.reuse,
-                no_test_container_cleanup_after_success=True,
-                no_test_container_cleanup_after_failure=False
-            )
+            self.create_child_task(task_class=SpawnTestContainer,
+                                   environment_name="test_environment_TestContainerReuseTest",
+                                   test_container_name="test_container_TestContainerReuseTest",
+                                   network_info=self.docker_network_future_1.get_output(),
+                                   ip_address_index_in_subnet=2,
+                                   attempt=self.attempt,
+                                   reuse_test_container=self.reuse,
+                                   no_test_container_cleanup_after_success=True,
+                                   no_test_container_cleanup_after_failure=False
+                                   )
         test_container_future_1 = yield from self.run_dependencies(test_container_task_1)
         container_info = test_container_future_1.get_output()  # type: ContainerInfo
-        container = self._client.containers.get(container_info.container_name)
-
-        self.return_object({"container_id": container.image.id, "image_id": container.image.id})
+        with ContextDockerClient() as docker_client:
+            container = docker_client.containers.get(container_info.container_name)
+            self.return_object({"container_id": container.image.id, "image_id": container.image.id})
 
 
 class TestContainerReuseTest(unittest.TestCase):
 
     def setUp(self):
-        self.client = docker.from_env()
         resource_directory = Path(Path(__file__).parent, "resources/test_test_container_reuse")
         print("resource_directory content", list(Path(resource_directory).iterdir()))
         self.temp_directory = tempfile.mkdtemp()
         self.working_directory = shutil.copytree(resource_directory,
                                                  Path(self.temp_directory, "test_test_container_reuse"))
-        print("working_directory",self.working_directory)
+        print("working_directory", self.working_directory)
         self.old_working_directory = os.getcwd()
         os.chdir(self.working_directory)
-        print("working_directory content",list(Path(self.working_directory).iterdir()))
+        print("working_directory content", list(Path(self.working_directory).iterdir()))
         self.docker_repository_name = self.__class__.__name__.lower()
-        print("docker_repository_name",self.docker_repository_name)
+        print("docker_repository_name", self.docker_repository_name)
         luigi_utils.clean(self.docker_repository_name)
+        self.setup_luigi_config()
 
     def tearDown(self):
         os.chdir(self.old_working_directory)
         luigi_utils.clean(self.docker_repository_name)
         shutil.rmtree(self.temp_directory)
-        self.client.close()
 
     def setup_luigi_config(self):
         set_build_config(force_rebuild=False,
@@ -94,9 +93,7 @@ def setup_luigi_config(self):
         )
 
     def run1(self):
-        self.setup_luigi_config()
-        luigi_utils.set_job_id(SpawnTestContainer)
-        task = TestTask(reuse=False, attempt=1)
+        task = generate_root_task(task_class=TestTask, reuse=False, attempt=1)
         try:
             success = luigi.build([task], workers=1, local_scheduler=True, log_level="INFO")
             if success:
@@ -104,15 +101,13 @@ def run1(self):
                 task.cleanup(True)
                 return result
             else:
-                task.cleanup(False)
-                Exception("Task failed")
+                raise Exception("Task failed")
         except Exception as e:
             task.cleanup(False)
+            raise RuntimeError("Error spawning test environment") from e
 
     def run2(self):
-        self.setup_luigi_config()
-        luigi_utils.set_job_id(SpawnTestContainer)
-        task = TestTask(reuse=True, attempt=2)
+        task = generate_root_task(task_class=TestTask, reuse=True, attempt=2)
         try:
             success = luigi.build([task], workers=1, local_scheduler=True, log_level="INFO")
 
@@ -121,22 +116,23 @@ def run2(self):
                 return result
             else:
                 raise Exception("Task failed")
-        finally:
+        except Exception as e:
             task.cleanup(False)
+            raise RuntimeError("Error spawning test environment") from e
 
     def test_test_container_no_reuse_after_change(self):
-        p1 = process_spawn_utils.run_in_process(self.run1)
+        p1 = self.run1()
         dockerfile = Path(self.working_directory, "tests/Dockerfile")
         with dockerfile.open("a") as f:
             f.write("\n#Test\n")
-        p2 = process_spawn_utils.run_in_process(self.run2)
+        p2 = self.run2()
         print(p1)
         print(p2)
         assert p1 != p2
 
     def test_test_container_reuse(self):
-        p1 = process_spawn_utils.run_in_process(self.run1)
-        p2 = process_spawn_utils.run_in_process(self.run2)
+        p1 = self.run1()
+        p2 = self.run2()
         print(p1)
         print(p2)
         assert p1 == p2
diff --git a/exasol_integration_test_docker_environment/test/test_test_env_reuse.py b/exasol_integration_test_docker_environment/test/test_test_env_reuse.py
index 62e282575..02a3fa403 100644
--- a/exasol_integration_test_docker_environment/test/test_test_env_reuse.py
+++ b/exasol_integration_test_docker_environment/test/test_test_env_reuse.py
@@ -1,14 +1,13 @@
 import os
 import unittest
 
-import docker
 import luigi
 
 from exasol_integration_test_docker_environment.lib.data.environment_type import EnvironmentType
+from exasol_integration_test_docker_environment.lib.docker import ContextDockerClient
 from exasol_integration_test_docker_environment.lib.test_environment.spawn_test_environment import SpawnTestEnvironment
-from exasol_integration_test_docker_environment.test.utils import process_spawn_utils
-from exasol_integration_test_docker_environment.cli.common import set_docker_repository_config
-from exasol_integration_test_docker_environment.test.utils import luigi_utils
+from exasol_integration_test_docker_environment.cli.common import set_docker_repository_config, generate_root_task
+from exasol_integration_test_docker_environment.testing import luigi_utils
 from exasol_integration_test_docker_environment.cli.options import test_environment_options
 
 
@@ -25,7 +24,6 @@ def env_name(self):
         return self.__class__.__name__.lower()
 
     def setUp(self):
-        self._client = docker.from_env()
         self._docker_repository_name = self.env_name()
         print("docker_repository_name", self._docker_repository_name)
         luigi_utils.clean(self._docker_repository_name)
@@ -34,9 +32,10 @@ def setUp(self):
         if "EXASOL_VERSION" in os.environ and os.environ["EXASOL_VERSION"] != "default":
             self.docker_db_version_parameter = os.environ["EXASOL_VERSION"]
 
+        self.setup_luigi_config()
+
     def tearDown(self):
         luigi_utils.clean(self._docker_repository_name)
-        self._client.close()
 
     def setup_luigi_config(self):
         set_docker_repository_config(
@@ -47,44 +46,38 @@ def setup_luigi_config(self):
             kind="target"
         )
 
-    def run_spawn_test_env_with_cleanup(self):
-        return self.run_spawn_test_env(cleanup=True)
-
-    def run_spawn_test_env_without_cleanup(self):
-        return self.run_spawn_test_env(cleanup=False)
-
     def run_spawn_test_env(self, cleanup: bool):
         result = None
-        self.setup_luigi_config()
-        luigi_utils.set_job_id(SpawnTestEnvironment)
-        task = SpawnTestEnvironment(reuse_database_setup=True,
-                                    reuse_database=True,
-                                    reuse_test_container=True,
-                                    no_test_container_cleanup_after_success=not cleanup,
-                                    no_database_cleanup_after_success=not cleanup,
-                                    external_exasol_xmlrpc_host="",
-                                    external_exasol_db_host="",
-                                    external_exasol_xmlrpc_port=0,
-                                    external_exasol_db_user="",
-                                    external_exasol_db_password="",
-                                    external_exasol_xmlrpc_user="",
-                                    external_exasol_xmlrpc_password="",
-                                    external_exasol_xmlrpc_cluster_name="",
-                                    external_exasol_bucketfs_write_password="",
-                                    environment_type=EnvironmentType.docker_db,
-                                    environment_name=self.env_name(),
-                                    docker_db_image_version=self.docker_db_version_parameter,
-                                    docker_db_image_name="exasol/docker-db"
-                                    )
+
+        task = generate_root_task(task_class=SpawnTestEnvironment,
+                                  reuse_database_setup=True,
+                                  reuse_database=True,
+                                  reuse_test_container=True,
+                                  no_test_container_cleanup_after_success=not cleanup,
+                                  no_database_cleanup_after_success=not cleanup,
+                                  external_exasol_xmlrpc_host="",
+                                  external_exasol_db_host="",
+                                  external_exasol_xmlrpc_port=0,
+                                  external_exasol_db_user="",
+                                  external_exasol_db_password="",
+                                  external_exasol_xmlrpc_user="",
+                                  external_exasol_xmlrpc_password="",
+                                  external_exasol_xmlrpc_cluster_name="",
+                                  external_exasol_bucketfs_write_password="",
+                                  environment_type=EnvironmentType.docker_db,
+                                  environment_name=self.env_name(),
+                                  docker_db_image_version=self.docker_db_version_parameter,
+                                  docker_db_image_name="exasol/docker-db"
+                                  )
         try:
             success = luigi.build([task], workers=1, local_scheduler=True, log_level="INFO")
             if success:
                 result = task
             else:
-                task.cleanup(False)
-                Exception("Task failed")
+                raise Exception("Task failed")
         except Exception as e:
             task.cleanup(False)
+            raise RuntimeError("Error spawning test environment") from e
         return result
 
     def _create_exaplus_check_cmd(self, test_environment_info):
@@ -96,12 +89,13 @@ def _create_exaplus_check_cmd(self, test_environment_info):
         return f"""$EXAPLUS -c '{database_host}:{database_port}' -u '{username}' -p '{password}' -sql \\\"{q}\\\""""
 
     def _exec_cmd_in_test_container(self, test_environment_info, cmd):
-        bash_cmd = f"""bash -c "{cmd}" """
-        test_container = self._client.containers.get(test_environment_info.test_container_info.container_name)
+        with ContextDockerClient() as docker_client:
+            bash_cmd = f"""bash -c "{cmd}" """
+            test_container = docker_client.containers.get(test_environment_info.test_container_info.container_name)
 
-        exit_code, output = test_container.exec_run(cmd=bash_cmd)
-        self.assertEquals(exit_code, 0)
-        return output.decode('utf-8')
+            exit_code, output = test_container.exec_run(cmd=bash_cmd)
+            self.assertEquals(exit_code, 0)
+            return output.decode('utf-8')
 
     def _verify_test_data(self, test_environment_info):
         cmd = self._create_exaplus_check_cmd(test_environment_info)
@@ -112,24 +106,26 @@ def _verify_test_data(self, test_environment_info):
         self.assertIn("ENGINETABLE", [output_entry.strip() for output_entry in output_str.split(sep='\n')])
 
     def test_initial_reuse_database_setup_populates_data(self):
-        task = process_spawn_utils.run_in_process(self.run_spawn_test_env_with_cleanup)
+        task = self.run_spawn_test_env(cleanup=True)
         self._verify_test_data(task.get_return_object())
         task.cleanup(True)
 
     def get_instance_ids(self, test_environment_info):
-        test_container = self._client.containers.get(test_environment_info.test_container_info.container_name)
-        db_container = self._client.containers.get(test_environment_info.database_info.container_info.container_name)
-        network = self._client.networks.get(test_environment_info.network_info.network_name)
-        return test_container.id, db_container.id, network.id
+        with ContextDockerClient() as docker_client:
+            test_container = docker_client.containers.get(test_environment_info.test_container_info.container_name)
+            db_container = \
+                docker_client.containers.get(test_environment_info.database_info.container_info.container_name)
+            network = docker_client.networks.get(test_environment_info.network_info.network_name)
+            return test_container.id, db_container.id, network.id
 
     def test_reuse_env_same_instances(self):
-        task = process_spawn_utils.run_in_process(self.run_spawn_test_env_without_cleanup)
+        task = self.run_spawn_test_env(cleanup=False)
         test_environment_info = task.get_return_object()
         old_instance_ids = self.get_instance_ids(test_environment_info)
         # This clean is supposed to not remove docker instances
         task.cleanup(True)
 
-        task = process_spawn_utils.run_in_process(self.run_spawn_test_env_with_cleanup)
+        task = self.run_spawn_test_env(cleanup=True)
         test_environment_info = task.get_return_object()
         new_instance_ids = self.get_instance_ids(test_environment_info)
         self.assertEquals(old_instance_ids, new_instance_ids)
diff --git a/exasol_integration_test_docker_environment/test/test_test_environment.py b/exasol_integration_test_docker_environment/test/test_test_environment.py
index 022bc5a2d..a5a5d1e3c 100644
--- a/exasol_integration_test_docker_environment/test/test_test_environment.py
+++ b/exasol_integration_test_docker_environment/test/test_test_environment.py
@@ -1,9 +1,9 @@
 import unittest
 from pathlib import Path
 
-import docker
-
+from exasol_integration_test_docker_environment.lib.docker import ContextDockerClient
 from exasol_integration_test_docker_environment.testing import utils
+from exasol_integration_test_docker_environment.testing.exaslct_test_environment import ExaslctTestEnvironment
 
 
 class DockerTestEnvironmentTest(unittest.TestCase):
@@ -14,65 +14,58 @@ def setUpClass(cls):
         # We can't use start-test-env. because it only mounts ./ and
         # doesn't work with --build_ouput-directory
         cls.test_environment = \
-            utils.ExaslctTestEnvironment(
+            ExaslctTestEnvironment(
                 cls,
                 utils.INTEGRATION_TEST_DOCKER_ENVIRONMENT_DEFAULT_BIN,
                 clean_images_at_close=False)
         # TODO cls.test_environment.clean_images()
         cls.docker_environment_name = cls.__name__
-        cls.on_host_docker_environment, cls.google_cloud_docker_environment = \
-            cls.test_environment.spawn_docker_test_environment(cls.docker_environment_name)
-
-    def setUp(self):
-        self.client = docker.from_env()
+        cls.spawned_docker_test_environments = \
+            cls.test_environment.spawn_docker_test_environments(cls.docker_environment_name)
 
     @classmethod
     def tearDownClass(cls):
-        try:
-            cls.on_host_docker_environment.close()
-        except Exception as e:
-            print(e)
-        try:
-            cls.google_cloud_docker_environment.close()
-        except Exception as e:
-            print(e)
-        try:
-            cls.test_environment.close()
-        except Exception as e:
-            print(e)
-
-    def tearDown(self):
-        self.client.close()
+        utils.close_environments(cls.spawned_docker_test_environments, cls.test_environment)
 
     def test_all_containers_started(self):
-        containers = [c.name for c in self.client.containers.list() if self.docker_environment_name in c.name]
-        self.assertEqual(len(containers), 2,
-                         f"Not exactly 2 containers in {containers}.\nStartup log was: {self.on_host_docker_environment.completed_process.stdout.decode('utf8')}")
-        db_container = [c for c in containers if "db_container" in c]
-        self.assertEqual(len(db_container), 1,
-                         f"Found no db container in {containers}.\nStartup log was: {self.on_host_docker_environment.completed_process.stdout.decode('utf8')}")
-        test_container = [c for c in containers if "test_container" in c]
-        self.assertEqual(len(test_container), 1,
-                         f"Found no test container in {containers}.\nStartup log was: {self.on_host_docker_environment.completed_process.stdout.decode('utf8')}")
+        on_host_docker_environment = self.spawned_docker_test_environments.on_host_docker_environment
+        with ContextDockerClient() as docker_client:
+            containers = [c.name for c in docker_client.containers.list() if self.docker_environment_name in c.name]
+            self.assertEqual(len(containers), 2,
+                             f"Not exactly 2 containers in {containers}.\nStartup log was: "
+                             f"{on_host_docker_environment.completed_process.stdout.decode('utf8')}")
+            db_container = [c for c in containers if "db_container" in c]
+            self.assertEqual(len(db_container), 1,
+                             f"Found no db container in {containers}.\nStartup log was: "
+                             f"{on_host_docker_environment.completed_process.stdout.decode('utf8')}")
+            test_container = [c for c in containers if "test_container" in c]
+            self.assertEqual(len(test_container), 1,
+                             f"Found no test container in {containers}.\nStartup log was: "
+                             f"{on_host_docker_environment.completed_process.stdout.decode('utf8')}")
 
     def test_docker_available_in_test_container(self):
-        containers = [c.name for c in self.client.containers.list() if self.docker_environment_name in c.name]
-        test_container = [c for c in containers if "test_container" in c]
-        exit_result = self.client.containers.get(test_container[0]).exec_run("docker ps")
-        exit_code = exit_result[0]
-        output = exit_result[1]
-        self.assertEqual(exit_code, 0,
-                         f"Error while executing 'docker ps' in test container got output\n {output}.\nStartup log was: {self.on_host_docker_environment.completed_process.stdout.decode('utf8')}")
+        on_host_docker_environment = self.spawned_docker_test_environments.on_host_docker_environment
+        with ContextDockerClient() as docker_client:
+            containers = [c.name for c in docker_client.containers.list() if self.docker_environment_name in c.name]
+            test_container = [c for c in containers if "test_container" in c]
+            exit_result = docker_client.containers.get(test_container[0]).exec_run("docker ps")
+            exit_code = exit_result[0]
+            output = exit_result[1]
+            self.assertEqual(exit_code, 0,
+                             f"Error while executing 'docker ps' in test container got output\n {output}.\n"
+                             f"Startup log was: {on_host_docker_environment.completed_process.stdout.decode('utf8')}")
 
     def test_envrionment_info_available(self):
+        on_host_docker_environment = self.spawned_docker_test_environments.on_host_docker_environment
         temp_dir_glob = list(Path(self.test_environment.temp_dir).glob("**/*"))
-        path_filter = f"cache/environments/{self.on_host_docker_environment.name}"
+        path_filter = f"cache/environments/{on_host_docker_environment.name}"
         actual_environment_infos = sorted([path.name
                                            for path in temp_dir_glob
                                            if path_filter in str(path) and "environment_info" in path.name])
         expected_environment_infos = sorted(["environment_info.sh", "environment_info.conf", "environment_info.json"])
         self.assertEqual(actual_environment_infos, expected_environment_infos,
-                         f"Did not found environment_info files on host.\nStartup log was: {self.on_host_docker_environment.completed_process.stdout.decode('utf8')}")
+                         f"Did not found environment_info files on host.\n"
+                         f"Startup log was: {on_host_docker_environment.completed_process.stdout.decode('utf8')}")
 
 
 if __name__ == '__main__':
diff --git a/exasol_integration_test_docker_environment/test/test_test_environment_db_disk_size.py b/exasol_integration_test_docker_environment/test/test_test_environment_db_disk_size.py
index ee0bfc313..2362b6b52 100644
--- a/exasol_integration_test_docker_environment/test/test_test_environment_db_disk_size.py
+++ b/exasol_integration_test_docker_environment/test/test_test_environment_db_disk_size.py
@@ -1,8 +1,8 @@
 import unittest
 
-import docker
-
+from exasol_integration_test_docker_environment.lib.docker import ContextDockerClient
 from exasol_integration_test_docker_environment.testing import utils
+from exasol_integration_test_docker_environment.testing.exaslct_test_environment import ExaslctTestEnvironment
 
 
 class DockerTestEnvironmentDBDiskSizeTest(unittest.TestCase):
@@ -13,64 +13,46 @@ def setUpClass(cls):
         # We can't use start-test-env. because it only mounts ./ and
         # doesn't work with --build_ouput-directory
         cls.test_environment = \
-            utils.ExaslctTestEnvironment(
+            ExaslctTestEnvironment(
                 cls,
                 utils.INTEGRATION_TEST_DOCKER_ENVIRONMENT_DEFAULT_BIN,
                 clean_images_at_close=False)
 
-    def setUp(self):
-        self.client = docker.from_env()
-
     @classmethod
     def tearDownClass(cls):
-        try:
-            cls.test_environment.close()
-        except Exception as e:
-            print(e)
-
-    def tearDown(self):
-        try:
-            self.on_host_docker_environment.close()
-        except Exception as e:
-            print(e)
-        try:
-            self.google_cloud_docker_environment.close()
-        except Exception as e:
-            print(e)
-        self.client.close()
+        utils.close_environments(cls.test_environment)
 
     def assert_disk_size(self, size: str):
-        containers = [c.name for c in self.client.containers.list() if self.docker_environment_name in c.name]
-        db_container = [c for c in containers if "db_container" in c]
-        exit_result = self.client.containers.get(db_container[0]).exec_run("cat /exa/etc/EXAConf")
-        output = exit_result[1].decode("UTF-8")
-        return_code = exit_result[0]
-        if output == '':
-            exit_result = self.client.containers.get(db_container[0]).exec_run("cat /exa/etc/EXAConf")
+        with ContextDockerClient() as docker_client:
+            containers = [c.name for c in docker_client.containers.list() if self.docker_environment_name in c.name]
+            db_container = [c for c in containers if "db_container" in c]
+            exit_result = docker_client.containers.get(db_container[0]).exec_run("cat /exa/etc/EXAConf")
             output = exit_result[1].decode("UTF-8")
             return_code = exit_result[0]
-        self.assertEquals(return_code, 0)
-        self.assertIn(" Size = %s" % size, output)
+            if output == '':
+                exit_result = docker_client.containers.get(db_container[0]).exec_run("cat /exa/etc/EXAConf")
+                output = exit_result[1].decode("UTF-8")
+                return_code = exit_result[0]
+            self.assertEqual(return_code, 0)
+            self.assertIn(" Size = %s" % size, output)
 
     def test_default_db_disk_size(self):
         self.docker_environment_name = "test_default_db_disk_size"
-        self.on_host_docker_environment, self.google_cloud_docker_environment = \
-            self.test_environment.spawn_docker_test_environment(self.docker_environment_name)
-        self.assert_disk_size("2 GiB")
+        with self.test_environment.spawn_docker_test_environments(self.docker_environment_name):
+            self.assert_disk_size("2 GiB")
 
     def test_smallest_valid_db_disk_size(self):
         self.docker_environment_name = "test_smallest_valid_db_disk_size"
-        self.on_host_docker_environment, self.google_cloud_docker_environment = \
-            self.test_environment.spawn_docker_test_environment(self.docker_environment_name,
-                                                                ["--db-disk-size", "'100 MiB'"])
-        self.assert_disk_size("100 MiB")
+        with self.test_environment.spawn_docker_test_environments(self.docker_environment_name,
+                                                                  ["--db-disk-size", "'100 MiB'"]):
+            self.assert_disk_size("100 MiB")
 
     def test_invalid_db_mem_size(self):
         self.docker_environment_name = "test_invalid_db_disk_size"
         with self.assertRaises(Exception) as context:
-            self.on_host_docker_environment, self.google_cloud_docker_environment = \
-                self.test_environment.spawn_docker_test_environment(self.docker_environment_name,
-                                                                    ["--db-disk-size", "'90 MiB'"])
+            with self.test_environment.spawn_docker_test_environments(self.docker_environment_name,
+                                                                      ["--db-disk-size", "'90 MiB'"]):
+                pass
 
 
 if __name__ == '__main__':
diff --git a/exasol_integration_test_docker_environment/test/test_test_environment_db_mem_size.py b/exasol_integration_test_docker_environment/test/test_test_environment_db_mem_size.py
index 54b10b8a6..dddb96dad 100644
--- a/exasol_integration_test_docker_environment/test/test_test_environment_db_mem_size.py
+++ b/exasol_integration_test_docker_environment/test/test_test_environment_db_mem_size.py
@@ -1,8 +1,8 @@
 import unittest
 
-import docker
-
+from exasol_integration_test_docker_environment.lib.docker import ContextDockerClient
 from exasol_integration_test_docker_environment.testing import utils
+from exasol_integration_test_docker_environment.testing.exaslct_test_environment import ExaslctTestEnvironment
 
 
 class DockerTestEnvironmentDBMemSizeTest(unittest.TestCase):
@@ -13,64 +13,47 @@ def setUpClass(cls):
         # We can't use start-test-env. because it only mounts ./ and
         # doesn't work with --build_ouput-directory
         cls.test_environment = \
-            utils.ExaslctTestEnvironment(
+            ExaslctTestEnvironment(
                 cls,
                 utils.INTEGRATION_TEST_DOCKER_ENVIRONMENT_DEFAULT_BIN,
                 clean_images_at_close=False)
 
-    def setUp(self):
-        self.client = docker.from_env()
-
     @classmethod
     def tearDownClass(cls):
-        try:
-            cls.test_environment.close()
-        except Exception as e:
-            print(e)
-
-    def tearDown(self):
-        try:
-            self.on_host_docker_environment.close()
-        except Exception as e:
-            print(e)
-        try:
-            self.google_cloud_docker_environment.close()
-        except Exception as e:
-            print(e)
-        self.client.close()
+        utils.close_environments(cls.test_environment)
 
     def assert_mem_size(self, size: str):
-        containers = [c.name for c in self.client.containers.list() if self.docker_environment_name in c.name]
-        db_container = [c for c in containers if "db_container" in c]
-        exit_result = self.client.containers.get(db_container[0]).exec_run("cat /exa/etc/EXAConf")
-        output = exit_result[1].decode("UTF-8")
-        if output == '':
-            exit_result = self.client.containers.get(db_container[0]).exec_run("cat /exa/etc/EXAConf")
+        with ContextDockerClient() as docker_client:
+            containers = [c.name for c in docker_client.containers.list() if self.docker_environment_name in c.name]
+            db_container = [c for c in containers if "db_container" in c]
+            exit_result = docker_client.containers.get(db_container[0]).exec_run("cat /exa/etc/EXAConf")
             output = exit_result[1].decode("UTF-8")
+            if output == '':
+                exit_result = docker_client.containers.get(db_container[0]).exec_run("cat /exa/etc/EXAConf")
+                output = exit_result[1].decode("UTF-8")
+                return_code = exit_result[0]
             return_code = exit_result[0]
-        return_code = exit_result[0]
-        self.assertEquals(return_code, 0)
-        self.assertIn("MemSize = %s" % size, output)
+            self.assertEquals(return_code, 0)
+            self.assertIn("MemSize = %s" % size, output)
 
     def test_default_db_mem_size(self):
         self.docker_environment_name = "test_default_db_mem_size"
-        self.on_host_docker_environment, self.google_cloud_docker_environment = \
-            self.test_environment.spawn_docker_test_environment(self.docker_environment_name)
-        self.assert_mem_size("2 GiB")
+        with self.test_environment.spawn_docker_test_environments(self.docker_environment_name):
+            self.assert_mem_size("2 GiB")
 
     def test_smallest_valid_db_mem_size(self):
         self.docker_environment_name = "test_smallest_valid_db_mem_size"
-        self.on_host_docker_environment, self.google_cloud_docker_environment = \
-            self.test_environment.spawn_docker_test_environment(self.docker_environment_name,
-                                                                ["--db-mem-size", "'1 GiB'"])
-        self.assert_mem_size("1 GiB")
+
+        with self.test_environment.spawn_docker_test_environments(self.docker_environment_name,
+                                                                  ["--db-mem-size", "'1 GiB'"]):
+            self.assert_mem_size("1 GiB")
 
     def test_invalid_db_mem_size(self):
         self.docker_environment_name = "test_invalid_db_mem_size"
         with self.assertRaises(Exception) as context:
-            self.on_host_docker_environment, self.google_cloud_docker_environment = \
-                self.test_environment.spawn_docker_test_environment(self.docker_environment_name,
-                                                                    ["--db-mem-size", "'999 MiB'"])
+            with self.test_environment.spawn_docker_test_environments(self.docker_environment_name,
+                                                                      ["--db-mem-size", "'999 MiB'"]):
+                pass
 
 
 if __name__ == '__main__':
diff --git a/exasol_integration_test_docker_environment/test/test_test_environment_docker_runtime.py b/exasol_integration_test_docker_environment/test/test_test_environment_docker_runtime.py
index 8de37d65c..08cab520f 100644
--- a/exasol_integration_test_docker_environment/test/test_test_environment_docker_runtime.py
+++ b/exasol_integration_test_docker_environment/test/test_test_environment_docker_runtime.py
@@ -1,32 +1,35 @@
 import unittest
 
-import docker
-
-from exasol_integration_test_docker_environment.test import utils
+from exasol_integration_test_docker_environment.lib.docker import ContextDockerClient
 from exasol_integration_test_docker_environment.testing import utils
+from exasol_integration_test_docker_environment.testing.exaslct_test_environment import ExaslctTestEnvironment
 
 
-def assertContainerRuntime(self, container_name, expected_runtime):
-    try:
-        container = self.client.containers.get(container_name)
-        container.reload()
-        actual_runtime = container.attrs['HostConfig']['Runtime']
-    except Exception as e:
-        startup_log = self.on_host_docker_environment.completed_process.stdout.decode("utf8")
-        raise Exception(f"Startup log: {startup_log}") from e
-    self.assertEqual(actual_runtime, expected_runtime,
-                     f"{container_name} has the wrong runtime expected {expected_runtime} got {actual_runtime}."
-                     f"\n Startup log is {self.on_host_docker_environment.completed_process.stdout.decode('utf8')}")
+def assert_container_runtime(self, container_name, expected_runtime):
+    on_host_docker_environment = self.spawned_docker_test_environments.on_host_docker_environment
+    with ContextDockerClient() as docker_client:
+        try:
+            container = docker_client.containers.get(container_name)
+            container.reload()
+            actual_runtime = container.attrs['HostConfig']['Runtime']
+        except Exception as e:
+            startup_log = on_host_docker_environment.completed_process.stdout.decode("utf8")
+            raise Exception(f"Startup log: {startup_log}") from e
+        self.assertEqual(actual_runtime, expected_runtime,
+                         f"{container_name} has the wrong runtime expected {expected_runtime} got {actual_runtime}."
+                         f"\n Startup log is "
+                         f"{on_host_docker_environment.completed_process.stdout.decode('utf8')}")
 
 
-def get_default_docker_runtime(cls):
-    tmp_container = cls.client.containers.create("ubuntu:18.04", "echo")
-    try:
-        tmp_container.reload()
-        default_docker_runtime = tmp_container.attrs['HostConfig']['Runtime']
-    finally:
-        tmp_container.remove(force=True)
-    return default_docker_runtime
+def get_default_docker_runtime():
+    with ContextDockerClient() as docker_client:
+        tmp_container = docker_client.containers.create("ubuntu:18.04", "echo")
+        try:
+            tmp_container.reload()
+            default_docker_runtime = tmp_container.attrs['HostConfig']['Runtime']
+        finally:
+            tmp_container.remove(force=True)
+        return default_docker_runtime
 
 
 class DockerTestEnvironmentDockerRuntimeNoRuntimeGivenTest(unittest.TestCase):
@@ -37,52 +40,39 @@ def setUpClass(cls):
         # We can't use start-test-env. because it only mounts ./ and
         # doesn't work with --build_ouput-directory
         cls.test_environment = \
-            utils.ExaslctTestEnvironment(
+            ExaslctTestEnvironment(
                 cls,
                 utils.INTEGRATION_TEST_DOCKER_ENVIRONMENT_DEFAULT_BIN,
                 clean_images_at_close=False)
         cls.docker_environment_name = "test_no_runtime_given"
-        cls.on_host_docker_environment, cls.google_cloud_docker_environment = \
-            cls.test_environment.spawn_docker_test_environment(cls.docker_environment_name)
-        cls.client = docker.from_env()
-        cls.default_docker_runtime = get_default_docker_runtime(cls)
+        cls.spawned_docker_test_environments = \
+            cls.test_environment.spawn_docker_test_environments(cls.docker_environment_name)
+        cls.default_docker_runtime = get_default_docker_runtime()
 
     @classmethod
     def tearDownClass(cls):
-        try:
-            cls.test_environment.close()
-        except Exception as e:
-            print(e)
-        try:
-            cls.on_host_docker_environment.close()
-        except Exception as e:
-            print(e)
-        try:
-            cls.google_cloud_docker_environment.close()
-        except Exception as e:
-            print(e)
-        try:
-            cls.client.close()
-        except Exception as e:
-            print(e)
+        utils.close_environments(cls.spawned_docker_test_environments, cls.test_environment)
 
     def test_test_container_runtime(self):
         try:
-            environment_info = self.on_host_docker_environment.environment_info
+            environment_info = self.spawned_docker_test_environments.on_host_docker_environment.environment_info
             test_container_name = environment_info.test_container_info.container_name
         except Exception as e:
-            startup_log = self.on_host_docker_environment.completed_process.stdout.decode("utf8")
+            startup_log = self.spawned_docker_test_environments\
+                .on_host_docker_environment.completed_process.stdout.decode("utf8")
             raise Exception(f"Startup log: {startup_log}") from e
-        assertContainerRuntime(self, test_container_name, self.default_docker_runtime)
+        assert_container_runtime(self, test_container_name, self.default_docker_runtime)
 
     def test_database_container_runtime(self):
         try:
-            environment_info = self.on_host_docker_environment.environment_info
+            environment_info = self.spawned_docker_test_environments.on_host_docker_environment.environment_info
             database_container_name = environment_info.database_info.container_info.container_name
         except Exception as e:
-            startup_log = self.on_host_docker_environment.completed_process.stdout.decode("utf8")
+            startup_log = \
+                self.spawned_docker_test_environments.on_host_docker_environment.completed_process.stdout.decode("utf8")
             raise Exception(f"Startup log: {startup_log}") from e
-        assertContainerRuntime(self, database_container_name, self.default_docker_runtime)
+        assert_container_runtime(self, database_container_name, self.default_docker_runtime)
+
 
 class DockerTestEnvironmentDockerRuntimeDefaultRuntimeGivenTest(unittest.TestCase):
 
@@ -92,54 +82,40 @@ def setUpClass(cls):
         # We can't use start-test-env. because it only mounts ./ and
         # doesn't work with --build_ouput-directory
         cls.test_environment = \
-            utils.ExaslctTestEnvironment(
+            ExaslctTestEnvironment(
                 cls,
                 utils.INTEGRATION_TEST_DOCKER_ENVIRONMENT_DEFAULT_BIN,
                 clean_images_at_close=False)
-        cls.client = docker.from_env()
-        cls.default_docker_runtime = get_default_docker_runtime(cls)
+        cls.default_docker_runtime = get_default_docker_runtime()
         cls.docker_environment_name = "test_default_runtime_given"
-        cls.on_host_docker_environment, cls.google_cloud_docker_environment = \
-            cls.test_environment.spawn_docker_test_environment(
+        cls.spawned_docker_test_environments = cls.test_environment.spawn_docker_test_environments(
                 cls.docker_environment_name,
                 additional_parameter=["--docker-runtime", cls.default_docker_runtime])
 
     @classmethod
     def tearDownClass(cls):
-        try:
-            cls.test_environment.close()
-        except Exception as e:
-            print(e)
-        try:
-            cls.on_host_docker_environment.close()
-        except Exception as e:
-            print(e)
-        try:
-            cls.google_cloud_docker_environment.close()
-        except Exception as e:
-            print(e)
-        try:
-            cls.client.close()
-        except Exception as e:
-            print(e)
+        utils.close_environments(cls.spawned_docker_test_environments, cls.test_environment)
 
     def test_test_container_runtime(self):
+        on_host_docker_environment = self.spawned_docker_test_environments.on_host_docker_environment
         try:
-            environment_info = self.on_host_docker_environment.environment_info
+            environment_info = on_host_docker_environment.environment_info
             test_container_name = environment_info.test_container_info.container_name
         except Exception as e:
-            startup_log = self.on_host_docker_environment.completed_process.stdout.decode("utf8")
+            startup_log = on_host_docker_environment.completed_process.stdout.decode("utf8")
             raise Exception(f"Startup log: {startup_log}") from e
-        assertContainerRuntime(self, test_container_name, self.default_docker_runtime)
+        assert_container_runtime(self, test_container_name, self.default_docker_runtime)
 
     def test_database_container_runtime(self):
+        on_host_docker_environment = self.spawned_docker_test_environments.on_host_docker_environment
         try:
-            environment_info = self.on_host_docker_environment.environment_info
+            environment_info = on_host_docker_environment.environment_info
             database_container_name = environment_info.database_info.container_info.container_name
         except Exception as e:
-            startup_log = self.on_host_docker_environment.completed_process.stdout.decode("utf8")
+            startup_log = on_host_docker_environment.completed_process.stdout.decode("utf8")
             raise Exception(f"Startup log: {startup_log}") from e
-        assertContainerRuntime(self, database_container_name, self.default_docker_runtime)
+        assert_container_runtime(self, database_container_name, self.default_docker_runtime)
+
 
 class DockerTestEnvironmentDockerRuntimeInvalidRuntimeGivenTest(unittest.TestCase):
 
@@ -149,16 +125,15 @@ def setUpClass(cls):
         # We can't use start-test-env. because it only mounts ./ and
         # doesn't work with --build_ouput-directory
         cls.test_environment = \
-            utils.ExaslctTestEnvironment(
+            ExaslctTestEnvironment(
                 cls,
                 utils.INTEGRATION_TEST_DOCKER_ENVIRONMENT_DEFAULT_BIN,
                 clean_images_at_close=False)
-        cls.client = docker.from_env()
-        cls.default_docker_runtime = get_default_docker_runtime(cls)
+        cls.default_docker_runtime = get_default_docker_runtime()
         cls.docker_environment_name = "test_default_runtime_given"
+        cls.docker_environments = ()
         try:
-            cls.on_host_docker_environment, cls.google_cloud_docker_environment = \
-                cls.test_environment.spawn_docker_test_environment(
+            cls.spawned_docker_test_environments = cls.test_environment.spawn_docker_test_environments(
                     cls.docker_environment_name,
                     additional_parameter=["--docker-runtime", "AAAABBBBCCCC_INVALID_RUNTIME_111122223333"])
         except Exception as e:
@@ -167,24 +142,13 @@ def setUpClass(cls):
     @classmethod
     def tearDownClass(cls):
         try:
-            cls.test_environment.close()
-        except Exception as e:
-            print(e)
-        try:
-            cls.on_host_docker_environment.close()
-        except Exception as e:
-            print(e)
-        try:
-            cls.google_cloud_docker_environment.close()
-        except Exception as e:
-            print(e)
-        try:
-            cls.client.close()
+            utils.close_environments(cls.spawned_docker_test_environments)
         except Exception as e:
             print(e)
+        utils.close_environments(cls.test_environment)
 
     def test_docker_environment_not_available(self):
-        self.assertFalse("on_host_docker_environment" in self.__dict__)
+        self.assertFalse("spawned_docker_test_environments" in self.__dict__)
 
 
 if __name__ == '__main__':
diff --git a/exasol_integration_test_docker_environment/test/test_test_environment_nameservers.py b/exasol_integration_test_docker_environment/test/test_test_environment_nameservers.py
index 8629d6a20..b99d82ad7 100644
--- a/exasol_integration_test_docker_environment/test/test_test_environment_nameservers.py
+++ b/exasol_integration_test_docker_environment/test/test_test_environment_nameservers.py
@@ -1,8 +1,9 @@
 import unittest
 
-import docker
 
+from exasol_integration_test_docker_environment.lib.docker import ContextDockerClient
 from exasol_integration_test_docker_environment.testing import utils
+from exasol_integration_test_docker_environment.testing.exaslct_test_environment import ExaslctTestEnvironment
 
 
 class DockerTestEnvironmentDBMemSizeTest(unittest.TestCase):
@@ -13,76 +14,50 @@ def setUpClass(cls):
         # We can't use start-test-env. because it only mounts ./ and
         # doesn't work with --build_ouput-directory
         cls.test_environment = \
-            utils.ExaslctTestEnvironment(
+            ExaslctTestEnvironment(
                 cls,
                 utils.INTEGRATION_TEST_DOCKER_ENVIRONMENT_DEFAULT_BIN,
                 clean_images_at_close=False)
 
-    def setUp(self):
-        self.client = docker.from_env()
-
     @classmethod
     def tearDownClass(cls):
-        try:
-            cls.test_environment.close()
-        except Exception as e:
-            print(e)
-
-    def tearDown(self):
-        try:
-            self.on_host_docker_environment.close()
-        except Exception as e:
-            print(e)
-        try:
-            self.google_cloud_docker_environment.close()
-        except Exception as e:
-            print(e)
-        self.client.close()
+        utils.close_environments(cls.test_environment)
 
     def assert_nameserver(self, nameservers: str):
-        containers = [c.name for c in self.client.containers.list() if self.docker_environment_name in c.name]
-        db_container = [c for c in containers if "db_container" in c]
-        exit_result = self.client.containers.get(db_container[0]).exec_run("cat /exa/etc/EXAConf")
-        output = exit_result[1].decode("UTF-8")
-        if output == '':
-            exit_result = self.client.containers.get(db_container[0]).exec_run("cat /exa/etc/EXAConf")
+        with ContextDockerClient() as docker_client:
+            containers = [c.name for c in docker_client.containers.list() if self.docker_environment_name in c.name]
+            db_container = [c for c in containers if "db_container" in c]
+            exit_result = docker_client.containers.get(db_container[0]).exec_run("cat /exa/etc/EXAConf")
             output = exit_result[1].decode("UTF-8")
+            if output == '':
+                exit_result = docker_client.containers.get(db_container[0]).exec_run("cat /exa/etc/EXAConf")
+                output = exit_result[1].decode("UTF-8")
+                return_code = exit_result[0]
             return_code = exit_result[0]
-        return_code = exit_result[0]
-        self.assertEquals(return_code, 0)
-        self.assertIn("NameServers = %s" % nameservers, output)
+            self.assertEquals(return_code, 0)
+            self.assertIn("NameServers = %s" % nameservers, output)
 
     def test_no_nameserver(self):
         self.docker_environment_name = "test_no_nameserver"
-        self.on_host_docker_environment, self.google_cloud_docker_environment = \
-            self.test_environment.spawn_docker_test_environment(self.docker_environment_name)
-        self.assert_nameserver("")
+        with self.test_environment.spawn_docker_test_environments(self.docker_environment_name):
+            self.assert_nameserver("")
 
     def test_single_nameserver(self):
         self.docker_environment_name = "test_single_nameserver"
-        self.on_host_docker_environment, self.google_cloud_docker_environment = \
-            self.test_environment.spawn_docker_test_environment(self.docker_environment_name,
-                                                                [
+        with self.test_environment.spawn_docker_test_environments(self.docker_environment_name,
+                                                                  [
                                                                     "--nameserver", "'8.8.8.8'",
-                                                                ])
-        self.assert_nameserver("8.8.8.8")
+                                                                  ]):
+            self.assert_nameserver("8.8.8.8")
 
     def test_multiple_nameserver(self):
         self.docker_environment_name = "test_multiple_nameserver"
-        self.on_host_docker_environment, self.google_cloud_docker_environment = \
-            self.test_environment.spawn_docker_test_environment(self.docker_environment_name,
-                                                                [
+        with self.test_environment.spawn_docker_test_environments(self.docker_environment_name,
+                                                                  [
                                                                     "--nameserver", "'8.8.8.8'",
                                                                     "--nameserver", "'8.8.8.9'",
-                                                                ])
-        self.assert_nameserver("8.8.8.8,8.8.8.9")
-
-    # def test_invalid_db_mem_size(self):
-    #     self.docker_environment_name = "test_invalid_db_mem_size"
-    #     with self.assertRaises(Exception) as context:
-    #         self.on_host_docker_environment, self.google_cloud_docker_environment = \
-    #             self.test_environment.spawn_docker_test_environment(self.docker_environment_name,
-                                                                    # ["--db-mem-size", "'999 MiB'"])
+                                                                  ]):
+            self.assert_nameserver("8.8.8.8,8.8.8.9")
 
 
 if __name__ == '__main__':
diff --git a/exasol_integration_test_docker_environment/test/utils/__init__.py b/exasol_integration_test_docker_environment/test/utils/__init__.py
deleted file mode 100644
index e69de29bb..000000000
diff --git a/exasol_integration_test_docker_environment/test/utils/luigi_utils.py b/exasol_integration_test_docker_environment/test/utils/luigi_utils.py
deleted file mode 100644
index 865772d8a..000000000
--- a/exasol_integration_test_docker_environment/test/utils/luigi_utils.py
+++ /dev/null
@@ -1,22 +0,0 @@
-import shutil
-from datetime import datetime
-
-import luigi
-
-from exasol_integration_test_docker_environment.lib.docker.images.clean.clean_images import CleanImagesStartingWith
-
-
-def set_job_id(task_cls):
-    strftime = datetime.now().strftime('%Y_%m_%d_%H_%M_%S')
-    job_id = f"{strftime}_{task_cls.__name__}"
-    config = luigi.configuration.get_config()
-    config.set('job_config', 'job_id', job_id)
-    return job_id
-
-
-def clean(docker_repository_name):
-    jobid = set_job_id(CleanImagesStartingWith)
-    task = CleanImagesStartingWith(starts_with_pattern=docker_repository_name, jobid=jobid)
-    luigi.build([task], workers=1, local_scheduler=True, log_level="INFO")
-    if task._get_tmp_path_for_job().exists():
-        shutil.rmtree(str(task._get_tmp_path_for_job()))
diff --git a/exasol_integration_test_docker_environment/test/utils/process_spawn_utils.py b/exasol_integration_test_docker_environment/test/utils/process_spawn_utils.py
deleted file mode 100644
index 0315ef907..000000000
--- a/exasol_integration_test_docker_environment/test/utils/process_spawn_utils.py
+++ /dev/null
@@ -1,22 +0,0 @@
-from multiprocessing import Process, Queue
-from typing import Callable
-
-
-def process_main(queue: Queue, func: Callable):
-    try:
-        result = func()
-        queue.put(("result", result))
-    except Exception as e:
-        queue.put(("exception", e))
-
-
-def run_in_process(func):
-    queue = Queue()
-    p1 = Process(target=process_main, args=(queue, func))
-    p1.start()
-    p1.join()
-    result = queue.get()
-    if result[0] == "result":
-        return result[1]
-    else:
-        raise result[1]
diff --git a/exasol_integration_test_docker_environment/testing/exaslct_docker_test_environment.py b/exasol_integration_test_docker_environment/testing/exaslct_docker_test_environment.py
new file mode 100644
index 000000000..011b0ade5
--- /dev/null
+++ b/exasol_integration_test_docker_environment/testing/exaslct_docker_test_environment.py
@@ -0,0 +1,29 @@
+import subprocess
+from typing import Optional
+
+from exasol_integration_test_docker_environment.lib.data.environment_info import EnvironmentInfo
+from exasol_integration_test_docker_environment.testing.utils import remove_docker_container, remove_docker_volumes
+
+
+class ExaslctDockerTestEnvironment:
+    def __init__(self, name: str, database_host: str,
+                 db_username: str, db_password: str,
+                 bucketfs_username: str, bucketfs_password: str,
+                 database_port: int, bucketfs_port: int,
+                 environment_info: Optional[EnvironmentInfo] = None,
+                 completed_process: Optional[subprocess.CompletedProcess] = None):
+        self.db_password = db_password
+        self.db_username = db_username
+        self.database_port = database_port
+        self.bucketfs_port = bucketfs_port
+        self.bucketfs_username = bucketfs_username
+        self.bucketfs_password = bucketfs_password
+        self.database_host = database_host
+        self.name = name
+        self.environment_info = environment_info
+        self.completed_process = completed_process
+
+    def close(self):
+        remove_docker_container([f"test_container_{self.name}",
+                                 f"db_container_{self.name}"])
+        remove_docker_volumes([f"db_container_{self.name}_volume"])
diff --git a/exasol_integration_test_docker_environment/test/utils/utils.py b/exasol_integration_test_docker_environment/testing/exaslct_test_environment.py
similarity index 70%
rename from exasol_integration_test_docker_environment/test/utils/utils.py
rename to exasol_integration_test_docker_environment/testing/exaslct_test_environment.py
index 7f06af42f..161dca28c 100644
--- a/exasol_integration_test_docker_environment/test/utils/utils.py
+++ b/exasol_integration_test_docker_environment/testing/exaslct_test_environment.py
@@ -1,46 +1,19 @@
 import inspect
-import json
 import os
-import shlex
 import shutil
-import socket
 import subprocess
 import tempfile
 import time
-from contextlib import closing
 from pathlib import Path
-from typing import List, Tuple
-
-import docker
-import requests
+import shlex
+from typing import List
 
 from exasol_integration_test_docker_environment.lib.data.environment_info import EnvironmentInfo
-
-INTEGRATION_TEST_DOCKER_ENVIRONMENT_DEFAULT_BIN = "./start-test-env-without-poetry"
-
-
-class ExaslctDockerTestEnvironment:
-    def __init__(self, name: str, database_host: str,
-                 db_username: str, db_password: str,
-                 bucketfs_username: str, bucketfs_password: str,
-                 database_port: int, bucketfs_port: int,
-                 environment_info: EnvironmentInfo = None,
-                 completed_process: subprocess.CompletedProcess = None):
-        self.db_password = db_password
-        self.db_username = db_username
-        self.database_port = database_port
-        self.bucketfs_port = bucketfs_port
-        self.bucketfs_username = bucketfs_username
-        self.bucketfs_password = bucketfs_password
-        self.database_host = database_host
-        self.name = name
-        self.environment_info = environment_info
-        self.completed_process = completed_process
-
-    def close(self):
-        remove_docker_container([f"test_container_{self.name}",
-                                 f"db_container_{self.name}"])
-        remove_docker_volumes([f"db_container_{self.name}_volume"])
+from exasol_integration_test_docker_environment.lib.docker import ContextDockerClient
+from exasol_integration_test_docker_environment.testing.exaslct_docker_test_environment import \
+    ExaslctDockerTestEnvironment
+from exasol_integration_test_docker_environment.testing.spawned_test_environments import SpawnedTestEnvironments
+from exasol_integration_test_docker_environment.testing.utils import find_free_port
 
 
 class ExaslctTestEnvironment:
@@ -57,9 +30,9 @@ def __init__(self, test_object, executable="./exaslct", clean_images_at_close=Tr
         self.name = self.test_class.__name__
         self._repository_prefix = "exaslct_test"
         if "GOOGLE_CLOUD_BUILD" in os.environ:
-            # We need to put the output directories into the workdir, 
+            # We need to put the output directories into the workdir,
             # because only this is shared between the current container and
-            # host. Only path within this shared directory can be mounted 
+            # host. Only path within this shared directory can be mounted
             # to docker container started by exaslct
             temp_dir_prefix_path = Path("./temp_outputs")
             temp_dir_prefix_path.mkdir(exist_ok=True)
@@ -84,11 +57,9 @@ def repository_prefix(self, value):
         self._update_attributes()
 
     def _update_attributes(self):
-        # docker repository names must be lowercase
-        self.repository_name = f"{self._repository_prefix.lower()}/{self.name.lower()}"
+        self.repository_name = f"{self._repository_prefix.lower()}/{self.name.lower()}"  # docker repository names must be lowercase
         self.flavor_path_argument = f"--flavor-path {self.get_test_flavor()}"
-        self.docker_repository_arguments = f"--source-docker-repository-name {self.repository_name} " \
-                                           f"--target-docker-repository-name {self.repository_name}"
+        self.docker_repository_arguments = f"--source-docker-repository-name {self.repository_name} --target-docker-repository-name {self.repository_name}"
         self.clean_docker_repository_arguments = f"--docker-repository-name {self.repository_name}"
         self.output_directory_arguments = f"--output-directory {self.temp_dir}"
         self.task_dependencies_argument = " ".join([f"--task-dependencies-dot-file {self.name}.dot", ])
@@ -137,8 +108,8 @@ def close(self):
         except Exception as e:
             print(e)
 
-    def spawn_docker_test_environment(self, name, additional_parameter: List[str] = None) \
-            -> Tuple[ExaslctDockerTestEnvironment, ExaslctDockerTestEnvironment]:
+    def spawn_docker_test_environments(self, name: str, additional_parameter: List[str] = None) \
+            -> SpawnedTestEnvironments:
         on_host_parameter = ExaslctDockerTestEnvironment(
             name=self.name + "_" + name,
             database_host="localhost",
@@ -181,25 +152,21 @@ def spawn_docker_test_environment(self, name, additional_parameter: List[str] =
                 environment_info=on_host_parameter.completed_process,
                 completed_process=on_host_parameter.completed_process
             )
-            docker_client = docker.from_env()
-            try:
+            with ContextDockerClient() as docker_client:
                 db_container = docker_client.containers.get(f"db_container_{google_cloud_parameter.name}")
                 cloudbuild_network = docker_client.networks.get("cloudbuild")
                 cloudbuild_network.connect(db_container)
                 db_container.reload()
                 google_cloud_parameter.database_host = \
                     db_container.attrs["NetworkSettings"]["Networks"][cloudbuild_network.name]["IPAddress"]
-                return on_host_parameter, google_cloud_parameter
-            finally:
-                docker_client.close()
+                return SpawnedTestEnvironments(on_host_parameter, google_cloud_parameter)
         else:
-            return on_host_parameter, None
+            return SpawnedTestEnvironments(on_host_parameter, None)
 
     def create_registry(self):
         registry_port = find_free_port()
         registry_container_name = self.name.replace("/", "_") + "_registry"
-        docker_client = docker.from_env()
-        try:
+        with ContextDockerClient() as docker_client:
             print("Start pull of registry:2")
             docker_client.images.pull(repository="registry", tag="2")
             print(f"Start container of {registry_container_name}")
@@ -226,53 +193,3 @@ def create_registry(self):
             else:
                 self.repository_prefix = f"localhost:{registry_port}"
                 return registry_container, "localhost", registry_port
-        finally:
-            docker_client.close()
-
-
-def find_free_port():
-    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
-        s.bind(('', 0))
-        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        port = s.getsockname()[1]
-    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
-        s.bind(('', port))
-        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-    return port
-
-
-def remove_docker_container(containers):
-    docker_client = docker.from_env()
-    try:
-        for container in containers:
-            try:
-                docker_client.containers.get(container).remove(force=True)
-            except Exception as e:
-                print(e)
-    finally:
-        docker_client.close()
-
-
-def remove_docker_volumes(volumes):
-    docker_client = docker.from_env()
-    try:
-        for volume in volumes:
-            try:
-                docker_client.volumes.get(volume).remove(force=True)
-            except Exception as e:
-                print(e)
-    finally:
-        docker_client.close()
-
-
-def request_registry_images(registry_host, registry_port, repo_name):
-    url = f"http://{registry_host}:{registry_port}/v2/{repo_name}/tags/list"
-    result = requests.request("GET", url)
-    images = json.loads(result.content.decode("UTF-8"))
-    return images
-
-
-def request_registry_repositories(registry_host, registry_port):
-    result = requests.request("GET", f"http://{registry_host}:{registry_port}/v2/_catalog/")
-    repositories_ = json.loads(result.content.decode("UTF-8"))["repositories"]
-    return repositories_
diff --git a/exasol_integration_test_docker_environment/testing/luigi_utils.py b/exasol_integration_test_docker_environment/testing/luigi_utils.py
new file mode 100644
index 000000000..7963bcc51
--- /dev/null
+++ b/exasol_integration_test_docker_environment/testing/luigi_utils.py
@@ -0,0 +1,13 @@
+import shutil
+
+import luigi
+
+from exasol_integration_test_docker_environment.cli.common import generate_root_task
+from exasol_integration_test_docker_environment.lib.docker.images.clean.clean_images import CleanImagesStartingWith
+
+
+def clean(docker_repository_name):
+    task = generate_root_task(task_class=CleanImagesStartingWith, starts_with_pattern=docker_repository_name)
+    luigi.build([task], workers=1, local_scheduler=True, log_level="INFO")
+    if task._get_tmp_path_for_job().exists():
+        shutil.rmtree(str(task._get_tmp_path_for_job()))
diff --git a/exasol_integration_test_docker_environment/testing/spawned_test_environments.py b/exasol_integration_test_docker_environment/testing/spawned_test_environments.py
new file mode 100644
index 000000000..78ee57845
--- /dev/null
+++ b/exasol_integration_test_docker_environment/testing/spawned_test_environments.py
@@ -0,0 +1,24 @@
+from typing import Optional
+
+from exasol_integration_test_docker_environment.testing.exaslct_docker_test_environment import \
+    ExaslctDockerTestEnvironment
+
+
+class SpawnedTestEnvironments:
+    def __init__(self, on_host_environment: ExaslctDockerTestEnvironment,
+                 google_cloud_environment: Optional[ExaslctDockerTestEnvironment]):
+        self.on_host_docker_environment = on_host_environment
+        self.google_cloud_environment = google_cloud_environment
+
+    def __enter__(self):
+        pass
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.close()
+
+    def close(self):
+        if self.on_host_docker_environment is not None:
+            self.on_host_docker_environment.close()
+
+        if self.google_cloud_environment is not None:
+            self.google_cloud_environment.close()
diff --git a/exasol_integration_test_docker_environment/testing/utils.py b/exasol_integration_test_docker_environment/testing/utils.py
index 6fc38360d..96aa1b841 100644
--- a/exasol_integration_test_docker_environment/testing/utils.py
+++ b/exasol_integration_test_docker_environment/testing/utils.py
@@ -1,231 +1,12 @@
-import inspect
 import json
-import os
-import shlex
-import shutil
 import socket
-import subprocess
-import tempfile
-import time
 from contextlib import closing
-from pathlib import Path
-from typing import List, Tuple
 
-import docker
 import requests
 
-from exasol_integration_test_docker_environment.lib.data.environment_info import EnvironmentInfo
+from exasol_integration_test_docker_environment.lib.docker import ContextDockerClient
 
-
-INTEGRATION_TEST_DOCKER_ENVIRONMENT_DEFAULT_BIN="./start-test-env-without-poetry"
-
-class ExaslctDockerTestEnvironment():
-    def __init__(self, name: str, database_host: str,
-                 db_username: str, db_password: str,
-                 bucketfs_username: str, bucketfs_password: str,
-                 database_port: int, bucketfs_port: int,
-                 environment_info: EnvironmentInfo = None,
-                 completed_process: subprocess.CompletedProcess = None):
-        self.db_password = db_password
-        self.db_username = db_username
-        self.database_port = database_port
-        self.bucketfs_port = bucketfs_port
-        self.bucketfs_username = bucketfs_username
-        self.bucketfs_password = bucketfs_password
-        self.database_host = database_host
-        self.name = name
-        self.environment_info = environment_info
-        self.completed_process = completed_process
-
-    def close(self):
-        remove_docker_container([f"test_container_{self.name}",
-                                 f"db_container_{self.name}"])
-        remove_docker_volumes([f"db_container_{self.name}_volume"])
-
-
-class ExaslctTestEnvironment():
-
-    def __init__(self, test_object, executable="./exaslct", clean_images_at_close=True):
-        self.clean_images_at_close = clean_images_at_close
-        self.executable = executable
-        self.test_object = test_object
-        if not inspect.isclass(self.test_object):
-            self.test_class = self.test_object.__class__
-        else:
-            self.test_class = self.test_object
-        self.flavor_path = self.get_test_flavor()
-        self.name = self.test_class.__name__
-        self._repository_prefix = "exaslct_test"
-        if "GOOGLE_CLOUD_BUILD" in os.environ:
-            # We need to put the output directories into the workdir, 
-            # because only this is shared between the current container and
-            # host. Only path within this shared directory can be mounted 
-            # to docker container started by exaslct
-            temp_dir_prefix_path = Path("./temp_outputs")
-            temp_dir_prefix_path.mkdir(exist_ok=True)
-            self.temp_dir = tempfile.mkdtemp(dir=temp_dir_prefix_path)
-        else:
-            self.temp_dir = tempfile.mkdtemp()
-        self._update_attributes()
-
-    def get_test_flavor(self):
-        source_file_of_test_object = inspect.getsourcefile(self.test_class)
-        flavor_path = Path(os.path.realpath(source_file_of_test_object)).parent.joinpath(
-            "resources/test-flavor")
-        return flavor_path
-
-    @property
-    def repository_prefix(self):
-        return self._repository_prefix
-
-    @repository_prefix.setter
-    def repository_prefix(self, value):
-        self._repository_prefix = value
-        self._update_attributes()
-
-    def _update_attributes(self):
-        self.repository_name = f"{self._repository_prefix.lower()}/{self.name.lower()}"  # docker repository names must be lowercase
-        self.flavor_path_argument = f"--flavor-path {self.get_test_flavor()}"
-        self.docker_repository_arguments = f"--source-docker-repository-name {self.repository_name} --target-docker-repository-name {self.repository_name}"
-        self.clean_docker_repository_arguments = f"--docker-repository-name {self.repository_name}"
-        self.output_directory_arguments = f"--output-directory {self.temp_dir}"
-        self.task_dependencies_argument = " ".join([f"--task-dependencies-dot-file {self.name}.dot", ])
-
-    def clean_images(self):
-        self.run_command(f"{self.executable} clean-flavor-images", clean=True)
-
-    def run_command(self, command: str,
-                    use_output_directory: bool = True,
-                    use_flavor_path: bool = True,
-                    use_docker_repository: bool = True,
-                    track_task_dependencies: bool = False,
-                    clean: bool = False,
-                    capture_output: bool = False):
-        if use_output_directory:
-            command = f"{command} {self.output_directory_arguments}"
-        if track_task_dependencies:
-            command = f"{command} {self.task_dependencies_argument}"
-        if use_flavor_path:
-            command = f"{command} {self.flavor_path_argument}"
-        if use_docker_repository and not clean:
-            command = f"{command} {self.docker_repository_arguments}"
-        if use_docker_repository and clean:
-            command = f"{command} {self.clean_docker_repository_arguments}"
-        print()
-        print(f"command: {command}")
-        if capture_output:
-            completed_process = subprocess.run(shlex.split(command), stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-        else:
-            completed_process = subprocess.run(shlex.split(command))
-        try:
-            completed_process.check_returncode()
-        except subprocess.CalledProcessError as e:
-            print(e.stdout.decode("UTF-8"))
-            raise e
-        return completed_process
-
-    def close(self):
-        try:
-            if self.clean_images_at_close:
-                self.clean_images()
-        except Exception as e:
-            print(e)
-        try:
-            shutil.rmtree(self.temp_dir)
-        except Exception as e:
-            print(e)
-
-    def spawn_docker_test_environment(self, name, additional_parameter: List[str] = None) \
-            -> Tuple[ExaslctDockerTestEnvironment, ExaslctDockerTestEnvironment]:
-        on_host_parameter = ExaslctDockerTestEnvironment(
-            name=self.name + "_" + name,
-            database_host="localhost",
-            db_username="sys",
-            db_password="exasol",
-            bucketfs_username="w",
-            bucketfs_password="write",
-            database_port=find_free_port(),
-            bucketfs_port=find_free_port())
-        docker_db_version_parameter = ""
-        if "EXASOL_VERSION" in os.environ and os.environ["EXASOL_VERSION"] != "default":
-            docker_db_version_parameter = f'--docker-db-image-version "{os.environ["EXASOL_VERSION"]}"'
-        if additional_parameter is None:
-            additional_parameter = []
-        arguments = " ".join([f"--environment-name {on_host_parameter.name}",
-                              f"--database-port-forward {on_host_parameter.database_port}",
-                              f"--bucketfs-port-forward {on_host_parameter.bucketfs_port}",
-                              docker_db_version_parameter] + additional_parameter)
-
-        command = f"{self.executable} spawn-test-environment {arguments}"
-        completed_process = self.run_command(command, use_flavor_path=False, use_docker_repository=False,
-                                             capture_output=True)
-        on_host_parameter.completed_process = completed_process
-        environment_info_json_path = Path(self.temp_dir,
-                                          f"cache/environments/{on_host_parameter.name}/environment_info.json")
-        if environment_info_json_path.exists():
-            with environment_info_json_path.open() as f:
-                environment_info = EnvironmentInfo.from_json(f.read())
-                on_host_parameter.environment_info = environment_info
-        if "GOOGLE_CLOUD_BUILD" in os.environ:
-            google_cloud_parameter = ExaslctDockerTestEnvironment(
-                name=on_host_parameter.name,
-                database_host="localhost",
-                db_username=on_host_parameter.db_username,
-                db_password=on_host_parameter.db_password,
-                bucketfs_username=on_host_parameter.bucketfs_username,
-                bucketfs_password=on_host_parameter.bucketfs_password,
-                database_port=8888,
-                bucketfs_port=6583,
-                environment_info=on_host_parameter.completed_process,
-                completed_process=on_host_parameter.completed_process
-            )
-            docker_client = docker.from_env()
-            try:
-                db_container = docker_client.containers.get(f"db_container_{google_cloud_parameter.name}")
-                cloudbuild_network = docker_client.networks.get("cloudbuild")
-                cloudbuild_network.connect(db_container)
-                db_container.reload()
-                google_cloud_parameter.database_host = \
-                    db_container.attrs["NetworkSettings"]["Networks"][cloudbuild_network.name]["IPAddress"]
-                return on_host_parameter, google_cloud_parameter
-            finally:
-                docker_client.close()
-        else:
-            return on_host_parameter, None
-
-    def create_registry(self):
-        registry_port = find_free_port()
-        registry_container_name = self.name.replace("/", "_") + "_registry"
-        docker_client = docker.from_env()
-        try:
-            print("Start pull of registry:2")
-            docker_client.images.pull(repository="registry", tag="2")
-            print(f"Start container of {registry_container_name}")
-            try:
-                docker_client.containers.get(registry_container_name).remove(force=True)
-            except:
-                pass
-            registry_container = docker_client.containers.run(
-                image="registry:2", name=registry_container_name,
-                ports={5000: registry_port},
-                detach=True
-            )
-            time.sleep(10)
-            print(f"Finished start container of {registry_container_name}")
-            if "GOOGLE_CLOUD_BUILD" in os.environ:
-                cloudbuild_network = docker_client.networks.get("cloudbuild")
-                cloudbuild_network.connect(registry_container)
-                registry_container.reload()
-                registry_host = registry_container.attrs["NetworkSettings"]["Networks"][cloudbuild_network.name][
-                    "IPAddress"]
-                # self.repository_prefix = f"{registry_host}:5000"
-                self.repository_prefix = f"localhost:{registry_port}"
-                return registry_container, registry_host, "5000"
-            else:
-                self.repository_prefix = f"localhost:{registry_port}"
-                return registry_container, "localhost", registry_port
-        finally:
-            docker_client.close()
+INTEGRATION_TEST_DOCKER_ENVIRONMENT_DEFAULT_BIN = "./start-test-env-without-poetry"
 
 
 def find_free_port():
@@ -240,27 +21,21 @@ def find_free_port():
 
 
 def remove_docker_container(containers):
-    docker_client = docker.from_env()
-    try:
+    with ContextDockerClient() as docker_client:
         for container in containers:
             try:
                 docker_client.containers.get(container).remove(force=True)
             except Exception as e:
                 print(e)
-    finally:
-        docker_client.close()
 
 
 def remove_docker_volumes(volumes):
-    docker_client = docker.from_env()
-    try:
+    with ContextDockerClient() as docker_client:
         for volume in volumes:
             try:
                 docker_client.volumes.get(volume).remove(force=True)
             except Exception as e:
                 print(e)
-    finally:
-        docker_client.close()
 
 
 def request_registry_images(registry_host, registry_port, repo_name):
@@ -274,3 +49,12 @@ def request_registry_repositories(registry_host, registry_port):
     result = requests.request("GET", f"http://{registry_host}:{registry_port}/v2/_catalog/")
     repositories_ = json.loads(result.content.decode("UTF-8"))["repositories"]
     return repositories_
+
+
+def close_environments(*args):
+    for env in args:
+        try:
+            if env is not None:
+                env.close()
+        except Exception as e:
+            print(e)