Adjust against mypy type errors

aqt/archives.py

@@ -248,7 +248,6 @@ class Updates:
     def _get_text(self, item) -> str:
         if item is not None and item.text is not None:
             return item.text
-        else:
         return ""
 
     def _get_list(self, item) -> Iterable[str]:
@@ -407,10 +406,11 @@ class QtArchives:
         for packageupdate in package_updates:
             if not self.all_extra:
                 target_packages.remove_module_for_package(packageupdate.name)
-            should_filter_archives: bool = self.subarchives and self.should_filter_archives(packageupdate.name)
+            should_filter_archives: bool = (self.subarchives is not None) and self.should_filter_archives(packageupdate.name)
 
             for archive in packageupdate.downloadable_archives:
                 archive_name = archive.split("-", maxsplit=1)[0]
+                if self.subarchives is not None:
                     if should_filter_archives and archive_name not in self.subarchives:
                         continue
                 archive_path = posixpath.join(
@@ -508,7 +508,7 @@ class SrcDocExamplesArchives(QtArchives):
 
     def __init__(
         self,
-        flavor,
+        flavor: str,
         os_name,
         target,
         version,
@@ -519,7 +519,7 @@ class SrcDocExamplesArchives(QtArchives):
         is_include_base_package: bool = True,
         timeout=(5, 5),
     ):
-        self.flavor = flavor
+        self.flavor: str = flavor
         self.target = target
         self.os_name = os_name
         self.base = base
@@ -584,7 +584,7 @@ class ToolArchives(QtArchives):
         tool_name: str,
         base: str,
         version_str: Optional[str] = None,
-        arch: Optional[str] = None,
+        arch: str = "",
         timeout: Tuple[int, int] = (5, 5),
     ):
         self.tool_name = tool_name

aqt/helper.py

@@ -27,10 +27,10 @@ import os
 import posixpath
 import secrets
 import sys
-from logging import getLogger
+from logging import Handler, getLogger
 from logging.handlers import QueueListener
 from pathlib import Path
-from typing import Callable, Dict, Generator, List, Optional, Tuple
+from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Union
 from urllib.parse import urlparse
 from xml.etree.ElementTree import Element
 
@@ -142,7 +142,7 @@ def downloadBinaryFile(url: str, out: Path, hash_algo: str, exp: bytes, timeout)
                 )
 
 
-def retry_on_errors(action: Callable[[], any], acceptable_errors: Tuple, num_retries: int, name: str):
+def retry_on_errors(action: Callable[[], Any], acceptable_errors: Tuple, num_retries: int, name: str):
     logger = getLogger("aqt.helper")
     for i in range(num_retries):
         try:
@@ -158,7 +158,7 @@ def retry_on_errors(action: Callable[[], any], acceptable_errors: Tuple, num_ret
             raise e from e
 
 
-def retry_on_bad_connection(function: Callable[[str], any], base_url: str):
+def retry_on_bad_connection(function: Callable[[str], Any], base_url: str):
     logger = getLogger("aqt.helper")
     fallback_url = secrets.choice(Settings.fallbacks)
     try:
@@ -250,7 +250,7 @@ def altlink(url: str, alt: str):
 
 class MyQueueListener(QueueListener):
     def __init__(self, queue):
-        handlers = []
+        handlers: List[Handler] = []
         super().__init__(queue, *handlers)
 
     def handle(self, record):
@@ -284,9 +284,9 @@ def xml_to_modules(
         parsed_xml = ElementTree.fromstring(xml_text)
     except ElementTree.ParseError as perror:
         raise ArchiveListError(f"Downloaded metadata is corrupted. {perror}") from perror
-    packages = {}
+    packages: Dict[str, Dict[str, str]] = {}
     for packageupdate in parsed_xml.iter("PackageUpdate"):
-        if predicate and not predicate(packageupdate):
+        if not predicate(packageupdate):
             continue
         name = packageupdate.find("Name").text
         packages[name] = {}

aqt/installer.py

@@ -257,9 +257,8 @@ class Cli:
             self._warn_on_deprecated_command("install", "install-qt")
         target: str = args.target
         os_name: str = args.host
-        arch: str = self._set_arch(
+        qt_version_or_spec: str = getattr(args, "qt_version", getattr(args, "qt_version_spec", ""))
-            args.arch, os_name, target, getattr(args, "qt_version", getattr(args, "qt_version_spec", None))
+        arch: str = self._set_arch(args.arch, os_name, target, qt_version_or_spec)
-        )
         keep: bool = args.keep or Settings.always_keep_archives
         archive_dest: Optional[str] = args.archive_dest
         output_dir = args.outputdir
@@ -288,7 +287,7 @@ class Cli:
         if hasattr(args, "qt_version_spec"):
             qt_version: str = str(Cli._determine_qt_version(args.qt_version_spec, os_name, target, arch, base_url=base))
         else:
-            qt_version: str = args.qt_version
+            qt_version = args.qt_version
             Cli._validate_version_str(qt_version)
         archives = args.archives
         if args.noarchives:
@@ -582,7 +581,7 @@ class Cli:
         is_fetch_modules: bool = getattr(args, "modules", False)
         meta = MetadataFactory(
             archive_id=ArchiveId("qt", args.host, target),
-            src_doc_examples_query=(cmd_type, version, is_fetch_modules),
+            src_doc_examples_query=MetadataFactory.SrcDocExamplesQuery(cmd_type, version, is_fetch_modules),
         )
         show_list(meta)
 

aqt/metadata.py

@@ -28,7 +28,7 @@ from abc import ABC, abstractmethod
 from functools import reduce
 from logging import getLogger
 from pathlib import Path
-from typing import Callable, Dict, Generator, Iterable, Iterator, List, Optional, Tuple, Union
+from typing import Callable, Dict, Generator, Iterable, Iterator, List, NamedTuple, Optional, Set, Tuple, Union, cast
 from urllib.parse import ParseResult, urlparse
 from xml.etree.ElementTree import Element
 
@@ -135,11 +135,11 @@ class Versions:
         versions: Union[None, Version, Iterable[Tuple[int, Iterable[Version]]]],
     ):
         if versions is None:
-            self.versions = list()
+            self.versions: List[List[Version]] = list()
         elif isinstance(versions, Version):
             self.versions = [[versions]]
         else:
-            self.versions: List[List[Version]] = [list(versions_iterator) for _, versions_iterator in versions]
+            self.versions = [list(versions_iterator) for _, versions_iterator in versions]
 
     def __str__(self) -> str:
         return str(self.versions)
@@ -494,6 +494,12 @@ class QtRepoProperty:
 class MetadataFactory:
     """Retrieve metadata of Qt variations, versions, and descriptions from Qt site."""
 
+    Metadata = Union[List[str], Versions, ToolData, ModuleData]
+    Action = Callable[[], Metadata]
+    SrcDocExamplesQuery = NamedTuple(
+        "SrcDocExamplesQuery", [("cmd_type", str), ("version", Version), ("is_modules_query", bool)]
+    )
+
     def __init__(
         self,
         archive_id: ArchiveId,
@@ -504,7 +510,7 @@ class MetadataFactory:
         modules_query: Optional[Tuple[str, str]] = None,
         architectures_ver: Optional[str] = None,
         archives_query: Optional[List[str]] = None,
-        src_doc_examples_query: Optional[Tuple[str, Version, bool]] = None,
+        src_doc_examples_query: Optional[SrcDocExamplesQuery] = None,
         tool_name: Optional[str] = None,
         is_long_listing: bool = False,
     ):
@@ -527,15 +533,14 @@ class MetadataFactory:
         self.base_url = base_url
 
         if archive_id.is_tools():
-            if tool_name:
+            if tool_name is not None:
-                if not tool_name.startswith("tools_"):
+                _tool_name: str = "tools_" + tool_name if not tool_name.startswith("tools_") else tool_name
-                    tool_name = "tools_" + tool_name
                 if is_long_listing:
                     self.request_type = "tool long listing"
-                    self._action = lambda: self.fetch_tool_long_listing(tool_name)
+                    self._action: MetadataFactory.Action = lambda: self.fetch_tool_long_listing(_tool_name)
                 else:
                     self.request_type = "tool variant names"
-                    self._action = lambda: self.fetch_tool_modules(tool_name)
+                    self._action = lambda: self.fetch_tool_modules(_tool_name)
             else:
                 self.request_type = "tools"
                 self._action = self.fetch_tools
@@ -551,28 +556,29 @@ class MetadataFactory:
                 self.request_type = "modules"
                 version, arch = modules_query
                 self._action = lambda: self.fetch_modules(self._to_version(version, arch), arch)
-        elif architectures_ver:
+        elif architectures_ver is not None:
+            ver_str: str = architectures_ver
             self.request_type = "architectures"
-            self._action = lambda: self.fetch_arches(self._to_version(architectures_ver, None))
+            self._action = lambda: self.fetch_arches(self._to_version(ver_str, None))
         elif archives_query:
             if len(archives_query) < 2:
                 raise CliInputError("The '--archives' flag requires a 'QT_VERSION' and an 'ARCHITECTURE' parameter.")
             self.request_type = "archives for modules" if len(archives_query) > 2 else "archives for qt"
             version, arch, modules = archives_query[0], archives_query[1], archives_query[2:]
             self._action = lambda: self.fetch_archives(self._to_version(version, arch), arch, modules)
-        elif src_doc_examples_query:
+        elif src_doc_examples_query is not None:
-            cmd_type, version, is_modules_query = src_doc_examples_query
+            q: MetadataFactory.SrcDocExamplesQuery = src_doc_examples_query
-            if is_modules_query:
+            if q.is_modules_query:
-                self.request_type = f"modules for {cmd_type}"
+                self.request_type = f"modules for {q.cmd_type}"
-                self._action = lambda: self.fetch_modules_sde(cmd_type, version)
+                self._action = lambda: self.fetch_modules_sde(q.cmd_type, q.version)
             else:
-                self.request_type = f"archives for {cmd_type}"
+                self.request_type = f"archives for {q.cmd_type}"
-                self._action = lambda: self.fetch_archives_sde(cmd_type, version)
+                self._action = lambda: self.fetch_archives_sde(q.cmd_type, q.version)
         else:
             self.request_type = "versions"
             self._action = self.fetch_versions
 
-    def getList(self) -> Union[List[str], Versions, ToolData]:
+    def getList(self) -> Metadata:
         return self._action()
 
     def fetch_arches(self, version: Version) -> List[str]:
@@ -591,15 +597,13 @@ class MetadataFactory:
     def fetch_versions(self, extension: str = "") -> Versions:
         def filter_by(ver_ext: Tuple[Optional[Version], str]) -> bool:
             version, ext = ver_ext
-            return version and (self.spec is None or version in self.spec) and (ext == extension)
+            return version is not None and (self.spec is None or version in self.spec) and (ext == extension)
-
-        def get_version(ver_ext: Tuple[Version, str]):
-            return ver_ext[0]
 
         versions_extensions = self.get_versions_extensions(
             self.fetch_http(self.archive_id.to_url(), False), self.archive_id.category
         )
-        versions = sorted(filter(None, map(get_version, filter(filter_by, versions_extensions))))
+        opt_versions = map(lambda _tuple: _tuple[0], filter(filter_by, versions_extensions))
+        versions: List[Version] = sorted(filter(None, opt_versions))
         iterables = itertools.groupby(versions, lambda version: version.minor)
         return Versions(iterables)
 
@@ -635,7 +639,7 @@ class MetadataFactory:
             return None
 
         # Remove items that don't conform to simple_spec
-        tools_versions = filter(lambda tool_item: tool_item[2] in simple_spec, tools_versions)
+        tools_versions = list(filter(lambda tool_item: tool_item[2] in simple_spec, tools_versions))
 
         try:
             # Return the conforming item with the highest version.
@@ -674,22 +678,25 @@ class MetadataFactory:
         timeout = (Settings.connection_timeout, Settings.response_timeout)
         expected_hash = get_hash(rest_of_url, "sha256", timeout) if is_check_hash else None
         base_urls = self.base_url, random.choice(Settings.fallbacks)
+
+        err: BaseException = AssertionError("unraisable")
+
         for i, base_url in enumerate(base_urls):
             try:
                 url = posixpath.join(base_url, rest_of_url)
                 return getUrl(url=url, timeout=timeout, expected_hash=expected_hash)
 
             except (ArchiveDownloadError, ArchiveConnectionError) as e:
-                if i == len(base_urls) - 1:
+                err = e
-                    raise e from e
+                if i < len(base_urls) - 1:
-                else:
                     getLogger("aqt.metadata").debug(
                         f"Connection to '{base_url}' failed. Retrying with fallback '{base_urls[i + 1]}'."
                     )
+        raise err from err
 
     def iterate_folders(self, html_doc: str, html_url: str, *, filter_category: str = "") -> Generator[str, None, None]:
         def link_to_folder(link: bs4.element.Tag) -> str:
-            raw_url: str = link.get("href", default="")
+            raw_url: str = str(link.get("href", default=""))
             url: ParseResult = urlparse(raw_url)
             if url.scheme or url.netloc:
                 return ""
@@ -738,7 +745,7 @@ class MetadataFactory:
         if downloads is None or update_file is None:
             return False
         uncompressed_size = int(update_file.attrib["UncompressedSize"])
-        return downloads.text and uncompressed_size >= Settings.min_module_size
+        return downloads.text is not None and uncompressed_size >= Settings.min_module_size
 
     def _get_qt_version_str(self, version: Version) -> str:
         """Returns a Qt version, without dots, that works in the Qt repo urls and Updates.xml files"""
@@ -778,13 +785,20 @@ class MetadataFactory:
                 module = module[len("addons.") :]
             return module, arch
 
-        modules = set()
+        modules: Set[str] = set()
         for name in modules_meta.keys():
             module, _arch = to_module_arch(name)
             if _arch == arch:
-                modules.add(module)
+                modules.add(cast(str, module))
         return sorted(modules)
 
+    @staticmethod
+    def require_text(element: Element, key: str) -> str:
+        node = element.find(key)
+        if node is None:
+            raise ArchiveListError(f"Downloaded metadata does not match the expected structure. Missing key: {key}")
+        return node.text or ""
+
     def fetch_long_modules(self, version: Version, arch: str) -> ModuleData:
         """Returns long listing of modules"""
         extension = QtRepoProperty.extension_for_arch(arch, version >= Version("6.0.0"))
@@ -801,11 +815,16 @@ class MetadataFactory:
         )
 
         def matches_arch(element: Element) -> bool:
