refactor(core): authentication for get_quicklook (#1608)

Author: cauriol

eodag/api/product/_product.py

@@ -313,7 +313,6 @@ def download(
                 "EO product is unable to download itself due to lacking of a "
                 "download plugin"
             )
-
         auth = (
             self.downloader_auth.authenticate()
             if self.downloader_auth is not None
@@ -323,7 +322,6 @@ def download(
         progress_callback, close_progress_callback = self._init_progress_bar(
             progress_callback
         )
-
         fs_path = self.downloader.download(
             self,
             auth=auth,
@@ -369,6 +367,47 @@ def _init_progress_bar(
         progress_callback.refresh()
         return (progress_callback, close_progress_callback)
 
+    def _download_quicklook(
+        self,
+        quicklook_file: str,
+        progress_callback: ProgressCallback,
+        ssl_verify: Optional[bool] = None,
+        auth: Optional[AuthBase] = None,
+    ):
+
+        """Download the quicklook image from the EOProduct's quicklook URL.
+
+        This method performs an HTTP GET request to retrieve the quicklook image and saves it
+        locally at the specified path. It optionally verifies SSL certificates, uses HTTP
+        authentication, and can display a download progress if a callback is provided.
+
+        :param quicklook_file: The full path (including filename) where the quicklook will be saved.
+        :param progress_callback: A callable that accepts the current and total download sizes
+                                to display or log the download progress. It must support `reset(total)`
+                                and be callable with downloaded chunk sizes.
+        :param ssl_verify: (optional) Whether to verify SSL certificates. Defaults to True.
+        :param auth: (optional) Authentication credentials (e.g., tuple or object) used for the
+                        HTTP request if the resource requires authentication.
+        :raises HTTPError: If the HTTP request to the quicklook URL fails.
+        """
+        with requests.get(
+            self.properties["quicklook"],
+            stream=True,
+            auth=auth,
+            headers=USER_AGENT,
+            timeout=DEFAULT_STREAM_REQUESTS_TIMEOUT,
+            verify=ssl_verify,
+        ) as stream:
+            stream.raise_for_status()
+            stream_size = int(stream.headers.get("content-length", 0))
+            progress_callback.reset(stream_size)
+            with open(quicklook_file, "wb") as fhandle:
+                for chunk in stream.iter_content(chunk_size=64 * 1024):
+                    if chunk:
+                        fhandle.write(chunk)
+                        progress_callback(len(chunk))
+            logger.info("Download recorded in %s", quicklook_file)
+
     def get_quicklook(
         self,
         filename: Optional[str] = None,
@@ -378,6 +417,9 @@ def get_quicklook(
         """Download the quicklook image of a given EOProduct from its provider if it
         exists.
 
+        This method retrieves the quicklook URL from the EOProduct metadata and delegates
+        the download to the internal `download_quicklook` method.
+
         :param filename: (optional) The name to give to the downloaded quicklook. If not
                          given, it defaults to the product's ID (without file extension).
         :param output_dir: (optional) The absolute path of the directory where to store
@@ -403,18 +445,6 @@ def format_quicklook_address() -> None:
                     }
                 )
 
-        # progress bar init
-        if progress_callback is None:
-            progress_callback = ProgressCallback()
-            # one shot progress callback to close after download
-            close_progress_callback = True
-        else:
-            close_progress_callback = False
-            # update units as bar may have been previously used for extraction
-            progress_callback.unit = "B"
-            progress_callback.unit_scale = True
-        progress_callback.desc = "quicklooks/%s" % self.properties.get("id", "")
-
         if self.properties.get("quicklook", None) is None:
             logger.warning(
                 "Missing information to retrieve quicklook for EO product: %s",
@@ -442,6 +472,18 @@ def format_quicklook_address() -> None:
         )
 
         if not os.path.isfile(quicklook_file):
+
+            # progress bar init
+            if progress_callback is None:
+                progress_callback = ProgressCallback()
+                # one shot progress callback to close after download
+                close_progress_callback = True
+            else:
+                close_progress_callback = False
+                # update units as bar may have been previously used for extraction
+                progress_callback.unit = "B"
+                progress_callback.unit_scale = True
+            progress_callback.desc = "quicklooks/%s" % self.properties.get("id", "")
             # VERY SPECIAL CASE (introduced by the onda provider): first check if
             # it is a HTTP URL. If not, we assume it is a base64 string, in which case
             # we just decode the content, write it into the quicklook_file and return it.
@@ -468,30 +510,25 @@ def format_quicklook_address() -> None:
                 if self.downloader
                 else True
             )
-            with requests.get(
-                self.properties["quicklook"],
-                stream=True,
-                auth=auth,
-                headers=USER_AGENT,
-                timeout=DEFAULT_STREAM_REQUESTS_TIMEOUT,
-                verify=ssl_verify,
-            ) as stream:
-                try:
-                    stream.raise_for_status()
-                except RequestException as e:
-                    import traceback as tb
+            try:
+                self._download_quicklook(
+                    quicklook_file, progress_callback, ssl_verify, auth
+                )
+            except RequestException as e:
 
-                    logger.error("Error while getting resource :\n%s", tb.format_exc())
+                logger.debug(
+                    f"Error while getting resource with authentication. {e} \nTrying without authentication..."
+                )
+                try:
+                    self._download_quicklook(
+                        quicklook_file, progress_callback, ssl_verify, None
+                    )
+                except RequestException as e_no_auth:
+                    logger.error(
+                        f"Failed to get resource with authentication: {e} \n \
+                        Failed to get resource even without authentication. {e_no_auth}"
+                    )
                     return str(e)
-                else:
-                    stream_size = int(stream.headers.get("content-length", 0))
-                    progress_callback.reset(stream_size)
-                    with open(quicklook_file, "wb") as fhandle:
-                        for chunk in stream.iter_content(chunk_size=64 * 1024):
-                            if chunk:
-                                fhandle.write(chunk)
-                                progress_callback(len(chunk))
-                    logger.info("Download recorded in %s", quicklook_file)
 
             # close progress bar if needed
             if close_progress_callback:

tests/units/test_eoproduct.py

@@ -178,6 +178,7 @@ def test_eoproduct_get_quicklook_http_error(self):
         product.register_downloader(mock_downloader, None)
 
         quicklook_file_path = product.get_quicklook()
+        self.assertEqual(self.requests_http_get.call_count, 2)
         self.requests_http_get.assert_called_with(
             "https://fake.url.to/quicklook",
             stream=True,
@@ -188,6 +189,35 @@ def test_eoproduct_get_quicklook_http_error(self):
         )
         self.assertEqual(quicklook_file_path, "")
 
+    def test_eoproduct_get_quicklook_ok_without_auth(self):
+        """EOProduct.get_quicklook must retrieve the quicklook without authentication."""  # noqa
+        product = self._dummy_product()
+        product.properties["quicklook"] = "https://fake.url.to/quicklook"
+
+        self.requests_http_get.return_value.__enter__.return_value.raise_for_status.side_effect = (  # noqa
+            requests.HTTPError,
+            None,
+        )
+        mock_downloader = mock.MagicMock(
+            spec_set=Download(provider=self.provider, config=None)
+        )
+        mock_downloader.config = config.PluginConfig.from_mapping(
+            {"output_dir": tempfile.gettempdir()}
+        )
+        product.register_downloader(mock_downloader, None)
+
+        quicklook_file_path = product.get_quicklook()
+        self.assertEqual(self.requests_http_get.call_count, 2)
+        self.requests_http_get.assert_called_with(
+            "https://fake.url.to/quicklook",
+            stream=True,
+            auth=None,
+            headers=USER_AGENT,
+            timeout=DEFAULT_STREAM_REQUESTS_TIMEOUT,
+            verify=True,
+        )
+        os.remove(quicklook_file_path)
+
     def test_eoproduct_get_quicklook_ok(self):
         """EOProduct.get_quicklook must return the path to the successfully downloaded quicklook"""  # noqa
         product = self._dummy_product()