Addon Manager: Fix pip on Snap and AppImage

Also fixes various issues with dependency updater
This commit is contained in:
Chris Hennes
2025-02-27 18:12:26 -03:00
committed by Adrián Insaurralde Avalos
parent f1e86b9304
commit 2462f72ebf
6 changed files with 212 additions and 85 deletions
@@ -136,12 +136,6 @@ class TestDependencyInstaller(unittest.TestCase):
self.assertTrue(ff_required.called)
self.assertTrue(ff_optional.called)
def test_verify_pip_no_python(self):
self.test_object._get_python = lambda: None
should_continue = self.test_object._verify_pip()
self.assertFalse(should_continue)
self.assertEqual(len(self.signals_caught), 0)
def test_verify_pip_no_pip(self):
sm = SubprocessMock()
sm.succeed = False
@@ -27,8 +27,6 @@ import os
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 @@ class DependencyInstaller(QObject):
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 @@ class DependencyInstaller(QObject):
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 @@ class DependencyInstaller(QObject):
proc = self._run_pip(
[
"install",
"--disable-pip-version-check",
"--target",
vendor_path,
pymod,
@@ -144,7 +149,6 @@ class DependencyInstaller(QObject):
proc = self._run_pip(
[
"install",
"--disable-pip-version-check",
"--target",
vendor_path,
pymod,
@@ -160,22 +164,13 @@ class DependencyInstaller(QObject):
)
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:
@@ -46,9 +46,13 @@ try:
getUserMacroDir = FreeCAD.getUserMacroDir
getUserCachePath = FreeCAD.getUserCachePath
translate = FreeCAD.Qt.translate
loadUi = None
if FreeCAD.GuiUp:
import FreeCADGui
if hasattr(FreeCADGui, "PySideUic"):
loadUi = FreeCADGui.PySideUic.loadUi
else:
FreeCADGui = None
@@ -77,14 +77,26 @@ class AddonInstallerGUI(QtCore.QObject):
self.installer.failure.connect(self._installation_failed)
def __del__(self):
if self.worker_thread and hasattr(self.worker_thread, "quit"):
self.worker_thread.quit()
self.worker_thread.wait(500)
if self.worker_thread.isRunning():
self._stop_thread(self.worker_thread)
self._stop_thread(self.dependency_worker_thread)
@staticmethod
def _stop_thread(thread: QtCore.QThread):
if thread and hasattr(thread, "quit"):
if thread.isRunning():
FreeCAD.Console.PrintMessage(
"INTERNAL ERROR: a QThread is still running when it should have finished"
)
thread.requestInterruption()
thread.wait(100)
thread.quit()
thread.wait(500)
if thread.isRunning():
FreeCAD.Console.PrintError(
"INTERNAL ERROR: Thread did not quit() cleanly, using terminate()\n"
)
self.worker_thread.terminate()
thread.terminate()
def run(self):
"""Instructs this class to begin displaying the necessary dialogs to guide a user through
@@ -300,13 +312,11 @@ class AddonInstallerGUI(QtCore.QObject):
self.dependency_installer.no_python_exe.connect(self._report_no_python_exe)
self.dependency_installer.no_pip.connect(self._report_no_pip)
self.dependency_installer.failure.connect(self._report_dependency_failure)
self.dependency_installer.finished.connect(self._cleanup_dependency_worker)
self.dependency_installer.finished.connect(self._report_dependency_success)
self.dependency_installer.finished.connect(self._dependencies_finished)
self.dependency_worker_thread = QtCore.QThread(self)
self.dependency_installer.moveToThread(self.dependency_worker_thread)
self.dependency_worker_thread.started.connect(self.dependency_installer.run)
self.dependency_installer.finished.connect(self.dependency_worker_thread.quit)
self.dependency_installation_dialog = QtWidgets.QMessageBox(
QtWidgets.QMessageBox.Information,
@@ -319,16 +329,6 @@ class AddonInstallerGUI(QtCore.QObject):
self.dependency_installation_dialog.show()
self.dependency_worker_thread.start()
def _cleanup_dependency_worker(self) -> None:
return
self.dependency_worker_thread.quit()
self.dependency_worker_thread.wait(500)
if self.dependency_worker_thread.isRunning():
FreeCAD.Console.PrintError(
"INTERNAL ERROR: Thread did not quit() cleanly, using terminate()\n"
)
self.dependency_worker_thread.terminate()
def _report_no_python_exe(self) -> None:
"""Callback for the dependency installer failing to locate a Python executable."""
if self.dependency_installation_dialog is not None:
@@ -409,6 +409,11 @@ class AddonInstallerGUI(QtCore.QObject):
self.dependency_installation_dialog.hide()
self.install()
def _dependencies_finished(self, success: bool):
if success:
self._report_dependency_success()
self.dependency_worker_thread.quit()
def _dependency_dialog_ignore_clicked(self) -> None:
"""Callback for when dependencies are ignored."""
self.install()
+148 -13
View File
@@ -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,14 +478,23 @@ 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()
raise ProcessInterrupted() from timeout_exception
if time.time() - start_time >= timeout_secs: # The real timeout
p.kill()
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
@@ -433,7 +502,39 @@ def run_interruptable_subprocess(args) -> subprocess.CompletedProcess:
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
@@ -33,6 +33,7 @@ import subprocess
import sys
from functools import partial
from typing import Dict, List, Tuple
from addonmanager_utilities import create_pip_call
import addonmanager_freecad_interface as fci
@@ -92,28 +93,21 @@ def call_pip(args) -> List[str]:
"""Tries to locate the appropriate Python executable and run pip with version checking
disabled. Fails if Python can't be found or if pip is not installed."""
python_exe = get_python_exe()
pip_failed = False
if python_exe:
call_args = [python_exe, "-m", "pip", "--disable-pip-version-check"]
call_args.extend(args)
proc = None
try:
proc = utils.run_interruptable_subprocess(call_args)
except subprocess.CalledProcessError:
pip_failed = True
try:
call_args = create_pip_call(args)
except RuntimeError as exception:
raise PipFailed() from exception
result = []
if not pip_failed:
data = proc.stdout
result = data.split("\n")
elif proc:
raise PipFailed(proc.stderr)
else:
raise PipFailed("pip timed out")
else:
raise PipFailed("Could not locate Python executable on this system")
return result
try:
proc = utils.run_interruptable_subprocess(call_args)
except subprocess.CalledProcessError as exception:
raise PipFailed("pip timed out") from exception
if proc.returncode != 0:
raise PipFailed(proc.stderr)
data = proc.stdout
return data.split("\n")
class PythonPackageManager: