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

Issue:
Hash value of docker images change when using volatile absolute paths
## RCA

The [BuildContextHasher](https://github.com/exasol/integration-test-docker-environment/blob/main/exasol_integration_test_docker_environment/lib/docker/images/create/utils/build_context_hasher.py) forwards the list of mapping **values** to the [FileDirectoryListHasher](https://github.com/exasol/integration-test-docker-environment/blob/main/exasol_integration_test_docker_environment/lib/docker/images/create/utils/file_directory_list_hasher.py) which then uses the file paths as input for calculating the Hash value. If the files of the mapping are stored in volatile directories (for example temporary directories), the hash value will change for each execution of the (respective) docker build task. This is a problem when the build content needs to be copied to a temporary directory, or if the user uses absolute paths as arguments (in case of the flavor path of exaslct for example).

## Solution

In order to fix this the [FileDirectoryListHasher](https://github.com/exasol/integration-test-docker-environment/blob/main/exasol_integration_test_docker_environment/lib/docker/images/create/utils/file_directory_list_hasher.py) needs to get the list of mappings as input. It will then use the relative paths, stored as keys in the mappings, as input for the hasher; and use the real file system paths (values) only to read the file content, as input for the hasher.
Hash value of docker images change when using volatile absolute paths
## RCA

The [BuildContextHasher](https://github.com/exasol/integration-test-docker-environment/blob/main/exasol_integration_test_docker_environment/lib/docker/images/create/utils/build_context_hasher.py) forwards the list of mapping **values** to the [FileDirectoryListHasher](https://github.com/exasol/integration-test-docker-environment/blob/main/exasol_integration_test_docker_environment/lib/docker/images/create/utils/file_directory_list_hasher.py) which then uses the file paths as input for calculating the Hash value. If the files of the mapping are stored in volatile directories (for example temporary directories), the hash value will change for each execution of the (respective) docker build task. This is a problem when the build content needs to be copied to a temporary directory, or if the user uses absolute paths as arguments (in case of the flavor path of exaslct for example).

## Solution

In order to fix this the [FileDirectoryListHasher](https://github.com/exasol/integration-test-docker-environment/blob/main/exasol_integration_test_docker_environment/lib/docker/images/create/utils/file_directory_list_hasher.py) needs to get the list of mappings as input. It will then use the relative paths, stored as keys in the mappings, as input for the hasher; and use the real file system paths (values) only to read the file content, as input for the hasher.


Discussion / hints:

Fix patch:
diff --git a/doc/changes/changes_0.12.0.md b/doc/changes/changes_0.12.0.md
index 0f2573a2d..0ddef081a 100644
--- a/doc/changes/changes_0.12.0.md
+++ b/doc/changes/changes_0.12.0.md
@@ -22,6 +22,7 @@ TBD
 - #230: Fixed default value of click option external-exasol-xmlrpc-port should be int
 - #211: Fixed output of test DockerTestEnvironmentDockerRuntimeInvalidRuntimeGivenTest
 - #244: Fixed bug in generate graph plot if task uses luigi.ListParameter
+- #249: Fixed bug of calculation of hash value of docker images when using volatile absolute paths
 
 ## Features / Enhancements:
 


Test patch:
diff --git a/exasol_integration_test_docker_environment/lib/docker/images/create/utils/build_context_hasher.py b/exasol_integration_test_docker_environment/lib/docker/images/create/utils/build_context_hasher.py
index eba21c4a0..78929a998 100644
--- a/exasol_integration_test_docker_environment/lib/docker/images/create/utils/build_context_hasher.py
+++ b/exasol_integration_test_docker_environment/lib/docker/images/create/utils/build_context_hasher.py
@@ -1,10 +1,11 @@
 import base64
 import hashlib
+from pathlib import Path
 from typing import Dict
 
 # TODO add hash config to the hash
 from exasol_integration_test_docker_environment.lib.docker.images.create.utils.file_directory_list_hasher import \
-    FileDirectoryListHasher
+    FileDirectoryListHasher, PathMapping
 from exasol_integration_test_docker_environment.lib.docker.images.image_info import ImageDescription, ImageInfo
 
 
@@ -25,10 +26,14 @@ def _generate_build_context_hash(self):
                                     hashfunc="sha256",
                                     hash_file_names=True,
                                     hash_directory_names=True,
-                                    hash_permissions=True,
-                                    use_relative_paths=True)
-        files_directories_to_hash = list(self.image_description.mapping_of_build_files_and_directories.values()) + \
-                                    [str(self.image_description.dockerfile)]
+                                    hash_permissions=True)
+        files_directories_to_hash = [PathMapping(destination, source) for destination, source in
+                                     self.image_description.mapping_of_build_files_and_directories.items()]
+        # Use only the dockerfile itself for hashing. In order to accomplish that,
+        # set the dockerfile only as destination in the mapping
+        dockerfile = Path(self.image_description.dockerfile).name
+        files_directories_to_hash.append(PathMapping(str(dockerfile),
+                                                     self.image_description.dockerfile))
         self.logger.debug("files_directories_list_hasher %s", files_directories_to_hash)
         hash_of_build_context = files_directories_list_hasher.hash(files_directories_to_hash)
         self.logger.debug("hash_of_build_context %s", self._encode_hash(hash_of_build_context))
diff --git a/exasol_integration_test_docker_environment/lib/docker/images/create/utils/character_length_checker.py b/exasol_integration_test_docker_environment/lib/docker/images/create/utils/character_length_checker.py
new file mode 100644
index 000000000..c453a2f21
--- /dev/null
+++ b/exasol_integration_test_docker_environment/lib/docker/images/create/utils/character_length_checker.py
@@ -0,0 +1,34 @@
+from pathlib import PurePath
+from typing import List
+
+
+class CharacterLengthChecker:
+    """
+    This helper class counts the number of character of directory- and file-names.
+    If this number exceeds a given limit it raises an exception.
+    The goal is to avoid high memory consumption for calculation of huge directories
+    which contain many sub-directories/files (broad directory trees) and long names.
+    """
+    def __init__(self, root_directory: PurePath, max_characters_paths: int,
+                 count_directory_names: bool, count_file_names: bool):
+        self._num_characters = 0
+        self._max_characters_paths = max_characters_paths
+        self._root_directory = root_directory
+        self._count_directory_names = count_directory_names
+        self._count_file_names = count_file_names
+
+    def add_and_check(self, directories: List[str], files: List[str]) -> None:
+        """
+        Adds number of characters of the names of all directories and files to the internal counter
+        and applies the check if the internal counter becomes greater than the given limit.
+        :param directories Characters of all names of this list will be counted
+        :param files Characters of all names of this list will be counted
+        raises: OSError if the counted character is greater than the given limit
+        """
+        if self._count_directory_names:
+            self._num_characters += sum([len(d) for d in directories])
+        if self._count_file_names:
+            self._num_characters += sum([len(f) for f in files])
+
+        if self._num_characters > self._max_characters_paths:
+            raise OSError(f"Walking through too many directories. Aborting. Please verify: {self._root_directory}")
diff --git a/exasol_integration_test_docker_environment/lib/docker/images/create/utils/file_directory_list_hasher.py b/exasol_integration_test_docker_environment/lib/docker/images/create/utils/file_directory_list_hasher.py
index 38c46b806..43b8f7e38 100644
--- a/exasol_integration_test_docker_environment/lib/docker/images/create/utils/file_directory_list_hasher.py
+++ b/exasol_integration_test_docker_environment/lib/docker/images/create/utils/file_directory_list_hasher.py
@@ -1,12 +1,18 @@
 import hashlib
 import os
 import stat
+from dataclasses import dataclass
 from multiprocessing import Pool
-from pathlib import Path
-from typing import List, Tuple
+from pathlib import Path, PurePath
+from typing import List, Callable
 
 import humanfriendly
 
+from exasol_integration_test_docker_environment.lib.docker.images.create.utils.character_length_checker import \
+    CharacterLengthChecker
+from exasol_integration_test_docker_environment.lib.docker.images.create.utils.symlink_loop_checker import \
+    SymlinkLoopChecker
+
 HASH_FUNCTIONS = {
     'md5': hashlib.md5,
     'sha1': hashlib.sha1,
@@ -15,6 +21,45 @@
 }
 
 
+@dataclass(frozen=True)
+class PathMapping:
+    """
+    Describes a mapping of a file or directory from source to destination.
+    """
+    destination: PurePath
+    source: Path
+
+
+@dataclass(frozen=True)
+class DestinationMapping:
+    """
+    Represents one file/directory found by traversing the source directory tree.
+    """
+    destination_path: PurePath
+    """The path of the file/directory in the destination (including destination_root_path)"""
+    source_path: Path
+    """The path of the file/directory in the source"""
+
+    def use_for_hashing(self, hash_files: bool, hash_directories: bool) -> bool:
+        """
+        Returns if this file/directory entry should be used for file hashing, considering parameters
+        "hash_files" and "hash_directories".
+        """
+        if self.source_path.is_file():
+            return hash_files
+        else:
+            return hash_directories
+
+
+@dataclass(frozen=True)
+class DirectoryMappingResult:
+    """
+    Contains all entries found by traversing all mapping directory paths.
+    """
+    sources: List[Path]
+    paths_for_hashing: List[DestinationMapping]
+
+
 class FileDirectoryListHasher:
 
     def __init__(self,
@@ -23,7 +68,6 @@ def __init__(self,
                  hash_file_names: bool = False,
                  hash_permissions: bool = False,
                  hash_directory_names: bool = False,
-                 use_relative_paths: bool = False,
                  excluded_directories=None,
                  excluded_files=None,
                  excluded_extensions=None,
@@ -32,7 +76,6 @@ def __init__(self,
                  max_characters_paths: int = 500000000
                  ):
         self.MAX_CHARACTERS_PATHS = max_characters_paths
-        self.use_relative_paths = use_relative_paths
         self.workers = workers
         self.excluded_files = excluded_files
         self.excluded_extensions = excluded_extensions
@@ -56,50 +99,93 @@ def __init__(self,
             self.excluded_extensions = []
 
         self.path_hasher = PathHasher(hashfunc,
-                                      hash_permissions=hash_permissions,
-                                      use_relative_paths=use_relative_paths)
+                                      hash_permissions=hash_permissions)
         self.file_content_hasher = FileContentHasher(hashfunc, blocksize)
 
-    def hash(self, files_and_directories: List[str]) -> bytes:
-        collected_directories = []
-        collected_files = []
+    def hash(self, files_and_directories: List[PathMapping]) -> bytes:
         if not isinstance(files_and_directories, List):
             raise Exception("List with paths expected and not '%s' with type %s"
                             % (files_and_directories, type(files_and_directories)))
+
+        directory_mapping_result = self.collect_dest_path_and_src_files(files_and_directories)
+        hashes = self.compute_hashes(directory_mapping_result)
+        return self._reduce_hash(hashes)
+
+    @staticmethod
+    def check_no_duplicate_destinations(mappings: List[DestinationMapping]) -> None:
+        """
+        Verify that there are no duplicate mappings to same destination.
+        This can happen, if the destination of two mappings are equal and the two sources contains the same sub-path
+        structure; or if subpaths of destinations and sources somehow match;
+        or if the destination of two mappings is the same file.
+        """
+        destination_paths_set = set(p.destination_path for p in mappings)
+        if len(destination_paths_set) != len(mappings):
+            raise AssertionError(f"Directory content for hashing contains duplicates: {mappings}")
+
+    def collect_dest_path_and_src_files(self, files_and_directories: List[PathMapping]) -> DirectoryMappingResult:
+        """
+        Traverse the source paths of all mappings and assemble two lists:
+        1) Destination paths-names (for Hashing):
+            Mapping of source file/directory -> destination of file/directory,
+            ordered by destination path. List[DestinationMapping].
+        2) Source files, ordered by destination path
+        Collection is done considering the current configuration
+         - for both: excluded_directory/excluded_extensions/excluded_file
+         - for 1): hash_file_names/hash_directory_names
+        """
+        collected_dest_paths: List[DestinationMapping] = list()
+
+        def replace_src_by_dest_path(src: PurePath, dest: PurePath, target: str) -> PurePath:
+            if not target.startswith(str(src)):
+                raise RuntimeError(f"path target {target} does not start with source: {src}")
+            p = Path(target).relative_to(src)
+            return Path(dest) / p
+
         for file_or_directory in files_and_directories:
-            if os.path.isdir(file_or_directory):
-                self.collect_files_and_directories(
-                    file_or_directory, collected_directories, collected_files)
-            elif os.path.isfile(file_or_directory):
-                collected_files.append(("", file_or_directory))
+            source = file_or_directory.source
+            destination = file_or_directory.destination
+
+            def handle_directory(directories: List[str]) -> None:
+                new_dest_paths_mappings = [DestinationMapping(
+                                            destination_path=replace_src_by_dest_path(source, destination, p),
+                                            source_path=Path(p)) for p in directories]
+                collected_dest_paths.extend(new_dest_paths_mappings)
+
+            def handle_files(files: List[str]) -> None:
+                collected_dest_paths.extend([DestinationMapping(
+                                           destination_path=replace_src_by_dest_path(source, destination, f),
+                                           source_path=Path(f)) for f in files])
+
+            if os.path.isdir(source):
+                self.traverse_directory(source, handle_directory, handle_files)
+            elif os.path.isfile(source):
+                collected_dest_paths.append(DestinationMapping(destination_path=Path(destination),
+                                                               source_path=Path(source)))
             else:
-                raise FileNotFoundError("Could not find file or directory %s" % file_or_directory)
-        hashes = self.compute_hashes(collected_directories, collected_files)
-        return self._reduce_hash(hashes)
+                raise FileNotFoundError("Could not find file or directory %s" % source)
+        collected_dest_paths.sort(key=lambda x: x.destination_path)
+        self.check_no_duplicate_destinations(collected_dest_paths)
 
-    def compute_hashes(self, collected_directories, collected_files):
+        # Now, after we sorted the collected paths, filter out paths which are not needed for current configuration
+        filtered_dest_paths = [d for d in collected_dest_paths
+                               if d.use_for_hashing(self.hash_file_names, self.hash_directory_names)]
+
+        collected_src_files = [p.source_path for p in collected_dest_paths if p.source_path.is_file()]
+        return DirectoryMappingResult(sources=collected_src_files, paths_for_hashing=filtered_dest_paths)
+
+    def compute_hashes(self, directory_mapping_result: DirectoryMappingResult) -> List[str]:
+        paths_for_hashing = directory_mapping_result.paths_for_hashing
+        collected_src_files = directory_mapping_result.sources
         pool = Pool(processes=self.workers)
-        if self.hash_directory_names:
-            file_path_hashes_of_directories_future = \
-                pool.map_async(self.path_hasher.hash, collected_directories, chunksize=2)
-        else:
-            file_path_hashes_of_directories_future = None
-        if self.hash_file_names:
-            file_path_hashes_of_files_future = \
-                pool.map_async(self.path_hasher.hash, collected_files, chunksize=2)
-        else:
-            file_path_hashes_of_files_future = None
-        file_list = [path for root, path in collected_files]
+        dest_path_hashes_future = \
+            pool.map_async(self.path_hasher.hash, paths_for_hashing, chunksize=2)
         file_content_hashes_future = \
-            pool.map_async(self.file_content_hasher.hash, file_list, chunksize=2)
+            pool.map_async(self.file_content_hasher.hash, collected_src_files, chunksize=2)
         hashes = []
         hashes.extend(file_content_hashes_future.get())
-        if self.hash_directory_names:
-            file_path_hashes_of_directories = file_path_hashes_of_directories_future.get()
-            hashes.extend(file_path_hashes_of_directories)
-        if self.hash_file_names:
-            file_path_hashes_of_files = file_path_hashes_of_files_future.get()
-            hashes.extend(file_path_hashes_of_files)
+        file_path_hashes_of_directories = dest_path_hashes_future.get()
+        hashes.extend(file_path_hashes_of_directories)
         return hashes
 
     def has_excluded_extension(self, f: str):
@@ -111,39 +197,23 @@ def is_excluded_file(self, f: str):
     def is_excluded_directory(self, f: str):
         return f in self.excluded_directories
 
-    def collect_files_and_directories(
-            self, directory: str,
-            collected_directories: List[Tuple[str, str]],
-            collected_files: List[Tuple[str, str]]):
-        tmp_collected_directories = []
-        tmp_collected_files = []
-        if self.hash_directory_names:
-            tmp_collected_directories.append((directory, directory))
-        inodes = set()
-        numCharacters = 0
+    def traverse_directory(self, directory: PurePath,
+                           directory_handler: Callable[[List[str]], None],
+                           file_handler: Callable[[List[str]], None]) -> None:
 
-        for root, dirs, files in os.walk(directory, topdown=True, followlinks=self.followlinks):
-            stat = os.stat(root)
-            if stat.st_ino > 0 and stat.st_ino in inodes:
-                raise OSError(
-                    f"Directory: {directory} contains symlink loops (Symlinks pointing to a parent directory). Please fix!")
-            inodes.add(stat.st_ino)
-
-            if self.hash_directory_names:
-                new_directories = [d for d in dirs if not self.is_excluded_directory(d)]
-                tmp_collected_directories.extend([(directory, os.path.join(root, d)) for d in new_directories])
-                numCharacters += sum([len(d) for d in new_directories])
-
-            new_files = [f for f in files
-                         if not self.is_excluded_file(f) and not self.has_excluded_extension(f)]
-            tmp_collected_files.extend([(directory, os.path.join(root, f)) for f in new_files])
-            numCharacters += sum([len(f) for f in new_files])
+        symlink_loop_checker = SymlinkLoopChecker()
+        character_length_checker = CharacterLengthChecker(directory, self.MAX_CHARACTERS_PATHS,
+                                                          self.hash_directory_names, self.hash_file_names)
 
-            if numCharacters > self.MAX_CHARACTERS_PATHS:
-                raise OSError(f"Walking through too many directories. Aborting. Please verify: {directory}")
+        for root, dirs, files in os.walk(directory, topdown=True, followlinks=self.followlinks):
+            symlink_loop_checker.check_and_add(root)
 
-        collected_directories.extend(sorted(tmp_collected_directories, key=lambda x: x[1]))
-        collected_files.extend(sorted(tmp_collected_files, key=lambda x: x[1]))
+            new_directories = [os.path.join(root, d) for d in dirs if not self.is_excluded_directory(d)]
+            directory_handler(new_directories)
+            new_files = [os.path.join(root, f) for f in files
+                         if not self.is_excluded_file(f) and not self.has_excluded_extension(f)]
+            character_length_checker.add_and_check(new_directories, new_files)
+            file_handler(new_files)
 
     def _reduce_hash(self, hashes):
         hasher = self.hash_func()
@@ -154,24 +224,19 @@ def _reduce_hash(self, hashes):
 
 class PathHasher:
     def __init__(self, hashfunc: str = 'md5',
-                 hash_permissions: bool = False,
-                 use_relative_paths: bool = False, ):
-        self.use_relative_paths = use_relative_paths
+                 hash_permissions: bool = False):
         self.hash_permissions = hash_permissions
         self.hash_func = HASH_FUNCTIONS.get(hashfunc)
         if not self.hash_func:
             raise NotImplementedError('{} not implemented.'.format(hashfunc))
 
-    def hash(self, root_path_pair: Tuple[str, str]):
-        original_path = Path(root_path_pair[1])
-        if self.use_relative_paths and len(root_path_pair[0]) > 0:
-            path = original_path.relative_to(Path(root_path_pair[0]))
-        else:
-            path = original_path
+    def hash(self, path_mapping: DestinationMapping):
+        src_path = path_mapping.source_path
+        dest_path = path_mapping.destination_path
         hasher = self.hash_func()
-        hasher.update(str(path).encode('utf-8'))
+        hasher.update(str(dest_path).encode('utf-8'))
         if self.hash_permissions:
-            stat_result = os.stat(original_path)
+            stat_result = os.stat(src_path)
             # we only check the executable right of the user, because git only remembers this
             user_has_executable_rights = stat.S_IXUSR & stat_result[stat.ST_MODE]
             hasher.update(str(user_has_executable_rights).encode("utf-8"))
@@ -186,9 +251,9 @@ def __init__(self, hashfunc: str = 'md5', blocksize: str = "64kb"):
         if not self.hash_func:
             raise NotImplementedError('{} not implemented.'.format(hashfunc))
 
-    def hash(self, filepath: str):
+    def hash(self, filepath: Path):
         hasher = self.hash_func()
-        with open(os.path.join(filepath), 'rb') as fp:
+        with open(filepath, 'rb') as fp:
             while True:
                 data = fp.read(self.blocksize)
                 if not data:
diff --git a/exasol_integration_test_docker_environment/lib/docker/images/create/utils/symlink_loop_checker.py b/exasol_integration_test_docker_environment/lib/docker/images/create/utils/symlink_loop_checker.py
new file mode 100644
index 000000000..afe8e10f9
--- /dev/null
+++ b/exasol_integration_test_docker_environment/lib/docker/images/create/utils/symlink_loop_checker.py
@@ -0,0 +1,23 @@
+import os
+
+
+class SymlinkLoopChecker:
+    """
+    Collects all visited directory inodes in a set and checks for duplicates.
+    Can be used to check for symlink loops.
+    """
+    def __init__(self):
+        self._inodes = set()
+
+    def check_and_add(self, directory: str) -> None:
+        """
+        Checks if given parameter directory has already been added to the internal set of inodes and raises exception
+        if found.
+        :param directory: The directory which should be checked.
+        :raises OSError if the directory already has been passed before.
+        """
+        stat = os.stat(directory)
+        if stat.st_ino > 0 and stat.st_ino in self._inodes:
+            raise OSError(
+                f"Directory: {directory} contains symlink loops (Symlinks pointing to a parent directory). Please fix!")
+        self._inodes.add(stat.st_ino)
diff --git a/exasol_integration_test_docker_environment/test/test_hash_temp_dir.py b/exasol_integration_test_docker_environment/test/test_hash_temp_dir.py
index 76d20f158..002559da8 100644
--- a/exasol_integration_test_docker_environment/test/test_hash_temp_dir.py
+++ b/exasol_integration_test_docker_environment/test/test_hash_temp_dir.py
@@ -3,13 +3,23 @@
 import shutil
 import tempfile
 import unittest
+from pathlib import Path, PurePath
 
 from exasol_integration_test_docker_environment.lib.docker.images.create.utils.file_directory_list_hasher import \
-    FileDirectoryListHasher
+    FileDirectoryListHasher, PathMapping
 
 TEST_FILE = "/tmp/SEFQWEFWQEHDUWEFDGZWGDZWEFDUWESGRFUDWEGFUDWAFGWAZESGFDWZA"
 
 
+def simple_path_mapping(src: str) -> PathMapping:
+    """
+    Helper function which maps a directory-path to: PathMapping(directory-name, directory-path),
+    e.g. /tmp/tmp123/test1 becomes PathMapping('test1', '/tmp/tmp123/test1')
+    """
+    p = Path(src)
+    return PathMapping(destination=PurePath(p.name), source=p)
+
+
 class HashTempDirTest(unittest.TestCase):
     @classmethod
     def setUpClass(self):
@@ -54,92 +64,73 @@ def tearDownClass(self):
 
     def test_single_character_directory_name(self):
         hasher = FileDirectoryListHasher(hashfunc="sha256",
-                                         use_relative_paths=False,
                                          hash_directory_names=True,
                                          hash_file_names=True)
         old_pwd = os.getcwd()
         os.chdir(self.test_dir3)
-        hash = hasher.hash(["."])
+        hash = hasher.hash([simple_path_mapping(".")])
         ascii_hash = base64.b32encode(hash).decode("ASCII")
-        self.assertEqual("W5VNU3LMI7OM2ZM7CNBW52YQ7FWZRCW5Z7PHTEUP3KSNOF3OPZXQ====", ascii_hash)
+        self.assertEqual("LVE2ZFQRMP6QLY43MKMZRHIEHE7KNSUS5LFWVJKPOWMI6JUPZHEQ====", ascii_hash)
         os.chdir(old_pwd)
 
     def test_file_content_only_fixed_hash(self):
         hasher = FileDirectoryListHasher(hashfunc="sha256",
-                                         use_relative_paths=False,
                                          hash_directory_names=False,
                                          hash_file_names=False)
-        hash = hasher.hash([TEST_FILE])
+        hash = hasher.hash([simple_path_mapping(TEST_FILE)])
         ascii_hash = base64.b32encode(hash).decode("ASCII")
         self.assertEqual("SVGVUSP5ODM3RPG3GXJFEJTYFGKX67XX7JWHJ6EEDG64L2BCBH2A====", ascii_hash)
 
     def test_file_with_path(self):
         hasher = FileDirectoryListHasher(hashfunc="sha256",
-                                         use_relative_paths=False,
                                          hash_directory_names=True,
                                          hash_file_names=True)
-        hash = hasher.hash([TEST_FILE])
+        hash = hasher.hash([simple_path_mapping(TEST_FILE)])
         ascii_hash = base64.b32encode(hash).decode("ASCII")
-        self.assertEqual("AOBF7HELTYAPHBEW6HQQ74N3BKGSNZTJXB4MTOEROHO5VH6YYJOA====", ascii_hash)
+        self.assertEqual("7D34CBUU2SNSWF3UFM6A7BYFJVV5ZFEY5F6THIMGJY725WC45KEA====", ascii_hash)
 
     def test_directory_with_relative_paths_fixed_hash(self):
         hasher = FileDirectoryListHasher(hashfunc="sha256",
-                                         use_relative_paths=True,
                                          hash_directory_names=True,
                                          hash_file_names=True)
-        hash = hasher.hash([self.test_dir1])
+        hash = hasher.hash([simple_path_mapping(self.test_dir1)])
         ascii_hash = base64.b32encode(hash).decode("ASCII")
-        self.assertEqual("VIN3VCPDX7DAC4GD37IDF4KQTCDNNH72QV5PARVGGQ4OMB4DZTLA====", ascii_hash)
+        self.assertEqual("EZ3ER6KZHAAYG4JNFGFLHUI7TVHZVIRVOV4QWJT4ERQ4XGI2GLUA====", ascii_hash)
 
     def test_directory_content_only_fixed_hash(self):
         hasher = FileDirectoryListHasher(hashfunc="sha256",
-                                         use_relative_paths=False,
                                          hash_directory_names=False,
                                          hash_file_names=False)
-        hash = hasher.hash([self.test_dir1])
+        hash = hasher.hash([simple_path_mapping(self.test_dir1)])
         ascii_hash = base64.b32encode(hash).decode("ASCII")
         self.assertEqual("TM2V22T326TCTLQ537BZAOR3I5NVHXE6IDJ4TXPCJPTUGDTI5WYQ====", ascii_hash)
 
-    def test_directory_with_relative_paths_equal(self):
+    def test_directory_to_same_destination_equal(self):
         hasher = FileDirectoryListHasher(hashfunc="sha256",
-                                         use_relative_paths=True,
                                          hash_directory_names=True,
                                          hash_file_names=True)
-        hash1 = hasher.hash([self.test_dir1])
-        hash2 = hasher.hash([self.test_dir2])
+        hash1 = hasher.hash([PathMapping(PurePath("test"), Path(self.test_dir1))])
+        hash2 = hasher.hash([PathMapping(PurePath("test"), Path(self.test_dir2))])
         ascii_hash1 = base64.b32encode(hash1).decode("ASCII")
         ascii_hash2 = base64.b32encode(hash2).decode("ASCII")
         self.assertEqual(ascii_hash1, ascii_hash2)
 
     def test_directory_without_relative_paths_not_equal(self):
         hasher = FileDirectoryListHasher(hashfunc="sha256",
-                                         use_relative_paths=False,
                                          hash_directory_names=True,
                                          hash_file_names=True)
-        hash1 = hasher.hash([self.test_dir1])
-        hash2 = hasher.hash([self.test_dir2])
+        hash1 = hasher.hash([simple_path_mapping(self.test_dir1)])
+        hash2 = hasher.hash([simple_path_mapping(self.test_dir2)])
         ascii_hash1 = base64.b32encode(hash1).decode("ASCII")
         ascii_hash2 = base64.b32encode(hash2).decode("ASCII")
         self.assertNotEqual(ascii_hash1, ascii_hash2)
 
     def test_directory_content_only_equal(self):
         hasher = FileDirectoryListHasher(hashfunc="sha256",
-                                         use_relative_paths=False,
-                                         hash_directory_names=False,
-                                         hash_file_names=False)
-        hash1 = hasher.hash([self.test_dir1])
-        hash2 = hasher.hash([self.test_dir2])
-        ascii_hash1 = base64.b32encode(hash1).decode("ASCII")
-        ascii_hash2 = base64.b32encode(hash2).decode("ASCII")
-        self.assertEqual(ascii_hash1, ascii_hash2)
-
-    def test_directory_relative_paths_equal(self):
-        hasher = FileDirectoryListHasher(hashfunc="sha256",
-                                         use_relative_paths=True,
                                          hash_directory_names=False,
                                          hash_file_names=False)
-        hash1 = hasher.hash([self.test_dir1])
-        hash2 = hasher.hash([self.test_dir2])
+        hash1 = hasher.hash([simple_path_mapping(self.test_dir1)])
+        hash2 = hasher.hash([simple_path_mapping(self.test_dir2)])
         ascii_hash1 = base64.b32encode(hash1).decode("ASCII")
         ascii_hash2 = base64.b32encode(hash2).decode("ASCII")
         self.assertEqual(ascii_hash1, ascii_hash2)
@@ -147,16 +138,14 @@ def test_directory_relative_paths_equal(self):
     def test_directory_content_only_not_equal_to_with_paths(self):
         hasher_content_only = \
             FileDirectoryListHasher(hashfunc="sha256",
-                                    use_relative_paths=False,
                                     hash_directory_names=False,
                                     hash_file_names=False)
         hasher_with_paths = \
             FileDirectoryListHasher(hashfunc="sha256",
-                                    use_relative_paths=True,
                                     hash_directory_names=True,
                                     hash_file_names=True)
-        hash1_content_only = hasher_content_only.hash([self.test_dir1])
-        hash2_with_paths = hasher_with_paths.hash([self.test_dir2])
+        hash1_content_only = hasher_content_only.hash([simple_path_mapping(self.test_dir1)])
+        hash2_with_paths = hasher_with_paths.hash([simple_path_mapping(self.test_dir2)])
         ascii_hash1_content_only = base64.b32encode(hash1_content_only).decode("ASCII")
         ascii_hash2_with_paths = base64.b32encode(hash2_with_paths).decode("ASCII")
         self.assertNotEqual(ascii_hash1_content_only, ascii_hash2_with_paths)
@@ -164,16 +153,14 @@ def test_directory_content_only_not_equal_to_with_paths(self):
     def test_directory_content_only_not_equal_to_dir_names(self):
         hasher_content_only = \
             FileDirectoryListHasher(hashfunc="sha256",
-                                    use_relative_paths=False,
                                     hash_directory_names=False,
                                     hash_file_names=False)
         hasher_with_paths = \
             FileDirectoryListHasher(hashfunc="sha256",
-                                    use_relative_paths=False,
                                     hash_directory_names=True,
                                     hash_file_names=False)
-        hash1_content_only = hasher_content_only.hash([self.test_dir1])
-        hash2_with_paths = hasher_with_paths.hash([self.test_dir1])
+        hash1_content_only = hasher_content_only.hash([simple_path_mapping(self.test_dir1)])
+        hash2_with_paths = hasher_with_paths.hash([simple_path_mapping(self.test_dir1)])
         ascii_hash1_content_only = base64.b32encode(hash1_content_only).decode("ASCII")
         ascii_hash2_with_paths = base64.b32encode(hash2_with_paths).decode("ASCII")
         self.assertNotEqual(ascii_hash1_content_only, ascii_hash2_with_paths)
@@ -181,16 +168,14 @@ def test_directory_content_only_not_equal_to_dir_names(self):
     def test_directory_content_only_not_equal_to_file_names(self):
         hasher_content_only = \
             FileDirectoryListHasher(hashfunc="sha256",
-                                    use_relative_paths=False,
                                     hash_directory_names=False,
                                     hash_file_names=False)
         hasher_with_paths = \
             FileDirectoryListHasher(hashfunc="sha256",
-                                    use_relative_paths=False,
                                     hash_directory_names=False,
                                     hash_file_names=True)
-        hash1_content_only = hasher_content_only.hash([self.test_dir1])
-        hash2_with_paths = hasher_with_paths.hash([self.test_dir1])
+        hash1_content_only = hasher_content_only.hash([simple_path_mapping(self.test_dir1)])
+        hash2_with_paths = hasher_with_paths.hash([simple_path_mapping(self.test_dir1)])
         ascii_hash1_content_only = base64.b32encode(hash1_content_only).decode("ASCII")
         ascii_hash2_with_paths = base64.b32encode(hash2_with_paths).decode("ASCII")
         self.assertNotEqual(ascii_hash1_content_only, ascii_hash2_with_paths)
@@ -198,16 +183,14 @@ def test_directory_content_only_not_equal_to_file_names(self):
     def test_directory_file_names_not_equal_to_dir_names(self):
         hasher_content_only = \
             FileDirectoryListHasher(hashfunc="sha256",
-                                    use_relative_paths=True,
                                     hash_directory_names=False,
                                     hash_file_names=True)
         hasher_with_paths = \
             FileDirectoryListHasher(hashfunc="sha256",
-                                    use_relative_paths=True,
                                     hash_directory_names=True,
                                     hash_file_names=False)
-        hash1_content_only = hasher_content_only.hash([self.test_dir1])
-        hash2_with_paths = hasher_with_paths.hash([self.test_dir2])
+        hash1_content_only = hasher_content_only.hash([simple_path_mapping(self.test_dir1)])
+        hash2_with_paths = hasher_with_paths.hash([simple_path_mapping(self.test_dir2)])
         ascii_hash1_content_only = base64.b32encode(hash1_content_only).decode("ASCII")
         ascii_hash2_with_paths = base64.b32encode(hash2_with_paths).decode("ASCII")
         self.assertNotEqual(ascii_hash1_content_only, ascii_hash2_with_paths)
diff --git a/exasol_integration_test_docker_environment/test/test_hash_temp_dir_with_files.py b/exasol_integration_test_docker_environment/test/test_hash_temp_dir_with_files.py
new file mode 100644
index 000000000..292064037
--- /dev/null
+++ b/exasol_integration_test_docker_environment/test/test_hash_temp_dir_with_files.py
@@ -0,0 +1,272 @@
+import base64
+import os
+import tempfile
+import unittest
+from pathlib import Path, PurePath
+
+from exasol_integration_test_docker_environment.lib.docker.images.create.utils.file_directory_list_hasher import \
+    FileDirectoryListHasher, PathMapping
+
+
+class HashTempDirTest(unittest.TestCase):
+    
+    def setUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.temp_path = Path(self.temp_dir.name)
+        self.test_path1 = self.temp_path / "test1"
+        self.test_path1.mkdir()
+        self.test_path2 = self.temp_path / "test2"
+        self.test_path2.mkdir()
+
+    def tearDown(self):
+        self.temp_dir.cleanup()
+
+    def test_file_name_with_relative_path(self):
+        """
+        Test that hashing of same files in different paths gives same result.
+        1. Mapping dest="test.txt", src="/tmp/.../$tmpA/test.txt"
+        2. Mapping dest="test.txt", src="/tmp/.../$tmpB/test.txt"
+        """
+        hasher_content_only = \
+            FileDirectoryListHasher(followlinks=True,
+                                    hashfunc="sha256",
+                                    hash_file_names=True,
+                                    hash_directory_names=True,
+                                    hash_permissions=True)
+        test_file1 = Path(f"{self.test_path1}/test.txt")
+        with open(test_file1, "w") as f:
+            f.write("test")
+        mapping = PathMapping(PurePath("test.txt"), test_file1)
+        hash1_content_only = hasher_content_only.hash([mapping])
+
+        test_file2 = Path(f"{self.test_path2}/test.txt")
+        with open(test_file2, "w") as f:
+            f.write("test")
+        mapping = PathMapping(PurePath("test.txt"), test_file2)
+        hash2_content_only = hasher_content_only.hash([mapping])
+
+        ascii_hash1_content_only = base64.b32encode(hash1_content_only).decode("ASCII")
+        ascii_hash2_content_only = base64.b32encode(hash2_content_only).decode("ASCII")
+        self.assertEqual(ascii_hash1_content_only, ascii_hash2_content_only)
+
+    def test_file_name_with_relative_path_in_same_sub_path(self):
+        """
+        Test that hashing of same files in different paths, but under same subpath, gives same result
+        1. Mapping dest="level0/test.txt", src="/tmp/.../$tmpA/level0/test.txt"
+        2. Mapping dest="level0/test.txt", src="/tmp/.../$tmpB/level0/test.txt"
+        """
+        hasher_content_only = \
+            FileDirectoryListHasher(followlinks=True,
+                                    hashfunc="sha256",
+                                    hash_file_names=True,
+                                    hash_directory_names=True,
+                                    hash_permissions=True)
+        p1 = self.test_path1 / "level0"
+        p1.mkdir()
+        test_file1 = p1 / "test.txt"
+        with open(test_file1, "w") as f:
+            f.write("test")
+        mapping = PathMapping(PurePath("level0/test.txt"), test_file1)
+        hash1_content_only = hasher_content_only.hash([mapping])
+
+        p2 = self.test_path2 / "level0"
+        p2.mkdir()
+        test_file2 = p2 / "test.txt"
+        with open(test_file2, "w") as f:
+            f.write("test")
+        mapping = PathMapping(PurePath("level0/test.txt"), test_file2)
+        hash2_content_only = hasher_content_only.hash([mapping])
+
+        ascii_hash1_content_only = base64.b32encode(hash1_content_only).decode("ASCII")
+        ascii_hash2_content_only = base64.b32encode(hash2_content_only).decode("ASCII")
+        self.assertEqual(ascii_hash1_content_only, ascii_hash2_content_only)
+
+    def test_file_name_with_relative_path_in_different_sub_path(self):
+        """
+        Test that hashing of same files in different paths, and different subpaths, gives different result.
+        1. Mapping dest="level0/test.txt", src="/tmp/.../level0/test.txt"
+        2. Mapping dest="level0/level1_0/test.txt", src="/tmp/.../level0/level1_0/test.txt"
+        """
+        hasher_content_only = \
+            FileDirectoryListHasher(followlinks=True,
+                                    hashfunc="sha256",
+                                    hash_file_names=True,
+                                    hash_directory_names=True,
+                                    hash_permissions=True)
+        p1 = self.test_path1 / "level0"
+        p1.mkdir()
+        test_file1 = p1 / "test.txt"
+        with open(test_file1, "w") as f:
+            f.write("test")
+        hash1_content_only = hasher_content_only.hash([PathMapping(PurePath("level0/test.txt"), test_file1)])
+
+        p2 = self.test_path2 / "level0" / "level1_0"
+        p2.mkdir(parents=True)
+        test_file2 = p2 / "test.txt"
+
+        with open(test_file2, "w") as f:
+            f.write("test")
+        hash2_content_only = hasher_content_only.hash([PathMapping(PurePath("level0/level1_0/test.txt"),
+                                                                   test_file2)])
+
+        ascii_hash1_content_only = base64.b32encode(hash1_content_only).decode("ASCII")
+        ascii_hash2_content_only = base64.b32encode(hash2_content_only).decode("ASCII")
+        self.assertNotEqual(ascii_hash1_content_only, ascii_hash2_content_only)
+
+    def test_file_name_with_relative_path_in_relative_path_as_argument(self):
+        """
+        Test that hashing of same files in different paths, gives same result, using relative paths as argument
+        for source and destination path in the mapping.
+        For that, we need to change pwd before running hasher_content_only.hash.
+        1. Mapping dest="test.txt", src="test.txt"
+        2. Mapping dest="test.txt", src="test.txt"
+        """
+        hasher_content_only = \
+            FileDirectoryListHasher(followlinks=True,
+                                    hashfunc="sha256",
+                                    hash_file_names=True,
+                                    hash_directory_names=True,
+                                    hash_permissions=True)
+        test_file = f"test.txt"
+        old_pwd = os.getcwd()
+        os.chdir(self.test_path1)
+        with open(test_file, "w") as f:
+            f.write("test")
+        mapping = PathMapping(PurePath("test.txt"), Path(test_file))
+        hash1_content_only = hasher_content_only.hash([mapping])
+        os.chdir(self.test_path2)
+
+        with open(test_file, "w") as f:
+            f.write("test")
+        mapping = PathMapping(PurePath("test.txt"), Path(test_file))
+        hash2_content_only = hasher_content_only.hash([mapping])
+        os.chdir(old_pwd)
+        ascii_hash1_content_only = base64.b32encode(hash1_content_only).decode("ASCII")
+        ascii_hash2_content_only = base64.b32encode(hash2_content_only).decode("ASCII")
+        self.assertEqual(ascii_hash1_content_only, ascii_hash2_content_only)
+
+    def test_duplicated_file_mapping_raises_exception(self):
+        """
+        Test that a duplicated mapping raises an exception.
+        1. Mapping dest="test.txt", src="/tmp/.../$tmpB/level0/level1_0/test.txt"
+        2. Mapping dest="test.txt", src="/tmp/.../$tmpB/level0/level1_1/test.txt"
+        """
+        hasher_content_only = \
+            FileDirectoryListHasher(followlinks=True,
+                                    hashfunc="sha256",
+                                    hash_file_names=True,
+                                    hash_directory_names=True,
+                                    hash_permissions=True)
+        p1 = self.test_path1 / "level0" / "level1_0"
+        p1.mkdir(parents=True)
+        test_file1 = p1 / "test.txt"
+
+        with open(test_file1, "w") as f:
+            f.write("test")
+
+        p2 = self.test_path2 / "level0" / "level1_1"
+        p2.mkdir(parents=True)
+        test_file2 = p2 / "test.txt"
+        with open(test_file2, "w") as f:
+            f.write("test")
+
+        path_mappings = [PathMapping(PurePath("test.txt"), test_file1),
+                         PathMapping(PurePath("test.txt"), test_file2)]
+        self.assertRaises(AssertionError, lambda: hasher_content_only.hash(path_mappings))
+
+    def test_duplicated_path_mapping_raises_exception(self):
+        """
+        Test that a duplicated mapping raises an exception. Mapping source is here a directory containing one file.
+        1. Mapping dest="test", src="/tmp/.../$tmpA/level0/level1_0", content under src="test/test.txt"
+        2. Mapping dest="test", src="/tmp/.../$tmpB/level0/level1_1", content under src="test/test.txt"
+        """
+        hasher_content_only = \
+            FileDirectoryListHasher(followlinks=True,
+                                    hashfunc="sha256",
+                                    hash_file_names=True,
+                                    hash_directory_names=True,
+                                    hash_permissions=True)
+
+        p1 = self.test_path1 / "level0" / "level1_0" / "test"
+        test_file1 = p1 / "test.txt"
+        p1.mkdir(parents=True)
+        with open(test_file1, "w") as f:
+            f.write("test")
+        p2 = self.test_path2 / "level0" / "level1_1" / "test"
+        p2.mkdir(parents=True)
+        test_file2 = p2 / "test.txt"
+        with open(test_file2, "w") as f:
+            f.write("test")
+
+        path1 = p1.parent
+        path2 = p2.parent
+        path_mappings = [PathMapping(PurePath("test"), path1),
+                         PathMapping(PurePath("test"), path2)]
+        self.assertRaises(AssertionError, lambda: hasher_content_only.hash(path_mappings))
+
+    def test_duplicated_path_mapping_with_subpath_raises_exception(self):
+        """
+        Test that a duplicated mapping raises an exception. Mapping source is here a directory containing one file.
+        1. Mapping dest="test/abc", src="/tmp/.../$tmpA/level0/level1_0", content under src="test/test.txt"
+        2. Mapping dest="test/abc", src="/tmp/.../$tmpB/level0/level1_1", content under src="test/test.txt"
+        """
+        hasher_content_only = \
+            FileDirectoryListHasher(followlinks=True,
+                                    hashfunc="sha256",
+                                    hash_file_names=True,
+                                    hash_directory_names=True,
+                                    hash_permissions=True)
+
+        p1 = self.test_path1 / "level0" / "level1_0" / "test"
+        test_file1 = p1 / "test.txt"
+        p1.mkdir(parents=True)
+        with open(test_file1, "w") as f:
+            f.write("test")
+        p2 = self.test_path2 / "level0" / "level1_1" / "test"
+        p2.mkdir(parents=True)
+        test_file2 = p2 / "test.txt"
+        with open(test_file2, "w") as f:
+            f.write("test")
+
+        path1 = p1.parent
+        path2 = p2.parent
+        destination_path = PurePath("test/abc")
+        path_mappings = [PathMapping(destination_path, path1), PathMapping(destination_path, path2)]
+        self.assertRaises(AssertionError, lambda: hasher_content_only.hash(path_mappings))
+
+    def test_duplicated_path_mapping_with_destination_subpath_raises_exception(self):
+        """
+        Test that a duplicated mapping raises an exception.
+        In this scenario we have one path which maps to a destination containing a subpath;
+        the second path maps to root destination of the first, but contains the subdirectory of the first in the
+        source directory.
+        1. Mapping dest="test", src="/tmp/.../$tmpA", content under src="abc/level0/level1_0/test/test.txt"
+        2. Mapping dest="test/abc", src="/tmp/.../$tmpB", content under src="level0/level1_0/test/test.txt"
+        """
+        hasher_content_only = \
+            FileDirectoryListHasher(followlinks=True,
+                                    hashfunc="sha256",
+                                    hash_file_names=True,
+                                    hash_directory_names=True,
+                                    hash_permissions=True)
+
+        p1 = self.test_path1 / "level1_0" / "test"
+        test_file1 = p1 / "test.txt"
+        p1.mkdir(parents=True)
+        with open(test_file1, "w") as f:
+            f.write("test")
+        p2 = self.test_path2 / "abc" /"level1_0" / "test"
+        p2.mkdir(parents=True)
+        test_file2 = p2 / "test.txt"
+        with open(test_file2, "w") as f:
+            f.write("test")
+
+        path1 = self.test_path1
+        path2 = self.test_path2
+
+        path_mappings = [PathMapping(PurePath("test/abc"), path1), PathMapping(PurePath("test"), path2)]
+        self.assertRaises(AssertionError, lambda: hasher_content_only.hash(path_mappings))
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/exasol_integration_test_docker_environment/test/test_symlink_loops.py b/exasol_integration_test_docker_environment/test/test_symlink_loops.py
index c5579c39c..d39325074 100644
--- a/exasol_integration_test_docker_environment/test/test_symlink_loops.py
+++ b/exasol_integration_test_docker_environment/test/test_symlink_loops.py
@@ -4,7 +4,7 @@
 import tempfile
 
 from exasol_integration_test_docker_environment.lib.docker.images.create.utils.file_directory_list_hasher import \
-    FileDirectoryListHasher
+    FileDirectoryListHasher, PathMapping
 
 
 class TestSymlinkLoops(unittest.TestCase):
@@ -47,13 +47,12 @@ def tearDown(self):
     def test_symlink_detection(self):
         self.generate_test_dir(True)
         hasher = FileDirectoryListHasher(hashfunc="sha256",
-                                         use_relative_paths=False,
                                          hash_directory_names=True,
                                          hash_file_names=True,
                                          followlinks=True)
         exception_thrown = False
         try:
-            hash = hasher.hash([self.temp_dir])
+            hash = hasher.hash([PathMapping(destination=self.__class__.__name__, source=self.temp_dir)])
         except OSError as e:
             if "contains symlink loops" in str(e):
                 exception_thrown = True
@@ -62,14 +61,13 @@ def test_symlink_detection(self):
     def test_symlink_detection_size(self):
         self.generate_test_dir(True)
         hasher = FileDirectoryListHasher(hashfunc="sha256",
-                                         use_relative_paths=False,
                                          hash_directory_names=True,
                                          hash_file_names=True,
                                          followlinks=True,
                                          max_characters_paths=100)
         exception_thrown = False
         try:
-            hash = hasher.hash([self.temp_dir])
+            hash = hasher.hash([PathMapping(destination=self.__class__.__name__, source=self.temp_dir)])
         except OSError as e:
             if "Walking through too many directories." in str(e):
                 exception_thrown = True
@@ -79,11 +77,10 @@ def test_symlink_no_loop(self):
         symlink_dest = self.generate_test_dir(False)
         os.symlink(self.temp_dir_dummy, symlink_dest)
         hasher = FileDirectoryListHasher(hashfunc="sha256",
-                                         use_relative_paths=False,
                                          hash_directory_names=True,
                                          hash_file_names=True,
                                          followlinks=True)
-        hash = hasher.hash([self.temp_dir])
+        hash = hasher.hash([PathMapping(destination=self.__class__.__name__, source=self.temp_dir)])
 
 
 if __name__ == '__main__':