WIP: this platforms.wrapper code is confusing...

Author: kinow

.github/workflows/ci.yaml

@@ -92,7 +92,7 @@ jobs:
           cache: 'pip' # caching pip dependencies
 
       - name: Install system dependencies
-        run: sudo apt-get install -y curl git graphviz rsync
+        run: sudo apt-get install -y curl git graphviz rsync xvfb
 
       - name: Install dependencies
         run: |
@@ -112,6 +112,9 @@ jobs:
       # CONTRIBUTING.md file for details how to set up your environment to run these.
       - name: Integration tests
         run: |
+          Xvfb :99 -screen 0 1024x768x24 -ac &
+          export DISPLAY=:99
+          
           pytest \
             --cov=autosubmit --cov-config=.coveragerc \
             --cov-report=xml:test/coverage.xml --cov-append \

.github/workflows/lint.yml

@@ -31,6 +31,7 @@ jobs:
 
   mypy-changes-in-file:
     runs-on: ubuntu-latest
+    continue-on-error: true
     name: mypy run
     steps:
       - uses: actions/checkout@v5

autosubmit/platforms/locplatform.py

@@ -26,7 +26,6 @@
 from autosubmit.log.log import Log, AutosubmitError
 from autosubmit.platforms.headers.local_header import LocalHeader
 from autosubmit.platforms.paramiko_platform import ParamikoPlatform
-from autosubmit.platforms.wrappers.wrapper_factory import LocalWrapperFactory
 
 if TYPE_CHECKING:
     from autosubmit.config.configcommon import AutosubmitConfig
