[Bug] Add error handling for submitting jobs using pysqa (#961)

* Add error handling for submitting jobs using pysqa

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Add more tests

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: jan-janssen

src/executorlib/task_scheduler/file/spawner_pysqa.py

@@ -1,4 +1,5 @@
 import os
+from subprocess import CalledProcessError
 from typing import Optional
 
 from pysqa import QueueAdapter
@@ -90,8 +91,14 @@ def execute_with_pysqa(
             )
         submit_kwargs.update(resource_dict)
         set_current_directory_in_environment()
-        queue_id = qa.submit_job(**submit_kwargs)
-        dump(file_name=file_name, data_dict={"queue_id": queue_id})
+        try:
+            queue_id = qa.submit_job(**submit_kwargs)
+        except (ValueError, CalledProcessError) as error:
+            dump(file_name=file_name, data_dict={"error": error})
+            file_name_out = os.path.splitext(file_name)[0][:-2]
+            os.rename(file_name_out + "_i.h5", file_name_out + "_o.h5")
+        else:
+            dump(file_name=file_name, data_dict={"queue_id": queue_id})
     return queue_id
 
 

tests/unit/executor/test_flux_cluster.py

@@ -25,6 +25,27 @@
 skip_mpi4py_test = importlib.util.find_spec("mpi4py") is None
 
 
+template = """\
+#!/bin/bash
+# flux: --job-name={{job_name}}
+# flux: --env=CORES={{cores}}
+# flux: --output=time.out
+# flux: --error=error.out
+# flux: --nslots={{cores}}
+# flux: --queue={{queue}}
+{%- if run_time_max %}
+# flux: --time-limit={{run_time_max}}s
+{%- endif %}
+{%- if dependency_list %}
+{%- for jobid in dependency_list %}
+# flux: --dependency=afterok:{{jobid}}
+{%- endfor %}
+{%- endif %}
+
+{{command}}
+"""
+
+
 def echo(i):
     sleep(1)
     return i
@@ -71,6 +92,23 @@ def test_executor(self):
             self.assertEqual(len(os.listdir("executorlib_cache")), 4)
             self.assertTrue(fs1.done())
 
+    def test_executor_wrong_queue_name(self):
+        with self.assertRaises(ValueError):
+            with FluxClusterExecutor(
+                resource_dict={
+                    "cores": 2,
+                    "cwd": "executorlib_cache",
+                    "submission_template": template,
+                    "queue": "test",
+                },
+                block_allocation=False,
+                cache_directory="executorlib_cache",
+                pmi_mode=pmi,
+            ) as exe:
+                cloudpickle_register(ind=1)
+                fs1 = exe.submit(mpi_funct, 1)
+                print(fs1.result())
+
     def test_executor_no_cwd(self):
         with FluxClusterExecutor(
             resource_dict={"cores": 2},