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

Issue:
The check for an existing upload target doesn't work in upload_file_to_db.py
# Background

- I think, the check works, because we use for the bucket_file_path in list_files the whole file path together with the file name

https://github.com/exasol/integration-test-docker-environment/blob/0283ddea7570818dea74726e52c56ddf68906494/exasol_integration_test_docker_environment/lib/test_environment/database_setup/upload_file_to_db.py#L112


Discussion / hints:

Fix patch:
diff --git a/doc/changes/changes_1.6.0.md b/doc/changes/changes_1.6.0.md
index 7e48d68fe..0263d39eb 100644
--- a/doc/changes/changes_1.6.0.md
+++ b/doc/changes/changes_1.6.0.md
@@ -17,4 +17,5 @@ If you need further versions, please open an issue.
 ## Changes
 
  - #294: Fixed that the client logging config gets overwritten
- - #295: Fixed that the test_container_content parameter pollutes the logs 
\ No newline at end of file
+ - #295: Fixed that the test_container_content parameter pollutes the logs
+ - #300: Fixed BucketFS uploads during reusing of an environment
\ No newline at end of file


Test patch:
diff --git a/exasol_integration_test_docker_environment/lib/api/common.py b/exasol_integration_test_docker_environment/lib/api/common.py
index 14ec50830..4e123fcf5 100644
--- a/exasol_integration_test_docker_environment/lib/api/common.py
+++ b/exasol_integration_test_docker_environment/lib/api/common.py
@@ -117,8 +117,8 @@ def generate_root_task(task_class, *args, **kwargs) -> DependencyLoggerBaseTask:
     return task_class(**params)
 
 
