Fix/some rework on remote service & remote resource (#1182)

- turn remote service start into steps, thus delay remote driver
  construction on remote host
- fix certain remote cmd gen utils
- fix empty report check in certain exporters
- add cleanup for dangling symlink due to workspace imitation on remote
  host, rearrange some source in remote resource as well
- fix a bug in driver dep graph type check
- kill ssh proc at rmt svc stop failure

Author: zhenyu-ms

doc/newsfragments/3193_changed.adjust_remote_service.rst

@@ -0,0 +1 @@
+Improve error logging for :py:class:`~testplan.common.remote.remote_service.RemoteService`; fix incorrect imitated workspace on remote due to leftover symlink from previous run.

pyproject.toml

@@ -62,7 +62,8 @@
         "gherkin-official==4.1.3",
         "parse",
         "orjson; python_version>='3.8'",
-        "flask-orjson; python_version>='3.8'"
+        "flask-orjson; python_version>='3.8'",
+        "exceptiongroup"
     ]
     requires-python = ">=3.7"
 

testplan/common/remote/remote_driver.py

@@ -12,20 +12,38 @@ class RemoteDriver:
 
     def __init__(self, remote_service, driver_cls, **options):
 
-        self.__dict__["rpyc_connection"] = remote_service.rpyc_connection
-        self.__dict__["_driver_cls"] = getattr(
-            self.rpyc_connection.modules[driver_cls.__module__],
-            driver_cls.__name__,
-        )
+        self.__dict__["_remote_service"] = remote_service
+        self.__dict__["_driver_cls"] = driver_cls
+        self.__dict__["_options"] = options
+        self.__dict__["_handle"] = None
 
