diff --git a/eodag/plugins/authentication/aws_auth.py b/eodag/plugins/authentication/aws_auth.py
index fa0d1b8ceb..ebfb3b1e7d 100644
--- a/eodag/plugins/authentication/aws_auth.py
+++ b/eodag/plugins/authentication/aws_auth.py
@@ -165,6 +165,22 @@ def authenticate(self) -> S3ServiceResource:
         self.s3_resource = self._create_s3_resource()
         return self.s3_resource
 
+    def get_bucket_objects(self, bucket_name: str) -> BucketObjectsCollection:
+        """Get the object collection for an S3 bucket.
+
+        The returned collection is configured for requester-pays buckets when
+        required by the authentication configuration.
+
+        :param bucket_name: Name of the S3 bucket
+        :returns: Collection of objects in the bucket, filtered for requester-pays if necessary
+        """
+        if not self.s3_resource:
+            self.s3_resource = self._create_s3_resource()
+        objects = self.s3_resource.Bucket(bucket_name).objects
+        if self.config.requester_pays:
+            return objects.filter(RequestPayer="requester")
+        return objects
+
     def _get_authenticated_objects(
         self, bucket_name: str, prefix: str
     ) -> BucketObjectsCollection:
@@ -174,15 +190,8 @@ def _get_authenticated_objects(
         :param prefix: Prefix used to filter objects
         :returns: The boto3 authenticated objects
         """
-        if not self.s3_resource:
-            self.s3_resource = self._create_s3_resource()
         try:
-            if self.config.requester_pays:
-                objects = self.s3_resource.Bucket(bucket_name).objects.filter(
-                    RequestPayer="requester"
-                )
-            else:
-                objects = self.s3_resource.Bucket(bucket_name).objects
+            objects = self.get_bucket_objects(bucket_name)
             list(objects.filter(Prefix=prefix).limit(1))
             if objects:
                 logger.debug(
@@ -217,7 +226,7 @@ def authenticate_objects(
         :return: authenticated objects per bucket
         """
 
-        authenticated_objects: dict[str, Any] = {}
+        authenticated_objects: dict[str, BucketObjectsCollection] = {}
         auth_error_messages: set[str] = set()
         for _, pack in enumerate(bucket_names_and_prefixes):
 
diff --git a/eodag/plugins/download/aws.py b/eodag/plugins/download/aws.py
index 54675f267e..a9a28f14fb 100644
--- a/eodag/plugins/download/aws.py
+++ b/eodag/plugins/download/aws.py
@@ -59,13 +59,19 @@
     NotAvailableError,
     TimeOutError,
 )
-from eodag.utils.s3 import S3FileInfo, open_s3_zipped_object, stream_download_from_s3
+from eodag.utils.s3 import (
+    S3FileInfo,
+    list_files_in_s3_prefix,
+    open_s3_zipped_object,
+    stream_download_from_s3,
+)
 
 from .base import Download
 
 if TYPE_CHECKING:
     from mypy_boto3_s3 import S3ServiceResource
     from mypy_boto3_s3.client import S3Client
+    from mypy_boto3_s3.service_resource import BucketObjectsCollection, ObjectSummary
 
     from eodag.api.product import EOProduct
     from eodag.config import PluginConfig
@@ -256,7 +262,6 @@ def download(
                         file or with environment variables.
         :returns: The absolute path to the downloaded product in the local filesystem
         """
-
         if progress_callback is None:
             logger.info(
                 "Progress bar unavailable, please call product.download() instead of plugin.download()"
@@ -430,9 +435,7 @@ def _download_file_in_zip(
         progress_callback: ProgressCallback,
         executor: ThreadPoolExecutor,
     ):
-        """
-        Download file in zip from a prefix like `foo/bar.zip!file.txt`
-        """
+        """Download file in zip from a prefix like `foo/bar.zip!file.txt`."""
         if downloader_auth.s3_resource is None:
             logger.debug("Cannot check files in s3 zip without s3 resource")
             return bucket_names_and_prefixes
@@ -504,9 +507,9 @@ def _download_preparation(
         progress_callback: ProgressCallback,
         **kwargs: Unpack[DownloadConf],
     ) -> tuple[Optional[str], Optional[str]]:
-        """
-        Preparation for the download:
+        """Prepare the download.
 
+        It means:
         - check if file was already downloaded
         - get file path
         - create directories
@@ -533,8 +536,7 @@ def _download_preparation(
         return product_local_path, record_filename
 
     def _configure_safe_build(self, build_safe: bool, product: EOProduct):
-        """
-        Updates the product properties with fetch metadata if safe build is enabled
+        """Update the product properties with fetch metadata if safe build is enabled.
 
         :param build_safe: if safe build is enabled
         :param product: product to be updated
@@ -579,8 +581,7 @@ def _get_bucket_names_and_prefixes(
         ignore_assets: bool,
         complementary_url_keys: list[str],
     ) -> list[tuple[str, Optional[str]]]:
-        """
-        Retrieves the bucket names and path prefixes for the assets
+        """Retrieve the bucket names and path prefixes for the assets.
 
         :param product: product for which the assets shall be downloaded
         :param asset_filter: text for which the assets should be filtered
@@ -635,14 +636,13 @@ def _get_bucket_names_and_prefixes(
     def _get_unique_products(
         self,
         bucket_names_and_prefixes: list[tuple[str, Optional[str]]],
-        authenticated_objects: dict[str, Any],
+        authenticated_objects: dict[str, BucketObjectsCollection],
         asset_filter: Optional[str],
         ignore_assets: bool,
         product: EOProduct,
         raise_error: bool = True,
-    ) -> set[Any]:
-        """
-        Retrieve unique product chunks based on authenticated objects and asset filters
+    ) -> set[ObjectSummary]:
+        """Retrieve unique product chunks based on authenticated objects and asset filters.
 
         :param bucket_names_and_prefixes: list of bucket names and corresponding path prefixes
         :param authenticated_objects: available objects per bucket
@@ -652,12 +652,15 @@ def _get_unique_products(
         :param raise_error: raise error if there is nothing to download
         :return: set of product chunks that can be downloaded
         """
-        product_chunks: list[Any] = []
+        product_chunks: list[ObjectSummary] = []
         for bucket_name, prefix in bucket_names_and_prefixes:
             # unauthenticated items filtered out
             if bucket_name in authenticated_objects.keys():
                 product_chunks.extend(
-                    authenticated_objects[bucket_name].filter(Prefix=prefix)
+                    list_files_in_s3_prefix(
+                        prefix or "",
+                        authenticated_objects[bucket_name],
+                    )
                 )
 
         unique_product_chunks = set(product_chunks)
@@ -691,8 +694,7 @@ def stream_download(
         timeout: float = DEFAULT_DOWNLOAD_TIMEOUT,
         **kwargs: Unpack[DownloadConf],
     ) -> StreamResponse:
-        """
-        Stream EO product data as a FastAPI-compatible `StreamResponse`, with support for partial downloads,
+        """Stream EO product data as a FastAPI-compatible `StreamResponse`, with support for partial downloads,
         asset filtering, and on-the-fly compression.
 
         This method streams data from one or more S3 objects that belong to a given EO product.
