diff --git a/aqt/installer.py b/aqt/installer.py index 2a8684d8..6646d279 100644 --- a/aqt/installer.py +++ b/aqt/installer.py @@ -660,16 +660,34 @@ def run_list_src_doc_examples(self, args: ListArgumentParser, cmd_type: str): ) show_list(meta) - def run_install_qt_commercial(self, args): + def run_install_qt_commercial(self, args: list[str]) -> None: """Execute commercial Qt installation""" self.show_aqt_version() + if args.base is not None: + base = args.base + else: + base = Settings.baseurl + if args.timeout is not None: + timeout = args.timeout + else: + timeout = Settings.response_timeout + target = args.target arch = args.arch version = args.version username = args.user password = args.password output_dir = args.outputdir + timeout = timeout + base_url = base + operation_does_not_exist_error = args.operation_does_not_exist_error + overwrite_target_dir = args.overwrite_target_dir + stop_processes_for_updates = args.stop_processes_for_updates + installation_error_with_cancel = args.installation_error_with_cancel + installation_error_with_ignore = args.installation_error_with_ignore + associate_common_filetypes = args.associate_common_filetypes + telemetry = args.telemetry commercial_installer = CommercialInstaller( target=target, @@ -679,6 +697,15 @@ def run_install_qt_commercial(self, args): password=password, output_dir=output_dir, logger=self.logger, + timeout=timeout, + base_url=base_url, + operation_does_not_exist_error=operation_does_not_exist_error, + overwrite_target_dir=overwrite_target_dir, + stop_processes_for_updates=stop_processes_for_updates, + installation_error_with_cancel=installation_error_with_cancel, + installation_error_with_ignore=installation_error_with_ignore, + associate_common_filetypes=associate_common_filetypes, + telemetry=telemetry, ) try: @@ -697,7 +724,7 @@ def _format_aqt_version(self) -> str: py_build = platform.python_compiler() return f"aqtinstall(aqt) v{aqt.__version__} on Python {py_version} [{py_impl} {py_build}]" - def show_aqt_version(self, args=None): + def show_aqt_version(self, args: list[str] = None) -> None: """Display version information""" self.logger.info(self._format_aqt_version()) @@ -780,7 +807,7 @@ def _set_install_tool_parser(self, install_tool_parser): ) self._set_common_options(install_tool_parser) - def _set_install_qt_commercial_parser(self, install_qt_commercial_parser): + def _set_install_qt_commercial_parser(self, install_qt_commercial_parser) -> None: install_qt_commercial_parser.set_defaults(func=self.run_install_qt_commercial) install_qt_commercial_parser.add_argument( "target", @@ -803,6 +830,48 @@ def _set_install_qt_commercial_parser(self, install_qt_commercial_parser): "--password", help="Qt account password", ) + install_qt_commercial_parser.add_argument( + "--operation_does_not_exist_error", + choices=["Abort", "Ignore"], + default="Ignore", + help="OperationDoesNotExistError: Abort, Ignore. Default: Ignore", + ) + install_qt_commercial_parser.add_argument( + "--overwrite_target_dir", + choices=["Yes", "No"], + default="No", + help="OverwriteTargetDirectory: Yes, No. Default: No", + ) + install_qt_commercial_parser.add_argument( + "--stop_processes_for_updates", + choices=["Retry", "Ignore", "Cancel"], + default="Cancel", + help="stopProcessesForUpdates: Retry, Ignore, Cancel. Default: Cancel", + ) + install_qt_commercial_parser.add_argument( + "--installation_error_with_cancel", + choices=["Retry", "Ignore", "Cancel"], + default="Cancel", + help="installationErrorWithCancel: Retry, Ignore, Cancel. Default: Cancel", + ) + install_qt_commercial_parser.add_argument( + "--installation_error_with_ignore", + choices=["Retry", "Ignore"], + default="Ignore", + help="installationErrorWithIgnore: Retry, Ignore. Default: Ignore", + ) + install_qt_commercial_parser.add_argument( + "--associate_common_filetypes", + choices=["Yes", "No"], + default="Yes", + help="AssociateCommonFiletypes: Yes, No. Default: Yes", + ) + install_qt_commercial_parser.add_argument( + "--telemetry", + choices=["Yes", "No"], + default="No", + help="telemetry-question: Yes, No. Default: No", + ) self._set_common_options(install_qt_commercial_parser) def _warn_on_deprecated_command(self, old_name: str, new_name: str) -> None: @@ -1019,7 +1088,7 @@ def _make_common_parsers(self, subparsers: argparse._SubParsersAction): version_parser = subparsers.add_parser("version") version_parser.set_defaults(func=self.show_aqt_version) - def _set_common_options(self, subparser): + def _set_common_options(self, subparser: argparse._SubParsersAction) -> None: subparser.add_argument( "-O", "--outputdir", @@ -1389,6 +1458,15 @@ def __init__( password: Optional[str] = None, output_dir: Optional[str] = None, logger: Optional[Logger] = None, + timeout: int = 10, + base_url: str = "https://download.qt.io", + operation_does_not_exist_error="Ignore", + overwrite_target_dir: str = "Yes", + stop_processes_for_updates: str = "Cancel", + installation_error_with_cancel: str = "Cancel", + installation_error_with_ignore: str = "Ignore", + associate_common_filetypes: str = "Yes", + telemetry: str = "No", ): self.target = target self.arch = arch @@ -1397,6 +1475,29 @@ def __init__( self.password = password self.output_dir = output_dir self.logger = logger or getLogger(__name__) + self.timeout = timeout + self.base_url = base_url + + # Validate parameters + self.operation_does_not_exist_error = self._validate_auto_answer_param( + operation_does_not_exist_error, ["Abort", "Ignore"], "OperationDoesNotExistError" + ) + self.overwrite_target_dir = self._validate_auto_answer_param( + overwrite_target_dir, ["Yes", "No"], "OverwriteTargetDirectory" + ) + self.stop_processes_for_updates = self._validate_auto_answer_param( + stop_processes_for_updates, ["Retry", "Ignore", "Cancel"], "stopProcessesForUpdates" + ) + self.installation_error_with_cancel = self._validate_auto_answer_param( + installation_error_with_cancel, ["Retry", "Ignore", "Cancel"], "installationErrorWithCancel" + ) + self.installation_error_with_ignore = self._validate_auto_answer_param( + installation_error_with_ignore, ["Retry", "Ignore"], "installationErrorWithIgnore" + ) + self.associate_common_filetypes = self._validate_auto_answer_param( + associate_common_filetypes, ["Yes", "No"], "AssociateCommonFiletypes" + ) + self.telemetry = self._validate_auto_answer_param(telemetry, ["Yes", "No"], "telemetry-question") # Map platform names consistently system = platform.system() @@ -1410,6 +1511,12 @@ def __init__( self.installer_filename = self._get_installer_filename() self.qt_account = self._get_qt_account_path() + def _validate_auto_answer_param(self, value: str, valid_choices: list[str], param_name: str) -> str: + """Validate auto-answer parameter""" + if value not in valid_choices: + raise ValueError(f"Invalid value '{value}' for {param_name}. Expected one of {valid_choices}.") + return value + def _get_installer_filename(self) -> str: """Get OS-specific installer filename""" base = "qt-unified" @@ -1430,12 +1537,12 @@ def _get_qt_account_path(self) -> Path: else: return Path.home() / ".local" / "share" / "Qt" / "qtaccount.ini" - def _download_installer(self, target_path: Path): + def _download_installer(self, target_path: Path) -> None: """Download Qt online installer""" - url = f"https://download.qt.io/official_releases/online_installers/{self.installer_filename}" + url = f"{self.base_url}/official_releases/online_installers/{self.installer_filename}" try: - response = requests.get(url, stream=True) + response = requests.get(url, stream=True, timeout=self.timeout) response.raise_for_status() total = response.headers.get("content-length", 0) @@ -1447,17 +1554,27 @@ def _download_installer(self, target_path: Path): f.write(chunk) if self.os_name != "windows": - os.chmod(target_path, 0o755) + os.chmod(target_path, 0o700) + except requests.exceptions.Timeout: + print(f"Request to {url} timed out.") + raise except requests.exceptions.RequestException as e: - raise ArchiveDownloadError(f"Failed to download installer: {str(e)}") + print(f"An error occurred during download: {e}") + raise + except OSError as e: + print(f"File operation error: {e}") + raise + finally: + if not target_path.exists(): + print(f"Download of {self.installer_filename} was not completed.") def _get_package_name(self) -> str: """Convert aqt parameters to Qt package name""" qt_version = f"{self.version.major}{self.version.minor}{self.version.patch}" return f"qt.qt{self.version.major}.{qt_version}.{self.arch}" - def _get_install_command(self, installer_path: Path) -> list: + def _get_install_command(self, installer_path: Path) -> list[str]: """Build installation command""" cmd = [str(installer_path)] @@ -1470,12 +1587,27 @@ def _get_install_command(self, installer_path: Path) -> list: cmd.extend(["--root", str(self.output_dir)]) # Unattended options + auto_answer_options = [ + f"OperationDoesNotExistError={self.operation_does_not_exist_error}", + f"OverwriteTargetDirectory={self.overwrite_target_dir}", + f"stopProcessesForUpdates={self.stop_processes_for_updates}", + f"installationErrorWithCancel={self.installation_error_with_cancel}", + f"installationErrorWithIgnore={self.installation_error_with_ignore}", + f"AssociateCommonFiletypes={self.associate_common_filetypes}", + f"telemetry-question={self.telemetry}", + ] + + # Filter out invalid or empty options + auto_answer_string = ",".join(filter(None, auto_answer_options)) + cmd.extend( [ + "--unattended", "--accept-licenses", "--accept-obligations", "--confirm-command", - "--default-answer", + "--auto-answer", + auto_answer_string, "install", self._get_package_name(), ] @@ -1483,7 +1615,7 @@ def _get_install_command(self, installer_path: Path) -> list: return cmd - def install(self): + def install(self) -> None: """Run commercial installation""" # Verify auth if not self.qt_account.exists() and not (self.username and self.password): @@ -1504,7 +1636,11 @@ def install(self): self.logger.info("Starting Qt installation") cmd = self._get_install_command(installer_path) - self.logger.info(f"Running: {cmd}") + # Ensure the command is a list of trusted strings + if not all(isinstance(arg, str) for arg in cmd): + raise ValueError("Command contains non-string arguments") + + self.logger.info(f"Running: {' '.join(cmd)}") try: subprocess.check_call(cmd)