run tests in a propery isolated soma-workflow

populse · Jun 13, 2024 · afe7c46 · afe7c46
1 parent de62c81
commit afe7c46
Showing 1 changed file with 22 additions and 7 deletions.
diff --git a/capsul/pipeline/test/test_pipeline_workflow.py b/capsul/pipeline/test/test_pipeline_workflow.py
@@ -157,12 +157,22 @@ def pipeline_definition(self):
 
 class TestPipelineWorkflow(unittest.TestCase):
 
+    _swf_dir = None
+
     def setUp(self):
+        # use a custom temporary soma-workflow dir to avoid concurrent
+        # access problems
+        if TestPipelineWorkflow._swf_dir is None:
+            TestPipelineWorkflow._swf_dir = tempfile.mkdtemp()
+            change_soma_workflow_directory(TestPipelineWorkflow._swf_dir,
+                                           socket.gethostname())
+
+        self.tmpdir = tempfile.mkdtemp()
+
         study_config = StudyConfig() #modules=StudyConfig.default_modules \
-                                   #+ ['FomConfig'])
+                                     #+ ['FomConfig'])
         self.pipeline = DummyPipeline()
         self.pipeline.set_study_config(study_config)
-        self.tmpdir = tempfile.mkdtemp()
         self.pipeline.input = osp.join(self.tmpdir, 'file_in.nii')
         self.pipeline.output1 = osp.join(self.tmpdir, 'file_out1.nii')
         self.pipeline.output2 = osp.join(self.tmpdir, 'file_out2.nii')
@@ -189,19 +199,24 @@ def setUp(self):
         study_config.somaworkflow_keep_succeeded_workflows = False
         self.exec_ids = []
 
-        # use a custom temporary soma-workflow dir to avoid concurrent
-        # access problems
-        change_soma_workflow_directory(self.tmpdir, socket.gethostname())
-
     def tearDown(self):
         for exec_id in self.exec_ids:
             self.study_config.engine.dispose(exec_id)
-        restore_soma_workflow_directory()
+        del self.study_config
         try:
             shutil.rmtree(self.tmpdir)
         except Exception:
             pass
 
+    def tearDownClass():
+        if TestPipelineWorkflow._swf_dir is not None:
+            restore_soma_workflow_directory()
+        try:
+            shutil.rmtree(TestPipelineWorkflow._swf_dir)
+            TestPipelineWorkflow._swf_dir = None
+        except Exception:
+            pass
+
     def test_requirements(self):
         engine = self.study_config.engine
         with engine.settings as session: