diff --git a/aqt/archives.py b/aqt/archives.py
index 4aadcbe..587d5ef 100644
--- a/aqt/archives.py
+++ b/aqt/archives.py
@@ -248,8 +248,7 @@ class Updates:
def _get_text(self, item) -> str:
if item is not None and item.text is not None:
return item.text
- else:
- return ""
+ return ""
def _get_list(self, item) -> Iterable[str]:
if item is not None and item.text is not None:
@@ -407,12 +406,13 @@ 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 should_filter_archives and archive_name not in self.subarchives:
- continue
+ if self.subarchives is not None:
+ if should_filter_archives and archive_name not in self.subarchives:
+ continue
archive_path = posixpath.join(
# online/qtsdkrepository/linux_x64/desktop/qt5_5150/
os_target_folder,
@@ -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
diff --git a/aqt/helper.py b/aqt/helper.py
index bd524c6..e88d31d 100644
--- a/aqt/helper.py
+++ b/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] = {}
diff --git a/aqt/installer.py b/aqt/installer.py
index 1f5584c..e2259f8 100644
--- a/aqt/installer.py
+++ b/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(
- args.arch, os_name, target, getattr(args, "qt_version", getattr(args, "qt_version_spec", None))
- )
+ qt_version_or_spec: str = getattr(args, "qt_version", getattr(args, "qt_version_spec", ""))
+ 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)
diff --git a/aqt/metadata.py b/aqt/metadata.py
index 6ae5128..93cfba9 100644
--- a/aqt/metadata.py
+++ b/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 not tool_name.startswith("tools_"):
- tool_name = "tools_" + tool_name
+ if tool_name is not None:
+ _tool_name: str = "tools_" + tool_name if not tool_name.startswith("tools_") else 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:
- cmd_type, version, is_modules_query = src_doc_examples_query
- if is_modules_query:
- self.request_type = f"modules for {cmd_type}"
- self._action = lambda: self.fetch_modules_sde(cmd_type, version)
+ elif src_doc_examples_query is not None:
+ q: MetadataFactory.SrcDocExamplesQuery = src_doc_examples_query
+ if q.is_modules_query:
+ self.request_type = f"modules for {q.cmd_type}"
+ self._action = lambda: self.fetch_modules_sde(q.cmd_type, q.version)
else:
- self.request_type = f"archives for {cmd_type}"
- self._action = lambda: self.fetch_archives_sde(cmd_type, version)
+ self.request_type = f"archives for {q.cmd_type}"
+ 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)
-
- def get_version(ver_ext: Tuple[Version, str]):
- return ver_ext[0]
+ return version is not None and (self.spec is None or version in self.spec) and (ext == extension)
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:
- raise e from e
- else:
+ err = e
+ if i < len(base_urls) - 1:
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(name_node is not None) and bool(pattern.match(str(name_node.text)))
+ return bool(pattern.match(MetadataFactory.require_text(element, "Name")))
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
- return name and name.endswith(f".{qt_version_str}.{arch}") and nonempty(element)
+ name: Optional[str] = getattr(element.find("Name"), "text", None)
+ 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?
diff --git a/aqt/updater.py b/aqt/updater.py
index 9667bfe..60a31c0 100644
--- a/aqt/updater.py
+++ b/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.qconfigs = {}
+ self.qmake_path: Optional[Path] = None
+ 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,
diff --git a/pyproject.toml b/pyproject.toml
index 9970b35..831e602 100644
--- a/pyproject.toml
+++ b/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"]
diff --git a/tests/test_list.py b/tests/test_list.py
index 0d74c64..83a8085 100644
--- a/tests/test_list.py
+++ b/tests/test_list.py
@@ -342,19 +342,22 @@ 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",
+ (
+ "",
+ "",
+ "",
+ ),
+)
+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 = ""
- xml_empty_name = ""
- xml_broken = ""
-
- 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
+ monkeypatch.setattr(MetadataFactory, "fetch_http", lambda self, _: xml_content)
+ with pytest.raises(ArchiveListError) as e:
+ MetadataFactory(archive_id, archives_query=archives_query).getList()
+ assert e.type == ArchiveListError
@pytest.mark.parametrize(