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

Issue:
Cleanup methods can be called multiple times when the task is a child task of multiple other tasks
Situation:

Cleanup methods can be called multiple times when the task is a child task of multiple other tasks.

Fix:

One solution could be to remember if as task already got cleaned up and return directly from the cleanup function if this is the case. For example, in the class BaseTask.
Cleanup methods can be called multiple times when the task is a child task of multiple other tasks
Situation:

Cleanup methods can be called multiple times when the task is a child task of multiple other tasks.

Fix:

One solution could be to remember if as task already got cleaned up and return directly from the cleanup function if this is the case. For example, in the class BaseTask.


Discussion / hints:

Fix patch:
diff --git a/doc/changes/changes_0.7.0.md b/doc/changes/changes_0.7.0.md
index 914042d52..5957fac71 100644
--- a/doc/changes/changes_0.7.0.md
+++ b/doc/changes/changes_0.7.0.md
@@ -21,7 +21,8 @@ If you need further versions, please open an issue.
 
 ## Bug Fixes:
 
- - #134: Release: Upload of artifacts must not run within matrix build 
+ - #134: Release: Upload of artifacts must not run within matrix build
+ - #25: Avoid that cleanup methods can be called multiple times when the task is a child task of multiple other task
 
 ## Features / Enhancements:
 


Test patch:
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 f3bcd11de..2e27e09a7 100644
--- a/exasol_integration_test_docker_environment/lib/base/base_task.py
+++ b/exasol_integration_test_docker_environment/lib/base/base_task.py
@@ -3,7 +3,7 @@
 import logging
 import shutil
 from pathlib import Path
-from typing import Dict, List, Generator, Any, Union
+from typing import Dict, List, Generator, Any, Union, Set
 
 import luigi
 import six
@@ -317,24 +317,31 @@ def create_child_task(self, task_class, **kwargs):
         params.update(kwargs)
         return task_class(**params)
 
-    def cleanup(self, success:bool):
+    def cleanup(self, success: bool):
+        self.cleanup_internal(success, set())
+
+    def cleanup_internal(self, success: bool, cleanup_checklist: Set[str]):
         self.logger.debug("Cleaning up")
-        if self._task_state != TaskState.CLEANUP and self._task_state != TaskState.CLEANED:
-            self._task_state = TaskState.CLEANUP
-            try:
-                self.cleanup_child_task(success)
-            except Exception as e:
-                self.logger.error("Error during cleaning up child tasks", e)
-            try:
-                self.cleanup_task(success)
-            except Exception as e:
-                self.logger.error("Error during cleaning up task", e)
-            if self._get_tmp_path_for_task().exists():
-                shutil.rmtree(self._get_tmp_path_for_task())
-            self._task_state = TaskState.CLEANED
-        self.logger.debug("Cleanup finished")
-
-    def cleanup_child_task(self, success:bool):
+        if str(self) not in cleanup_checklist:
+            cleanup_checklist.add(str(self))
+            if self._task_state != TaskState.CLEANUP and self._task_state != TaskState.CLEANED:
+                self._task_state = TaskState.CLEANUP
+                try:
+                    self.cleanup_child_task(success, cleanup_checklist)
+                except Exception as e:
+                    self.logger.error("Error during cleaning up child tasks", e)
+                try:
+                    self.cleanup_task(success)
+                except Exception as e:
+                    self.logger.error("Error during cleaning up task", e)
+                if self._get_tmp_path_for_task().exists():
+                    shutil.rmtree(self._get_tmp_path_for_task())
+                self._task_state = TaskState.CLEANED
+            self.logger.debug("Cleanup finished")
+        else:
+            self.logger.debug("Cleanup skipped")
+
+    def cleanup_child_task(self, success: bool, cleanup_checklist: Set[str]):
         if self._run_dependencies_target.exists():
             _run_dependencies_tasks_from_target = self._run_dependencies_target.read()
         else:
@@ -343,12 +350,12 @@ def cleanup_child_task(self, success:bool):
         reversed_run_dependencies_task_list = list(_run_dependencies_tasks)
         reversed_run_dependencies_task_list.reverse()
         for task in reversed_run_dependencies_task_list:
-            task.cleanup(success)
+            task.cleanup_internal(success, cleanup_checklist)
 
         reversed_registered_task_list = list(self._registered_tasks)
         reversed_registered_task_list.reverse()
         for task in reversed_registered_task_list:
-            task.cleanup(success)
+            task.cleanup_internal(success, cleanup_checklist)
 
-    def cleanup_task(self, success:bool):
+    def cleanup_task(self, success: bool):
         pass
diff --git a/exasol_integration_test_docker_environment/test/test_base_task_cleanup.py b/exasol_integration_test_docker_environment/test/test_base_task_cleanup.py
new file mode 100644
index 000000000..77a1a5ffd
--- /dev/null
+++ b/exasol_integration_test_docker_environment/test/test_base_task_cleanup.py
@@ -0,0 +1,142 @@
+import shutil
+import unittest
+
+import luigi
+from luigi import BoolParameter, IntParameter
+
+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
+
+TestBaseTask = DependencyLoggerBaseTask
+
+
+class TestTaskBase(TestBaseTask):
+    different_grandchild = BoolParameter()
+    use_dynamic_dependency = BoolParameter()
+
+    def _build_child_task(self, use_dynamic_dependency):
+        index_for_grandchild = 0
+        child_a = self.create_child_task(task_class=TestTaskChildA,
+                                         index_for_grandchild=index_for_grandchild,
+                                         use_dynamic_dependency=use_dynamic_dependency)
+        if self.different_grandchild:
+            index_for_grandchild = 1
+        child_b = self.create_child_task(task_class=TestTaskChildB,
+                                         index_for_grandchild=index_for_grandchild,
+                                         use_dynamic_dependency=use_dynamic_dependency)
+        return child_a, child_b
+
+    def register_required(self):
+        if not self.use_dynamic_dependency:
+            child_a, child_b = self._build_child_task(False)
+            self.child_tasks = self.register_dependencies([child_a, child_b])
+
+    def run_task(self):
+        if self.use_dynamic_dependency:
+            child_a, child_b = self._build_child_task(True)
+            yield from self.run_dependencies([child_a, child_b])
+
+
+class TestTaskChildA(TestBaseTask):
+    index_for_grandchild = IntParameter()
+    use_dynamic_dependency = BoolParameter()
+
+    def register_required(self):
+        if not self.use_dynamic_dependency:
+            grandchild = self.create_child_task(task_class=TestTaskGrandchild,
+                                                index_for_grandchild=self.index_for_grandchild)
+            self.register_dependency(grandchild)
+
+    def run_task(self):
+        if self.use_dynamic_dependency:
+            grandchild = self.create_child_task(task_class=TestTaskGrandchild,
+                                                index_for_grandchild=self.index_for_grandchild)
+            yield from self.run_dependencies(grandchild)
+
+    def cleanup_task(self, success: bool):
+        pass
+
+
+class TestTaskChildB(TestBaseTask):
+    index_for_grandchild = IntParameter()
+    use_dynamic_dependency = BoolParameter()
+
+    def register_required(self):
+        if not self.use_dynamic_dependency:
+            grandchild = self.create_child_task(task_class=TestTaskGrandchild,
+                                                index_for_grandchild=self.index_for_grandchild)
+            self.register_dependency(grandchild)
+
+    def run_task(self):
+        if self.use_dynamic_dependency:
+            grandchild = self.create_child_task(task_class=TestTaskGrandchild,
+                                                index_for_grandchild=self.index_for_grandchild)
+            yield from self.run_dependencies(grandchild)
+
+    def cleanup_task(self, success: bool):
+        pass
+
+
+global_counter = 0
+
+
+class TestTaskGrandchild(TestBaseTask):
+    index_for_grandchild = IntParameter()
+
+    def run_task(self):
+        pass
+
+    def cleanup_task(self, success: bool):
+        global global_counter
+        global_counter = global_counter + 1
+
+
+class BaseTaskCleanupTest(unittest.TestCase):
+
+    def _run_it(self, different_grandchild, use_dynamic_dependency, expected_result):
+        global global_counter
+        global_counter = 0
+        task = generate_root_task(task_class=TestTaskBase,
+                                  different_grandchild=different_grandchild,
+                                  use_dynamic_dependency=use_dynamic_dependency)
+        try:
+            luigi.build([task], workers=3, local_scheduler=True, log_level="INFO")
+            task.cleanup(success=True)
+        finally:
+            if task._get_tmp_path_for_job().exists():
+                shutil.rmtree(str(task._get_tmp_path_for_job()))
+        self.assertEqual(global_counter, expected_result, "number of Cleanups not matching")
+
+    def test_cleanup_of_grandchildren_called_only_once(self):
+        """
+        Test that creating the same grandchild task by two different parent tasks, will invoke cleanup of grandchild
+        task only once! Luigi takes care of invoking run only once, we take care to invoke cleanup() only once.
+        """
+        self._run_it(different_grandchild=False, use_dynamic_dependency=False, expected_result=1)
+
+    def test_cleanup_of_grandchildren_called_twice(self):
+        """
+        Test that creating grandchild task with different parameters by two different parent tasks,
+        will invoke cleanup of grandchild twice.
+        """
+        self._run_it(different_grandchild=True, use_dynamic_dependency=False, expected_result=2)
+
+    def test_cleanup_of_grandchildren_called_only_once_dynamic(self):
+        """
+        Test that creating the same grandchild task by two different parent tasks, will invoke cleanup of grandchild
+        task only once! Luigi takes care of invoking run only once, we take care to invoke cleanup() only once.
+        In this test all child tasks are created dynamically.
+        """
+        self._run_it(different_grandchild=False, use_dynamic_dependency=True, expected_result=1)
+
+    def test_cleanup_of_grandchildren_called_twice_dynamic(self):
+        """
+        Test that creating grandchild task with different parameters by two different parent tasks,
+        will invoke cleanup of grandchild twice.
+        In this test all child tasks are created dynamically.
+        """
+        self._run_it(different_grandchild=True, use_dynamic_dependency=True, expected_result=2)
+
+
+if __name__ == '__main__':
+    unittest.main()