Merge pull request #545 from ddalcino/metadatafactory-use-baseurl

Make MetadataFactory respect baseurl set in arguments

aqt/installer.py

@@ -197,15 +197,17 @@ def _check_modules_arg(self, qt_version, modules):
         return all([m in available for m in modules])
 
     @staticmethod
-    def _determine_qt_version(qt_version_or_spec: str, host: str, target: str, arch: str) -> Version:
+    def _determine_qt_version(
+        qt_version_or_spec: str, host: str, target: str, arch: str, base_url: str = Settings.baseurl
+    ) -> Version:
         def choose_highest(x: Optional[Version], y: Optional[Version]) -> Optional[Version]:
             if x and y:
                 return max(x, y)
             return x or y
 
         def opt_version_for_spec(ext: str, _spec: SimpleSpec) -> Optional[Version]:
             try:
-                return MetadataFactory(ArchiveId("qt", host, target, ext), spec=_spec).getList().latest()
+                return MetadataFactory(ArchiveId("qt", host, target, ext), spec=_spec, base_url=base_url).getList().latest()
             except AqtException:
                 return None
 
@@ -257,11 +259,6 @@ def run_install_qt(self, args):
         arch: str = self._set_arch(
             args.arch, os_name, target, getattr(args, "qt_version", getattr(args, "qt_version_spec", None))
         )
-        if hasattr(args, "qt_version_spec"):
-            qt_version: str = str(Cli._determine_qt_version(args.qt_version_spec, os_name, target, arch))
-        else:
-            qt_version: str = args.qt_version
-            Cli._validate_version_str(qt_version)
         keep: bool = args.keep or Settings.always_keep_archives
         archive_dest: Optional[str] = args.archive_dest
         output_dir = args.outputdir
@@ -287,6 +284,11 @@ def run_install_qt(self, args):
             base = args.base
         else:
             base = Settings.baseurl
+        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
+            Cli._validate_version_str(qt_version)
         archives = args.archives
         if args.noarchives:
             if modules is None:
@@ -342,11 +344,6 @@ def _run_src_doc_examples(self, flavor, args, cmd_name: Optional[str] = None):
             self._warn_on_deprecated_parameter("target", args.target)
         target = "desktop"  # The only valid target for src/doc/examples is "desktop"
         os_name = args.host
-        if hasattr(args, "qt_version_spec"):
-            qt_version = str(Cli._determine_qt_version(args.qt_version_spec, os_name, target, arch=""))
-        else:
-            qt_version = args.qt_version
-            Cli._validate_version_str(qt_version)
         output_dir = args.outputdir
         if output_dir is None:
             base_dir = os.getcwd()
@@ -358,6 +355,11 @@ def _run_src_doc_examples(self, flavor, args, cmd_name: Optional[str] = None):
             base = args.base
         else:
             base = Settings.baseurl
+        if hasattr(args, "qt_version_spec"):
+            qt_version = str(Cli._determine_qt_version(args.qt_version_spec, os_name, target, arch="", base_url=base))
+        else:
+            qt_version = args.qt_version
+            Cli._validate_version_str(qt_version)
         if args.timeout is not None:
             timeout = (args.timeout, args.timeout)
         else:
@@ -394,7 +396,10 @@ def _run_src_doc_examples(self, flavor, args, cmd_name: Optional[str] = None):
     def run_install_src(self, args):
         """Run src subcommand"""
         if not hasattr(args, "qt_version"):
-            args.qt_version = str(Cli._determine_qt_version(args.qt_version_spec, args.host, args.target, arch=""))
+            base = args.base if hasattr(args, "base") else Settings.baseurl
+            args.qt_version = str(
+                Cli._determine_qt_version(args.qt_version_spec, args.host, args.target, arch="", base_url=base)
+            )
         if args.kde and args.qt_version != "5.15.2":
             raise CliInputError("KDE patch: unsupported version!!")
         start_time = time.perf_counter()
@@ -452,7 +457,7 @@ def run_install_tool(self, args):
             timeout = (Settings.connection_timeout, Settings.response_timeout)
         if args.tool_variant is None:
             archive_id = ArchiveId("tools", os_name, target, "")
-            meta = MetadataFactory(archive_id, is_latest_version=True, tool_name=tool_name)
+            meta = MetadataFactory(archive_id, base_url=base, is_latest_version=True, tool_name=tool_name)
             try:
                 archs = meta.getList()
             except ArchiveDownloadError as e:

aqt/metadata.py