-def run_task(task_creator: Callable[[], DependencyLoggerBaseTask], workers: int,
-             task_dependencies_dot_file: Optional[str],
+def run_task(task_creator: Callable[[], DependencyLoggerBaseTask], workers: int = 2,
+             task_dependencies_dot_file: Optional[str] = None,
              log_level: str = None, use_job_specific_log_file: bool = False) \
         -> Any:
     setup_worker()
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 2cfb66559..d6f622284 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
@@ -1,3 +1,4 @@
+import dataclasses
 from pathlib import Path
 from typing import Tuple
 
@@ -21,6 +22,12 @@
     TimeBasedBucketFSSyncWaiter
 
 
+@dataclasses.dataclass
+class UploadResult:
+    upload_target: str
+    reused: bool
+
+
 class UploadFileToBucketFS(DockerBaseTask):
     environment_name = luigi.Parameter()
     test_environment_info = JsonPickleParameter(
@@ -53,10 +60,18 @@ def run_task(self):
                                      log_file,
                                      pattern_to_wait_for,
                                      sync_time_estimation)
+                self.return_object(UploadResult(
+                    upload_target=upload_target,
+                    reused=False
+                ))
             else:
                 self.logger.warning("Reusing uploaded target %s instead of file %s",
                                     upload_target, file_to_upload)
                 self.write_logs("Reusing")
+                self.return_object(UploadResult(
+                    upload_target=upload_target,
+                    reused=True
+                ))
 
     def upload_and_wait(self, database_container,
                         file_to_upload: str, upload_target: str,
@@ -97,27 +112,25 @@ def should_be_reused(self, upload_target: str):
         return self.reuse_uploaded and self.exist_file_in_bucketfs(upload_target)
 
     @staticmethod
-    def split_upload_target(upload_target: str) -> Tuple[str, str]:
+    def split_upload_target(upload_target: str) -> Tuple[str, str, str]:
         upload_parts = upload_target.split("/")
         bucket_name = upload_parts[0]
-        upload_target_in_bucket = "/".join(upload_parts[1:])
-        return bucket_name, upload_target_in_bucket
+        path_in_bucket = "/".join(upload_parts[1:-1])
+        file_in_bucket = upload_parts[-1]
+        return bucket_name, path_in_bucket, file_in_bucket
 
     def exist_file_in_bucketfs(self, upload_target: str) -> bool:
         self.logger.info("Check if file %s exist in bucketfs", upload_target)
-        bucket_name, upload_target = self.split_upload_target(upload_target)
+        bucket_name, path_in_bucket, file_in_bucket = self.split_upload_target(upload_target)
 
         bucket_config = self.generate_bucket_config(bucket_name)
         try:
             files = list_files.list_files_in_bucketfs(
                 bucket_config=bucket_config,
-                bucket_file_path=upload_target)
-            if upload_target not in files:
-                raise RuntimeError(f"Unexpected behavior of list_files.list_files_in_bucketfs. "
-                                   f"bucket_file_path='{upload_target}' was requested, but not returned.")
+                bucket_file_path=path_in_bucket)
+            return file_in_bucket in files
         except FileNotFoundError as ex:
             return False
-        return True
 
     def generate_bucket_config(self, bucket_name: str) -> BucketConfig:
         connection_config = BucketFSConnectionConfig(
@@ -135,12 +148,12 @@ def generate_bucket_config(self, bucket_name: str) -> BucketConfig:
     def upload_file(self, file_to_upload: str, upload_target: str):
         self.logger.info("upload file %s to %s",
                          file_to_upload, upload_target)
-        bucket_name, upload_target = self.split_upload_target(upload_target)
+        bucket_name, path_in_bucket, file_in_bucket = self.split_upload_target(upload_target)
 
         bucket_config = self.generate_bucket_config(bucket_name)
         upload.upload_file_to_bucketfs(
             bucket_config=bucket_config,
-            bucket_file_path=upload_target,
+            bucket_file_path=f"{path_in_bucket}/{file_in_bucket}",
             local_file_path=Path(file_to_upload))
         return f"File '{file_to_upload}' to '{upload_target}'"
 
diff --git a/exasol_integration_test_docker_environment/test/test_upload.py b/exasol_integration_test_docker_environment/test/test_upload.py
index b9769058d..781c8a5bb 100644
--- a/exasol_integration_test_docker_environment/test/test_upload.py
+++ b/exasol_integration_test_docker_environment/test/test_upload.py
@@ -2,16 +2,15 @@
 import unittest
 from pathlib import Path
 from sys import stderr
+from typing import List
 
 import luigi
-from exasol_bucketfs_utils_python import list_files, download
-from exasol_bucketfs_utils_python.bucket_config import BucketConfig
-from exasol_bucketfs_utils_python.bucketfs_config import BucketFSConfig
-from exasol_bucketfs_utils_python.bucketfs_connection_config import BucketFSConnectionConfig
+from exasol.bucketfs import Service, Bucket, as_string
+from exasol_bucketfs_utils_python import list_files
 
-from exasol_integration_test_docker_environment.lib.api.common import generate_root_task
+from exasol_integration_test_docker_environment.lib.api.common import generate_root_task, run_task
 from exasol_integration_test_docker_environment.lib.test_environment.database_setup.upload_file_to_db import \
-    UploadFileToBucketFS
+    UploadFileToBucketFS, UploadResult
 from exasol_integration_test_docker_environment.testing import utils
 from exasol_integration_test_docker_environment.testing.api_test_environment import ApiTestEnvironment
 
@@ -19,9 +18,13 @@
 PATH_IN_BUCKET = "upload_test"
 
 
+def construct_upload_target(file_to_upload: str) -> str:
+    return f"{BUCKET_NAME}/{PATH_IN_BUCKET}/{file_to_upload}"
+
+
 class TestfileUpload(UploadFileToBucketFS):
     path = luigi.Parameter()
-    file_to_upload = luigi.Parameter()
+    file_to_upload = luigi.Parameter()  # type: str
 
     def get_log_file(self) -> str:
         return "/exa/logs/cored/*bucketfsd*"
@@ -33,7 +36,7 @@ def get_file_to_upload(self) -> str:
         return f"{Path(str(self.path)) / str(self.file_to_upload)}"
 
     def get_upload_target(self) -> str:
-        return f"{BUCKET_NAME}/{PATH_IN_BUCKET}/{self.file_to_upload}"
+        return construct_upload_target(self.file_to_upload)
 
     def get_sync_time_estimation(self) -> int:
         """Estimated time in seconds which the bucketfs needs to extract and sync a uploaded file"""
@@ -54,68 +57,140 @@ def setUpClass(cls):
     def tearDownClass(cls):
         utils.close_environments(cls.environment, cls.test_environment)
 
-    def _upload(self, temp_dir: str, file_to_upload: str):
-        task = generate_root_task(task_class=TestfileUpload,
-                                  path=temp_dir,
-                                  file_to_upload=file_to_upload,
-                                  environment_name=self.environment.name,
-                                  test_environment_info=self.environment.environment_info,
-                                  bucketfs_write_password=self.environment.bucketfs_password
-                                  )
-        try:
-            success = luigi.build([task], workers=1, local_scheduler=True, log_level="INFO")
-            if not success:
-                raise Exception("Task failed")
-        except Exception as e:
-            task.cleanup(False)
-            raise RuntimeError("Error uploading test file.") from e
-
-    def get_bucket_config(self):
+    def _upload(self, temp_dir: str, file_to_upload: str, reuse: bool) -> UploadResult:
+        task_creator = lambda: generate_root_task(task_class=TestfileUpload,
+                                                  path=temp_dir,
+                                                  file_to_upload=file_to_upload,
+                                                  environment_name=self.environment.name,
+                                                  test_environment_info=self.environment.environment_info,
+                                                  bucketfs_write_password=self.environment.bucketfs_password,
+                                                  reuse_uploaded=reuse)
+        result = run_task(task_creator=task_creator, log_level="INFO")
+        return result
+
+    def _get_bucket(self) -> Bucket:
         db_info = self.environment.environment_info.database_info
-        connection_config = BucketFSConnectionConfig(
-            host=db_info.host, port=int(db_info.bucketfs_port),
-            user=self.environment.bucketfs_username, pwd=self.environment.bucketfs_password,
-            is_https=False)
-        bucketfs_config = BucketFSConfig(
-            connection_config=connection_config,
-            bucketfs_name="bfsdefault")
-
-        bucket_config = BucketConfig(
-            bucket_name=BUCKET_NAME,
-            bucketfs_config=bucketfs_config)
-
-        return bucket_config
-
-    def download_file_and_read(self, output_path: Path, filename: str):
-        local_output_file_path = output_path / filename
-        download.download_from_bucketfs_to_file(
-            bucket_config=self.get_bucket_config(),
-            bucket_file_path=f"{PATH_IN_BUCKET}/{filename}",
-            local_file_path=local_output_file_path)
-        file_content = local_output_file_path.read_text()
+        URL = f"http://{db_info.host}:{db_info.bucketfs_port}"
+        CREDENTAILS = {BUCKET_NAME: {"username": self.environment.bucketfs_username,
+                                     "password": self.environment.bucketfs_password}}
+        bucketfs = Service(URL, CREDENTAILS)
+        return bucketfs.buckets[BUCKET_NAME]
+
+    def _download_file(self, filename: str) -> str:
+        file_content = as_string(self._get_bucket().download(f"{PATH_IN_BUCKET}/{filename}"))
         return file_content
 
-    def test_upload(self):
-        with tempfile.TemporaryDirectory() as d:
+    def _assert_file_upload(self, file_name: str,
+                            upload_result: UploadResult,
+                            expected_reuse: bool,
+                            expected_content: str):
+        files = self._get_bucket().files
+        self.assertEqual(upload_result, UploadResult(
+            upload_target=construct_upload_target(file_name),
+            reused=expected_reuse
+        ))
+        self.assertIn(f"{PATH_IN_BUCKET}/{file_name}", list(files))
+        content = self._download_file(file_name)
+        self.assertEqual(content, expected_content)
+
+    def _create_file_and_upload(self,
+                                local_directory: str,
+                                content: str,
+                                upload_target: str,
+                                reuse: bool) -> UploadResult:
+        with open(f"{local_directory}/{upload_target}", "w") as f:
+            f.write(content)
+        return self._upload(local_directory, upload_target, reuse)
+
+    def test_upload_without_reuse(self):
+        with tempfile.TemporaryDirectory() as temp_directory:
             file_one = "test1.txt"
+            result_file_one = self._create_file_and_upload(
+                local_directory=temp_directory,
+                content=file_one,
+                upload_target=file_one,
+                reuse=False
+            )
             file_two = "test2.txt"
-            with open(f"{d}/{file_one}", "w") as f:
-                f.write(file_one)
-            with open(f"{d}/{file_two}", "w") as f:
-                f.write(file_two)
-
-            self._upload(d, file_one)
-            self._upload(d, file_two)
-
-        bucket_config = self.get_bucket_config()
-        files = list_files.list_files_in_bucketfs(bucket_config=bucket_config, bucket_file_path=PATH_IN_BUCKET)
-        self.assertEquals(sorted(list(files)), [file_one, file_two])
-        with tempfile.TemporaryDirectory() as d_target:
-            p_target = Path(d_target)
-            file_one_content = self.download_file_and_read(p_target, file_one)
-            assert file_one_content == file_one
-            file_two_content = self.download_file_and_read(p_target, file_two)
-            assert file_two_content == file_two
+            result_file_two = self._create_file_and_upload(
+                local_directory=temp_directory,
+                content=file_two,
+                upload_target=file_two,
+                reuse=False
+            )
+        self._assert_file_upload(file_name=file_one,
+                                 upload_result=result_file_one,
+                                 expected_reuse=False,
+                                 expected_content=file_one)
+        self._assert_file_upload(file_name=file_two,
+                                 upload_result=result_file_two,
+                                 expected_reuse=False,
+                                 expected_content=file_two)
+
+    def test_upload_with_reuse(self):
+        with tempfile.TemporaryDirectory() as temp_directory:
+            file_reuse = "test_reuse.txt"
+            result = self._create_file_and_upload(
+                local_directory=temp_directory,
+                content=file_reuse,
+                upload_target=file_reuse,
+                reuse=True
+            )
+        self._assert_file_upload(file_name=file_reuse,
+                                 upload_result=result,
+                                 expected_reuse=False,
+                                 expected_content=file_reuse)
+
+    def test_reupload_with_reuse(self):
+        with tempfile.TemporaryDirectory() as temp_directory:
+            file_reupload_reuse = "test_reupload_reuse.txt"
+            result = self._create_file_and_upload(
+                local_directory=temp_directory,
+                content=file_reupload_reuse,
+                upload_target=file_reupload_reuse,
+                reuse=True
+            )
+            self._assert_file_upload(file_name=file_reupload_reuse,
+                                     upload_result=result,
+                                     expected_reuse=False,
+                                     expected_content=file_reupload_reuse)
+
+            result = self._create_file_and_upload(
+                local_directory=temp_directory,
+                content=file_reupload_reuse,
+                upload_target=file_reupload_reuse,
+                reuse=True
+            )
+            self._assert_file_upload(file_name=file_reupload_reuse,
+                                     upload_result=result,
+                                     expected_reuse=True,
+                                     expected_content=file_reupload_reuse)
+
+    def test_reupload_without_reuse(self):
+        with tempfile.TemporaryDirectory() as temp_directory:
+            file_reupload_no_reuse = "test_reupload_no_reuse.txt"
+            result = self._create_file_and_upload(
+                local_directory=temp_directory,
+                content=file_reupload_no_reuse,
+                upload_target=file_reupload_no_reuse,
+                reuse=False
+            )
+            self._assert_file_upload(file_name=file_reupload_no_reuse,
+                                     upload_result=result,
+                                     expected_reuse=False,
+                                     expected_content=file_reupload_no_reuse)
+
+            expected_content_after_reupload = file_reupload_no_reuse + "2"
+            result = self._create_file_and_upload(
+                local_directory=temp_directory,
+                content=expected_content_after_reupload,
+                upload_target=file_reupload_no_reuse,
+                reuse=False
+            )
+            self._assert_file_upload(file_name=file_reupload_no_reuse,
+                                     upload_result=result,
+                                     expected_reuse=False,
+                                     expected_content=expected_content_after_reupload)
 
 
 if __name__ == '__main__':