-        options["runpath"] = "/".join(
-            [remote_service._remote_resource_runpath, options["name"]]
+    def _init_handle(self):
+        rmt_driver_cls = getattr(
+            self._remote_service.rpyc_connection.modules[
+                self._driver_cls.__module__
+            ],
+            self._driver_cls.__name__,
+        )
+        self._options["runpath"] = "/".join(
+            [
+                self._remote_service._remote_resource_runpath,
+                self._options["name"],
+            ]
         )
+        self.__dict__["_handle"] = rmt_driver_cls(**self._options)
 
-        self.__dict__["_handle"] = self._driver_cls(**options)
+    def uid(self) -> str:
+        # driver uid needed in env constructing check
+        # whether remote or local, uid should be the same since it's cfg based
+        # XXX: would it be a bad idea to have it inited locally?
+        return self._driver_cls(**self._options).uid()
 
     def __getattr__(self, name):
+        if self._handle is None:
+            self._init_handle()
         return getattr(self._handle, name)
 
     def __setattr__(self, name, value):
+        if self._handle is None:
+            self._init_handle()
         setattr(self._handle, name, value)

testplan/common/remote/remote_resource.py

@@ -1,6 +1,7 @@
 import getpass
 import itertools
 import os
+import shlex
 import sys
 from typing import Callable, Iterator, List, Tuple, Union, Dict, Optional
 
@@ -70,6 +71,8 @@ def get_options(cls):
             ConfigOption("remote_runpath", default=None): str,
             ConfigOption("testplan_path", default=None): str,
             ConfigOption("remote_workspace", default=None): str,
+            # proposing cfg clobber_remote_workspace,
+            # to overwrite what's in remote_workspace when set to True
             ConfigOption("clean_remote", default=False): bool,
             ConfigOption("push", default=[]): Or(list, None),
             ConfigOption("push_exclude", default=[]): Or(list, None),
@@ -152,33 +155,43 @@ def __init__(
         status_wait_timeout: int = 60,
         **options,
     ) -> None:
-
         if not worker_is_remote(remote_host):
+            # TODO: allow connecting to local for testing purpose?
             raise RuntimeError(
                 "Cannot create remote resource on the same host that Testplan runs."
             )
         options.update(self.filter_locals(locals()))
         super(RemoteResource, self).__init__(**options)
 
+        self.ssh_cfg = {
+            "host": self.cfg.remote_host,
+            "port": self.cfg.ssh_port,
+        }
+        # if 0 != self._execute_cmd_remote("uname"):
+        #     raise NotImplementedError(
+        #         "RemoteResource not supported on Windows remote hosts."
+        #     )
+
         self._remote_plan_runpath = None
         self._remote_resource_runpath = None
         self._child_paths = _LocationPaths()
         self._testplan_import_path = _LocationPaths()
         self._workspace_paths = _LocationPaths()
         self._working_dirs = _LocationPaths()
 
-        self.ssh_cfg = {
-            "host": self.cfg.remote_host,
-            "port": self.cfg.ssh_port,
-        }
-
         self.setup_metadata = WorkerSetupMetadata()
         self._user = getpass.getuser()
+        # TODO: allow specifying remote env
         self.python_binary = (
             os.environ["PYTHON3_REMOTE_BINARY"] if IS_WIN else sys.executable
         )
         self._error_exec = []
 
+        # remote file system obj outside runpath that needs to be cleaned upon
+        # exit when clean_remote is True, otherwise it will break workspace
+        # detect etc. in next run
+        self._dangling_remote_fs_obj = None
+
     @property
     def error_exec(self) -> list:
         return self._error_exec
@@ -252,8 +265,6 @@ def _remote_working_dir(self) -> None:
     def _create_remote_dirs(self) -> None:
         """Create mandatory directories in remote host."""
 
-        exist_on_remote = self._check_workspace()
-
         if 0 != self._execute_cmd_remote(
             cmd=filepath_exist_cmd(self._remote_runid_file),
             label="runid file availability check",
@@ -275,7 +286,18 @@ def _create_remote_dirs(self) -> None:
                 label="create remote runid file",
             )
 
+            # TODO: remote venv setup
+            # TODO: testplan_lib will resolved to site-packages under venv,
+            # TODO: while rpyc_classic.py under bin isn't included
+
+            exist_on_remote = self._check_workspace()
             self._prepare_workspace(exist_on_remote)
+
+            # NOTE: if workspace under testplan_lib (testplan installed in
+            # NOTE: editable mode), actual testplan package will never be
+            # NOTE: transferred
+            # NOTE: nevertheless, this impl won't work with venv in the first
+            # NOTE: place
             self._copy_testplan_package()
 
         self._execute_cmd_remote(
@@ -382,6 +404,7 @@ def _prepare_workspace(self, exist_on_remote: bool) -> None:
                 self,
                 self.ssh_cfg["host"],
             )
+            # proposed: if clobber_remote_workspace set, overwrite remote
             self._execute_cmd_remote(
                 cmd=link_cmd(
                     path=self._workspace_paths.local,
@@ -400,31 +423,68 @@ def _prepare_workspace(self, exist_on_remote: bool) -> None:
             )
             self._transfer_data(
                 # join with "" to add trailing "/" to source
-                # this will copy everything under local import path to to testplan_lib
+                # this will copy everything under local workspace path to fetched_workspace
                 source=os.path.join(self._workspace_paths.local, ""),
                 target=self._workspace_paths.remote,
                 remote_target=True,
                 exclude=self.cfg.workspace_exclude,
             )
+
+            if IS_WIN:
+                return
+
             self.logger.info(
                 "%s: creating symlink to imitate local workspace path on %s, "
                 "pointing to runpath/fetched_workspace",
                 self,
                 self.ssh_cfg["host"],
             )
-            self._execute_cmd_remote(
+            rmt_non_existing = None
+            # TODO: uncomment later
+            # TODO: there is another issue related to created dir cleanup
+            # TODO: if push given, pushed files under created dir and delete_pushed
+            # TODO: set to False, what to do?
+            # r, w = os.pipe()
+            # if 0 == self._execute_cmd_remote(
+            #     cmd="/bin/bash -c "
+            #     + shlex.quote(
+            #         'e=""; for i in '
+            #         + " ".join(
+            #             map(
+            #                 shlex.quote,
+            #                 self._workspace_paths.local.split(os.sep)[1:],
+            #             )
+            #         )
+            #         + '; do e+="/${i}"; if [ ! -e "$e" ]; then echo "$e"; break; fi; done'
+            #     ),
+            #     label="imitate local workspace path on remote - detect non-existing",
+            #     stdout=os.fdopen(w),
+            #     check=False,  # XXX: bash might not be there
+            # ):
+            #     rmt_non_existing = os.fdopen(r).read().strip() or None
+            if 0 == self._execute_cmd_remote(
                 cmd=mkdir_cmd(os.path.dirname(self._workspace_paths.local)),
                 label="imitate local workspace path on remote - mkdir",
                 check=False,  # just best effort
-            )
-            self._execute_cmd_remote(
+            ) and 0 == self._execute_cmd_remote(
                 cmd=link_cmd(
                     path=self._workspace_paths.remote,
                     link=self._workspace_paths.local,
                 ),
                 label="imitate local workspace path on remote - ln",
                 check=False,  # just best effort
-            )
+            ):
+                # NOTE: we shall always remove created symlink
+                self._dangling_remote_fs_obj = (
+                    rmt_non_existing or self._workspace_paths.local
+                )
+                self.logger.info(
+                    "%s: on %s, %s and its possible descendants are "
+                    "created to imitate local workspace path",
+                    self,
+                    self.ssh_cfg["host"],
+                    self._dangling_remote_fs_obj,
+                )
 
     def _push_files(self) -> None:
         """Push files and directories to remote host."""
@@ -601,10 +661,21 @@ def _clean_remote(self) -> None:
             )
 
             self._execute_cmd_remote(
-                cmd=f"/bin/rm -rf {self._remote_plan_runpath}",
+                cmd=rm_cmd(self._remote_plan_runpath),
                 label="Clean remote root runpath",
             )
 
+            if self._dangling_remote_fs_obj:
+                self._execute_cmd_remote(
+                    cmd=rm_cmd(self._dangling_remote_fs_obj),
+                    label=f"Remove imitated workspace outside runpath",
+                )
+                self._dangling_remote_fs_obj = None
+
+        if self.cfg.delete_pushed:
+            # TODO
+            ...
+
     def _pull_files(self) -> None:
         """Pull custom files from remote host."""
 

testplan/common/remote/remote_service.py

@@ -240,15 +240,14 @@ def stopping(self) -> None:
         """
         Stops remote rpyc process.
         """
-        remote_pid = self.rpyc_connection.modules.os.getpid()
         try:
-            self.rpyc_connection.modules.os.kill(remote_pid, signal.SIGTERM)
+            self.rpyc_connection.modules.os.kill(self.rpyc_pid, signal.SIGTERM)
         except EOFError:
             pass
-
-        # actually if remote rpyc server is shutdown, ssh proc is also finished
-        # but calling kill_process just in case
-        if self.proc:
-            kill_process(self.proc, self.cfg.stop_timeout)
+        finally:
+            # if remote rpyc server is shutdown successfully, ssh proc is also finished
+            # otherwise we need to manual kill this orphaned ssh procc
+            if self.proc:
+                kill_process(self.proc, self.cfg.stop_timeout)
 
         self.status.change(self.STATUS.STOPPED)

testplan/common/utils/path.py

@@ -254,7 +254,7 @@ def rebase_path(path, old_base, new_base):
     """
 
     rel_path = os.path.relpath(path, old_base).split(os.sep)
-    return "/".join([new_base] + rel_path)
+    return os.path.join(new_base, *rel_path)
 
 
 def is_subdir(child, parent):

testplan/common/utils/remote.py

@@ -1,11 +1,12 @@
 """Remote execution utilities."""
 
+import getpass
 import os
 import platform
+import shlex
 import socket
-import sys
-import getpass
 import subprocess
+import sys
 
 IS_WIN = platform.system() == "Windows"
 USER = getpass.getuser()
@@ -119,19 +120,19 @@ def copy_cmd(source, target, exclude=None, port=None, deref_links=False):
 
 def link_cmd(path, link):
     """Returns link creation command."""
-    return " ".join(["ln", "-sfn", path, link])
+    return " ".join(["/bin/ln", "-sfn", shlex.quote(path), shlex.quote(link)])
 
 
 def mkdir_cmd(path):
     """Return mkdir command"""
-    return " ".join(["/bin/mkdir", "-p", path])
+    return " ".join(["/bin/mkdir", "-p", shlex.quote(path)])
 
 
 def rm_cmd(path):
     """Return rm command"""
-    return " ".join(["/bin/rm", "-rf", path])
+    return " ".join(["/bin/rm", "-rf", shlex.quote(path)])
 
 
 def filepath_exist_cmd(path):
     """Checks if filepath exists."""
-    return " ".join(["test", "-e", path])
+    return " ".join(["/bin/test", "-e", shlex.quote(path)])

testplan/exporters/testing/json/base.py

@@ -113,7 +113,7 @@ def export(
         result = None
         json_path = pathlib.Path(self.cfg.json_path).resolve()
 
-        if len(source):
+        if not source.is_empty():
             json_path.parent.mkdir(parents=True, exist_ok=True)
 
             test_plan_schema = TestReportSchema()

testplan/exporters/testing/pdf/__init__.py

@@ -250,7 +250,7 @@ def export(
             exporter=self, export_context=export_context
         )
         result = None
-        if len(source):
+        if not source.is_empty():
             pdf_path = self.create_pdf(source)
             self.logger.user_info("PDF generated at %s", pdf_path)
 

testplan/exporters/testing/webserver/base.py

@@ -87,7 +87,7 @@ def export(
             exporter=self, export_context=export_context
         )
         result = None
-        if len(source):
+        if not source.is_empty():
             exporter = JSONExporter(
                 json_path=self.cfg.json_path, split_json_report=False
             )