-            name_node = element.find("Name")
+            return bool(pattern.match(MetadataFactory.require_text(element, "Name")))
-            return bool(name_node is not None) and bool(pattern.match(str(name_node.text)))
 
         modules_meta = self._fetch_module_metadata(self.archive_id.to_folder(qt_ver_str, extension), matches_arch)
-        m = {pattern.match(key).group("module"): value for key, value in modules_meta.items()}
+        m: Dict[str, Dict[str, str]] = {}
+        for key, value in modules_meta.items():
+            match = pattern.match(key)
+            if match is not None:
+                module = match.group("module")
+                if module is not None:
+                    m[module] = value
 
         return ModuleData(m)
 
@@ -839,23 +858,23 @@ class MetadataFactory:
         nonempty = MetadataFactory._has_nonempty_downloads
 
         def all_modules(element: Element) -> bool:
-            _module, _arch = element.find("Name").text.split(".")[-2:]
+            _module, _arch = MetadataFactory.require_text(element, "Name").split(".")[-2:]
             return _arch == arch and _module != qt_version_str and nonempty(element)
 
         def specify_modules(element: Element) -> bool:
-            _module, _arch = element.find("Name").text.split(".")[-2:]
+            _module, _arch = MetadataFactory.require_text(element, "Name").split(".")[-2:]
             return _arch == arch and _module in modules and nonempty(element)
 
         def no_modules(element: Element) -> bool:
-            name: Optional[str] = element.find("Name").text
+            name: Optional[str] = getattr(element.find("Name"), "text", None)
-            return name and name.endswith(f".{qt_version_str}.{arch}") and nonempty(element)
+            return name is not None and name.endswith(f".{qt_version_str}.{arch}") and nonempty(element)
 
         predicate = no_modules if not modules else all_modules if "all" in modules else specify_modules
         try:
             mod_metadata = self._fetch_module_metadata(
                 self.archive_id.to_folder(qt_version_str, extension), predicate=predicate
             )
-        except (AttributeError,) as e:
+        except (AttributeError, ValueError) as e:
             raise ArchiveListError(f"Downloaded metadata is corrupted. {e}") from e
 
         # Did we find all requested modules?

aqt/updater.py

@@ -23,7 +23,7 @@ import os
 import subprocess
 from logging import getLogger
 from pathlib import Path
-from typing import Optional
+from typing import Dict, Optional
 
 import patch
 
@@ -43,8 +43,8 @@ class Updater:
     def __init__(self, prefix: Path, logger):
         self.logger = logger
         self.prefix = prefix
-        self.qmake_path = None
+        self.qmake_path: Optional[Path] = None
-        self.qconfigs = {}
+        self.qconfigs: Dict[str, str] = {}
 
     def _patch_binfile(self, file: Path, key: bytes, newpath: bytes):
         """Patch binary file with key/value"""
@@ -149,6 +149,8 @@ class Updater:
     def patch_qmake(self):
         """Patch to qmake binary"""
         if self._detect_qmake():
+            if self.qmake_path is None:
+                return
             self.logger.info("Patching {}".format(str(self.qmake_path)))
             self._patch_binfile(
                 self.qmake_path,

pyproject.toml

@@ -151,6 +151,12 @@ warn_return_any = true
 warn_unreachable = true
 warn_unused_ignores = true
 
+# TODO: Remove this `ignore_missing_imports` and add type stubs.
+# See https://mypy.readthedocs.io/en/stable/running_mypy.html#missing-imports
+[[tool.mypy.overrides]]
+module = "texttable"
+ignore_missing_imports = true
+
 [tool.pytest.ini_options]
 minversion = "6.0"
 testpaths = ["tests"]

tests/test_list.py

@@ -342,16 +342,19 @@ def test_list_archives_insufficient_args(capsys):
     assert err.strip() == "ERROR   : The '--archives' flag requires a 'QT_VERSION' and an 'ARCHITECTURE' parameter."
 
 
-def test_list_archives_bad_xml(monkeypatch):
+@pytest.mark.parametrize(
+    "xml_content",
+    (
+        "<Updates><PackageUpdate><badname></badname></PackageUpdate></Updates>",
+        "<Updates><PackageUpdate><Name></Name></PackageUpdate></Updates>",
+        "<Updates></PackageUpdate><PackageUpdate></Updates><Name></Name>",
+    ),
+)
+def test_list_archives_bad_xml(monkeypatch, xml_content: str):
     archive_id = ArchiveId("qt", "windows", "desktop")
     archives_query = ["5.15.2", "win32_mingw81", "qtcharts"]
 
-    xml_no_name = "<Updates><PackageUpdate><badname></badname></PackageUpdate></Updates>"
+    monkeypatch.setattr(MetadataFactory, "fetch_http", lambda self, _: xml_content)
-    xml_empty_name = "<Updates><PackageUpdate><Name></Name></PackageUpdate></Updates>"
-    xml_broken = "<Updates></PackageUpdate><PackageUpdate></Updates><Name></Name>"
-
-    for _xml in (xml_no_name, xml_empty_name, xml_broken):
-        monkeypatch.setattr(MetadataFactory, "fetch_http", lambda self, _: _xml)
     with pytest.raises(ArchiveListError) as e:
         MetadataFactory(archive_id, archives_query=archives_query).getList()
     assert e.type == ArchiveListError