@@ -360,6 +360,7 @@ def __init__(
         self,
         archive_id: ArchiveId,
         *,
+        base_url: str = Settings.baseurl,
         spec: Optional[SimpleSpec] = None,
         is_latest_version: bool = False,
         modules_query: Optional[Tuple[str, str]] = None,
@@ -387,6 +388,7 @@ def __init__(
         self.logger = getLogger("aqt.metadata")
         self.archive_id = archive_id
         self.spec = spec
+        self.base_url = base_url
 
         if archive_id.is_tools():
             if tool_name:
@@ -480,7 +482,7 @@ def fetch_latest_version(self) -> Optional[Version]:
 
     def fetch_tools(self) -> List[str]:
         html_doc = self.fetch_http(self.archive_id.to_url(), False)
-        return list(self.iterate_folders(html_doc, "tools"))
+        return list(self.iterate_folders(html_doc, self.base_url, filter_category="tools"))
 
     def fetch_tool_modules(self, tool_name: str) -> List[str]:
         tool_data = self._fetch_module_metadata(tool_name)
@@ -571,11 +573,10 @@ def _to_version(self, qt_ver: str) -> Version:
             raise CliInputError(e) from e
         return version
 
-    @staticmethod
-    def fetch_http(rest_of_url: str, is_check_hash: bool = True) -> str:
+    def fetch_http(self, rest_of_url: str, is_check_hash: bool = True) -> str:
         timeout = (Settings.connection_timeout, Settings.response_timeout)
         expected_hash = get_hash(rest_of_url, "sha256", timeout) if is_check_hash else None
-        base_urls = Settings.baseurl, random.choice(Settings.fallbacks)
+        base_urls = self.base_url, random.choice(Settings.fallbacks)
         for i, base_url in enumerate(base_urls):
             try:
                 url = posixpath.join(base_url, rest_of_url)
@@ -589,7 +590,7 @@ def fetch_http(rest_of_url: str, is_check_hash: bool = True) -> str:
                         f"Connection to '{base_url}' failed. Retrying with fallback '{base_urls[i + 1]}'."
                     )
 
-    def iterate_folders(self, html_doc: str, filter_category: str = "") -> Generator[str, None, None]:
+    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="")
             url: ParseResult = urlparse(raw_url)
@@ -609,12 +610,11 @@ def link_to_folder(link: bs4.element.Tag) -> str:
                 if folder.startswith(filter_category):
                     yield folder
         except Exception as e:
-            url = posixpath.join(Settings.baseurl, self.archive_id.to_url())
             raise ArchiveConnectionError(
-                f"Failed to retrieve the expected HTML page at {url}",
+                f"Failed to retrieve the expected HTML page at {html_url}",
                 suggested_action=[
                     "Check your network connection.",
-                    f"Make sure that you can access {url} in your web browser.",
+                    f"Make sure that you can access {html_url} in your web browser.",
                 ],
             ) from e
 
@@ -630,7 +630,7 @@ def folder_to_version_extension(folder: str) -> Tuple[Optional[Version], str]:
 
         return map(
             folder_to_version_extension,
-            self.iterate_folders(html_doc, category),
+            self.iterate_folders(html_doc, self.base_url, filter_category=category),
         )
 
     @staticmethod

tests/test_install.py

