adding tests for job and pjm platform files

Author: VindeeR

autosubmit/platforms/paramiko_platform.py

@@ -257,8 +257,6 @@ def agent_auth(self, port: int) -> bool:
     # noinspection PyUnusedLocal
     def interactive_auth_handler(self, title, instructions, prompt_list):
         answers = []
-        # Walk the list of prompts that the server sent that we need to answer
-        twofactor_nonpush = None
         for prompt_, _ in prompt_list:
             prompt = str(prompt_).strip().lower()
             # str() used to make sure that we're dealing with a string rather than a unicode string
@@ -272,8 +270,7 @@ def interactive_auth_handler(self, title, instructions, prompt_list):
                     # Sometimes the server may ask for the 2FA code more than once. This is to avoid asking the
                     # user again. If it is wrong, just run again autosubmit run because the issue could be in
                     # the password step.
-                    if twofactor_nonpush is None:
-                        twofactor_nonpush = input("Please type the 2FA/OTP/token code: ")
+                    twofactor_nonpush = input("Please type the 2FA/OTP/token code: ")
                     answers.append(twofactor_nonpush)
         return tuple(answers)
 

autosubmit/platforms/pjmplatform.py

@@ -263,30 +263,6 @@ def update_cmds(self):
         self.get_cmd = "scp"
         self.mkdir_cmd = "mkdir -p " + self.remote_log_dir
 
-    # TODO: Not used, but it's in Slurm. To confirm later if it must stay...
-    def hold_job(self, job):
-        try:
-            cmd = f"pjrls {job.id} ; sleep 2 ; pjhold -R ASHOLD {job.id}"
-            self.send_command(cmd)
-            job_status = self.check_job(job, submit_hold_check=True)
-            if job_status == Status.RUNNING:
-                self.send_command(f"{self.cancel_cmd} {job.id}")
-                return False
-            elif job_status == Status.FAILED:
-                return False
-            cmd = self.get_queue_status_cmd(job.id)
-            self.send_command(cmd)
-        except BaseException as e:
-            try:
-                self.send_command(f"{self.cancel_cmd} {job.id}")
-                raise AutosubmitError(
-                    f"Can't hold jobid:{job.id}, canceling job", 6000, str(e))
-            except BaseException as e:
-                raise AutosubmitError(
-                    f"Can't cancel the jobid: {job.id}", 6000, str(e))
-            except AutosubmitError as e:
-                raise
-
     def get_checkhost_cmd(self):
         return self._checkhost_cmd
 

test/unit/test_job.py

@@ -51,6 +51,7 @@
 from autosubmit.platforms.platform import Platform
 from autosubmit.platforms.psplatform import PsPlatform
 from autosubmit.platforms.slurmplatform import SlurmPlatform
+from test.conftest import AutosubmitConfigFactory
 
 """Tests for the Autosubmit ``Job`` class."""
 
@@ -2414,3 +2415,36 @@ def test_submit_ready_jobs(autosubmit_exp, mocker):
     assert error_message == ''
     assert len(valid_packages_to_submit) == 1
     assert any_job_submitted
+
+
+@pytest.mark.parametrize("create_jobs", [[1, 2]], indirect=True)
+@pytest.mark.parametrize(
+    'status,failed_file',
+    [
+        (Status.RUNNING, False),
+        (Status.QUEUING, False),
+        (Status.HELD, False),
+        (Status.FAILED, False),
+        (Status.FAILED, True),
+        (Status.UNKNOWN, False),
+        (Status.SUBMITTED, False)
+    ]
+)
+def test_update_status(create_jobs: list[Job], status: Status, failed_file,
+                            autosubmit_config: 'AutosubmitConfigFactory', local: 'LocalPlatform'):
+    as_conf = autosubmit_config('t000', experiment_data={
+        'PLATFORMS': {
+            local.name: {
+                'DISABLE_RECOVERY_THREADS': False
+            }
+        }
+    })
+    job = create_jobs[0]
+    job.id = 0
+    job.platform = local
+    job.platform_name = local.name
+    job.new_status = status
+
+    assert job.status != status
+    job.update_status(as_conf=as_conf, failed_file=failed_file)
+    assert job.status == status

test/unit/test_pjm.py

@@ -20,6 +20,7 @@
 
 import pytest
 
+from autosubmit.config.configcommon import AutosubmitConfig
 from autosubmit.config.yamlparser import YAMLParserFactory
 from autosubmit.job.job import Job
 from autosubmit.job.job_common import Status
@@ -187,3 +188,26 @@ def test_process_batch_ready_jobs_valid_packages_to_submit(mocker, pjm_platform,
             else:
                 assert job.wrapper_name is None
     assert failed_packages == []
+
+
+@pytest.mark.parametrize("create_jobs", [[3, 5]], indirect=True)
+@pytest.mark.parametrize('parse_queue_reason, result', [
+    ('ASHOLD', 3),
+    ('WAITING ASHOLD', 6),
+    ('(Invalid)', -1)
+], ids=['ASHOLD', 'WAITING ASHOLD', 'AssociationJobLimit'])
+def test_get_queue_status(mocker, create_jobs: list[Job], pjm_platform: 'PJMPlatform', as_conf: 'AutosubmitConfig',
+                          parse_queue_reason, result):
+    in_queue_jobs = create_jobs
+    jobs_id = []
+    mocker.patch('autosubmit.platforms.platform.Platform.add_job_to_log_recover', return_value = True)
+    for job in in_queue_jobs:
+        job.platform = pjm_platform
+        jobs_id.append(job.id)
+        if 'WAITING' in parse_queue_reason:
+            job.hold = True
+    mocker.patch('autosubmit.platforms.pjmplatform.PJMPlatform.send_command', return_value = True)
+    mocker.patch('autosubmit.platforms.pjmplatform.PJMPlatform.parse_queue_reason', return_value = parse_queue_reason)
+    pjm_platform.get_queue_status(in_queue_jobs, jobs_id, as_conf)
+    assert result == in_queue_jobs[0].new_status
+