Merge branch 'main' into orca_parse_negfrq

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))