@@ -67,8 +66,7 @@ def __init__(self, expid: str, name: str, config: dict, auth_password: Optional[
         self.job_status['RUNNING'] = ['0']
         self.job_status['QUEUING'] = []
         self.job_status['FAILED'] = []
-        self._allow_wrappers = True
-        self._wrapper = LocalWrapperFactory(self)
+        self._allow_wrappers = False
 
         self.update_cmds()
 

autosubmit/platforms/paramiko_platform.py

@@ -30,7 +30,7 @@
 from pathlib import Path
 from threading import Thread
 from time import sleep
-from typing import Optional, Union, TYPE_CHECKING
+from typing import Any, Optional, Union, TYPE_CHECKING
 
 import Xlib.support.connect as xlib_connect
 import paramiko
@@ -143,9 +143,7 @@ def reset(self):
             self.poller = select.kqueue()
         else:
             self.poller = select.poll()
-        display = os.getenv('DISPLAY')
-        if display is None:
-            display = "localhost:0"
+        display = os.getenv('DISPLAY', "localhost:0")
         try:
             self.local_x11_display = xlib_connect.get_display(display)
         except Exception as e:
@@ -244,6 +242,7 @@ def agent_auth(self, port: int) -> bool:
             return False
         return True
 
+    # 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
@@ -254,7 +253,7 @@ def interactive_auth_handler(self, title, instructions, prompt_list):
             # strip() used to get rid of any padding spaces sent by the server
             if "password" in prompt:
                 answers.append(self.pw)
-            elif "token" in prompt or "2fa" in prompt or "otp" in prompt:
+            elif "token" in prompt or "2fa" in prompt or "otp" in prompt or "code":
                 if self.two_factor_method == "push":
                     answers.append("")
                 elif self.two_factor_method == "token":
@@ -506,7 +505,7 @@ def get_file(self, filename, must_exist=True, relative_path='', ignore_log=False
                     Log.printlog(f"Log file couldn't be retrieved: {filename}", 5000)
                 return False
 
-    def delete_file(self, filename: str) -> None:
+    def delete_file(self, filename: str) -> bool:
         """
         Deletes a file from this platform
 
@@ -515,7 +514,6 @@ def delete_file(self, filename: str) -> None:
         :return: True if successful or file does not exist
         :rtype: bool
         """
-        # TODO: pytests when the slurm container is avaliable
         remote_file = Path(self.get_files_path()) / filename
         try:
             self._ftpChannel.remove(str(remote_file))
@@ -524,12 +522,15 @@ def delete_file(self, filename: str) -> None:
             return False
         except Exception as e:
             # Change to Path
-            Log.error(f'Could not remove file {str(remote_file)}, something went wrong with the platform', 6004, str(e))
+            Log.error(f'Could not remove file {str(remote_file)}, something went wrong with the platform',
+                      6004, str(e))
 
             if str(e).lower().find("garbage") != -1:
                 raise AutosubmitCritical(
-                    "Wrong User or invalid .ssh/config. Or invalid user in the definition of PLATFORMS in YAML or public key not set ",
+                    "Wrong User or invalid .ssh/config. Or invalid user in the definition of PLATFORMS in "
+                    "YAML or public key not set ",
                     7051, str(e))
+        return False
 
     def move_file(self, src, dest, must_exist=False):
         """
@@ -776,26 +777,25 @@ def _check_jobid_in_queue(self, ssh_output, job_list_cmd):
                 return False
         return True
 
-    def parse_joblist(self, job_list: list[list['Job']]) -> str:
-        """
-        Convert a list of job_list to job_list_cmd
+    def parse_joblist(self, job_list: list[list['Job', Any]]) -> str:
+        """Return a string containing a comma-separated list of job IDs.
 
-        :param job_list: list of jobs
-        :type job_list: list
-        :return: job status
-        :rtype: str
+        If a job in the provided list is missing its ID, this function will initialize
+        it to a string containing the digit zero,``"0"``.
+
+        :param job_list: A list of jobs.
+        :return: A comma-separated string containing the job IDs.
         """
-        job_list_cmd = ""
-        for job, job_prev_status in job_list:
+        job_list_cmd: list[str] = []
+        # TODO: second item in tuple, _, is a ``job_prev_status``? What for?
+        for job, _ in job_list:
             if job.id is None:
                 job_str = "0"
             else:
                 job_str = str(job.id)
-            job_list_cmd += job_str + ","
-        if job_list_cmd[-1] == ",":
-            job_list_cmd = job_list_cmd[:-1]
+            job_list_cmd.append(job_str)
 
-        return job_list_cmd
+        return ','.join(job_list_cmd)
 
     def check_Alljobs(self, job_list: list[list['Job']], as_conf, retries=5):
         """
@@ -979,12 +979,15 @@ def get_queue_status_cmd(self, job_name):
 
     def x11_handler(self, channel, xxx_todo_changeme):
         """Handler for incoming x11 connections.
+
         For each x11 incoming connection:
 
         - get a connection to the local display
         - maintain bidirectional map of remote x11 channel to local x11 channel
         - add the descriptors to the poller
         - queue the channel (use transport.accept())
+
+        Incoming connections come from the server when we open an actual GUI application.
         """
         (src_addr, src_port) = xxx_todo_changeme
         x11_chanfd = channel.fileno()
@@ -1038,7 +1041,7 @@ def exec_command(
     ) -> Union[tuple[paramiko.Channel, paramiko.Channel, paramiko.Channel], tuple[bool, bool, bool]]:
         """
         Execute a command on the SSH server.  A new `.Channel` is opened and
-        the requested command is execed.  The command's input and output
+        the requested command is executed.  The command's input and output
         streams are returned as Python ``file``-like objects representing
         stdin, stdout, and stderr.
 
@@ -1052,7 +1055,6 @@ def exec_command(
         :param timeout: set command's channel timeout. See `Channel.settimeout`.settimeout.
         :type timeout: int
         :return: the stdin, stdout, and stderr of the executing command
-
         :raises SSHException: if the server fails to execute the command
         """
         while retries > 0:
@@ -1061,7 +1063,8 @@ def exec_command(
                     self._init_local_x11_display()
                     chan = self.transport.open_session()
                     if not chan.request_x11(single_connection=False, handler=self.x11_handler):
-                        # FIXME: test this!
+                        # FIXME: Apparently, the SSH session closes before reaching here?
+                        #        If so, we can just remove this check?
                         raise AutosubmitCritical("Remote platform does not support X11!")
                 else:
                     chan = self.transport.open_session()
@@ -1309,7 +1312,7 @@ def get_ssh_output(self):
     def get_ssh_output_err(self):
         return self._ssh_output_err
 
-    def get_call(self, job_script: str, job: 'Job', export="none", timeout=-1) -> str:
+    def get_call(self, job_script: str, job: Optional['Job'], export="none", timeout=-1) -> str:
         """Gets execution command for given job.
 
         :param job_script: script to run
@@ -1318,7 +1321,7 @@ def get_call(self, job_script: str, job: 'Job', export="none", timeout=-1) -> st
         :param timeout:
         :return: command to execute script
         """
-        # If job is None, it is a wrapper. ( 0 clarity there, to be improved in a rework TODO )
+        # If job is None, it is a wrapper. (TODO: 0 clarity there, to be improved in a rework)
         if job:
             if job.executable != '':
                 executable = ''  # Alternative: use job.executable with substituted placeholders
@@ -1512,10 +1515,10 @@ def check_remote_log_dir(self):
         except BaseException as e:
             raise AutosubmitError(f"Couldn't send the file {self.remote_log_dir} to HPC {self.host}", 6004, str(e))
 
-    def check_absolute_file_exists(self, src):
+    def check_absolute_file_exists(self, src) -> bool:
         try:
-            if self._ftpChannel.stat(src):
-                return True
+            self._ftpChannel.stat(src)
+            return True
         except Exception as e:
             Log.debug(f'Failed to check absolute file {src} exists: {str(e)}')
         return False

autosubmit/platforms/wrappers/wrapper_builder.py

@@ -1,19 +1,17 @@
-#!/usr/bin/env python3
-
-# Copyright 2015-2020 Earth Sciences Department, BSC-CNS
-
+# Copyright 2015-2025 Earth Sciences Department, BSC-CNS
+#
 # This file is part of Autosubmit.
-
+#
 # Autosubmit is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
-
+#
 # Autosubmit is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
-
+#
 # You should have received a copy of the GNU General Public License
 # along with Autosubmit.  If not, see <http://www.gnu.org/licenses/>.
 
@@ -30,22 +28,26 @@ class WrapperDirector:
     """
     def __init__(self):
         self._builder = None
+
     def construct(self, builder):
         self._builder = builder
 
         header = self._builder.build_header()
         job_thread = self._builder.build_job_thread()
-        #if "bash" not in header[0:15]:
+        # if "bash" not in header[0:15]:
 
         main = self._builder.build_main()
-        #else:
+        # else:
         #    nodes,main = self._builder.build_main() #What to do with nodes?
         # change to WrapperScript object
         wrapper_script = header + job_thread + main
         wrapper_script = wrapper_script.replace("_NEWLINE_", '\\n')
 
         return wrapper_script
+
+
 class WrapperBuilder(object):
+
     def __init__(self, **kwargs):
         if "retrials" in list(kwargs.keys()):
             self.retrials = kwargs['retrials']
@@ -66,7 +68,7 @@ def __init__(self, **kwargs):
     def build_header(self):
         return textwrap.dedent(self.header_directive) + self.build_imports()
 
-    def build_imports(self):
+    def build_imports(self) -> str:
         pass  # pragma: no cover
 
     def build_job_thread(self):