extensions

executorlib/interactive/flux.py

@@ -91,6 +91,9 @@ def bootup(
         if self._flux_log_files and self._cwd is not None:
             jobspec.stderr = os.path.join(self._cwd, "flux.err")
             jobspec.stdout = os.path.join(self._cwd, "flux.out")
+        elif self._flux_log_files:
+            jobspec.stderr = os.path.abspath("flux.err")
+            jobspec.stdout = os.path.abspath("flux.out")
         self._future = self._flux_executor.submit(jobspec)
 
     def shutdown(self, wait: bool = True):

tests/test_executor_backend_flux.py

@@ -100,7 +100,29 @@ def test_single_task(self):
             [[(1, 2, 0), (1, 2, 1)], [(2, 2, 0), (2, 2, 1)], [(3, 2, 0), (3, 2, 1)]],
         )
 
-    def test_output_files(self):
+    def test_output_files_cwd(self):
+        file_stdout = os.path.join(os.path.dirname(__file__), "flux.out")
+        file_stderr = os.path.join(os.path.dirname(__file__), "flux.err")
+        with Executor(
+            max_cores=2,
+            resource_dict={"cores": 2, "cwd": os.path.dirname(__file__)},
+            flux_executor=self.executor,
+            backend="flux_allocation",
+            block_allocation=True,
+            flux_executor_pmi_mode=pmi,
+            flux_log_files=True,
+        ) as p:
+            output = p.map(mpi_funct, [1, 2, 3])
+        self.assertEqual(
+            list(output),
+            [[(1, 2, 0), (1, 2, 1)], [(2, 2, 0), (2, 2, 1)], [(3, 2, 0), (3, 2, 1)]],
+        )
+        self.assertTrue(os.path.exists(file_stdout))
+        self.assertTrue(os.path.exists(file_stderr))
+        os.remove(file_stdout)
+        os.remove(file_stderr)
+
+    def test_output_files_abs(self):
         file_stdout = os.path.abspath("flux.out")
         file_stderr = os.path.abspath("flux.err")
         with Executor(