Addon Manager: Fix pip on Snap and AppImage

Also fixes various issues with dependency updater

Author: chennes

src/Mod/AddonManager/addonmanager_dependency_installer.py

@@ -27,8 +27,6 @@
 import subprocess
 from typing import List
 
-from freecad.utils import get_python_exe
-
 import addonmanager_freecad_interface as fci
 from addonmanager_pyside_interface import QObject, Signal, is_interruption_requested
 
@@ -46,7 +44,7 @@ class DependencyInstaller(QObject):
     no_python_exe = Signal()
     no_pip = Signal(str)  # Attempted command
     failure = Signal(str, str)  # Short message, detailed message
-    finished = Signal()
+    finished = Signal(bool)  # True if everything completed normally, otherwise false
 
     def __init__(
         self,
@@ -65,17 +63,25 @@ def __init__(
         self.python_requires = python_requires
         self.python_optional = python_optional
         self.location = location
+        self.required_succeeded = False
+        self.finished_successfully = False
 
     def run(self):
         """Normally not called directly, but rather connected to the worker thread's started
         signal."""
-        if self._verify_pip():
+        try:
             if self.python_requires or self.python_optional:
-                if not is_interruption_requested():
-                    self._install_python_packages()
-        if not is_interruption_requested():
-            self._install_addons()
-        self.finished.emit()
+                if self._verify_pip():
+                    if not is_interruption_requested():
+                        self._install_python_packages()
+            else:
+                self.required_succeeded = True
+            if not is_interruption_requested():
+                self._install_addons()
+                self.finished_successfully = self.required_succeeded
+        except RuntimeError:
+            pass
+        self.finished.emit(self.finished_successfully)
 
     def _install_python_packages(self):
         """Install required and optional Python dependencies using pip."""
@@ -87,20 +93,20 @@ def _install_python_packages(self):
         if not os.path.exists(vendor_path):
             os.makedirs(vendor_path)
 
-        self._install_required(vendor_path)
+        self.required_succeeded = self._install_required(vendor_path)
         self._install_optional(vendor_path)
 
     def _verify_pip(self) -> bool:
         """Ensure that pip is working -- returns True if it is, or False if not. Also emits the
         no_pip signal if pip cannot execute."""
-        python_exe = self._get_python()
-        if not python_exe:
-            return False
         try:
             proc = self._run_pip(["--version"])
             fci.Console.PrintMessage(proc.stdout + "\n")
+            if proc.returncode != 0:
+                return False
         except subprocess.CalledProcessError:
-            self.no_pip.emit(f"{python_exe} -m pip --version")
+            call = utils.create_pip_call([])
+            self.no_pip.emit(" ".join(call))
             return False
         return True
 
@@ -115,7 +121,6 @@ def _install_required(self, vendor_path: str) -> bool:
                 proc = self._run_pip(
                     [
                         "install",
-                        "--disable-pip-version-check",
                         "--target",
                         vendor_path,
                         pymod,
@@ -144,7 +149,6 @@ def _install_optional(self, vendor_path: str):
                 proc = self._run_pip(
                     [
                         "install",
-                        "--disable-pip-version-check",
                         "--target",
                         vendor_path,
                         pymod,
@@ -160,22 +164,13 @@ def _install_optional(self, vendor_path: str):
                 )
 
     def _run_pip(self, args):
-        python_exe = self._get_python()
-        final_args = [python_exe, "-m", "pip"]
-        final_args.extend(args)
+        final_args = utils.create_pip_call(args)
         return self._subprocess_wrapper(final_args)
 
     @staticmethod
     def _subprocess_wrapper(args) -> subprocess.CompletedProcess:
         """Wrap subprocess call so test code can mock it."""
-        return utils.run_interruptable_subprocess(args)
-
-    def _get_python(self) -> str:
-        """Wrap Python access so test code can mock it."""
-        python_exe = get_python_exe()
-        if not python_exe:
-            self.no_python_exe.emit()
-        return python_exe
+        return utils.run_interruptable_subprocess(args, timeout_secs=120)
 
     def _install_addons(self):
         for addon in self.addons:

src/Mod/AddonManager/addonmanager_utilities.py

@@ -22,42 +22,94 @@
 # *                                                                         *
 # ***************************************************************************
 
-""" Utilities to work across different platforms, providers and python versions """
+"""Utilities to work across different platforms, providers and python versions"""
 
+# pylint: disable=deprecated-module, ungrouped-imports
+
+from datetime import datetime
+from typing import Optional, Any, List
 import os
 import platform
 import shutil
 import stat
 import subprocess
+import time
 import re
 import ctypes
-from typing import Optional, Any
 
 from urllib.parse import urlparse
 
 try:
     from PySide import QtCore, QtGui, QtWidgets
 except ImportError:
-    QtCore = None
-    QtWidgets = None
-    QtGui = None
+    try:
+        from PySide6 import QtCore, QtGui, QtWidgets
+    except ImportError:
+        from PySide2 import QtCore, QtGui, QtWidgets
 
 import addonmanager_freecad_interface as fci
 
+try:
+    from freecad.utils import get_python_exe
+except ImportError:
+
+    def get_python_exe():
+        """Use shutil.which to find python executable"""
+        return shutil.which("python")
+
+
 if fci.FreeCADGui:
 
     # If the GUI is up, we can use the NetworkManager to handle our downloads. If there is no event
     # loop running this is not possible, so fall back to requests (if available), or the native
     # Python urllib.request (if requests is not available).
     import NetworkManager  # Requires an event loop, so is only available with the GUI
+
+    requests = None
+    ssl = None
+    urllib = None
 else:
+    NetworkManager = None
     try:
         import requests
+
+        ssl = None
+        urllib = None
     except ImportError:
         requests = None
         import urllib.request
         import ssl
 
+if fci.FreeCADGui:
+    loadUi = fci.loadUi
+else:
+    has_loader = False
+    try:
+        from PySide6.QtUiTools import QUiLoader
+
+        has_loader = True
+    except ImportError:
+        try:
+            from PySide2.QtUiTools import QUiLoader
+
+            has_loader = True
+        except ImportError:
+
+            def loadUi(ui_file: str):
+                """If there are no available versions of QtUiTools, then raise an error if this
+                method is used."""
+                raise RuntimeError("Cannot use QUiLoader without PySide or FreeCAD")
+
+    if has_loader:
+
+        def loadUi(ui_file: str) -> QtWidgets.QWidget:
+            """Load a Qt UI from an on-disk file."""
+            q_ui_file = QtCore.QFile(ui_file)
+            q_ui_file.open(QtCore.QFile.OpenModeFlag.ReadOnly)
+            loader = QUiLoader()
+            return loader.load(ui_file)
+
+
 #  @package AddonManager_utilities
 #  \ingroup ADDONMANAGER
 #  \brief Utilities to work across different platforms, providers and python versions
@@ -97,10 +149,13 @@ def symlink(source, link_name):
 
 
 def rmdir(path: str) -> bool:
+    """Remove a directory or symlink, even if it is read-only."""
     try:
         if os.path.islink(path):
             os.unlink(path)  # Remove symlink
         else:
+            # NOTE: the onerror argument was deprecated in Python 3.12, replaced by onexc -- replace
+            # when earlier versions are no longer supported.
             shutil.rmtree(path, onerror=remove_readonly)
     except (WindowsError, PermissionError, OSError):
         return False
@@ -175,7 +230,7 @@ def get_zip_url(repo):
 
 
 def recognized_git_location(repo) -> bool:
-    """Returns whether this repo is based at a known git repo location: works with github, gitlab,
+    """Returns whether this repo is based at a known git repo location: works with GitHub, gitlab,
     framagit, and salsa.debian.org"""
 
     parsed_url = urlparse(repo.url)
@@ -357,7 +412,7 @@ def is_float(element: Any) -> bool:
 
 
 def get_pip_target_directory():
-    # Get the default location to install new pip packages
+    """Get the default location to install new pip packages"""
     major, minor, _ = platform.python_version_tuple()
     vendor_path = os.path.join(
         fci.DataPaths().mod_dir, "..", "AdditionalPythonPackages", f"py{major}{minor}"
@@ -379,7 +434,12 @@ def blocking_get(url: str, method=None) -> bytes:
     succeeded, or an empty string if it failed, or returned no data. The method argument is
     provided mainly for testing purposes."""
     p = b""
-    if fci.FreeCADGui and method is None or method == "networkmanager":
+    if (
+        fci.FreeCADGui
+        and method is None
+        or method == "networkmanager"
+        and NetworkManager is not None
+    ):
         NetworkManager.InitializeNetworkManager()
         p = NetworkManager.AM_NETWORK_MANAGER.blocking_get(url, 10000)  # 10 second timeout
         if p:
@@ -398,7 +458,7 @@ def blocking_get(url: str, method=None) -> bytes:
     return p
 
 
-def run_interruptable_subprocess(args) -> subprocess.CompletedProcess:
+def run_interruptable_subprocess(args, timeout_secs: int = 10) -> subprocess.CompletedProcess:
     """Wrap subprocess call so it can be interrupted gracefully."""
     creation_flags = 0
     if hasattr(subprocess, "CREATE_NO_WINDOW"):
@@ -418,22 +478,63 @@ def run_interruptable_subprocess(args) -> subprocess.CompletedProcess:
     stdout = ""
     stderr = ""
     return_code = None
+    start_time = time.time()
     while return_code is None:
         try:
-            stdout, stderr = p.communicate(timeout=10)
+            # one second timeout allows interrupting the run once per second
+            stdout, stderr = p.communicate(timeout=1)
             return_code = p.returncode
-        except subprocess.TimeoutExpired:
-            if QtCore.QThread.currentThread().isInterruptionRequested():
+        except subprocess.TimeoutExpired as timeout_exception:
+            if (
+                hasattr(QtCore, "QThread")
+                and QtCore.QThread.currentThread().isInterruptionRequested()
+            ):
+                p.kill()
+                raise ProcessInterrupted() from timeout_exception
+            if time.time() - start_time >= timeout_secs:  # The real timeout
                 p.kill()
-                raise ProcessInterrupted()
+                stdout, stderr = p.communicate()
+                return_code = -1
     if return_code is None or return_code != 0:
         raise subprocess.CalledProcessError(
             return_code if return_code is not None else -1, args, stdout, stderr
         )
     return subprocess.CompletedProcess(args, return_code, stdout, stderr)
 
 
+def process_date_string_to_python_datetime(date_string: str) -> datetime:
+    """For modern macros the expected date format is ISO 8601, YYYY-MM-DD. For older macros this
+    standard was not always used, and various orderings and separators were used. This function
+    tries to match the majority of those older macros. Commonly-used separators are periods,
+    slashes, and dashes."""
+
+    def raise_error(bad_string: str, root_cause: Exception = None):
+        raise ValueError(
+            f"Unrecognized date string '{bad_string}' (expected YYYY-MM-DD)"
+        ) from root_cause
+
+    split_result = re.split(r"[ ./-]+", date_string.strip())
+    if len(split_result) != 3:
+        raise_error(date_string)
+
+    try:
+        split_result = [int(x) for x in split_result]
+        # The earliest possible year an addon can be created or edited is 2001:
+        if split_result[0] > 2000:
+            return datetime(split_result[0], split_result[1], split_result[2])
+        if split_result[2] > 2000:
+            # Generally speaking it's not possible to distinguish between DD-MM and MM-DD, so try
+            # the first, and only if that fails try the second
+            if split_result[1] <= 12:
+                return datetime(split_result[2], split_result[1], split_result[0])
+            return datetime(split_result[2], split_result[0], split_result[1])
+        raise ValueError(f"Invalid year in date string '{date_string}'")
+    except ValueError as exception:
+        raise_error(date_string, exception)
+
+
 def get_main_am_window():
+    """Find the Addon Manager's main window in the Qt widget hierarchy."""
     windows = QtWidgets.QApplication.topLevelWidgets()
     for widget in windows:
         if widget.objectName() == "AddonManager_Main_Window":
@@ -449,3 +550,37 @@ def get_main_am_window():
             return widget.centralWidget()
     # Why is this code even getting called?
     return None
+
+
+def remove_target_option(args: List[str]) -> List[str]:
+    # The Snap pip automatically adds the --user option, which is not compatible with the
+    # --target option, so we have to remove --target and its argument, if present
+    try:
+        index = args.index("--target")
+        del args[index : index + 2]  # The --target option and its argument
+    except ValueError:
+        pass
+    return args
+
+
+def create_pip_call(args: List[str]) -> List[str]:
+    """Choose the correct mechanism for calling pip on each platform. It currently supports
+    either `python -m pip` (most environments) or `pip` (Snap packages). Returns a list
+    of arguments suitable for passing directly to subprocess.Popen and related functions."""
+    snap_package = os.getenv("SNAP_REVISION")
+    appimage = os.getenv("APPIMAGE")
+    if snap_package:
+        args = remove_target_option(args)
+        call_args = ["pip", "--disable-pip-version-check"]
+        call_args.extend(args)
+    elif appimage:
+        python_exe = fci.DataPaths.home_dir + "bin/python"
+        call_args = [python_exe, "-m", "pip", "--disable-pip-version-check"]
+        call_args.extend(args)
+    else:
+        python_exe = get_python_exe()
+        if not python_exe:
+            raise RuntimeError("Could not locate Python executable on this system")
+        call_args = [python_exe, "-m", "pip", "--disable-pip-version-check"]
+        call_args.extend(args)
+    return call_args