@@ -855,7 +857,7 @@ def _get_commonpath(
     def get_product_bucket_name_and_prefix(
         self, product: EOProduct, url: Optional[str] = None
     ) -> tuple[str, Optional[str]]:
-        """Extract bucket name and prefix from product URL
+        """Extract bucket name and prefix from product URL.
 
         :param product: The EO product to download
         :param url: (optional) URL to use as product.location
@@ -880,7 +882,7 @@ def get_product_bucket_name_and_prefix(
         return bucket, prefix
 
     def check_manifest_file_list(self, product_path: str) -> None:
-        """Checks if products listed in manifest.safe exist"""
+        """Check if products listed in manifest.safe exist."""
         manifest_path_list = [
             os.path.join(d, x)
             for d, _, f in os.walk(product_path)
@@ -904,7 +906,7 @@ def check_manifest_file_list(self, product_path: str) -> None:
                 logger.warning("SAFE build: %s is missing" % safe_file.get("href"))
 
     def finalize_s2_safe_product(self, product_path: str) -> None:
-        """Add missing dirs to downloaded product"""
+        """Add missing dirs to downloaded product."""
         try:
             logger.debug("Finalize SAFE product")
             manifest_path_list = [
@@ -970,7 +972,7 @@ def get_chunk_dest_path(
         dir_prefix: Optional[str] = None,
         build_safe: bool = False,
     ) -> str:
-        """Get chunk SAFE destination path"""
+        """Get chunk SAFE destination path."""
         if not build_safe:
             if dir_prefix is None:
                 dir_prefix = chunk.key
diff --git a/eodag/utils/s3.py b/eodag/utils/s3.py
index 63f7767066..18d94061fa 100644
--- a/eodag/utils/s3.py
+++ b/eodag/utils/s3.py
@@ -51,6 +51,7 @@
     from zipfile import ZipInfo
 
     from mypy_boto3_s3.client import S3Client
+    from mypy_boto3_s3.service_resource import BucketObjectsCollection, ObjectSummary
 
     from eodag.api.product import EOProduct  # type: ignore
 
@@ -526,6 +527,42 @@ def stream_download_from_s3(
     )
 
 
+def list_files_in_s3_prefix(
+    prefix: str,
+    objects: BucketObjectsCollection,
+) -> list[ObjectSummary]:
+    """List files matching an S3 object key or a descendant prefix.
+
+    S3 prefix matching is lexical, so sibling keys such as ``product.txt`` are
+    excluded when listing ``product``. Directory marker objects are excluded,
+    except for a standalone exact object. Some S3-compatible services require
+    a trailing slash to list descendants, which is retried when needed.
+
+    :param prefix: S3 object key prefix to list
+    :param objects: Collection of S3 objects to filter
+    :returns: List of S3 objects matching the prefix criteria
+    """
+    descendant_prefix = f"{prefix.rstrip('/')}/"
+    matching_objects = list(objects.filter(Prefix=prefix))
+    if prefix and not matching_objects:
+        matching_objects = list(objects.filter(Prefix=descendant_prefix))
+
+    has_descendants = any(
+        item.key.startswith(descendant_prefix) and not item.key.endswith("/")
+        for item in matching_objects
+    )
+    return [
+        item
+        for item in matching_objects
+        if not item.key.endswith("/")
+        and (
+            not prefix
+            or (item.key == prefix and not (has_descendants and item.size == 0))
+            or item.key.startswith(descendant_prefix)
+        )
+    ]
+
+
 def update_assets_from_s3(
     product: EOProduct,
     auth: AwsAuth,
@@ -578,29 +615,25 @@ def update_assets_from_s3(
                     }
         else:
             # List files in prefix
-            s3_objects = s3_client.list_objects(
-                Bucket=bucket, Prefix=prefix, MaxKeys=300
-            )
-            for item_s3 in s3_objects.get("Contents", []):
+            bucket_objects = auth.get_bucket_objects(bucket)
+            for item_s3 in list_files_in_s3_prefix(prefix, bucket_objects):
 
-                url = "s3://{bucket}/{key}".format(bucket=bucket, key=item_s3["Key"])
+                url = "s3://{bucket}/{key}".format(bucket=bucket, key=item_s3.key)
                 key, roles = product.driver.guess_asset_key_and_roles(url, product)
                 if key is not None:
                     asset_data: dict[str, Any] = {
                         "title": key,
                         "roles": roles,
                         "href": url,
-                        "type": guess_file_type(item_s3["Key"]),
+                        "type": guess_file_type(item_s3.key),
                     }
-                    etag = item_s3.get("ETag")
+                    etag = item_s3.e_tag
                     # multipart-upload ETags ("<hash>-N") are not MD5 digests
                     if isinstance(etag, str) and "-" not in etag:
                         asset_data["file:checksum"] = etag.strip('"')
-                    size = item_s3.get("Size")
-                    if size is not None:
-                        asset_data["file:size"] = size
+                    asset_data["file:size"] = item_s3.size
 
-                    if last_modified := to_iso_utc_string(item_s3.get("LastModified")):
+                    if last_modified := to_iso_utc_string(item_s3.last_modified):
                         asset_data["updated"] = last_modified
 
                     assets_data[key] = asset_data
diff --git a/tests/context.py b/tests/context.py
index e2e4d9eae2..0ca77e305b 100644
--- a/tests/context.py
+++ b/tests/context.py
@@ -117,6 +117,7 @@
 from eodag.utils.import_system import import_all_modules, patch_owslib_requests
 from eodag.utils.notebook import NotebookWidgets, check_ipython, check_notebook
 from eodag.utils.s3 import (
+    list_files_in_s3_prefix,
     list_files_in_s3_zipped_object,
     update_assets_from_s3,
     open_s3_zipped_object,
diff --git a/tests/units/test_auth_plugins.py b/tests/units/test_auth_plugins.py
index 7dedb15d4f..5b56aef8e2 100644
--- a/tests/units/test_auth_plugins.py
+++ b/tests/units/test_auth_plugins.py
@@ -854,6 +854,29 @@ def test_plugins_auth_aws_authenticate_objects(
         auth_objects = plugin.authenticate_objects(buckets_prefixes)
         self.assertDictEqual({"a": auth_objects_a, "b": auth_objects_b}, auth_objects)
 
+    @mock.patch(
+        "eodag.plugins.authentication.aws_auth.AwsAuth._create_s3_resource",
+        autospec=True,
+    )
+    def test_plugins_auth_aws_get_bucket_objects_requester_pays(
+        self, mock_create_s3_resource
+    ):
+        """get_bucket_objects must configure requester-pays collections."""
+        plugin = self.get_auth_plugin("provider_with_auth_keys")
+        plugin.config.requester_pays = True
+        plugin.s3_resource = None
+        bucket_objects = (
+            mock_create_s3_resource.return_value.Bucket.return_value.objects
+        )
+
+        objects = plugin.get_bucket_objects("requester-pays-bucket")
+
+        mock_create_s3_resource.return_value.Bucket.assert_called_once_with(
+            "requester-pays-bucket"
+        )
+        bucket_objects.filter.assert_called_once_with(RequestPayer="requester")
+        self.assertIs(objects, bucket_objects.filter.return_value)
+
     @mock.patch(
         "eodag.plugins.authentication.aws_auth.AwsAuth._create_s3_resource",
         autospec=True,
diff --git a/tests/units/test_download_plugins.py b/tests/units/test_download_plugins.py
index 1c777bccc8..0ffd66ed00 100644
--- a/tests/units/test_download_plugins.py
+++ b/tests/units/test_download_plugins.py
@@ -115,9 +115,10 @@ def get_auth_plugin(self, associated_plugin, product, not_none=True):
 
 
 class TestDownloadPluginBase(BaseDownloadPluginTest):
-    def test_plugins_download_base_prepare_download_existing(self):
-        """Download._prepare_download must detect if product destination already exists"""
+    """Test cases for the base download plugin."""
 
+    def test_plugins_download_base_prepare_download_existing(self):
+        """Download._prepare_download must detect if product destination already exists."""
         product_file = NamedTemporaryFile()
         self.product.location = path_to_uri(product_file.name)
         Path(product_file.name).touch()
@@ -132,8 +133,7 @@ def test_plugins_download_base_prepare_download_existing(self):
             self.assertIn("Product already present on this platform", str(cm.output))
 
     def test_plugins_download_base_prepare_download_record_file(self):
-        """Download._prepare_download must check existing record files"""
-
+        """Download._prepare_download must check existing record files."""
         self.product.location = self.product.remote_location = "http://foo.bar"
         self.product.collection = "foo"
 
@@ -189,8 +189,7 @@ def test_plugins_download_base_prepare_download_record_file(self):
                 self.assertIn("Product already downloaded", str(cm.output))
 
     def test_plugins_download_base_prepare_download_record_file_collision_dir(self):
-        """Download._prepare_download must detect already-downloaded dir without collision suffix"""
-
+        """Download._prepare_download must detect already-downloaded dir without collision suffix."""
         self.product.properties["title"] = "title to sanitïze"
         self.product.properties["id"] = "id alsô"
         self.product.location = self.product.remote_location = "http://foo.bar"
@@ -221,8 +220,7 @@ def test_plugins_download_base_prepare_download_record_file_collision_dir(self):
                 )
 
     def test_plugins_download_base_prepare_download_no_url(self):
-        """Download._prepare_download must return None when no download url"""
-
+        """Download._prepare_download must return None when no download url."""
         self.assertEqual(self.product.remote_location, "")
 
         plugin = self.get_download_plugin(self.product)
@@ -233,8 +231,7 @@ def test_plugins_download_base_prepare_download_no_url(self):
         self.assertIsNone(record_filename)
 
     def test_plugins_download_base_prepare_download_collision_avoidance(self):
-        """Download._prepare_download must use collision avoidance suffix"""
-
+        """Download._prepare_download must use collision avoidance suffix."""
         self.product.properties["title"] = "needs sanitïze"
         self.product.properties["id"] = "alsô.zip"
         self.product.location = self.product.remote_location = "somewhere"
@@ -246,7 +243,7 @@ def test_plugins_download_base_prepare_download_collision_avoidance(self):
         self.assertEqual(fs_path, os.path.join(gettempdir(), "needs_sanitize-also.zip"))
 
     def test_plugins_download_base_prepare_download_dir_permission(self):
-        """Download._prepare_download must check output directory permissions"""
+        """Download._prepare_download must check output directory permissions."""
         if os.name == "nt":
             self.skipTest("windows permissions too complex to set for this test")
 
@@ -261,7 +258,7 @@ def test_plugins_download_base_prepare_download_dir_permission(self):
             self.assertIn("Unable to create records directory", str(cm.output))
 
     def test_plugins_download_base_finalize_extract_not_complete(self):
-        """Download._finalize must not delete archive if extract is True but file is not an archive"""
+        """Download._finalize must not delete archive if extract is True but file is not an archive."""
         plugin = self.get_download_plugin(self.product)
 
         with TemporaryDirectory() as output_dir:
@@ -277,7 +274,7 @@ def test_plugins_download_base_finalize_extract_not_complete(self):
             self.assertTrue(os.path.isfile(fs_path))
 
     def test_plugins_download_base_finalize_extract_complete(self):
-        """Download._finalize must delete archive if extract is True and file is an archive"""
+        """Download._finalize must delete archive if extract is True and file is an archive."""
         plugin = self.get_download_plugin(self.product)
 
         with TemporaryDirectory() as output_dir:
@@ -299,6 +296,8 @@ def test_plugins_download_base_finalize_extract_complete(self):
 
 
 class TestDownloadPluginHttp(BaseDownloadPluginTest):
+    """Test cases for the HTTP download plugin."""
+
     def _dummy_product(
         self, provider: str, properties: dict[str, Any], collection: str
     ):
@@ -330,8 +329,7 @@ def _dummy_downloadable_product(
 
     @responses.activate
     def test_plugins_download_http_zip_file_ok(self):
-        """HTTPDownload.download() must keep the output as it is when it is a zip file"""
-
+        """HTTPDownload.download() must keep the output as it is when it is a zip file."""
         provider = "creodias"
         download_url = (
             "https://zipper.creodias.eu/download/8ff765a2-e089-465d-a48f-cc27008a0962"
@@ -412,9 +410,10 @@ def test_plugins_download_http_zip_file_ok(self):
     def test_plugins_download_http_nonzip_file_with_zip_extension_ok(
         self, mock_stream, mock_stream_download
     ):
-        """HTTPDownload.download() must create an output directory
-        when the result is a non-zip file with a '.zip' outputs extension"""
+        """HTTPDownload.download() must create an output directory in certain conditions.
 
+        One is when the result is a non-zip file with a '.zip' outputs extension.
+        """
         plugin = self.get_download_plugin(self.product)
         self.product.location = self.product.remote_location = "http://somewhere"
         self.product.properties["id"] = "someproduct"
@@ -458,9 +457,10 @@ def test_plugins_download_http_nonzip_file_with_zip_extension_ok(
     def test_plugins_download_http_file_without_zip_extension_ok(
         self, mock_stream, mock_stream_download
     ):
-        """HTTPDownload.download() must create an output directory
-        when the result is a file without a '.zip' outputs extension"""
+        """HTTPDownload.download() must create an output directory in certain conditions.
 
+        One is when the result is a file without a '.zip' outputs extension.
+        """
         # we use a provider having '.nc' as outputs file extension in its configuration
         product = EOProduct(
             "meteoblue",
@@ -518,8 +518,7 @@ def test_plugins_download_http_file_without_zip_extension_ok(
     def test_plugins_download_http_ignore_assets(
         self, mock_requests_get, mock_requests_head, mock_stream, mock_stream_download
     ):
-        """HTTPDownload.download() must ignore assets if configured to"""
-
+        """HTTPDownload.download() must ignore assets if configured to."""
         plugin = self.get_download_plugin(self.product)
         self.product.location = self.product.remote_location = (
             "http://somewhere/download_from_location"
@@ -583,8 +582,7 @@ def test_plugins_download_http_ignore_assets(
     def test_plugins_download_http_ignore_assets_without_ssl(
         self, mock_requests_get, mock_requests_head
     ):
-        """HTTPDownload.download() must ignore assets if configured to"""
-
+        """HTTPDownload.download() must ignore assets if configured to."""
         plugin = self.get_download_plugin(self.product)
         self.product.location = self.product.remote_location = (
             "http://somewhere/download_from_location"
@@ -628,8 +626,7 @@ def test_plugins_download_http_ignore_assets_without_ssl(
     def test_plugins_download_http_ignore_assets_product_config_true(
         self, mock_stream_download, mock_stream, mock_download_assets
     ):
-        """HTTPDownload.download() must support product-level ignore_assets=True"""
-
+        """HTTPDownload.download() must support product-level ignore_assets=True."""
         product = self._dummy_downloadable_product(
             "dummy_provider",
             {
@@ -665,8 +662,7 @@ def test_plugins_download_http_ignore_assets_product_config_true(
     def test_plugins_download_http_ignore_assets_product_config_false(
         self, mock_stream_download, mock_download_assets
     ):
-        """HTTPDownload.download() must support product-level ignore_assets=False"""
-
+        """HTTPDownload.download() must support product-level ignore_assets=False."""
         product = self._dummy_downloadable_product(
             "dummy_provider",
             {
@@ -699,8 +695,7 @@ def test_plugins_download_http_ignore_assets_product_config_false(
     def test_plugins_download_http_assets_filename_from_href(
         self, mock_requests_get, mock_requests_head
     ):
-        """HTTPDownload.download() must create an outputfile"""
-
+        """HTTPDownload.download() must create an output file."""
         plugin = self.get_download_plugin(self.product)
         self.product.location = self.product.remote_location = "http://somewhere"
         self.product.properties["id"] = "someproduct"
@@ -752,8 +747,7 @@ def test_plugins_download_http_assets_filename_from_href(
     def test_plugins_download_http_assets_filename_from_get(
         self, mock_requests_get, mock_requests_head
     ):
-        """HTTPDownload.download() must create an outputfile"""
-
+        """HTTPDownload.download() must create an outputfile."""
         plugin = self.get_download_plugin(self.product)
         self.product.location = self.product.remote_location = "http://somewhere"
         self.product.properties["id"] = "someproduct"
@@ -785,8 +779,7 @@ def test_plugins_download_http_assets_filename_from_get(
     def test_plugins_download_http_assets_error(
         self, mock_requests_get, mock_requests_head, mock_asset_size
     ):
-        """HTTPDownload.download() must create an outputfile"""
-
+        """HTTPDownload.download() must create an outputfile."""
         plugin = self.get_download_plugin(self.product)
         self.product.location = self.product.remote_location = "http://somewhere"
         self.product.properties["id"] = "someproduct"
@@ -810,8 +803,7 @@ def test_plugins_download_http_assets_error(
     def test_plugins_download_http_assets_interrupt(
         self, mock_requests_get, mock_requests_head, mock_progress_callback
     ):
-        """HTTPDownload.download() must download assets to a temporary file"""
-
+        """HTTPDownload.download() must download assets to a temporary file."""
         plugin = self.get_download_plugin(self.product)
         self.product.location = self.product.remote_location = "http://somewhere"
         self.product.properties["id"] = "someproduct"
@@ -865,8 +857,7 @@ def test_plugins_download_http_assets_interrupt(
     def test_plugins_download_http_assets_stream_zip_interrupt(
         self, mock_requests_get, mock_requests_head, mock_progress_callback
     ):
-        """HTTPDownload.stream_download must raise an error if an error is returned by the provider"""
-
+        """HTTPDownload.stream_download must raise an error if an error is returned by the provider."""
         plugin = self.get_download_plugin(self.product)
         self.product.location = self.product.remote_location = "http://somewhere"
         self.product.properties["id"] = "someproduct"
@@ -896,8 +887,7 @@ def test_plugins_download_http_assets_stream_zip_interrupt(
     def test_plugins_download_http_assets_too_many_requests_error(
         self, mock_requests_get, mock_requests_head, mock_asset_size
     ):
-        """HTTPDownload.download() must handle a 429 (Too many requests) error"""
-
+        """HTTPDownload.download() must handle a 429 (Too many requests) error."""
         plugin = self.get_download_plugin(self.product)
         self.product.location = self.product.remote_location = "http://somewhere"
         self.product.properties["id"] = "someproduct"
@@ -913,8 +903,7 @@ def test_plugins_download_http_assets_too_many_requests_error(
             plugin.download(self.product, output_dir=self.output_dir)
 
     def test_plugins_download_http_stream_dict_misconfigured(self):
-        """HTTPDownload.stream_download() must raise an error if misconfigured"""
-
+        """HTTPDownload.stream_download() must raise an error if misconfigured."""
         plugin = self.get_download_plugin(self.product)
         with self.assertRaises(MisconfiguredError):
             # Wrong auth instance
@@ -924,8 +913,7 @@ def test_plugins_download_http_stream_dict_misconfigured(self):
             )
 
     def test_stream_download_single_asset(self):
-        """HTTPDownload.stream_download() must return a response with a single asset"""
-
+        """HTTPDownload.stream_download() must return a response with a single asset."""
         plugin = self.get_download_plugin(self.product)
 
         asset = mock.Mock()
@@ -951,8 +939,7 @@ def test_stream_download_single_asset(self):
         self.assertEqual(response.content, [b"chunk1"])
 
     def test_stream_download_multiple_assets_zip(self):
-        """HTTPDownload.stream_download() must return a zipped response with multiple assets"""
-
+        """HTTPDownload.stream_download() must return a zipped response with multiple assets."""
         plugin = self.get_download_plugin(self.product)
 
         asset1 = mock.Mock(filename="file1.txt")
@@ -999,8 +986,7 @@ def fake_iter(*args, **kwargs):
         self.assertIn(".zip", response.headers["Content-Disposition"])
 
     def test_stream_download_asset_not_available(self):
-        """HTTPDownload.stream_download() must raise NotAvailableError if asset not available"""
-
+        """HTTPDownload.stream_download() must raise NotAvailableError if asset not available."""
         plugin = self.get_download_plugin(self.product)
 
         self.product.assets = mock.Mock()
@@ -1013,8 +999,7 @@ def test_stream_download_asset_not_available(self):
             )
 
     def test_stream_download_single_asset_with_type(self):
-        """HTTPDownload.stream_download() must return a response with a single asset and its type"""
-
+        """HTTPDownload.stream_download() must return a response with a single asset and its type."""
         plugin = self.get_download_plugin(self.product)
 
         asset = mock.Mock()
@@ -1044,8 +1029,7 @@ def test_stream_download_single_asset_with_type(self):
         self.assertEqual(response.content, [b"chunk1"])
 
     def test_stream_download_fallback_to_product(self):
-        """HTTPDownload.stream_download() must return a response with product headers if no asset headers"""
-
+        """HTTPDownload.stream_download() must return a response with product headers if no asset headers."""
         plugin = self.get_download_plugin(self.product)
 
         self.product.assets = mock.Mock()
@@ -1062,8 +1046,7 @@ def test_stream_download_fallback_to_product(self):
         self.assertEqual(response.headers, self.product.headers)
 
     def test_stream_download_product_empty_raises(self):
-        """HTTPDownload.stream_download() must raise NotAvailableError if no asset and no product headers"""
-
+        """HTTPDownload.stream_download() must raise NotAvailableError if no asset and no product headers."""
         plugin = self.get_download_plugin(self.product)
 
         self.product.assets = mock.Mock()
@@ -1083,8 +1066,7 @@ def test_stream_download_product_empty_raises(self):
     def test_plugins_download_http_assets_resume(
         self, mock_requests_get, mock_requests_head
     ):
-        """HTTPDownload.download() must resume the interrupted download of assets"""
-
+        """HTTPDownload.download() must resume the interrupted download of assets."""
         plugin = self.get_download_plugin(self.product)
         self.product.location = self.product.remote_location = "http://somewhere"
         self.product.properties["id"] = "someproduct"
@@ -1128,8 +1110,7 @@ def test_plugins_download_http_assets_resume(
     def test_plugins_download_http_asset_filter(
         self, mock_requests_get, mock_requests_head
     ):
-        """HTTPDownload.download() must create an outputfile"""
-
+        """HTTPDownload.download() must create an outputfile."""
         plugin = self.get_download_plugin(self.product)
         self.product.location = self.product.remote_location = "http://somewhere"
         self.product.properties["id"] = "someproduct"
@@ -1169,8 +1150,7 @@ def test_plugins_download_http_asset_filter(
     def test_plugins_download_http_assets_filename_from_head(
         self, mock_requests_get, mock_requests_head
     ):
-        """HTTPDownload.download() must create an outputfile"""
-
+        """HTTPDownload.download() must create an outputfile."""
         plugin = self.get_download_plugin(self.product)
         self.product.location = self.product.remote_location = "http://somewhere"
         self.product.properties["id"] = "someproduct"
@@ -1202,8 +1182,7 @@ def test_plugins_download_http_assets_filename_from_head(
     def test_plugins_download_http_assets_size(
         self, mock_requests_get, mock_requests_head, mock_progress_callback_reset
     ):
-        """HTTPDownload.download() must get assets sizes"""
-
+        """HTTPDownload.download() must get assets sizes."""
         plugin = self.get_download_plugin(self.product)
         self.product.location = self.product.remote_location = "http://somewhere"
         self.product.assets.clear()
@@ -1307,8 +1286,7 @@ def test_plugins_download_http_assets_size(
     def test_plugins_download_http_one_local_asset(
         self,
     ):
-        """HTTPDownload.download() must handle one local asset"""
-
+        """HTTPDownload.download() must handle one local asset."""
         plugin = self.get_download_plugin(self.product)
         self.product.location = self.product.remote_location = "http://somewhere"
         self.product.properties["id"] = "someproduct"
@@ -1332,8 +1310,7 @@ def test_plugins_download_http_one_local_asset(
     def test_plugins_download_http_several_local_assets(
         self,
     ):
-        """HTTPDownload.download() must handle several local assets"""
-
+        """HTTPDownload.download() must handle several local assets."""
         plugin = self.get_download_plugin(self.product)
         self.product.location = self.product.remote_location = "http://somewhere"
         self.product.properties["id"] = "someproduct"
@@ -1375,8 +1352,7 @@ def test_plugins_download_http_several_local_assets(
     def test_plugins_download_http_order_download_cop_ads(
         self,
     ):
-        """HTTPDownload.download must order the product if needed"""
-
+        """HTTPDownload.download must order the product if needed."""
         self.product.provider = "cop_ads"
         self.product.collection = "CAMS_EAC4"
         product_dataset = "cams-global-reanalysis-eac4"
@@ -1472,7 +1448,7 @@ def run():
 
     @mock.patch("eodag.plugins.download.http.requests.request", autospec=True)
     def test_plugins_download_http_order_get(self, mock_request):
-        """HTTPDownload._order() must request using eodag:order_link and GET protocol"""
+        """HTTPDownload._order() must request using eodag:order_link and GET protocol."""
         plugin = self.get_download_plugin(self.product)
         self.product.properties["eodag:download_link"] = (
             "https://copernicus.nci.org.au/dummy"
@@ -1506,9 +1482,7 @@ def test_plugins_download_http_order_get(self, mock_request):
 
     @mock.patch("eodag.plugins.download.http.requests.request", autospec=True)
     def test_plugins_download_http_order_get_raises_if_request_500(self, mock_request):
-        """HTTPDownload._order() must raise an error if request to backend
-        provider failed"""
-
+        """HTTPDownload._order() must raise an error if request to backend provider failed."""
         # Configure mock to raise an error
         mock_request.return_value = MockResponse(status_code=500)
 
@@ -1571,7 +1545,7 @@ def test_plugins_download_http_order_get_raises_if_request_400(self, mock_reques
 
     @mock.patch("eodag.plugins.download.http.requests.request", autospec=True)
     def test_plugins_download_http_order_post(self, mock_request):
-        """HTTPDownload._order() must request using eodag:order_link and POST protocol"""
+        """HTTPDownload._order() must request using eodag:order_link and POST protocol."""
         plugin = self.get_download_plugin(self.product)
         self.product.properties["eodag:download_link"] = (
             "https://copernicus.nci.org.au/dummy"
@@ -1628,7 +1602,7 @@ def test_plugins_download_http_order_post(self, mock_request):
         )
 
     def test_plugins_download_http_order_status(self):
-        """HTTPDownload._order_status() must request status using eodag:status_link"""
+        """HTTPDownload._order_status() must request status using eodag:status_link."""
         plugin = self.get_download_plugin(self.product)
         plugin.config.order_status = {
             "metadata_mapping": {
@@ -1668,7 +1642,7 @@ def run():
 
     @mock.patch("eodag.plugins.download.http.requests.request", autospec=True)
     def test_plugins_download_http_order_status_from_head_headers(self, mock_request):
-        """HTTPDownload._order_status() must not parse an empty HEAD response"""
+        """HTTPDownload._order_status() must not parse an empty HEAD response."""
         plugin = HTTPDownload(
             provider=self.product.provider,
             config=PluginConfig.from_mapping(
@@ -1709,9 +1683,7 @@ def test_plugins_download_http_order_status_from_head_headers(self, mock_request
     def test_plugins_download_http_order_status_get_raises_if_request_500(
         self, mock_request
     ):
-        """HTTPDownload._order() must raise an error if request to backend
-        provider failed"""
-
+        """HTTPDownload._order() must raise an error if request to backend provider failed."""
         # Configure mock to raise an error
         mock_request.return_value = MockResponse(status_code=500)
 
@@ -1836,7 +1808,7 @@ def run():
     def test_plugins_download_http_order_status_search_again_raises_if_request_failed(
         self,
     ):
-        """HTTPDownload._order_status() must raise an error if the search request after success failed"""
+        """HTTPDownload._order_status() must raise an error if the search request after success failed."""
         plugin = self.get_download_plugin(self.product)
         plugin.config.order_status = {
             "metadata_mapping": {"eodag:order_status": "$.json.status"},
@@ -1887,6 +1859,8 @@ def run(error_code: int):
 
 
 class TestDownloadPluginHttpRetry(BaseDownloadPluginTest):
+    """Test cases for the HTTP download plugin with retry logic."""
+
     def setUp(self):
         super(TestDownloadPluginHttpRetry, self).setUp()
 
@@ -1896,7 +1870,7 @@ def setUp(self):
         self.product.properties["order:status"] = OFFLINE_STATUS
 
     def test_plugins_download_http_retry_error_timeout(self):
-        """HTTPDownload.download() must retry on error until timeout"""
+        """HTTPDownload.download() must retry on error until timeout."""
 
         @responses.activate(registry=responses.registries.FirstMatchRegistry)
         def run():
@@ -1932,7 +1906,7 @@ def run():
         run()
 
     def test_plugins_download_http_retry_notready_timeout(self):
-        """HTTPDownload.download() must retry if not ready until timeout"""
+        """HTTPDownload.download() must retry if not ready until timeout."""
 
         @responses.activate(registry=responses.registries.FirstMatchRegistry)
         def run():
@@ -1953,7 +1927,7 @@ def run():
         run()
 
     def test_plugins_download_http_retry_ok(self):
-        """HTTPDownload.download() must retry until request succeeds"""
+        """HTTPDownload.download() must retry until request succeeds."""
 
         @responses.activate(registry=responses.registries.OrderedRegistry)
         def run():
@@ -1983,7 +1957,7 @@ def run():
         run()
 
     def test_plugins_download_http_retry_short_timeout(self):
-        """HTTPDownload.download() must not retry on very short timeout"""
+        """HTTPDownload.download() must not retry on very short timeout."""
 
         @responses.activate(registry=responses.registries.FirstMatchRegistry)
         def run():
@@ -2004,7 +1978,7 @@ def run():
         run()
 
     def test_plugins_download_http_retry_once_timeout(self):
-        """HTTPDownload.download() must retry once if wait time is equal to timeout"""
+        """HTTPDownload.download() must retry once if wait time is equal to timeout."""
 
         @responses.activate(registry=responses.registries.FirstMatchRegistry)
         def run():
@@ -2025,7 +1999,7 @@ def run():
         run()
 
     def test_plugins_download_http_retry_timeout_disabled(self):
-        """HTTPDownload.download() must not retry on error if timeout is disabled"""
+        """HTTPDownload.download() must not retry on error if timeout is disabled."""
 
         @responses.activate(registry=responses.registries.FirstMatchRegistry)
         def run():
@@ -2044,6 +2018,8 @@ def run():
 
 
 class TestDownloadPluginAws(BaseDownloadPluginTest):
+    """Test cases for the AWS download plugin."""
+
     def setUp(self):
         super(TestDownloadPluginAws, self).setUp()
         self.product = EOProduct(
@@ -2061,8 +2037,7 @@ def setUp(self):
         )
 
     def test_plugins_download_aws_get_bucket_prefix(self):
-        """AwsDownload.get_product_bucket_name_and_prefix() must extract bucket & prefix from location"""
-
+        """AwsDownload.get_product_bucket_name_and_prefix() must extract bucket & prefix from location."""
         plugin = self.get_download_plugin(self.product)
         plugin.config.products["S2_MSI_L2A"]["default_bucket"] = "default_bucket"
 
@@ -2080,7 +2055,7 @@ def test_plugins_download_aws_get_bucket_prefix(self):
         self.assertEqual((bucket, prefix), ("default_bucket", "somewhere/else"))
 
     def test_plugins_download_aws_get_commonpath(self):
-        """AwsDownload._get_commonpath() must return common chunk destination path"""
+        """AwsDownload._get_commonpath() must return common chunk destination path."""
         plugin = self.get_download_plugin(self.product)
         product_chunks = {
             mock.Mock(key="path/to/some/product/file1.tif"),
@@ -2093,9 +2068,120 @@ def test_plugins_download_aws_get_commonpath(self):
 
         self.assertEqual(common_path, "path/to/some/product")
 
+    def test_plugins_download_aws_ignores_directory_markers_and_sibling_files(self):
+        """AwsDownload._get_unique_products() must ignore S3 directory markers and sibling files.
+
+        Objects that should be included are only:
+        - exact object at the product prefix which does not have any child object, even if it has zero size
+        - exact object at the product prefix which has children objects, but it must have a non-zero size
+        - actual product files (child below the product folder), even if they have zero size.
+        """
+        plugin = self.get_download_plugin(self.product)
+
+        # Test the case where the S3 bucket contains a mix of directory markers, actual product files, and sibling files
+
+        # Setup mock S3 objects representing directory markers and actual files:
+        # - Exact directory marker (zero-size object at the prefix itself)
+        # - Exact directory marker with non-zero size (exact object at the prefix itself)
+        # - Slash directory marker (object with trailing slash)
+        # - Actual product file with non-zero size (child below the product folder)
+        # - Actual product file with zero size (child below the product folder)
+        # - Sibling file outside the product directory with similar prefix
+        exact_directory_marker_zero_size = mock.Mock(key="path/to/some/product", size=0)
+        exact_directory_marker_non_zero_size = mock.Mock(
+            key="path/to/some/product", size=1
+        )
+        slash_directory_marker = mock.Mock(key="path/to/some/product/", size=1)
+        product_file_zero_size = mock.Mock(key="path/to/some/product/file2.tif", size=0)
+        product_file_non_zero_size = mock.Mock(
+            key="path/to/some/product/file1.tif", size=1
+        )
+        sibling_file = mock.Mock(key="path/to/some/product.txt", size=1)
+        bucket_objects = mock.Mock()
+        bucket_objects.filter.side_effect = lambda Prefix: [
+            chunk
+            for chunk in (
+                exact_directory_marker_zero_size,
+                exact_directory_marker_non_zero_size,
+                slash_directory_marker,
+                product_file_zero_size,
+                product_file_non_zero_size,
+                sibling_file,
+            )
+            if chunk.key.startswith(Prefix)
+        ]
+
+        product_chunks = plugin._get_unique_products(
+            [("somebucket", "path/to/some/product")],
+            {"somebucket": bucket_objects},
+            None,
+            False,
+            self.product,
+        )
+
+        bucket_objects.filter.assert_called_once_with(Prefix="path/to/some/product")
+
+        # Check the files that should be included in the product chunks are:
+        # - child below the product directory (actual product files), zero size included
+        # - non-empty folder with children objects (not a directory marker)
+        self.assertSetEqual(
+            product_chunks,
+            {
+                product_file_non_zero_size,
+                product_file_zero_size,
+                exact_directory_marker_non_zero_size,
+            },
+        )
+
+        # Test the case where the exact zero-size object is the only one
+        # matching the prefix (it does not have any child object).
+        bucket_objects.reset_mock()
+        bucket_objects.filter.side_effect = lambda Prefix: (
+            []
+            if Prefix == "path/to/some/product"
+            else [exact_directory_marker_zero_size]
+        )
+
+        product_chunks = plugin._get_unique_products(
+            [("somebucket", "path/to/some/product")],
+            {"somebucket": bucket_objects},
+            None,
+            False,
+            self.product,
+        )
+
+        # Check that when an exact object does not have any child object, it is still included
+        # in the product chunks, even it is zero-size (it is not an directory marker anymore).
+        self.assertSetEqual(product_chunks, {exact_directory_marker_zero_size})
+
+    def test_plugins_download_aws_falls_back_to_slash_prefix(self):
+        """AwsDownload._get_unique_products() must support backends requiring slash prefix."""
+        plugin = self.get_download_plugin(self.product)
+        product_file = mock.Mock(key="path/to/some/product/file.tif", size=1)
+        bucket_objects = mock.Mock()
+        bucket_objects.filter.side_effect = lambda Prefix: (
+            [] if Prefix == "path/to/some/product" else [product_file]
+        )
+
+        product_chunks = plugin._get_unique_products(
+            [("somebucket", "path/to/some/product")],
+            {"somebucket": bucket_objects},
+            None,
+            False,
+            self.product,
+        )
+
+        bucket_objects.filter.assert_has_calls(
+            [
+                mock.call(Prefix="path/to/some/product"),
+                mock.call(Prefix="path/to/some/product/"),
+            ]
+        )
+        self.assertSetEqual(product_chunks, {product_file})
+
     @mock.patch("eodag.plugins.download.aws.stream_download_from_s3", autospec=True)
     def test_plugins_download_aws_stream_download(self, mock_stream_download_from_s3):
-        """AwsDownload.stream_download() must stream S3 objects with flattened paths"""
+        """AwsDownload.stream_download() must stream S3 objects with flattened paths."""
         expected_response = StreamResponse(iter([b"content"]))
         mock_stream_download_from_s3.return_value = expected_response
         plugin = self.get_download_plugin(self.product)
@@ -2210,7 +2296,7 @@ def test_plugins_download_aws_no_safe_build_no_flatten_top_dirs(
         mock_flatten_top_directories: mock.Mock,
         mock__get_unique_products: mock.Mock,
     ):
-        """AwsDownload.download() must not call safe build methods if not needed"""
+        """AwsDownload.download() must not call safe build methods if not needed."""
         mock_aws_auth_init.return_value = None
 
         plugin = self.get_download_plugin(self.product)
@@ -2265,8 +2351,7 @@ def test_plugins_download_aws_no_safe_build_flatten_top_dirs(
         mock_flatten_top_directories: mock.Mock,
         mock__get_unique_products: mock.Mock,
     ):
-        """AwsDownload.download() must not call safe build methods if not needed"""
-
+        """AwsDownload.download() must not call safe build methods if not needed."""
         mock_aws_auth_init.return_value = None
         plugin = self.get_download_plugin(self.product)
         auth_plugin = self.get_auth_plugin(plugin, self.product)
@@ -2319,7 +2404,7 @@ def test_plugins_download_aws_in_zip(
         mock_s3_resource: mock.Mock,
         mock_s3_session: mock.Mock,
     ):
-        """AwsDownload.download() must handle files in zip"""
+        """AwsDownload.download() must handle files in zip."""
 
         def _open_zip(*args, **kwargs):
             return (
@@ -2400,8 +2485,7 @@ def test_plugins_download_aws_safe_build(
         mock_check_manifest_file_list,
         mock_flatten_top_directories,
     ):
-        """AwsDownload.download() must call safe build methods if needed"""
-
+        """AwsDownload.download() must call safe build methods if needed."""
         mock_aws_auth_init.return_value = None
         plugin = self.get_download_plugin(self.product)
         auth_plugin = self.get_auth_plugin(plugin, self.product)
@@ -2486,8 +2570,7 @@ def test_plugins_download_aws_safe_build_assets(
         mock_check_manifest_file_list,
         mock_flatten_top_directories,
     ):
-        """AwsDownload.download() must call safe build methods if needed"""
-
+        """AwsDownload.download() must call safe build methods if needed."""
         mock_aws_auth_init.return_value = None
         plugin = self.get_download_plugin(self.product)
         auth_plugin = self.get_auth_plugin(plugin, self.product)
@@ -2567,8 +2650,7 @@ def test_plugins_download_aws_no_matching_collection(
         mock_aws_auth_init,
         mock_get_authenticated_objects: mock.Mock,
     ):
-        """AwsDownload.download() must fail if no product chunk is available"""
-
+        """AwsDownload.download() must fail if no product chunk is available."""
         mock_aws_auth_init.return_value = None
         plugin = self.get_download_plugin(self.product)
         auth_plugin = self.get_auth_plugin(plugin, self.product)
@@ -2600,8 +2682,7 @@ def test_plugins_download_aws_get_rio_env(
         mock_s3_session: mock.Mock,
         mock_s3_client: mock.Mock,
     ):
-        """AwsDownload.get_rio_env() must return rio env dict"""
-
+        """AwsDownload.get_rio_env() must return rio env dict."""
         self.product.properties["eodag:download_link"] = "s3://some-bucket/some/prefix"
 
         plugin = self.get_download_plugin(self.product)
diff --git a/tests/units/test_search_plugins.py b/tests/units/test_search_plugins.py
index e41087d45f..faf9edf905 100644
--- a/tests/units/test_search_plugins.py
+++ b/tests/units/test_search_plugins.py
@@ -3368,105 +3368,122 @@ class TestSearchPluginCreodiasS3Search(BaseSearchPluginTest):
     def setUp(self):
         super(TestSearchPluginCreodiasS3Search, self).setUp()
         self.provider = "creodias_s3"
+        with open(
+            Path(TEST_RESOURCES_PATH) / "provider_responses/creodias_s3_objects.json"
+        ) as f:
+            self.s3_objects = json.load(f)["Contents"]
+        with open(Path(TEST_RESOURCES_PATH) / "eodag_search_result_creodias.json") as f:
+            self.search_result = json.load(f)
+
+    def _get_search_results(self, mock_request):
+        mock_request.return_value = MockResponse(self.search_result, 200)
+        search_plugin = self.get_search_plugin("S1_SAR_GRD", self.provider)
+        return search_plugin.query(collection="S1_SAR_GRD")
+
+    def _get_downloader_and_auth(self, product):
+        downloader = self.plugins_manager.get_download_plugin(product)
+        auth = self.plugins_manager.get_auth_plugin(downloader, product)
+        auth.config.credentials = {
+            "aws_access_key_id": "foo",
+            "aws_secret_access_key": "bar",
+        }
+        auth.s3_resource = boto3.resource(
+            "s3", aws_access_key_id="foo", aws_secret_access_key="bar"
+        )
+        return downloader, auth
+
+    def _stub_s3_list_objects_for_product(
+        self, product, auth, object_entries
+    ) -> Stubber:
+        bucket, prefix = product.remote_location.removeprefix("s3://").split("/", 1)
+        response = {
+            "Contents": [
+                {**item, "Key": f"{prefix}/{item['Key'].rsplit('/', 1)[-1]}"}
+                for item in object_entries
+            ]
+        }
+        stubber = Stubber(auth.s3_resource.meta.client)
+        for requested_prefix in (prefix, f"{prefix}/"):
+            stubber.add_response(
+                "list_objects",
+                response if object_entries else {},
+                {"Bucket": bucket, "Prefix": requested_prefix},
+            )
+        return stubber
+
+    def _register_product_with_stubbed_s3_objects(self, product, object_entries):
+        downloader, auth = self._get_downloader_and_auth(product)
+        stubber = self._stub_s3_list_objects_for_product(product, auth, object_entries)
+        stubber.activate()
+        try:
+            product.register_downloader(downloader, auth)
+        finally:
+            stubber.deactivate()
+        return product
 
-    @mock.patch(
-        "eodag.plugins.authentication.aws_auth.AwsAuth.get_s3_client", autospec=True
-    )
     @mock.patch(
         "eodag.plugins.search.qssearch.QueryStringSearch._request", autospec=True
     )
-    def test_plugins_search_creodias_s3_links(self, mock_request, mock_s3_client):
-        # s3 links should be added to products with register_downloader
-        search_plugin = self.get_search_plugin("S1_SAR_GRD", self.provider)
-        client = boto3.client("s3", aws_access_key_id="a", aws_secret_access_key="b")
-        mock_s3_client.return_value = client
-        stubber = Stubber(client)
-        s3_response_file = (
-            Path(TEST_RESOURCES_PATH) / "provider_responses/creodias_s3_objects.json"
+    def test_plugins_search_creodias_s3_links(self, mock_request):
+        """Registering a product must add its S3 object links as assets."""
+        product = self._register_product_with_stubbed_s3_objects(
+            self._get_search_results(mock_request).data[0], self.s3_objects
         )
-        with open(s3_response_file) as f:
-            list_objects_response = json.load(f)
-        creodias_search_result_file = (
-            Path(TEST_RESOURCES_PATH) / "eodag_search_result_creodias.json"
-        )
-        with open(creodias_search_result_file) as f:
-            creodias_search_result = json.load(f)
-        mock_request.return_value = MockResponse(creodias_search_result, 200)
 
-        res = search_plugin.query(collection="S1_SAR_GRD")
-        for product in res.data:
-            download_plugin = self.plugins_manager.get_download_plugin(product)
-            auth_plugin = self.plugins_manager.get_auth_plugin(download_plugin, product)
-            stubber.add_response("list_objects", list_objects_response)
-            stubber.activate()
-            # fails if credentials are missing
-            auth_plugin.config.credentials = {
-                "aws_access_key_id": "",
-                "aws_secret_access_key": "",
-            }
-            with self.assertRaisesRegex(
-                MisconfiguredError,
-                r"^Incomplete credentials .* \['aws_access_key_id', 'aws_secret_access_key'\]$",
-            ):
-                product.register_downloader(download_plugin, auth_plugin)
-            auth_plugin.config.credentials = {
-                "aws_access_key_id": "foo",
-                "aws_secret_access_key": "bar",
-            }
-            product.register_downloader(download_plugin, auth_plugin)
-        assets = res.data[0].assets
+        assets = product.assets
         self.assertEqual(3, len(assets))
-        # check if s3 links have been created correctly
         for asset in assets.values():
-            self.assertIn("s3://eodata/Sentinel-1/SAR/GRD/2014/10/10", asset["href"])
-
-        # no occur should occur and assets should be empty if list_objects does not have content
-        # (this situation will occur if the product does not have assets but is a tar file)
-        stubber.add_response("list_objects", {})
-        download_plugin = self.plugins_manager.get_download_plugin(res.data[0])
-        auth_plugin = self.plugins_manager.get_auth_plugin(download_plugin, res.data[0])
-        res.data[0].driver = None
-        res.data[0].assets = AssetsDict(res.data[0])
-        res.data[0].register_downloader(download_plugin, auth_plugin)
-        self.assertIsNotNone(res.data[0].driver)
-        self.assertEqual(0, len(res.data[0].assets))
+            self.assertTrue(asset["href"].startswith(product.remote_location))
 
     @mock.patch(
-        "eodag.plugins.authentication.aws_auth.AwsAuth.get_s3_client", autospec=True
+        "eodag.plugins.search.qssearch.QueryStringSearch._request", autospec=True
     )
+    def test_plugins_search_creodias_s3_links_empty(self, mock_request):
+        """Registering a product with no S3 objects must leave assets empty."""
+        product = self._get_search_results(mock_request).data[0]
+        product.driver = None
+        product.assets = AssetsDict(product)
+
+        self._register_product_with_stubbed_s3_objects(product, [])
+
+        self.assertIsNotNone(product.driver)
+        self.assertEqual(0, len(product.assets))
+
     @mock.patch(
         "eodag.plugins.search.qssearch.QueryStringSearch._request", autospec=True
     )
-    def test_plugins_search_creodias_s3_client_error(
-        self, mock_request, mock_s3_client
-    ):
-        # request error should be raised when there is an error when fetching data from the s3
-        search_plugin = self.get_search_plugin("S1_SAR_GRD", self.provider)
-        client = boto3.client("s3", aws_access_key_id="a", aws_secret_access_key="b")
-        mock_s3_client.return_value = client
-        stubber = Stubber(client)
+    def test_plugins_search_creodias_s3_links_require_credentials(self, mock_request):
+        """Registering a Creodias S3 product requires complete AWS credentials."""
+        product = self._get_search_results(mock_request).data[0]
+        downloader = self.plugins_manager.get_download_plugin(product)
+        auth = self.plugins_manager.get_auth_plugin(downloader, product)
+        auth.config.credentials = {"aws_access_key_id": "", "aws_secret_access_key": ""}
+
+        with self.assertRaisesRegex(
+            MisconfiguredError,
+            r"^Incomplete credentials .* \['aws_access_key_id', 'aws_secret_access_key'\]$",
+        ):
+            product.register_downloader(downloader, auth)
 
-        creodias_search_result_file = (
-            Path(TEST_RESOURCES_PATH) / "eodag_search_result_creodias.json"
+    @mock.patch(
+        "eodag.plugins.search.qssearch.QueryStringSearch._request", autospec=True
+    )
+    def test_plugins_search_creodias_s3_client_error(self, mock_request):
+        # request error should be raised when there is an error when fetching data from the s3
+        product = self._get_search_results(mock_request).data[0]
+        downloader, auth = self._get_downloader_and_auth(product)
+        bucket, prefix = product.remote_location.removeprefix("s3://").split("/", 1)
+        stubber = Stubber(auth.s3_resource.meta.client)
+        stubber.add_client_error(
+            "list_objects", expected_params={"Bucket": bucket, "Prefix": prefix}
         )
-        with open(creodias_search_result_file) as f:
-            creodias_search_result = json.load(f)
-        mock_request.return_value = MockResponse(creodias_search_result, 200)
 
         with self.assertRaises(NotAvailableError):
-            res = search_plugin.query(collection="S1_SAR_GRD")
-            for product in res.data:
-                download_plugin = self.plugins_manager.get_download_plugin(product)
-                auth_plugin = self.plugins_manager.get_auth_plugin(
-                    download_plugin, product
-                )
-                auth_plugin.config.credentials = {
-                    "aws_access_key_id": "foo",
-                    "aws_secret_access_key": "bar",
-                }
-                stubber.add_client_error("list_objects")
-                stubber.activate()
-                product.register_downloader(download_plugin, auth_plugin)
+            stubber.activate()
+            try:
+                product.register_downloader(downloader, auth)
+            finally:
+                stubber.deactivate()
 
 
 class TestSearchPluginECMWFSearch(unittest.TestCase):
diff --git a/tests/units/test_utils_s3.py b/tests/units/test_utils_s3.py
index e79a1e50f2..ad58e07d2c 100644
--- a/tests/units/test_utils_s3.py
+++ b/tests/units/test_utils_s3.py
@@ -21,6 +21,7 @@
     _compute_file_ranges,
     _prepare_file_in_zip,
     file_position_from_s3_zip,
+    list_files_in_s3_prefix,
     list_files_in_s3_zipped_object,
     open_s3_zipped_object,
     stream_download_from_s3,
@@ -177,6 +178,23 @@ def test_utils_s3_open_s3_zipped_object(self):
         self.assertIsInstance(cd_data, bytes)
         self.assertEqual(len(zip_file.filelist), 6)
 
+    def test_utils_s3_list_files_in_s3_prefix(self):
+        """list_files_in_s3_prefix must exclude directory markers and siblings."""
+        self.s3_client.put_object(Bucket="mybucket", Key="test-prefix", Body=b"")
+        self.s3_client.put_object(Bucket="mybucket", Key="test-prefix/", Body=b"")
+        self.s3_client.put_object(
+            Bucket="mybucket", Key="test-prefix/file.txt", Body=b"content"
+        )
+        self.s3_client.put_object(
+            Bucket="mybucket", Key="test-prefix.txt", Body=b"sibling"
+        )
+
+        objects = list_files_in_s3_prefix(
+            "test-prefix", self.auth_plugin.get_bucket_objects("mybucket")
+        )
+
+        self.assertListEqual([item.key for item in objects], ["test-prefix/file.txt"])
+
     def test_utils_s3_update_assets_from_s3_zip(self):
         """update_assets_from_s3 must update the assets of a product from a zipped object stored in S3"""
         update_assets_from_s3(