Updated ARC PBS to recognise Zeus submission error (#815)

Due to the changes in Zeus, unless the user is submitting arc via n170
host, they will not be informed there is a job submission error
occurring and ARC will continuously say it cannot find the output files
(even though it technically was never able to submit the ESS jobs). This
PR will therefore error ARC if it receives such a message about job
submission issues.

Author: calvinp0

arc/job/local.py

@@ -267,6 +267,11 @@ def submit_job(path: str,
                        )
         elif cluster_soft.lower() == 'pbs' and any('qsub: Illegal attribute or resource value' in err_line for err_line in stderr):
             raise ValueError(f'Got the following error when trying to submit job:\n{stderr}. Please check your submit script')
+        elif cluster_soft.lower() == 'pbs' and (
+                any('Please do NOT submit jobs on compute nodes' in err_line for err_line in stderr)
+                or any('Jobs should be submitted on login server' in err_line for err_line in stderr)
+        ):
+            raise ValueError('PBS job submission attempted from a compute node. Submit jobs from the login server.')
     if not len(stdout) or recursion:
         return None, None
     if len(stderr) > 0 or len(stdout) == 0:

arc/job/local_test.py

@@ -9,6 +9,7 @@
 import os
 import shutil
 import unittest
+from unittest.mock import patch
 
 import arc.job.local as local
 from arc.common import ARC_PATH
@@ -96,6 +97,16 @@ def test_parse_running_jobs_ids(self):
         running_job_ids = local.parse_running_jobs_ids(stdout, cluster_soft='htcondor')
         self.assertEqual(running_job_ids, ['11224', '11225', '11226', '11227', '11228', '11229', '11230', '11231'])
 
+    def test_submit_job_pbs_compute_node_error(self):
+        """Test submit_job() error handling for PBS compute node submissions."""
+        stderr = ['qsub: Unauthorized Request: Please do NOT submit jobs on compute nodes!',
+                  'Jobs should be submitted on login server, i.e. ZEUS.']
+        with patch('arc.job.local.execute_command', side_effect=[([], stderr), ([], stderr)]):
+            with patch('time.sleep', return_value=None):
+                with self.assertRaises(ValueError) as cm:
+                    local.submit_job(path='.', cluster_soft='pbs', submit_cmd='qsub', submit_filename='submit.sh')
+        self.assertIn('compute node', str(cm.exception))
+
 
 if __name__ == '__main__':
     unittest.main(testRunner=unittest.TextTestRunner(verbosity=2))