refactor: removal of duplicated download_installer functionality

Signed-off-by: Hiroshi Miura <miurahr@linux.com>
This commit is contained in:
Hiroshi Miura
2025-03-04 11:43:53 +09:00
parent edc423a239
commit 122377cb8a
4 changed files with 25 additions and 34 deletions

View File

@@ -23,16 +23,15 @@ import os
from dataclasses import dataclass
from logging import Logger, getLogger
from pathlib import Path
from typing import List, Optional, Tuple
from typing import List, Optional
from defusedxml import ElementTree
from aqt.exceptions import DiskAccessNotPermitted
from aqt.helper import (
Settings,
downloadBinaryFile,
download_installer,
extract_auth,
get_hash,
get_os_name,
get_qt_account_path,
get_qt_installer_name,
@@ -298,17 +297,6 @@ class CommercialInstaller:
return cmd
def download_installer(self, target_path: Path, timeout: Tuple[float, float]) -> None:
base_path = f"official_releases/online_installers/{self._installer_filename}"
url = f"{self.base_url}/{base_path}"
try:
hash = get_hash(base_path, Settings.hash_algorithm, timeout)
downloadBinaryFile(url, target_path, Settings.hash_algorithm, hash, timeout=timeout)
if self.os_name != "windows":
os.chmod(target_path, 0o500)
except Exception as e:
raise RuntimeError(f"Failed to download installer: {e}")
def install(self) -> None:
"""Run the Qt installation process."""
if (
@@ -351,7 +339,8 @@ class CommercialInstaller:
installer_path = temp_path / self._installer_filename
self.logger.info(f"Downloading Qt installer to {installer_path}")
self.download_installer(installer_path, Settings.qt_installer_timeout)
timeout = (Settings.connection_timeout, Settings.response_timeout)
download_installer(self.base_url, self._installer_filename, self.os_name, installer_path, timeout)
try:
if self.override:

View File

@@ -667,3 +667,15 @@ def extract_auth(args: List[str]) -> Tuple[str | None, str | None, List[str] | N
continue
i += 1
return username, password, args
def download_installer(base_url: str, installer_filename:str, os_name:str, target_path: Path, timeout: Tuple[float, float]) -> None:
base_path = f"official_releases/online_installers/{installer_filename}"
url = f"{base_url}/{base_path}"
try:
hash = get_hash(base_path, Settings.hash_algorithm, timeout)
downloadBinaryFile(url, target_path, Settings.hash_algorithm, hash, timeout=timeout)
if os_name != "windows":
os.chmod(target_path, 0o500)
except Exception as e:
raise RuntimeError(f"Failed to download installer: {e}")

View File

@@ -1,7 +1,7 @@
#!/usr/bin/env python3
#
# Copyright (C) 2018 Linus Jahn <lnj@kaidan.im>
# Copyright (C) 2019-2024 Hiroshi Miura <miurahr@linux.com>
# Copyright (C) 2019-2025 Hiroshi Miura <miurahr@linux.com>
# Copyright (C) 2020, Aurélien Gâteau
#
# Permission is hereby granted, free of charge, to any person obtaining a copy of
@@ -59,10 +59,10 @@ from aqt.exceptions import (
from aqt.helper import (
MyQueueListener,
Settings,
download_installer,
downloadBinaryFile,
extract_auth,
get_hash,
get_os_name,
get_qt_installer_name,
retry_on_bad_connection,
retry_on_errors,
@@ -903,20 +903,7 @@ class Cli:
try:
# Download installer
self.logger.info(f"Downloading Qt installer to {installer_path}")
base_url = Settings.baseurl
url = f"{base_url}/official_releases/online_installers/{installer_filename}"
import requests
response = requests.get(url, stream=True, timeout=Settings.qt_installer_timeout)
response.raise_for_status()
with open(installer_path, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
if get_os_name() != "windows":
os.chmod(installer_path, 0o500)
download_installer(Settings.baseurl, installer_path)
# Build command
cmd = [str(installer_path), "--accept-licenses", "--accept-obligations", "--confirm-command"]

View File

@@ -10,7 +10,7 @@ import requests
from aqt.commercial import CommercialInstaller, QtPackageInfo, QtPackageManager
from aqt.exceptions import DiskAccessNotPermitted
from aqt.helper import Settings, get_qt_account_path
from aqt.helper import Settings, download_installer, get_qt_account_path
from aqt.installer import Cli
from aqt.metadata import Version
from tests.test_helper import mocked_requests_get
@@ -219,7 +219,10 @@ def test_commercial_installer_download_sha256(tmp_path, monkeypatch, commercial_
monkeypatch.setattr(requests.Session, "get", mocked_requests_get)
target_path = tmp_path / "qt-installer"
commercial_installer.download_installer(target_path, timeout=Settings.qt_installer_timeout)
timeout = (Settings.connection_timeout, Settings.response_timeout)
download_installer(commercial_installer.base_url, commercial_installer._installer_filename,
commercial_installer.os_name, target_path, timeout)
assert target_path.exists()
@@ -284,7 +287,7 @@ def test_install_qt_commercial(
return CompletedProcess(args=args[0], returncode=0)
def mock_get_default_local_cache_path(*args, **kwargs):
return tmp_path.joinpath('cache')
return tmp_path.joinpath("cache")
monkeypatch.setattr("aqt.commercial.safely_run", mock_safely_run)
monkeypatch.setattr("aqt.helper.get_default_local_cache_path", mock_get_default_local_cache_path)