@@ -137,7 +137,7 @@ def make_mock_geturl_download_archive(
 
     xml = "<Updates>\n{}\n</Updates>".format("\n".join([archive.xml_package_update() for archive in archives]))
 
-    def mock_getUrl(url: str, *args) -> str:
+    def mock_getUrl(url: str, *args, **kwargs) -> str:
         if url.endswith(updates_url):
             return xml
         elif url.endswith(".sha256"):
@@ -940,3 +940,97 @@ def mock_extractor_that_fails(*args, **kwargs):
     assert err.type == ArchiveExtractionError
     err_msg = format(err.value).rstrip()
     assert err_msg == "Extraction error: 1\nout\nerr"
+
+
+@pytest.mark.parametrize(
+    "cmd, host, target, version, arch, arch_dir, base_url, updates_url, archives, expect_out",
+    (
+        (
+            "install-tool linux desktop tools_qtcreator qt.tools.qtcreator".split(),
+            "linux",
+            "desktop",
+            "1.2.3-0-197001020304",
+            "",
+            "",
+            "https://www.alt.qt.mirror.com",
+            "linux_x64/desktop/tools_qtcreator/Updates.xml",
+            [tool_archive("linux", "tools_qtcreator", "qt.tools.qtcreator")],
+            re.compile(
+                r"^INFO    : aqtinstall\(aqt\) v.* on Python 3.*\n"
+                r"INFO    : Downloading qt.tools.qtcreator...\n"
+                r"Finished installation of tools_qtcreator-linux-qt.tools.qtcreator.7z in .*\n"
+                r"INFO    : Finished installation\n"
+                r"INFO    : Time elapsed: .* second"
+            ),
+        ),
+        (
+            "install-qt windows desktop 5.12 win32_mingw73".split(),
+            "windows",
+            "desktop",
+            "5.12.10",
+            "win32_mingw73",
+            "mingw73_32",
+            "https://www.alt.qt.mirror.com",
+            "windows_x86/desktop/qt5_51210/Updates.xml",
+            [plain_qtbase_archive("qt.qt5.51210.win32_mingw73", "win32_mingw73")],
+            re.compile(
+                r"^INFO    : aqtinstall\(aqt\) v.* on Python 3.*\n"
+                r"INFO    : Resolved spec '5\.12' to 5\.12\.10\n"
+                r"INFO    : Downloading qtbase\.\.\.\n"
+                r"Finished installation of qtbase-windows-win32_mingw73\.7z in .*\n"
+                r"INFO    : Finished installation\n"
+                r"INFO    : Time elapsed: .* second"
+            ),
+        ),
+    ),
+)
+def test_installer_passes_base_to_metadatafactory(
+    monkeypatch,
+    capsys,
+    cmd: List[str],
+    host: str,
+    target: str,
+    version: str,
+    arch: str,
+    arch_dir: str,
+    base_url: str,
+    updates_url: str,
+    archives: List[MockArchive],
+    expect_out,  # type: re.Pattern
+):
+    # For convenience, fill in version and arch dir: prevents repetitive data declarations
+    for i in range(len(archives)):
+        archives[i].version = version
+        archives[i].arch_dir = arch_dir
+
+    basic_mock_get_url, mock_download_archive = make_mock_geturl_download_archive(archives, arch, host, updates_url)
+
+    def mock_get_url(url: str, *args, **kwargs) -> str:
+        # If we are fetching an index.html file, get it from tests/data/
+        if url == f"{base_url}/online/qtsdkrepository/{host}_x{'86' if host == 'windows' else '64'}/{target}/":
+            return (Path(__file__).parent / "data" / f"{host}-{target}.html").read_text("utf-8")
+
+        # Intercept and check the base url, but only if it's not a hash.
+        # Hashes must come from trusted mirrors only.
+        if not url.endswith(".sha256"):
+            assert url.startswith(base_url)
+
+        return basic_mock_get_url(url, *args, **kwargs)
+
+    monkeypatch.setattr("aqt.archives.getUrl", mock_get_url)
+    monkeypatch.setattr("aqt.helper.getUrl", mock_get_url)
+    monkeypatch.setattr("aqt.installer.downloadBinaryFile", mock_download_archive)
+
+    monkeypatch.setattr("aqt.metadata.getUrl", mock_get_url)
+
+    with TemporaryDirectory() as output_dir:
+        cli = Cli()
+        cli._setup_settings()
+
+        assert 0 == cli.run(cmd + ["--base", base_url, "--outputdir", output_dir])
+
+        out, err = capsys.readouterr()
+        sys.stdout.write(out)
+        sys.stderr.write(err)
+
+        assert expect_out.match(err)

tests/test_list.py

@@ -958,9 +958,16 @@ def _mock_fetch_http(_, rest_of_url, *args, **kwargs: str) -> str:
 
 
 def test_fetch_http_ok(monkeypatch):
-    monkeypatch.setattr("aqt.metadata.get_hash", lambda *args, **kwargs: hashlib.sha256(b"some_html_content").hexdigest())
-    monkeypatch.setattr("aqt.metadata.getUrl", lambda **kwargs: "some_html_content")
-    assert MetadataFactory.fetch_http("some_url") == "some_html_content"
+    html_content = b"some_html_content"
+    base_url = "https://alt.baseurl.com"
+
+    def mock_getUrl(url: str, *args, **kwargs) -> str:
+        assert url.startswith(base_url)
+        return str(html_content)
+
+    monkeypatch.setattr("aqt.metadata.get_hash", lambda *args, **kwargs: hashlib.sha256(html_content).hexdigest())
+    monkeypatch.setattr("aqt.metadata.getUrl", mock_getUrl)
+    assert MetadataFactory(mac_qt, base_url=base_url).fetch_http("some_url") == str(html_content)
 
 
 def test_fetch_http_failover(monkeypatch):
@@ -976,7 +983,7 @@ def _mock(url, **kwargs):
     monkeypatch.setattr("aqt.metadata.getUrl", _mock)
 
     # Require that the first attempt failed, but the second did not
-    assert MetadataFactory.fetch_http("some_url") == "some_html_content"
+    assert MetadataFactory(mac_qt).fetch_http("some_url") == "some_html_content"
     assert len(urls_requested) == 2
 
 
@@ -991,7 +998,7 @@ def _mock(url, **kwargs):
     monkeypatch.setattr("aqt.metadata.get_hash", lambda *args, **kwargs: hashlib.sha256(b"some_html_content").hexdigest())
     monkeypatch.setattr("aqt.metadata.getUrl", _mock)
     with pytest.raises(exception_on_error) as e:
-        MetadataFactory.fetch_http("some_url")
+        MetadataFactory(mac_qt).fetch_http("some_url")
     assert e.type == exception_on_error
 
     # Require that a fallback url was tried