diff --git a/AddonManagerTest/app/test_python_deps.py b/AddonManagerTest/app/test_python_deps.py
index 391ad4f..0aa197e 100644
--- a/AddonManagerTest/app/test_python_deps.py
+++ b/AddonManagerTest/app/test_python_deps.py
@@ -21,22 +21,27 @@
import os
import subprocess
+import tempfile
import unittest
from unittest.mock import MagicMock, patch
from AddonManagerTest.app.mocks import SignalCatcher
+from addonmanager_utilities import ProcessInterrupted
from addonmanager_python_deps import (
+ AsynchronousPipWorker,
PackageInfo,
+ PipCommand,
PythonPackageListModel,
parse_pip_list_output,
call_pip,
PipFailed,
+ PipInterrupted,
)
class TestPythonDepsStandaloneFunctions(unittest.TestCase):
- @patch("addonmanager_python_deps.run_interruptable_subprocess")
+ @patch("addonmanager_python_deps.run_monitored_subprocess")
def test_call_pip(self, mock_run_subprocess: MagicMock):
mock_run_subprocess.return_value = MagicMock()
mock_run_subprocess.return_value.returncode = 0
@@ -51,7 +56,7 @@ def test_call_pip_no_python(self, mock_get_python_exe: MagicMock):
with self.assertRaises(PipFailed):
call_pip(["arg1", "arg2", "arg3"])
- @patch("addonmanager_python_deps.run_interruptable_subprocess")
+ @patch("addonmanager_python_deps.run_monitored_subprocess")
def test_call_pip_exception_raised(self, mock_run_subprocess: MagicMock):
mock_run_subprocess.side_effect = subprocess.CalledProcessError(
-1, "dummy_command", "Fake contents of stdout", "Fake contents of stderr"
@@ -59,7 +64,24 @@ def test_call_pip_exception_raised(self, mock_run_subprocess: MagicMock):
with self.assertRaises(PipFailed):
call_pip(["arg1", "arg2", "arg3"])
- @patch("addonmanager_python_deps.run_interruptable_subprocess")
+ @patch("addonmanager_python_deps.run_monitored_subprocess")
+ def test_call_pip_interrupted(self, mock_run_subprocess: MagicMock):
+ """An interrupted pip call is reported as a cancellation, not a generic failure."""
+ mock_run_subprocess.side_effect = ProcessInterrupted()
+ with self.assertRaises(PipInterrupted):
+ call_pip(["arg1", "arg2", "arg3"])
+
+ @patch("addonmanager_python_deps.run_monitored_subprocess")
+ def test_call_pip_passes_line_callback(self, mock_run_subprocess: MagicMock):
+ """The caller's line callback is handed to the subprocess runner so that pip output can
+ be displayed as it is produced."""
+ mock_run_subprocess.return_value = MagicMock()
+ mock_run_subprocess.return_value.stdout = ""
+ callback = MagicMock()
+ call_pip(["list"], line_callback=callback)
+ self.assertIs(callback, mock_run_subprocess.call_args[1]["line_callback"])
+
+ @patch("addonmanager_python_deps.run_monitored_subprocess")
def test_call_pip_splits_results(self, mock_run_subprocess: MagicMock):
result_mock = MagicMock()
result_mock.stdout = "\n".join(["Value 1", "Value 2", "Value 3"])
@@ -85,6 +107,22 @@ def test_parse_pip_list_output_all_packages_no_updates(self):
self.assertEqual("41.2.0", results_list[1].installed_version)
self.assertEqual("", results_list[1].available_version)
+ def test_parse_pip_list_output_ignores_pip_log_lines(self):
+ """Because pip's error output is merged into its standard output, log lines can appear
+ alongside the package table and must not be mistaken for packages."""
+ results_list = parse_pip_list_output(
+ [
+ "WARNING: Ignoring invalid distribution ~umpy",
+ "Package Version",
+ "---------- -------",
+ "gitdb 4.0.9",
+ "ERROR: something went wrong",
+ "setuptools 41.2.0",
+ ],
+ {},
+ )
+ self.assertEqual(["gitdb", "setuptools"], [package.name for package in results_list])
+
def test_parse_pip_list_output_update_available_when_constrained_version_differs(self):
"""An update is available when the constrained version differs from what is installed;
a package without a constraint, or already at its constrained version, shows no update."""
@@ -203,205 +241,388 @@ def test_determine_new_python_dependencies_single_addon_given(self):
python_deps,
)
- class TestUpdateMultiplePackages(unittest.TestCase):
- @patch("addonmanager_python_deps.call_pip")
- @patch("addonmanager_python_deps.fci.Console.PrintLog")
- @patch("addonmanager_python_deps.fci.Console.PrintError")
- def test_update_all_packages(self, mock_print_error, mock_print_log, mock_call_pip):
- model = PythonPackageListModel([])
- model.vendor_path = "/vendor/path"
- model.package_list = [
- PackageInfo("pkg1", "1", "2", []),
- PackageInfo("pkg2", "1", "2", []),
- ]
-
- model.update_all_packages()
-
- mock_call_pip.assert_called_once_with(
- ["install", "--upgrade", "--target", "/vendor/path", "pkg1", "pkg2"]
- )
- mock_print_log.assert_called_once()
- mock_print_error.assert_not_called()
-
- @patch("addonmanager_python_deps.call_pip", side_effect=PipFailed("upgrade failed"))
- @patch("addonmanager_python_deps.fci.Console.PrintLog")
- @patch("addonmanager_python_deps.fci.Console.PrintError")
- def test_update_packages_pip_failure(self, mock_print_error, mock_print_log, mock_call_pip):
- model = PythonPackageListModel([])
- model.vendor_path = "/vendor/path"
- model.package_list = [PackageInfo("pkg1", "1", "2", [])]
- model.update_all_packages()
+@patch("addonmanager_python_deps.get_pip_target_directory", return_value="/vendor/path")
+@patch("addonmanager_python_deps.get_constraints")
+@patch("addonmanager_python_deps.using_system_pip_installation_location", return_value=True)
+class TestUpdateMultiplePackages(unittest.TestCase):
+ """Tests of the pip call used to install and update packages. The system installation
+ location is simulated, so no backup of the package directory is involved."""
- mock_call_pip.assert_called_once()
- mock_print_error.assert_called_once_with("upgrade failed\n")
+ @patch("addonmanager_python_deps.call_pip", return_value=[])
+ @patch("addonmanager_python_deps.fci.Console.PrintLog")
+ @patch("addonmanager_python_deps.fci.Console.PrintError")
+ def test_update_all_packages(self, mock_print_error, mock_print_log, mock_call_pip, *_):
+ model = PythonPackageListModel([])
+ model.vendor_path = "/vendor/path"
+ model.package_list = [
+ PackageInfo("pkg1", "1", "2", []),
+ PackageInfo("pkg2", "1", "2", []),
+ ]
- class TestCleanupOldPackageVersions(unittest.TestCase):
- """Tests for the _cleanup_old_package_versions method"""
+ model.update_all_packages()
- @patch("addonmanager_python_deps.os.path.exists")
- @patch("addonmanager_python_deps.os.listdir")
- @patch("addonmanager_python_deps.shutil.rmtree")
- @patch("addonmanager_python_deps.fci.Console.PrintLog")
- def test_cleanup_removes_old_versions_keeps_newest(
- self, mock_print_log, mock_rmtree, mock_listdir, mock_exists
- ):
- """Test that old package versions are removed and newest is kept"""
- mock_exists.return_value = True
- mock_listdir.return_value = [
- "requests-2.28.0.dist-info",
- "requests-2.31.0.dist-info",
- "numpy-1.24.0.dist-info",
- "numpy-1.26.0.dist-info",
- "numpy-1.25.2.dist-info",
- "other_file.txt",
- ]
+ self.assertEqual(
+ [
+ "install",
+ "--progress-bar",
+ "off",
+ "--upgrade",
+ "--target",
+ "/vendor/path",
+ "pkg1",
+ "pkg2",
+ ],
+ mock_call_pip.call_args_list[0][0][0],
+ )
+ mock_print_log.assert_called_once()
+ mock_print_error.assert_not_called()
- model = PythonPackageListModel([])
- model.vendor_path = "/fake/path"
- model._cleanup_old_package_versions()
-
- # Should remove old versions but keep newest
- self.assertEqual(mock_rmtree.call_count, 3)
- removed_paths = [call[0][0] for call in mock_rmtree.call_args_list]
-
- # Check old versions were removed (works on all platforms)
- self.assertIn(os.path.join("/fake/path", "requests-2.28.0.dist-info"), removed_paths)
- self.assertIn(os.path.join("/fake/path", "numpy-1.24.0.dist-info"), removed_paths)
- self.assertIn(os.path.join("/fake/path", "numpy-1.25.2.dist-info"), removed_paths)
-
- @patch("addonmanager_python_deps.os.path.exists")
- @patch("addonmanager_python_deps.os.listdir")
- @patch("addonmanager_python_deps.shutil.rmtree")
- def test_cleanup_single_version_no_removal(self, mock_rmtree, mock_listdir, mock_exists):
- """Test that packages with only one version are not touched"""
- mock_exists.return_value = True
- mock_listdir.return_value = [
- "requests-2.31.0.dist-info",
- "numpy-1.26.0.dist-info",
- "pandas-2.1.0.dist-info",
- ]
+ @patch("addonmanager_python_deps.call_pip", side_effect=PipFailed("upgrade failed"))
+ @patch("addonmanager_python_deps.fci.Console.PrintLog")
+ @patch("addonmanager_python_deps.fci.Console.PrintError")
+ def test_update_packages_pip_failure(self, mock_print_error, mock_print_log, mock_call_pip, *_):
+ model = PythonPackageListModel([])
+ model.vendor_path = "/vendor/path"
+ model.package_list = [PackageInfo("pkg1", "1", "2", [])]
- model = PythonPackageListModel([])
- model.vendor_path = "/fake/path"
- model._cleanup_old_package_versions()
+ model.update_all_packages()
+
+ mock_call_pip.assert_called()
+ mock_print_error.assert_called_once_with("upgrade failed\n")
- # No removals should happen when only one version exists per package
- mock_rmtree.assert_not_called()
- @patch("addonmanager_python_deps.os.path.exists")
- def test_cleanup_nonexistent_directory(self, mock_exists):
- """Test graceful handling when vendor path doesn't exist"""
- mock_exists.return_value = False
+class TestAsynchronousPipWorker(unittest.TestCase):
+ """Tests of the worker that runs pip off the GUI thread."""
- model = PythonPackageListModel([])
- model.vendor_path = "/nonexistent/path"
- model._cleanup_old_package_versions()
+ @patch("addonmanager_python_deps.fci.Console.PrintError")
+ @patch("addonmanager_python_deps.call_pip", side_effect=RuntimeError("something broke"))
+ def test_finished_is_emitted_after_an_unexpected_error(self, _mock_call_pip, _mock_print_error):
+ """Whatever goes wrong, the caller is told the run is over, so that it can restore the
+ package directory."""
+ worker = AsynchronousPipWorker(PipCommand.Upgrade, ["pkg1"])
+ catcher = SignalCatcher()
+ worker.finished.connect(catcher.catch_signal)
- @patch("addonmanager_python_deps.os.path.exists")
- @patch("addonmanager_python_deps.os.listdir")
- @patch("addonmanager_python_deps.shutil.rmtree")
- def test_cleanup_empty_directory(self, mock_rmtree, mock_listdir, mock_exists):
- """Test handling of empty vendor directory"""
- mock_exists.return_value = True
- mock_listdir.return_value = []
+ worker.run()
- model = PythonPackageListModel([])
- model.vendor_path = "/fake/path"
- model._cleanup_old_package_versions()
- mock_rmtree.assert_not_called()
-
- @patch("addonmanager_python_deps.os.path.exists")
- @patch("addonmanager_python_deps.os.listdir")
- @patch("addonmanager_python_deps.os.path.isdir")
- @patch("addonmanager_python_deps.shutil.rmtree")
- @patch("addonmanager_python_deps.fci.Console.PrintWarning")
- def test_cleanup_handles_permission_error(
- self, mock_print_warning, mock_rmtree, mock_isdir, mock_listdir, mock_exists
- ):
- """Test that permission errors are handled gracefully"""
- mock_exists.return_value = True
- mock_listdir.return_value = [
- "requests-2.28.0.dist-info",
- "requests-2.31.0.dist-info",
- ]
- mock_isdir.return_value = True
- mock_rmtree.side_effect = PermissionError("Permission denied")
+ self.assertTrue(catcher.caught)
+ self.assertIn("something broke", worker.error)
+ self.assertFalse(worker.is_running)
- model = PythonPackageListModel([])
- model.vendor_path = "/fake/path"
- model._cleanup_old_package_versions()
- mock_print_warning.assert_called()
-
- @patch("addonmanager_python_deps.os.path.exists")
- @patch("addonmanager_python_deps.os.listdir")
- @patch("addonmanager_python_deps.os.path.isdir")
- @patch("addonmanager_python_deps.shutil.rmtree")
- def test_cleanup_normalizes_package_names(
- self, mock_rmtree, mock_isdir, mock_listdir, mock_exists
- ):
- """Test that package names are normalized per PEP 503 (underscores to dashes)"""
- mock_exists.return_value = True
- mock_listdir.return_value = [
- "my_package-1.0.0.dist-info",
- "my-package-2.0.0.dist-info",
- ]
- mock_isdir.return_value = True
+ @patch("addonmanager_python_deps.fci.Console.PrintMessage")
+ @patch("addonmanager_python_deps.call_pip", side_effect=PipInterrupted("cancelled"))
+ def test_interrupted_installation_is_recorded_and_listing_skipped(
+ self, mock_call_pip, _mock_print_message
+ ):
+ worker = AsynchronousPipWorker(PipCommand.Install, ["pkg1"])
+ catcher = SignalCatcher()
+ worker.finished.connect(catcher.catch_signal)
- model = PythonPackageListModel([])
- model.vendor_path = "/fake/path"
- model._cleanup_old_package_versions()
-
- # Should remove old version (they're the same package after normalization)
- mock_rmtree.assert_called_once_with(
- os.path.join("/fake/path", "my_package-1.0.0.dist-info")
- )
-
- @patch("addonmanager_python_deps.os.path.exists")
- @patch("addonmanager_python_deps.os.listdir")
- @patch("addonmanager_python_deps.os.path.isdir")
- @patch("addonmanager_python_deps.shutil.rmtree")
- def test_cleanup_ignores_non_dist_info_directories(
- self, mock_rmtree, mock_isdir, mock_listdir, mock_exists
- ):
- """Test that only .dist-info directories are processed"""
- mock_exists.return_value = True
- mock_listdir.return_value = [
- "requests-2.28.0.dist-info",
- "requests-2.31.0.dist-info",
- "some_package",
- "__pycache__",
- "random_file.txt",
- ]
- mock_isdir.return_value = True
+ worker.run()
- model = PythonPackageListModel([])
- model.vendor_path = "/fake/path"
- model._cleanup_old_package_versions()
- mock_rmtree.assert_called_once_with(
- os.path.join("/fake/path", "requests-2.28.0.dist-info")
- )
-
- @patch("addonmanager_python_deps.os.path.exists")
- @patch("addonmanager_python_deps.os.listdir")
- @patch("addonmanager_python_deps.os.path.isdir")
- @patch("addonmanager_python_deps.shutil.rmtree")
- @patch("addonmanager_python_deps.fci.Console.PrintWarning")
- def test_cleanup_handles_invalid_version_format(
- self, mock_print_warning, mock_rmtree, mock_isdir, mock_listdir, mock_exists
- ):
- """Test handling of malformed version strings"""
- mock_exists.return_value = True
- mock_listdir.return_value = [
- "badpackage-invalid.version.dist-info",
- "goodpackage-1.0.0.dist-info",
- "goodpackage-2.0.0.dist-info",
- ]
- mock_isdir.return_value = True
+ self.assertTrue(worker.cancelled)
+ self.assertTrue(worker.error)
+ self.assertTrue(catcher.caught)
+ mock_call_pip.assert_called_once()
- model = PythonPackageListModel([])
- model.vendor_path = "/fake/path"
- model._cleanup_old_package_versions()
- mock_rmtree.assert_called_once_with(
- os.path.join("/fake/path", "goodpackage-1.0.0.dist-info")
- )
+ @patch("addonmanager_python_deps.get_constraints")
+ @patch("addonmanager_python_deps.fci.Console.PrintLog")
+ def test_pip_output_is_reported_as_progress(self, _mock_print_log, _mock_get_constraints):
+ def fake_call_pip(args, line_callback=None):
+ if line_callback is not None:
+ line_callback("Collecting numpy")
+ line_callback(" ")
+ return []
+
+ messages = []
+ worker = AsynchronousPipWorker(PipCommand.Install, ["numpy"])
+ worker.progress_message.connect(messages.append)
+
+ with patch("addonmanager_python_deps.call_pip", side_effect=fake_call_pip):
+ worker.run()
+
+ self.assertIn("Collecting numpy", messages)
+ self.assertNotIn(" ", messages)
+
+
+@patch("addonmanager_python_deps.using_system_pip_installation_location", return_value=False)
+class TestPackageDirectoryBackup(unittest.TestCase):
+ """Tests of the backup that protects the installed packages while pip runs."""
+
+ def setUp(self):
+ self.temp_directory = tempfile.TemporaryDirectory()
+ self.model = PythonPackageListModel([])
+ self.model.vendor_path = os.path.join(self.temp_directory.name, "py311")
+ self.backup_path = self.model.vendor_path + ".old"
+
+ def tearDown(self):
+ self.temp_directory.cleanup()
+
+ @staticmethod
+ def _create_directory_containing(path: str, filename: str) -> None:
+ os.makedirs(path, exist_ok=True)
+ with open(os.path.join(path, filename), "w", encoding="utf-8") as marker:
+ marker.write("marker")
+
+ def test_existing_directory_is_moved_aside(self, _mock_system_location):
+ self._create_directory_containing(self.model.vendor_path, "installed.txt")
+
+ self.assertTrue(self.model._set_aside_package_directory())
+
+ self.assertEqual(self.backup_path, self.model.backup_path)
+ self.assertTrue(os.path.exists(os.path.join(self.backup_path, "installed.txt")))
+ self.assertEqual([], os.listdir(self.model.vendor_path))
+
+ def test_missing_directory_is_created_without_a_backup(self, _mock_system_location):
+ self.assertTrue(self.model._set_aside_package_directory())
+
+ self.assertIsNone(self.model.backup_path)
+ self.assertTrue(os.path.isdir(self.model.vendor_path))
+
+ @patch("addonmanager_python_deps.fci.Console.PrintWarning")
+ def test_leftover_backup_is_recovered_when_packages_are_missing(
+ self, _mock_print_warning, _mock_system_location
+ ):
+ """A backup left behind by a run that never completed holds the only copy of the
+ packages, so it is put back rather than deleted."""
+ self._create_directory_containing(self.backup_path, "installed.txt")
+ os.makedirs(self.model.vendor_path)
+
+ self.assertTrue(self.model._set_aside_package_directory())
+
+ self.assertTrue(os.path.exists(os.path.join(self.backup_path, "installed.txt")))
+ self.assertEqual([], os.listdir(self.model.vendor_path))
+
+ def test_leftover_backup_is_discarded_when_packages_are_present(self, _mock_system_location):
+ self._create_directory_containing(self.backup_path, "stale.txt")
+ self._create_directory_containing(self.model.vendor_path, "installed.txt")
+
+ self.assertTrue(self.model._set_aside_package_directory())
+
+ self.assertTrue(os.path.exists(os.path.join(self.backup_path, "installed.txt")))
+ self.assertFalse(os.path.exists(os.path.join(self.backup_path, "stale.txt")))
+
+ def test_failed_run_restores_the_backup(self, _mock_system_location):
+ self._create_directory_containing(self.model.vendor_path, "installed.txt")
+ self.model._set_aside_package_directory()
+ self.model.update_worker = MagicMock(error="pip call failed", is_running=False)
+
+ self.model.finalize_package_directory()
+
+ self.assertTrue(os.path.exists(os.path.join(self.model.vendor_path, "installed.txt")))
+ self.assertFalse(os.path.exists(self.backup_path))
+ self.assertIsNone(self.model.backup_path)
+
+ def test_successful_run_discards_the_backup(self, _mock_system_location):
+ self._create_directory_containing(self.model.vendor_path, "installed.txt")
+ self.model._set_aside_package_directory()
+ self.model.update_worker = MagicMock(error="", is_running=False)
+
+ self.model.finalize_package_directory()
+
+ self.assertFalse(os.path.exists(self.backup_path))
+ self.assertFalse(os.path.exists(os.path.join(self.model.vendor_path, "installed.txt")))
+ self.assertIsNone(self.model.backup_path)
+
+ @patch("addonmanager_python_deps.fci.Console.PrintError")
+ def test_backup_is_kept_while_pip_is_still_running(
+ self, _mock_print_error, _mock_system_location
+ ):
+ self._create_directory_containing(self.model.vendor_path, "installed.txt")
+ self.model._set_aside_package_directory()
+ self.model.update_worker = MagicMock(error="", is_running=True)
+
+ self.model.finalize_package_directory()
+
+ self.assertTrue(os.path.exists(os.path.join(self.backup_path, "installed.txt")))
+ self.assertEqual(self.backup_path, self.model.backup_path)
+
+ @patch("addonmanager_python_deps.fci.Console.PrintError")
+ @patch("addonmanager_python_deps.call_pip")
+ def test_installation_is_abandoned_when_the_backup_fails(
+ self, mock_call_pip, _mock_print_error, _mock_system_location
+ ):
+ """If the packages cannot be moved to safety then pip is not run at all, because a
+ failure would otherwise destroy them."""
+ self._create_directory_containing(self.model.vendor_path, "installed.txt")
+ catcher = SignalCatcher()
+ self.model.update_complete.connect(catcher.catch_signal)
+
+ with patch("addonmanager_python_deps.os.rename", side_effect=OSError("locked")):
+ self.model.install_packages(["pkg1"])
+
+ mock_call_pip.assert_not_called()
+ self.assertTrue(catcher.caught)
+ self.assertTrue(os.path.exists(os.path.join(self.model.vendor_path, "installed.txt")))
+
+
+class TestCleanupOldPackageVersions(unittest.TestCase):
+ """Tests for the _cleanup_old_package_versions method"""
+
+ @patch("addonmanager_python_deps.os.path.exists")
+ @patch("addonmanager_python_deps.os.listdir")
+ @patch("addonmanager_python_deps.os.path.isdir")
+ @patch("addonmanager_python_deps.shutil.rmtree")
+ @patch("addonmanager_python_deps.fci.Console.PrintLog")
+ def test_cleanup_removes_old_versions_keeps_newest(
+ self, mock_print_log, mock_rmtree, mock_isdir, mock_listdir, mock_exists
+ ):
+ """Test that old package versions are removed and newest is kept"""
+ mock_exists.return_value = True
+ mock_isdir.return_value = True
+ mock_listdir.return_value = [
+ "requests-2.28.0.dist-info",
+ "requests-2.31.0.dist-info",
+ "numpy-1.24.0.dist-info",
+ "numpy-1.26.0.dist-info",
+ "numpy-1.25.2.dist-info",
+ "other_file.txt",
+ ]
+
+ model = PythonPackageListModel([])
+ model.vendor_path = "/fake/path"
+ model._cleanup_old_package_versions()
+
+ # Should remove old versions but keep newest
+ self.assertEqual(mock_rmtree.call_count, 3)
+ removed_paths = [call[0][0] for call in mock_rmtree.call_args_list]
+
+ # Check old versions were removed (works on all platforms)
+ self.assertIn(os.path.join("/fake/path", "requests-2.28.0.dist-info"), removed_paths)
+ self.assertIn(os.path.join("/fake/path", "numpy-1.24.0.dist-info"), removed_paths)
+ self.assertIn(os.path.join("/fake/path", "numpy-1.25.2.dist-info"), removed_paths)
+
+ @patch("addonmanager_python_deps.os.path.exists")
+ @patch("addonmanager_python_deps.os.listdir")
+ @patch("addonmanager_python_deps.shutil.rmtree")
+ def test_cleanup_single_version_no_removal(self, mock_rmtree, mock_listdir, mock_exists):
+ """Test that packages with only one version are not touched"""
+ mock_exists.return_value = True
+ mock_listdir.return_value = [
+ "requests-2.31.0.dist-info",
+ "numpy-1.26.0.dist-info",
+ "pandas-2.1.0.dist-info",
+ ]
+
+ model = PythonPackageListModel([])
+ model.vendor_path = "/fake/path"
+ model._cleanup_old_package_versions()
+
+ # No removals should happen when only one version exists per package
+ mock_rmtree.assert_not_called()
+
+ @patch("addonmanager_python_deps.os.path.exists")
+ def test_cleanup_nonexistent_directory(self, mock_exists):
+ """Test graceful handling when vendor path doesn't exist"""
+ mock_exists.return_value = False
+
+ model = PythonPackageListModel([])
+ model.vendor_path = "/nonexistent/path"
+ model._cleanup_old_package_versions()
+
+ @patch("addonmanager_python_deps.os.path.exists")
+ @patch("addonmanager_python_deps.os.listdir")
+ @patch("addonmanager_python_deps.shutil.rmtree")
+ def test_cleanup_empty_directory(self, mock_rmtree, mock_listdir, mock_exists):
+ """Test handling of empty vendor directory"""
+ mock_exists.return_value = True
+ mock_listdir.return_value = []
+
+ model = PythonPackageListModel([])
+ model.vendor_path = "/fake/path"
+ model._cleanup_old_package_versions()
+ mock_rmtree.assert_not_called()
+
+ @patch("addonmanager_python_deps.os.path.exists")
+ @patch("addonmanager_python_deps.os.listdir")
+ @patch("addonmanager_python_deps.os.path.isdir")
+ @patch("addonmanager_python_deps.shutil.rmtree")
+ @patch("addonmanager_python_deps.fci.Console.PrintWarning")
+ def test_cleanup_handles_permission_error(
+ self, mock_print_warning, mock_rmtree, mock_isdir, mock_listdir, mock_exists
+ ):
+ """Test that permission errors are handled gracefully"""
+ mock_exists.return_value = True
+ mock_listdir.return_value = [
+ "requests-2.28.0.dist-info",
+ "requests-2.31.0.dist-info",
+ ]
+ mock_isdir.return_value = True
+ mock_rmtree.side_effect = PermissionError("Permission denied")
+
+ model = PythonPackageListModel([])
+ model.vendor_path = "/fake/path"
+ model._cleanup_old_package_versions()
+ mock_print_warning.assert_called()
+
+ @patch("addonmanager_python_deps.os.path.exists")
+ @patch("addonmanager_python_deps.os.listdir")
+ @patch("addonmanager_python_deps.os.path.isdir")
+ @patch("addonmanager_python_deps.shutil.rmtree")
+ def test_cleanup_normalizes_package_names(
+ self, mock_rmtree, mock_isdir, mock_listdir, mock_exists
+ ):
+ """Test that package names are normalized per PEP 503 (underscores to dashes)"""
+ mock_exists.return_value = True
+ mock_listdir.return_value = [
+ "my_package-1.0.0.dist-info",
+ "my-package-2.0.0.dist-info",
+ ]
+ mock_isdir.return_value = True
+
+ model = PythonPackageListModel([])
+ model.vendor_path = "/fake/path"
+ model._cleanup_old_package_versions()
+
+ # Should remove old version (they're the same package after normalization)
+ mock_rmtree.assert_called_once_with(
+ os.path.join("/fake/path", "my_package-1.0.0.dist-info")
+ )
+
+ @patch("addonmanager_python_deps.os.path.exists")
+ @patch("addonmanager_python_deps.os.listdir")
+ @patch("addonmanager_python_deps.os.path.isdir")
+ @patch("addonmanager_python_deps.shutil.rmtree")
+ def test_cleanup_ignores_non_dist_info_directories(
+ self, mock_rmtree, mock_isdir, mock_listdir, mock_exists
+ ):
+ """Test that only .dist-info directories are processed"""
+ mock_exists.return_value = True
+ mock_listdir.return_value = [
+ "requests-2.28.0.dist-info",
+ "requests-2.31.0.dist-info",
+ "some_package",
+ "__pycache__",
+ "random_file.txt",
+ ]
+ mock_isdir.return_value = True
+
+ model = PythonPackageListModel([])
+ model.vendor_path = "/fake/path"
+ model._cleanup_old_package_versions()
+ mock_rmtree.assert_called_once_with(os.path.join("/fake/path", "requests-2.28.0.dist-info"))
+
+ @patch("addonmanager_python_deps.os.path.exists")
+ @patch("addonmanager_python_deps.os.listdir")
+ @patch("addonmanager_python_deps.os.path.isdir")
+ @patch("addonmanager_python_deps.shutil.rmtree")
+ @patch("addonmanager_python_deps.fci.Console.PrintWarning")
+ def test_cleanup_handles_invalid_version_format(
+ self, mock_print_warning, mock_rmtree, mock_isdir, mock_listdir, mock_exists
+ ):
+ """Test handling of malformed version strings"""
+ mock_exists.return_value = True
+ mock_listdir.return_value = [
+ "badpackage-invalid.version.dist-info",
+ "goodpackage-1.0.0.dist-info",
+ "goodpackage-2.0.0.dist-info",
+ ]
+ mock_isdir.return_value = True
+
+ model = PythonPackageListModel([])
+ model.vendor_path = "/fake/path"
+ model._cleanup_old_package_versions()
+ mock_rmtree.assert_called_once_with(
+ os.path.join("/fake/path", "goodpackage-1.0.0.dist-info")
+ )
diff --git a/AddonManagerTest/gui/test_python_deps_gui.py b/AddonManagerTest/gui/test_python_deps_gui.py
index 8dfbae5..1be2545 100644
--- a/AddonManagerTest/gui/test_python_deps_gui.py
+++ b/AddonManagerTest/gui/test_python_deps_gui.py
@@ -3,6 +3,7 @@
import sys
import unittest
+from unittest.mock import MagicMock
from PySideWrapper import QtCore, QtWidgets
@@ -15,6 +16,35 @@ class TestPythonPackageManagerGui(unittest.TestCase):
def setUp(self) -> None:
self.manager = PythonPackageManagerGui([])
+ def test_stop_button_is_only_enabled_while_pip_runs(self):
+ self.manager._working(True)
+ self.assertTrue(self.manager.dlg.buttonCancel.isEnabled())
+ self.manager._working(False)
+ self.assertFalse(self.manager.dlg.buttonCancel.isEnabled())
+
+ def test_progress_message_is_displayed(self):
+ self.manager._working(True)
+ self.manager._show_progress_message("Collecting numpy")
+ self.assertNotEqual("", self.manager.dlg.progressDetailsLabel.text())
+
+ def test_progress_message_is_cleared_when_the_run_ends(self):
+ self.manager._show_progress_message("Collecting numpy")
+ self.manager._working(False)
+ self.assertEqual("", self.manager.dlg.progressDetailsLabel.text())
+
+ def test_stop_button_cancels_the_run(self):
+ self.manager.model.cancel_update = MagicMock()
+ self.manager.dlg.buttonCancel.click()
+ self.manager.model.cancel_update.assert_called_once()
+ self.assertFalse(self.manager.dlg.buttonCancel.isEnabled())
+
+ def test_closing_the_dialog_waits_for_pip_to_stop(self):
+ """The model is destroyed with the dialog, so a running pip call must be stopped and its
+ backup dealt with before the dialog goes away."""
+ self.manager.model.cancel_update = MagicMock()
+ self.manager.dlg.reject()
+ self.manager.model.cancel_update.assert_called_once_with(wait_for_completion=True)
+
if __name__ == "__main__":
app = QtWidgets.QApplication(sys.argv)
diff --git a/PythonDependencyUpdateDialog.ui b/PythonDependencyUpdateDialog.ui
index 46bab39..b7cddec 100644
--- a/PythonDependencyUpdateDialog.ui
+++ b/PythonDependencyUpdateDialog.ui
@@ -63,6 +63,22 @@
+ -
+
+
+
+ 0
+ 0
+
+
+
+
+
+
+ false
+
+
+
-
@@ -89,6 +105,16 @@
+ -
+
+
+ Stop
+
+
+ Stop the running pip installation or update
+
+
+
diff --git a/addonmanager_python_deps.py b/addonmanager_python_deps.py
index b029615..fbe45b0 100644
--- a/addonmanager_python_deps.py
+++ b/addonmanager_python_deps.py
@@ -28,12 +28,13 @@
import re
import shutil
import subprocess
-from typing import Dict, Iterable, List, TypedDict, Optional, Set
+from typing import Callable, Dict, Iterable, List, TypedDict, Optional, Set
from enum import Enum
from addonmanager_metadata import Version
from addonmanager_utilities import (
+ ProcessInterrupted,
create_pip_call,
- run_interruptable_subprocess,
+ run_monitored_subprocess,
get_pip_target_directory,
pep503_normalize,
translate,
@@ -48,31 +49,38 @@
translate = fci.translate
+BACKUP_SUFFIX = ".old"
+CANCELLATION_TIMEOUT_MS = 10000
+
+
class PipFailed(Exception):
- """Exception thrown when pip times out or otherwise fails to return valid results"""
+ """Exception thrown when pip fails to return valid results"""
+
+class PipInterrupted(PipFailed):
+ """Exception thrown when a pip call is stopped by an interruption request."""
-def call_pip(args: List[str]) -> List[str]:
+
+def call_pip(args: List[str], line_callback: Optional[Callable[[str], None]] = None) -> 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."""
+ disabled. Fails if Python can't be found or if pip is not installed. Each line of output is
+ passed to line_callback as it is produced, if a callback is provided."""
try:
call_args = create_pip_call(args)
- fci.Console.PrintLog(f"Running pip with the following command:\n")
+ fci.Console.PrintLog("Running pip with the following command:\n")
fci.Console.PrintLog(" ".join(call_args) + "\n")
except RuntimeError as exception:
raise PipFailed() from exception
try:
- proc = run_interruptable_subprocess(call_args, timeout_secs=None)
+ proc = run_monitored_subprocess(call_args, line_callback=line_callback)
+ except ProcessInterrupted as exception:
+ raise PipInterrupted("The pip call was cancelled") from exception
except subprocess.CalledProcessError as exception:
raise PipFailed(f"pip call failed:\n{exception}") from exception
- if proc.returncode != 0:
- raise PipFailed(proc.stderr)
-
- data = proc.stdout
- return data.split("\n")
+ return proc.stdout.split("\n")
@dataclasses.dataclass
@@ -83,10 +91,14 @@ class PackageInfo:
dependencies: List[str]
+LOG_LINE_PREFIXES = ("WARNING:", "ERROR:", "DEPRECATION:", "NOTICE:")
+
+
def parse_pip_list_output(all_packages, constrained_versions: Dict[str, str]) -> List[PackageInfo]:
"""Parse 'pip list --path' output into package information, marking an update as available
whenever the vetted (constrained) version differs from the installed one. The pip output
- should be an array of lines of text.
+ should be an array of lines of text. Anything before the underlined header, and any log line
+ that pip mixed into its output, is ignored.
All Packages output looks like this:
Package Version
@@ -96,10 +108,12 @@ def parse_pip_list_output(all_packages, constrained_versions: Dict[str, str]) ->
"""
packages: Dict[str, PackageInfo] = {}
- skip_counter = 0
+ header_seen = False
for line in all_packages:
- if skip_counter < 2:
- skip_counter += 1
+ if line.startswith(LOG_LINE_PREFIXES):
+ continue
+ if not header_seen:
+ header_seen = line.startswith("---")
continue
entries = line.split()
if len(entries) > 1:
@@ -133,6 +147,7 @@ class AsynchronousPipWorker(QtCore.QObject):
"""A worker class that runs pip to install/update/list packages."""
finished = QtCore.Signal()
+ progress_message = QtCore.Signal(str) # A line of pip output, or a status message
def __init__(
self,
@@ -143,18 +158,27 @@ def __init__(
super().__init__(parent)
self.is_running = False
self.error = ""
+ self.cancelled = False
self.vendor_path = get_pip_target_directory()
self.package_list = package_list or []
self.command = command
def run(self):
- """Runs pip: when complete, either self.package_list is populated, or self.error is set."""
+ """Runs pip: when complete, either self.package_list is populated, or self.error is set.
+ The finished signal is emitted no matter how the run ends, so that callers can always
+ rely on it to restore whatever state they set up before starting the run."""
self.is_running = True
self.error = ""
+ self.cancelled = False
- if self.command in (PipCommand.Upgrade, PipCommand.Install):
- self._install_or_update()
- self._list()
+ try:
+ if self.command in (PipCommand.Upgrade, PipCommand.Install):
+ self._install_or_update()
+ if not self.cancelled:
+ self._list()
+ except Exception as e:
+ self.error = f"Unexpected failure while running pip: {e}"
+ fci.Console.PrintError(f"{self.error}\n")
self.is_running = False
self.finished.emit()
@@ -167,23 +191,37 @@ def _install_or_update(self) -> None:
action = "install" if self.command == PipCommand.Install else "upgrade"
log_message = f"Running pip to {action} the following packages in {self.vendor_path}: {update_string}\n"
upgrade = ["--upgrade"] if self.command == PipCommand.Upgrade else []
- command = ["install", *upgrade, "--target", self.vendor_path]
+ command = ["install", "--progress-bar", "off", *upgrade, "--target", self.vendor_path]
command.extend(self.package_list)
fci.Console.PrintLog(f"{log_message}\n")
+ self.progress_message.emit(translate("AddonsInstaller", "Starting pip"))
try:
- upgrade_stdout = call_pip(command)
+ upgrade_stdout = call_pip(command, line_callback=self._report_progress)
for line in upgrade_stdout:
fci.Console.PrintLog(f"{line}\n")
+ except PipInterrupted as e:
+ self.cancelled = True
+ self.error = str(e)
+ fci.Console.PrintMessage(f"{self.error}\n")
except PipFailed as e:
self.error = str(e)
fci.Console.PrintError(f"{self.error}\n")
+ def _report_progress(self, line: str) -> None:
+ """Forward a non-empty line of pip output to anyone displaying progress."""
+ stripped_line = line.strip()
+ if stripped_line:
+ self.progress_message.emit(stripped_line)
+
def _list(self) -> None:
try:
all_packages_stdout = call_pip(["list", "--path", self.vendor_path])
constrained_versions = get_constraints().constrained_versions()
self.package_list = parse_pip_list_output(all_packages_stdout, constrained_versions)
+ except PipInterrupted as e:
+ self.cancelled = True
+ self.error = str(e)
except PipFailed as e:
self.error = str(e)
@@ -194,6 +232,7 @@ class PythonPackageListModel(QtCore.QAbstractTableModel):
for the Qt view."""
update_complete = QtCore.Signal()
+ progress_message = QtCore.Signal(str)
def __init__(self, addons):
super().__init__()
@@ -205,6 +244,7 @@ def __init__(self, addons):
self.update_worker = None
self.reset_worker_thread = None
self.update_worker_thread = None
+ self.backup_path = None
def can_use_thread(self) -> bool:
threaded = (
@@ -219,6 +259,7 @@ def reset_package_list(self):
self.beginResetModel()
self.package_list.clear()
self.reset_worker = AsynchronousPipWorker(PipCommand.List)
+ self.reset_worker.progress_message.connect(self.progress_message)
if self.can_use_thread():
self.reset_worker_thread = QtCore.QThread()
self.reset_worker.moveToThread(self.reset_worker_thread)
@@ -319,12 +360,11 @@ def install_packages(self, packages: list[str]) -> None:
def _install_or_update_packages(self, packages: list[str], command: PipCommand) -> None:
"""Installs/Upgrade packages. Uses an asynchronous thread when possible."""
+ if not using_system_pip_installation_location() and not self._set_aside_package_directory():
+ self.update_complete.emit()
+ return
self.update_worker = AsynchronousPipWorker(command, packages)
- if not using_system_pip_installation_location():
- # pip doesn't properly update when using the target directory, so we have to delete
- # it and reinstall
- os.rename(self.vendor_path, self.vendor_path + ".old")
- os.mkdir(self.vendor_path)
+ self.update_worker.progress_message.connect(self.progress_message)
if self.can_use_thread():
self.update_worker_thread = QtCore.QThread()
self.update_worker.moveToThread(self.update_worker_thread)
@@ -337,25 +377,158 @@ def _install_or_update_packages(self, packages: list[str], command: PipCommand)
self.update_call_finished()
def update_call_finished(self):
+ """Put the package directory into its final state, then report that the run is over."""
+ self.finalize_package_directory()
self.update_complete.emit()
- if not using_system_pip_installation_location():
- if self.update_worker.error:
- try:
- os.rename(self.vendor_path + ".old", self.vendor_path)
- except Exception as err:
- fci.Console.PrintError(f"Backup restore failed: {self.vendor_path}.old.\n")
- fci.Console.PrintError(f"{err}\n")
+
+ def cancel_update(self, wait_for_completion: bool = False) -> None:
+ """Ask any running pip call to stop. When wait_for_completion is set, the call blocks
+ until the worker has stopped and the package directory has been dealt with, which is
+ required when the caller is about to destroy this model."""
+ for thread in (self.update_worker_thread, self.reset_worker_thread):
+ if thread is not None and thread.isRunning():
+ thread.requestInterruption()
+ if not wait_for_completion:
+ return
+ for worker, thread in (
+ (self.update_worker, self.update_worker_thread),
+ (self.reset_worker, self.reset_worker_thread),
+ ):
+ if thread is None or not thread.isRunning():
+ continue
+ worker.blockSignals(True)
+ thread.quit()
+ if not thread.wait(CANCELLATION_TIMEOUT_MS):
+ fci.Console.PrintWarning(
+ translate("AddonsInstaller", "A pip call did not stop when asked to") + "\n"
+ )
+ self.finalize_package_directory()
+
+ def finalize_package_directory(self) -> None:
+ """Restore the backup of the package directory if the run failed or was cancelled, and
+ discard it if the run succeeded. Does nothing if no backup was made."""
+ if self.backup_path is None:
+ return
+ if self.update_worker is not None and self.update_worker.is_running:
+ fci.Console.PrintError(
+ translate(
+ "AddonsInstaller",
+ "pip is still running, so the Python packages were left in {}",
+ ).format(self.backup_path)
+ + "\n"
+ )
+ return
+ if self.update_worker is not None and self.update_worker.error:
+ self._restore_package_directory_backup()
+ else:
+ self._discard_package_directory_backup()
+ self._cleanup_old_package_versions()
+
+ def _set_aside_package_directory(self) -> bool:
+ """Move the existing package directory aside so that it can be restored if pip does not
+ succeed, because pip cannot reliably upgrade in place when installing to a target
+ directory. Returns True if the installation may proceed."""
+ backup_path = self.vendor_path + BACKUP_SUFFIX
+ self.backup_path = None
+ if os.path.exists(backup_path):
+ self._resolve_leftover_backup(backup_path)
+ if not os.path.exists(self.vendor_path):
+ try:
+ os.makedirs(self.vendor_path)
+ except OSError as err:
+ fci.Console.PrintError(
+ translate(
+ "AddonsInstaller", "Failed to create the Python package directory {}"
+ ).format(self.vendor_path)
+ + f"\n{err}\n"
+ )
+ return False
+ return True
+ try:
+ os.rename(self.vendor_path, backup_path)
+ except OSError as err:
+ fci.Console.PrintError(
+ translate(
+ "AddonsInstaller",
+ "Failed to back up the Python package directory {}, so no packages were"
+ " installed or updated",
+ ).format(self.vendor_path)
+ + f"\n{err}\n"
+ )
+ return False
+ try:
+ os.mkdir(self.vendor_path)
+ except OSError as err:
+ fci.Console.PrintError(f"{err}\n")
+ self.backup_path = backup_path
+ self._restore_package_directory_backup()
+ return False
+ self.backup_path = backup_path
+ return True
+
+ def _resolve_leftover_backup(self, backup_path: str) -> None:
+ """Deal with a backup left behind by a run that never completed: it is put back when the
+ package directory is missing or empty, and discarded otherwise."""
+ try:
+ if not os.path.exists(self.vendor_path):
+ os.rename(backup_path, self.vendor_path)
+ elif not os.listdir(self.vendor_path):
+ os.rmdir(self.vendor_path)
+ os.rename(backup_path, self.vendor_path)
else:
- shutil.rmtree(self.vendor_path + ".old")
- # Clean up old package versions that may remain after update
- self._cleanup_old_package_versions()
+ shutil.rmtree(backup_path)
+ return
+ fci.Console.PrintWarning(
+ translate(
+ "AddonsInstaller",
+ "Recovered the Python packages left in {} by an interrupted update",
+ ).format(backup_path)
+ + "\n"
+ )
+ except OSError as err:
+ fci.Console.PrintError(f"{err}\n")
+
+ def _restore_package_directory_backup(self) -> None:
+ """Put the backed-up package directory back after a failed or cancelled run."""
+ backup_path = self.backup_path
+ self.backup_path = None
+ try:
+ if os.path.exists(self.vendor_path):
+ shutil.rmtree(self.vendor_path)
+ os.rename(backup_path, self.vendor_path)
+ return
+ except OSError as err:
+ fci.Console.PrintError(f"{err}\n")
+ try:
+ shutil.copytree(backup_path, self.vendor_path, dirs_exist_ok=True)
+ except OSError as err:
+ fci.Console.PrintError(
+ translate(
+ "AddonsInstaller",
+ "Failed to restore the Python packages: they remain in {}",
+ ).format(backup_path)
+ + f"\n{err}\n"
+ )
+
+ def _discard_package_directory_backup(self) -> None:
+ """Remove the backup of the package directory after a successful run."""
+ backup_path = self.backup_path
+ self.backup_path = None
+ try:
+ shutil.rmtree(backup_path)
+ except OSError as err:
+ fci.Console.PrintWarning(
+ translate("AddonsInstaller", "Failed to remove the backup directory {}").format(
+ backup_path
+ )
+ + f"\n{err}\n"
+ )
def _cleanup_old_package_versions(self):
"""Remove old package version metadata directories after an update.
- When pip updates packages with --target, it doesn't always remove old
- version metadata (.dist-info directories). This can cause version detection
- to find the old version instead of the new one, especially in Flatpak
+ When pip updates packages with --target, it doesn't always remove old version metadata (.dist-info directories).
+ This can cause version detection to find the old version instead of the new one, especially in Flatpak
installations where multiple versions accumulate.
"""
if not os.path.exists(self.vendor_path):
diff --git a/addonmanager_python_deps_gui.py b/addonmanager_python_deps_gui.py
index 596cce1..d68d560 100644
--- a/addonmanager_python_deps_gui.py
+++ b/addonmanager_python_deps_gui.py
@@ -25,7 +25,7 @@
import addonmanager_freecad_interface as fci
from addonmanager_python_deps import PythonPackageListModel
-from PySideWrapper import QtWidgets
+from PySideWrapper import QtCore, QtWidgets
translate = fci.translate
@@ -53,8 +53,11 @@ def __init__(self, addons):
self.dlg.buttonInstallPkgs.clicked.connect(self._install_button_clicked)
self.dlg.buttonUpdateAll.clicked.connect(self._update_button_clicked)
+ self.dlg.buttonCancel.clicked.connect(self._cancel_button_clicked)
+ self.dlg.rejected.connect(self._dialog_rejected)
self.model.modelReset.connect(self._model_was_reset)
self.model.update_complete.connect(self._update_complete)
+ self.model.progress_message.connect(self._show_progress_message)
def show(self):
self._working(True)
@@ -63,12 +66,36 @@ def show(self):
self.dlg.exec()
def _working(self, working: bool) -> None:
+ """Show or hide the progress display, and enable the buttons that make sense while pip is
+ running, or while it is not."""
self.dlg.buttonInstallPkgs.setEnabled(not working)
self.dlg.buttonUpdateAll.setEnabled(not working and self.model.updates_are_available())
+ self.dlg.buttonCancel.setEnabled(working)
if working:
self.dlg.updateInProgressLabel.show()
+ self.dlg.progressDetailsLabel.show()
else:
self.dlg.updateInProgressLabel.hide()
+ self.dlg.progressDetailsLabel.hide()
+ self.dlg.progressDetailsLabel.setText("")
+
+ def _show_progress_message(self, message: str) -> None:
+ """Display the most recent line of pip output, shortened to fit the available width."""
+ label = self.dlg.progressDetailsLabel
+ elided = label.fontMetrics().elidedText(
+ message, QtCore.Qt.TextElideMode.ElideRight, label.width()
+ )
+ label.setText(elided)
+
+ def _cancel_button_clicked(self):
+ """Ask the running pip call to stop, without waiting for it to do so."""
+ self.dlg.buttonCancel.setEnabled(False)
+ self._show_progress_message(translate("AddonsInstaller", "Stopping pip…"))
+ self.model.cancel_update()
+
+ def _dialog_rejected(self):
+ """Stop any running pip call before this dialog and its model are destroyed."""
+ self.model.cancel_update(wait_for_completion=True)
def _install_button_clicked(self):
title = translate("AddonsInstaller", "Install")