files.new_contents: #!/usr/bin/env python3
  #
  # Copyright (C) 2018 Linus Jahn <lnj@kaidan.im>
  # Copyright (C) 2019-2021 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
  # this software and associated documentation files (the "Software"), to deal in
  # the Software without restriction, including without limitation the rights to
  # use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
  # the Software, and to permit persons to whom the Software is furnished to do so,
  # subject to the following conditions:
  #
  # The above copyright notice and this permission notice shall be included in all
  # copies or substantial portions of the Software.
  #
  # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
  # FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
  # COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
  # IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
  # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  
  import argparse
  import gc
  import multiprocessing
  import os
  import platform
  import posixpath
  import signal
  import subprocess
  import sys
  import time
  from logging import getLogger
  from logging.handlers import QueueHandler
  from pathlib import Path
  from tempfile import TemporaryDirectory
  from typing import List, Optional
  
  import aqt
  from aqt.archives import QtArchives, QtPackage, SrcDocExamplesArchives, ToolArchives
  from aqt.exceptions import (
      AqtException,
      ArchiveChecksumError,
      ArchiveDownloadError,
      ArchiveExtractionError,
      ArchiveListError,
      CliInputError,
      CliKeyboardInterrupt,
      OutOfMemory,
  )
  from aqt.helper import (
      MyQueueListener,
      Settings,
      downloadBinaryFile,
      get_hash,
      retry_on_bad_connection,
      retry_on_errors,
      setup_logging,
  )
  from aqt.metadata import ArchiveId, MetadataFactory, QtRepoProperty, SimpleSpec, Version, show_list, suggested_follow_up
  from aqt.updater import Updater
  
  try:
      import py7zr
  
      EXT7Z = False
  except ImportError:
      EXT7Z = True
  
  
  class Cli:
      """CLI main class to parse command line argument and launch proper functions."""
  
      __slot__ = ["parser", "combinations", "logger"]
  
      UNHANDLED_EXCEPTION_CODE = 254
  
      def __init__(self):
          parser = argparse.ArgumentParser(
              prog="aqt",
              description="Another unofficial Qt Installer.\naqt helps you install Qt SDK, tools, examples and others\n",
              formatter_class=argparse.RawTextHelpFormatter,
              add_help=True,
          )
          parser.add_argument(
              "-c",
              "--config",
              type=argparse.FileType("r"),
              help="Configuration ini file.",
          )
          subparsers = parser.add_subparsers(
              title="subcommands",
              description="aqt accepts several subcommands:\n"
              "install-* subcommands are commands that install components\n"
              "list-* subcommands are commands that show available components\n\n"
              "commands {install|tool|src|examples|doc} are deprecated and marked for removal\n",
              help="Please refer to each help message by using '--help' with each subcommand",
          )
          self._make_all_parsers(subparsers)
          parser.set_defaults(func=self.show_help)
          self.parser = parser
  
      def run(self, arg=None) -> int:
          args = self.parser.parse_args(arg)
          self._setup_settings(args)
          try:
              args.func(args)
              return 0
          except AqtException as e:
              self.logger.error(format(e), exc_info=Settings.print_stacktrace_on_error)
              if e.should_show_help:
                  self.show_help()
              return 1
          except Exception as e:
              # If we didn't account for it, and wrap it in an AqtException, it's a bug.
              self.logger.exception(e)  # Print stack trace
              self.logger.error(
                 
  
  [... the rest is omitted ...]
files.new_path: aqt/installer.py
files.old_contents: #!/usr/bin/env python3
  #
  # Copyright (C) 2018 Linus Jahn <lnj@kaidan.im>
  # Copyright (C) 2019-2021 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
  # this software and associated documentation files (the "Software"), to deal in
  # the Software without restriction, including without limitation the rights to
  # use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
  # the Software, and to permit persons to whom the Software is furnished to do so,
  # subject to the following conditions:
  #
  # The above copyright notice and this permission notice shall be included in all
  # copies or substantial portions of the Software.
  #
  # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
  # FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
  # COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
  # IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
  # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  
  import argparse
  import gc
  import multiprocessing
  import os
  import platform
  import posixpath
  import signal
  import subprocess
  import sys
  import time
  from logging import getLogger
  from logging.handlers import QueueHandler
  from pathlib import Path
  from tempfile import TemporaryDirectory
  from typing import List, Optional
  
  import aqt
  from aqt.archives import QtArchives, QtPackage, SrcDocExamplesArchives, ToolArchives
  from aqt.exceptions import (
      AqtException,
      ArchiveChecksumError,
      ArchiveDownloadError,
      ArchiveExtractionError,
      ArchiveListError,
      CliInputError,
      CliKeyboardInterrupt,
      OutOfMemory,
  )
  from aqt.helper import (
      MyQueueListener,
      Settings,
      downloadBinaryFile,
      get_hash,
      retry_on_bad_connection,
      retry_on_errors,
      setup_logging,
  )
  from aqt.metadata import ArchiveId, MetadataFactory, QtRepoProperty, SimpleSpec, Version, show_list, suggested_follow_up
  from aqt.updater import Updater
  
  try:
      import py7zr
  
      EXT7Z = False
  except ImportError:
      EXT7Z = True
  
  
  class Cli:
      """CLI main class to parse command line argument and launch proper functions."""
  
      __slot__ = ["parser", "combinations", "logger"]
  
      UNHANDLED_EXCEPTION_CODE = 254
  
      def __init__(self):
          parser = argparse.ArgumentParser(
              prog="aqt",
              description="Another unofficial Qt Installer.\naqt helps you install Qt SDK, tools, examples and others\n",
              formatter_class=argparse.RawTextHelpFormatter,
              add_help=True,
          )
          parser.add_argument(
              "-c",
              "--config",
              type=argparse.FileType("r"),
              help="Configuration ini file.",
          )
          subparsers = parser.add_subparsers(
              title="subcommands",
              description="aqt accepts several subcommands:\n"
              "install-* subcommands are commands that install components\n"
              "list-* subcommands are commands that show available components\n\n"
              "commands {install|tool|src|examples|doc} are deprecated and marked for removal\n",
              help="Please refer to each help message by using '--help' with each subcommand",
          )
          self._make_all_parsers(subparsers)
          parser.set_defaults(func=self.show_help)
          self.parser = parser
  
      def run(self, arg=None) -> int:
          args = self.parser.parse_args(arg)
          self._setup_settings(args)
          try:
              args.func(args)
              return 0
          except AqtException as e:
              self.logger.error(format(e), exc_info=Settings.print_stacktrace_on_error)
              if e.should_show_help:
                  self.show_help()
              return 1
          except Exception as e:
              # If we didn't account for it, and wrap it in an AqtException, it's a bug.
              self.logger.exception(e)  # Print stack trace
              self.logger.error(
                 
  
  [... the rest is omitted ...]
files.old_path: aqt/installer.py
files.new_contents: import hashlib
  import logging
  import os
  import re
  import subprocess
  import sys
  import textwrap
  from dataclasses import dataclass
  from datetime import datetime
  from pathlib import Path
  from tempfile import TemporaryDirectory
  from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple
  
  import py7zr
  import pytest
  
  from aqt.archives import QtPackage
  from aqt.exceptions import ArchiveDownloadError, ArchiveExtractionError
  from aqt.helper import Settings
  from aqt.installer import Cli, installer
  
  
  class MockMultiprocessingContext:
      """
      By default, monkeypatch will only patch objects in the main process.
      When multiprocessing is used, child processes will not be patched.
      This class forces all work to be done in the main process, so that all
      patched objects remain patched.
  
      NOTE: This probably isn't the right way to solve the problem, but it is
      the only way I have found to solve it.
      """
  
      def __init__(self, *args):
          pass
  
      class Pool:
          def __init__(self, *args):
              pass
  
          def starmap(self, func: Callable, func_args: List[Tuple], *args):
              for set_of_args in func_args:
                  func(*set_of_args)
  
          def close(self):
              pass
  
          def join(self):
              pass
  
          def terminate(self):
              pass
  
  
  class MockMultiprocessingManager:
      class Queue:
          def __init__(self, *args):
              pass
  
          def put_nowait(self, log_record: Optional[logging.LogRecord]):
              # NOTE: This is certainly not the right way to do this, but it works locally
              if not log_record or log_record.levelno < logging.INFO:
                  return
              print(log_record.message, file=sys.stderr)
  
          def get(self, *args):
              return None
  
  
  GET_URL_TYPE = Callable[[str, Any], str]
  DOWNLOAD_ARCHIVE_TYPE = Callable[[str, str, str, bytes, Any], None]
  LIST_OF_FILES_AND_CONTENTS = Iterable[Dict[str, str]]
  
  
  @dataclass
  class PatchedFile:
      filename: str
      unpatched_content: str
      patched_content: Optional[str]
  
      def expected_content(self, **kwargs) -> str:
          if not self.patched_content:
              return self.unpatched_content
          return self.patched_content.format(**kwargs)
  
  
  @dataclass
  class MockArchive:
      filename_7z: str
      update_xml_name: str
      contents: Iterable[PatchedFile]
      version: str = ""
      arch_dir: str = ""
      should_install: bool = True
      date: datetime = datetime.now()
  
      def xml_package_update(self) -> str:
          return textwrap.dedent(
              f"""\
               <PackageUpdate>
                <Name>{self.update_xml_name}</Name>
                <Version>{self.version}-0-{self.date.strftime("%Y%m%d%H%M")}</Version>
                <Description>none</Description>
                <DownloadableArchives>{self.filename_7z}</DownloadableArchives>
               </PackageUpdate>"""
          )
  
      def write_compressed_archive(self, dest: Path) -> None:
          with TemporaryDirectory() as temp_dir, py7zr.SevenZipFile(dest / self.filename_7z, "w") as archive:
              temp_path = Path(temp_dir)
  
              for folder in ("bin", "lib", "mkspecs"):
                  (temp_path / self.arch_dir / folder).mkdir(parents=True, exist_ok=True)
  
              # Use `self.contents` to write qmake binary, qmake script, QtCore binaries, etc
              for patched_file in self.contents:
                  full_path = temp_path / self.arch_dir / patched_file.filename
                  if not full_path.parent.exists():
                      full_path.parent.mkdir(parents=True)
                  full_path.write_text(patched_file.unpatched_content, "utf_8")
  
              archive_name = "5.9" if self.version == "5.9.0" else self.version
              archive.writeall(path=temp_path, arcname=archive_name)
  
  
  def make_mock_geturl_download_archive(
      archives: Iterable[MockArchive],
      arch: str,
      os_name: str,
      updates_url: str,
  ) -> Tuple[GET_URL_TYPE, DOWNLOAD_ARCHIVE_TYPE]:
      """
      Returns a mock 'getUrl' and a mock 'downloadArchive'
  
  [... the rest is omitted ...]
files.new_path: tests/test_install.py
files.old_contents: import hashlib
  import logging
  import os
  import re
  import subprocess
  import sys
  import textwrap
  from dataclasses import dataclass
  from datetime import datetime
  from pathlib import Path
  from tempfile import TemporaryDirectory
  from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple
  
  import py7zr
  import pytest
  
  from aqt.archives import QtPackage
  from aqt.exceptions import ArchiveDownloadError, ArchiveExtractionError
  from aqt.helper import Settings
  from aqt.installer import Cli, installer
  
  
  class MockMultiprocessingContext:
      """
      By default, monkeypatch will only patch objects in the main process.
      When multiprocessing is used, child processes will not be patched.
      This class forces all work to be done in the main process, so that all
      patched objects remain patched.
  
      NOTE: This probably isn't the right way to solve the problem, but it is
      the only way I have found to solve it.
      """
  
      def __init__(self, *args):
          pass
  
      class Pool:
          def __init__(self, *args):
              pass
  
          def starmap(self, func: Callable, func_args: List[Tuple], *args):
              for set_of_args in func_args:
                  func(*set_of_args)
  
          def close(self):
              pass
  
          def join(self):
              pass
  
          def terminate(self):
              pass
  
  
  class MockMultiprocessingManager:
      class Queue:
          def __init__(self, *args):
              pass
  
          def put_nowait(self, log_record: Optional[logging.LogRecord]):
              # NOTE: This is certainly not the right way to do this, but it works locally
              if not log_record or log_record.levelno < logging.INFO:
                  return
              print(log_record.message, file=sys.stderr)
  
          def get(self, *args):
              return None
  
  
  GET_URL_TYPE = Callable[[str, Any], str]
  DOWNLOAD_ARCHIVE_TYPE = Callable[[str, str, str, bytes, Any], None]
  LIST_OF_FILES_AND_CONTENTS = Iterable[Dict[str, str]]
  
  
  @dataclass
  class PatchedFile:
      filename: str
      unpatched_content: str
      patched_content: Optional[str]
  
      def expected_content(self, **kwargs) -> str:
          if not self.patched_content:
              return self.unpatched_content
          return self.patched_content.format(**kwargs)
  
  
  @dataclass
  class MockArchive:
      filename_7z: str
      update_xml_name: str
      contents: Iterable[PatchedFile]
      version: str = ""
      arch_dir: str = ""
      should_install: bool = True
      date: datetime = datetime.now()
  
      def xml_package_update(self) -> str:
          return textwrap.dedent(
              f"""\
               <PackageUpdate>
                <Name>{self.update_xml_name}</Name>
                <Version>{self.version}-0-{self.date.strftime("%Y%m%d%H%M")}</Version>
                <Description>none</Description>
                <DownloadableArchives>{self.filename_7z}</DownloadableArchives>
               </PackageUpdate>"""
          )
  
      def write_compressed_archive(self, dest: Path) -> None:
          with TemporaryDirectory() as temp_dir, py7zr.SevenZipFile(dest / self.filename_7z, "w") as archive:
              temp_path = Path(temp_dir)
  
              for folder in ("bin", "lib", "mkspecs"):
                  (temp_path / self.arch_dir / folder).mkdir(parents=True, exist_ok=True)
  
              # Use `self.contents` to write qmake binary, qmake script, QtCore binaries, etc
              for patched_file in self.contents:
                  full_path = temp_path / self.arch_dir / patched_file.filename
                  if not full_path.parent.exists():
                      full_path.parent.mkdir(parents=True)
                  full_path.write_text(patched_file.unpatched_content, "utf_8")
  
              archive_name = "5.9" if self.version == "5.9.0" else self.version
              archive.writeall(path=temp_path, arcname=archive_name)
  
  
  def make_mock_geturl_download_archive(
      archives: Iterable[MockArchive],
      arch: str,
      os_name: str,
      updates_url: str,
  ) -> Tuple[GET_URL_TYPE, DOWNLOAD_ARCHIVE_TYPE]:
      """
      Returns a mock 'getUrl' and a mock 'downloadArchive'
  
  [... the rest is omitted ...]
files.old_path: tests/test_install.py