Addon Manager: Fix bug in ZIP isntallation

This commit is contained in:
Chris Hennes
2023-01-09 19:35:18 -06:00
parent 5196a4c01b
commit a7b4a6deba
5 changed files with 159 additions and 60 deletions

View File

@@ -145,16 +145,56 @@ class TestAddonInstaller(unittest.TestCase):
test_github_style_repo = os.path.join(
self.test_data_dir, "test_github_style_repo.zip"
)
installer = AddonInstaller(self.mock_addon, [])
installer.installation_path = temp_dir
self.mock_addon.url = test_github_style_repo
self.mock_addon.branch = "master"
installer = AddonInstaller(self.mock_addon, [])
installer.installation_path = temp_dir
installer._finalize_zip_installation(test_github_style_repo)
expected_location = os.path.join(temp_dir, self.mock_addon.name, "README")
self.assertTrue(
os.path.isfile(expected_location), "GitHub zip extraction failed"
)
def test_code_in_branch_subdirectory_true(self):
"""When there is a subdirectory with the branch name in it, find it"""
installer = AddonInstaller(self.mock_addon, [])
with tempfile.TemporaryDirectory() as temp_dir:
os.mkdir(os.path.join(temp_dir,f"{self.mock_addon.name}-{self.mock_addon.branch}"))
result = installer._code_in_branch_subdirectory(temp_dir)
self.assertTrue(result,"Failed to find ZIP subdirectory")
def test_code_in_branch_subdirectory_false(self):
"""When there is not a subdirectory with the branch name in it, don't find one"""
installer = AddonInstaller(self.mock_addon, [])
with tempfile.TemporaryDirectory() as temp_dir:
result = installer._code_in_branch_subdirectory(temp_dir)
self.assertFalse(result,"Found ZIP subdirectory when there was none")
def test_code_in_branch_subdirectory_more_than_one(self):
"""When there are multiple subdirectories, never find a branch subdirectory"""
installer = AddonInstaller(self.mock_addon, [])
with tempfile.TemporaryDirectory() as temp_dir:
os.mkdir(os.path.join(temp_dir,f"{self.mock_addon.name}-{self.mock_addon.branch}"))
os.mkdir(os.path.join(temp_dir,"AnotherSubdir"))
result = installer._code_in_branch_subdirectory(temp_dir)
self.assertFalse(result,"Found ZIP subdirectory when there were multiple subdirs")
def test_move_code_out_of_subdirectory(self):
"""All files are moved out and the subdirectory is deleted"""
installer = AddonInstaller(self.mock_addon, [])
with tempfile.TemporaryDirectory() as temp_dir:
subdir = os.path.join(temp_dir,f"{self.mock_addon.name}-{self.mock_addon.branch}")
os.mkdir(subdir)
with open(os.path.join(subdir,"README.txt"),"w",encoding="utf-8") as f:
f.write("# Test file for unit testing")
with open(os.path.join(subdir,"AnotherFile.txt"),"w",encoding="utf-8") as f:
f.write("# Test file for unit testing")
installer._move_code_out_of_subdirectory(temp_dir)
self.assertTrue(os.path.isfile(os.path.join(temp_dir,"README.txt")))
self.assertTrue(os.path.isfile(os.path.join(temp_dir,"AnotherFile.txt")))
self.assertFalse(os.path.isdir(subdir))
def test_install_by_git(self):
"""Test using git to install. Depends on there being a local git installation: the test
is skipped if there is no local git."""
@@ -292,44 +332,86 @@ class TestAddonInstaller(unittest.TestCase):
method = installer._determine_install_method(temp_file, InstallationMethod.ANY)
self.assertEqual(method, InstallationMethod.ZIP)
def test_determine_install_method_https_known_sites(self):
def test_determine_install_method_https_known_sites_copy(self):
"""Test which install methods are accepted for an https github URL"""
installer = AddonInstaller(self.mock_addon, [])
installer.git_manager = True
for site in ["github.org", "gitlab.org", "framagit.org", "salsa.debian.org"]:
temp_file = f"https://{site}/dummy/dummy" # Doesn't have to actually exist!
method = installer._determine_install_method(
temp_file, InstallationMethod.COPY
)
self.assertIsNone(method, f"Allowed copying from {site} URL")
method = installer._determine_install_method(
temp_file, InstallationMethod.GIT
)
self.assertEqual(
method,
InstallationMethod.GIT,
f"Failed to allow git access to {site} URL",
)
method = installer._determine_install_method(
temp_file, InstallationMethod.ZIP
)
self.assertEqual(
method,
InstallationMethod.ZIP,
f"Failed to allow zip access to {site} URL",
)
method = installer._determine_install_method(
temp_file, InstallationMethod.ANY
)
git_manager = initialize_git()
if git_manager:
with self.subTest(site=site):
temp_file = f"https://{site}/dummy/dummy" # Doesn't have to actually exist!
method = installer._determine_install_method(
temp_file, InstallationMethod.COPY
)
self.assertIsNone(method, f"Allowed copying from {site} URL")
def test_determine_install_method_https_known_sites_git(self):
"""Test which install methods are accepted for an https github URL"""
installer = AddonInstaller(self.mock_addon, [])
installer.git_manager = True
for site in ["github.org", "gitlab.org", "framagit.org", "salsa.debian.org"]:
with self.subTest(site=site):
temp_file = f"https://{site}/dummy/dummy" # Doesn't have to actually exist!
method = installer._determine_install_method(
temp_file, InstallationMethod.GIT
)
self.assertEqual(
method,
InstallationMethod.GIT,
f"Failed to allow git access to {site} URL",
)
else:
def test_determine_install_method_https_known_sites_zip(self):
"""Test which install methods are accepted for an https github URL"""
installer = AddonInstaller(self.mock_addon, [])
installer.git_manager = True
for site in ["github.org", "gitlab.org", "framagit.org", "salsa.debian.org"]:
with self.subTest(site=site):
temp_file = f"https://{site}/dummy/dummy" # Doesn't have to actually exist!
method = installer._determine_install_method(
temp_file, InstallationMethod.ZIP
)
self.assertEqual(
method,
InstallationMethod.ZIP,
f"Failed to allow zip access to {site} URL",
)
def test_determine_install_method_https_known_sites_any_gm(self):
"""Test which install methods are accepted for an https github URL"""
installer = AddonInstaller(self.mock_addon, [])
installer.git_manager = True
for site in ["github.org", "gitlab.org", "framagit.org", "salsa.debian.org"]:
with self.subTest(site=site):
temp_file = f"https://{site}/dummy/dummy" # Doesn't have to actually exist!
method = installer._determine_install_method(
temp_file, InstallationMethod.ANY
)
self.assertEqual(
method,
InstallationMethod.GIT,
f"Failed to allow git access to {site} URL",
)
def test_determine_install_method_https_known_sites_any_no_gm(self):
"""Test which install methods are accepted for an https github URL"""
installer = AddonInstaller(self.mock_addon, [])
installer.git_manager = None
for site in ["github.org", "gitlab.org", "framagit.org", "salsa.debian.org"]:
with self.subTest(site=site):
temp_file = f"https://{site}/dummy/dummy" # Doesn't have to actually exist!
method = installer._determine_install_method(
temp_file, InstallationMethod.ANY
)
self.assertEqual(
method,
InstallationMethod.ZIP,

View File

@@ -297,20 +297,13 @@ class AddonInstaller(QtCore.QObject):
else:
zip_url = utils.get_zip_url(self.addon_to_install)
FreeCAD.Console.PrintLog(f"Downloading ZIP file from {zip_url}...\n")
parse_result = urlparse(zip_url)
is_remote = parse_result.scheme in ["http", "https"]
if is_remote:
if FreeCAD.GuiUp:
NetworkManager.AM_NETWORK_MANAGER.progress_made.connect(
self._update_zip_status
)
NetworkManager.AM_NETWORK_MANAGER.progress_complete.connect(
self._finish_zip
)
self.zip_download_index = (
NetworkManager.AM_NETWORK_MANAGER.submit_monitored_get(zip_url)
)
self._run_zip_downloader_in_event_loop(zip_url)
else:
zip_data = utils.blocking_get(zip_url)
with tempfile.NamedTemporaryFile(delete=False) as f:
@@ -321,16 +314,33 @@ class AddonInstaller(QtCore.QObject):
self._finalize_zip_installation(zip_url)
return True
def _run_zip_downloader_in_event_loop(self, zip_url: str):
"""Runs the zip downloader in a private event loop. This function does not exit until the
ZIP download is complete. It requires the GUI to be up, and should not be run on the main
GUI thread."""
NetworkManager.AM_NETWORK_MANAGER.progress_made.connect(self._update_zip_status)
NetworkManager.AM_NETWORK_MANAGER.progress_complete.connect(self._finish_zip)
self.zip_download_index = (
NetworkManager.AM_NETWORK_MANAGER.submit_monitored_get(zip_url)
)
while self.zip_download_index is not None:
if QtCore.QThread.currentThread().isInterruptionRequested():
break
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents, 50)
def _update_zip_status(self, index: int, bytes_read: int, data_size: int):
"""Called periodically when downloading a zip file, emits a signal to display the
download progress."""
if index == self.zip_download_index:
self.progress_update.emit(bytes_read, data_size)
def _finish_zip(self, _: int, response_code: int, filename: os.PathLike):
def _finish_zip(self, index: int, response_code: int, filename: os.PathLike):
"""Once the zip download is finished, unzip it into the correct location. Only called if
the GUI is up, and the NetworkManager was responsible for the download. Do not call
directly."""
if index != self.zip_download_index:
return
self.zip_download_index = None
if response_code != 200:
self.failure.emit(
self.addon_to_install,
@@ -341,6 +351,7 @@ class AddonInstaller(QtCore.QObject):
return
QtCore.QCoreApplication.processEvents(QtCore.QEventLoop.AllEvents)
FreeCAD.Console.PrintLog("ZIP download complete. Installing...\n")
self._finalize_zip_installation(filename)
def _finalize_zip_installation(self, filename: os.PathLike):
@@ -348,7 +359,6 @@ class AddonInstaller(QtCore.QObject):
location. Has special handling for GitHub's zip structure, which places the data in a
subdirectory of the main directory."""
subdirectory = ""
destination = os.path.join(self.installation_path, self.addon_to_install.name)
with zipfile.ZipFile(filename, "r") as zfile:
zfile.extractall(destination)
@@ -356,24 +366,31 @@ class AddonInstaller(QtCore.QObject):
# GitHub (and possibly other hosts) put all files in the zip into a subdirectory named
# after the branch. If that is the setup that we just extracted, move all files out of
# that subdirectory.
subdirectories = os.listdir(destination)
if (
len(subdirectories) == 1
and subdirectories[0] == self.addon_to_install.branch
):
subdirectory = subdirectories[0]
if self._code_in_branch_subdirectory(destination):
self._move_code_out_of_subdirectory(destination)
if subdirectory:
for extracted_filename in os.listdir(
os.path.join(destination, subdirectory)
):
shutil.move(
os.path.join(destination, subdirectory, extracted_filename),
os.path.join(destination, extracted_filename),
)
os.rmdir(os.path.join(destination, subdirectory))
FreeCAD.Console.PrintLog("ZIP installation complete.\n")
self._finalize_successful_installation()
def _code_in_branch_subdirectory(self, destination: str) -> bool:
subdirectories = os.listdir(destination)
if len(subdirectories) == 1:
subdir_name = subdirectories[0]
if subdir_name.endswith(os.path.sep):
subdir_name = subdir_name[:-1] # Strip trailing slash if present
if subdir_name.endswith(self.addon_to_install.branch):
return True
return False
def _move_code_out_of_subdirectory(self, destination):
subdirectory = os.listdir(destination)[0]
for extracted_filename in os.listdir(os.path.join(destination, subdirectory)):
shutil.move(
os.path.join(destination, subdirectory, extracted_filename),
os.path.join(destination, extracted_filename),
)
os.rmdir(os.path.join(destination, subdirectory))
def _finalize_successful_installation(self):
"""Perform any necessary additional steps after installing the addon."""
self._update_metadata()

View File

@@ -87,13 +87,13 @@ class AddonInstallerGUI(QtCore.QObject):
# Dependency check
deps = MissingDependencies(self.addon_to_install, self.addons)
# Python interpreter version check
stop_installation = self._check_python_version(deps)
if stop_installation:
self.finished.emit()
return
# Required Python
if hasattr(deps, "python_requires") and deps.python_requires:
# Disallowed packages:
@@ -106,7 +106,7 @@ class AddonInstallerGUI(QtCore.QObject):
# Remove any disallowed packages from the optional list
if hasattr(deps, "python_optional") and deps.python_optional:
self._clean_up_optional(deps)
# Missing FreeCAD workbenches
if hasattr(deps, "wbs") and deps.wbs:
stop_installation = self._report_missing_workbenches(deps.wbs)

View File

@@ -66,7 +66,7 @@ translate = FreeCAD.Qt.translate
class ProcessInterrupted(RuntimeError):
"""An interruption request was received and the process killed because if it."""
"""An interruption request was received and the process killed because of it."""
def symlink(source, link_name):