Repository: fsspec/gcsfs
Pull request #572 (2023-08-09 06:27:59)

Better double asterisk support
This PR updates the `gcsfs` implementation to support updates made in the [Better double asterisks ** support](https://github.com/fsspec/filesystem_spec/pull/1329) pull request in the `filesystem_spec` repo.

Successful run: https://github.com/fsspec/filesystem_spec/actions/runs/5847462518/job/15915309794

@ianthomas23 Let me know if you prefer to merge your PR with `derived test` and I'll remove my updates regarding this.

ianthomas23: There is already a PR (#535) to add the derived tests to gcsfs.
john-jam: > There is already a PR (#535) to add the derived tests to gcsfs.

Oh sorry, I didn't see it. Do you plan to merge it soon? Having those is really convenient to test the repo changes!
ianthomas23:  > Oh sorry, I didn't see it. Do you plan to merge it soon? Having those is really convenient to test the repo changes!

I'd love to merge it soon, but there are 3 test failures that I haven't fixed yet.
john-jam: > > Oh sorry, I didn't see it. Do you plan to merge it soon? Having those is really convenient to test the repo changes!
> 
> I'd love to merge it soon, but there are 3 test failures that I haven't fixed yet.

The derived tests passes on my branch with my updates here (I temporarily updated the fsspec CI to clone my gcsfs branch): https://github.com/john-jam/filesystem_spec/actions/runs/5807521471/job/15742515177

Patch:
diff --git a/gcsfs/core.py b/gcsfs/core.py
index e8098bf..a975419 100644
--- a/gcsfs/core.py
+++ b/gcsfs/core.py
@@ -24,6 +24,7 @@ from fsspec.utils import setup_logging, stringify_path
 from . import __version__ as version
 from .checkers import get_consistency_checker
 from .credentials import GoogleCredentials
+from .inventory_report import InventoryReport
 from .retry import retry_request, validate_response
 
 logger = logging.getLogger("gcsfs")
@@ -537,25 +538,32 @@ class GCSFileSystem(AsyncFileSystem):
                 raise FileNotFoundError(path)
         return self._process_object(bucket, res)
 
-    async def _list_objects(self, path, prefix="", versions=False):
+    async def _list_objects(self, path, prefix="", versions=False, **kwargs):
+
         bucket, key, generation = self.split_path(path)
         path = path.rstrip("/")
 
-        try:
-            clisting = self._ls_from_cache(path)
-            hassubdirs = clisting and any(
-                c["name"].rstrip("/") == path and c["type"] == "directory"
-                for c in clisting
-            )
-            if clisting and not hassubdirs:
-                return clisting
-        except FileNotFoundError:
-            # not finding a bucket in list of "my" buckets is OK
-            if key:
-                raise
+        # NOTE: the inventory report logic is experimental.
+        inventory_report_info = kwargs.get("inventory_report_info", None)
+
+        # Only attempt to list from the cache when the user does not use
+        # the inventory report service.
+        if not inventory_report_info:
+            try:
+                clisting = self._ls_from_cache(path)
+                hassubdirs = clisting and any(
+                    c["name"].rstrip("/") == path and c["type"] == "directory"
+                    for c in clisting
+                )
+                if clisting and not hassubdirs:
+                    return clisting
+            except FileNotFoundError:
+                # not finding a bucket in list of "my" buckets is OK
+                if key:
+                    raise
 
         items, prefixes = await self._do_list_objects(
-            path, prefix=prefix, versions=versions
+            path, prefix=prefix, versions=versions, **kwargs
         )
 
         pseudodirs = [
@@ -574,28 +582,170 @@ class GCSFileSystem(AsyncFileSystem):
             else:
                 return []
         out = pseudodirs + items
-        # Don't cache prefixed/partial listings
-        if not prefix:
+
+        use_snapshot_listing = (
+            False
+            if not inventory_report_info
+            else inventory_report_info.get("use_snapshot_listing")
+        )
+
+        # Don't cache prefixed/partial listings, in addition to
+        # not using the inventory report service to do listing directly.
+        if not prefix and use_snapshot_listing is False:
             self.dircache[path] = out
         return out
 
     async def _do_list_objects(
-        self, path, max_results=None, delimiter="/", prefix="", versions=False
+        self, path, max_results=None, delimiter="/", prefix="", versions=False, **kwargs
     ):
+
         """Object listing for the given {bucket}/{prefix}/ path."""
         bucket, _path, generation = self.split_path(path)
         _path = "" if not _path else _path.rstrip("/") + "/"
         prefix = f"{_path}{prefix}" or None
 
+        # Page size of 5000 is officially supported across GCS.
+        default_page_size = 5000
+
+        # NOTE: the inventory report logic is experimental.
+        inventory_report_info = kwargs.get("inventory_report_info", None)
+
+        # Check if the user has configured inventory report option.
+        if inventory_report_info is not None:
+            items, prefixes = await InventoryReport.fetch_snapshot(
+                gcs_file_system=self,
+                inventory_report_info=inventory_report_info,
+                prefix=prefix,
+            )
+
+            use_snapshot_listing = inventory_report_info.get("use_snapshot_listing")
+
+            # If the user wants to rely on the snapshot from the inventory report
+            # for listing, directly return the results.
+            if use_snapshot_listing:
+                return items, prefixes
+
+            # Otherwise, use the snapshot to initiate concurrent listing.
+            return await self._concurrent_list_objects_helper(
+                items=items,
+                bucket=bucket,
+                delimiter=delimiter,
+                prefix=prefix,
+                versions=versions,
+                generation=generation,
+                page_size=default_page_size,
+            )
+
+        # If the user has not configured inventory report, proceed to use
+        # sequential listing.
+        else:
+            return await self._sequential_list_objects_helper(
+                bucket=bucket,
+                delimiter=delimiter,
+                start_offset=None,
+                end_offset=None,
+                prefix=prefix,
+                versions=versions,
+                generation=generation,
+                page_size=default_page_size,
+            )
+
+    async def _concurrent_list_objects_helper(
+        self, items, bucket, delimiter, prefix, versions, generation, page_size
+    ):
+        """
+        Lists objects using coroutines, using the object names from the inventory
+        report to split up the ranges.
+        """
+
+        # Extract out the names of the objects fetched from the inventory report.
+        snapshot_object_names = [item["name"] for item in items]
+        snapshot_object_names = sorted(snapshot_object_names)
+
+        # Determine the number of coroutines needed to concurrent listing.
+        # Ideally, want each coroutine to fetch a single page of objects.
+        num_coroutines = len(snapshot_object_names) // page_size + 1
+        num_objects_per_coroutine = len(snapshot_object_names) // num_coroutines
+
+        start_offsets = []
+        end_offsets = []
+
+        # Calculate the split splits of each coroutine (start offset and end offset).
+        for i in range(num_coroutines):
+            range_start = i * num_objects_per_coroutine
+            if i == num_coroutines - 1:
+                range_end = len(snapshot_object_names)
+            else:
+                range_end = range_start + num_objects_per_coroutine
+
+            if range_start == 0:
+                prefix_start = None
+            else:
+                prefix_start = snapshot_object_names[range_start]
+
+            if range_end == len(snapshot_object_names):
+                prefix_end = None
+            else:
+                prefix_end = snapshot_object_names[range_end]
+
+            start_offsets.append(prefix_start)
+            end_offsets.append(prefix_end)
+
+        # Assign the coroutine all at once, and wait for them to finish listing.
+        results = await asyncio.gather(
+            *[
+                self._sequential_list_objects_helper(
+                    bucket=bucket,
+                    delimiter=delimiter,
+                    start_offset=start_offsets[i],
+                    end_offset=end_offsets[i],
+                    prefix=prefix,
+                    versions=versions,
+                    generation=generation,
+                    page_size=page_size,
+                )
+                for i in range(0, len(start_offsets))
+            ]
+        )
+
+        items = []
+        prefixes = []
+
+        # Concatenate the items and prefixes from each coroutine for final results.
+        for i in range(len(results)):
+            items_from_process, prefixes_from_process = results[i]
+            items.extend(items_from_process)
+            prefixes.extend(prefixes_from_process)
+
+        return items, prefixes
+
+    async def _sequential_list_objects_helper(
+        self,
+        bucket,
+        delimiter,
+        start_offset,
+        end_offset,
+        prefix,
+        versions,
+        generation,
+        page_size,
+    ):
+        """
+        Sequential list objects within the start and end offset range.
+        """
+
         prefixes = []
         items = []
+
         page = await self._call(
             "GET",
             "b/{}/o",
             bucket,
             delimiter=delimiter,
             prefix=prefix,
-            maxResults=max_results,
+            startOffset=start_offset,
+            endOffset=end_offset,
+            maxResults=page_size,
             json_out=True,
             versions="true" if versions or generation else None,
         )
@@ -605,13 +755,16 @@ class GCSFileSystem(AsyncFileSystem):
         next_page_token = page.get("nextPageToken", None)
 
         while next_page_token is not None:
+
             page = await self._call(
                 "GET",
                 "b/{}/o",
                 bucket,
                 delimiter=delimiter,
                 prefix=prefix,
-                maxResults=max_results,
+                startOffset=start_offset,
+                endOffset=end_offset,
+                maxResults=page_size,
                 pageToken=next_page_token,
                 json_out=True,
                 versions="true" if generation else None,
@@ -623,6 +776,7 @@ class GCSFileSystem(AsyncFileSystem):
             next_page_token = page.get("nextPageToken", None)
 
         items = [self._process_object(bucket, i) for i in items]
+
         return items, prefixes
 
     async def _list_buckets(self):
@@ -856,7 +1010,7 @@ class GCSFileSystem(AsyncFileSystem):
         else:
             out = []
             for entry in await self._list_objects(
-                path, prefix=prefix, versions=versions
+                path, prefix=prefix, versions=versions, **kwargs
             ):
                 if versions and "generation" in entry:
                     entry = entry.copy()
@@ -1294,7 +1448,7 @@ class GCSFileSystem(AsyncFileSystem):
 
         if maxdepth:
             # Filter returned objects based on requested maxdepth
-            depth = path.count("/") + maxdepth
+            depth = path.rstrip("/").count("/") + maxdepth
             objects = list(filter(lambda o: o["name"].count("/") <= depth, objects))
 
         if detail:
diff --git a/gcsfs/inventory_report.py b/gcsfs/inventory_report.py
new file mode 100644
index 0000000..b16aa4c
--- /dev/null
+++ b/gcsfs/inventory_report.py
@@ -0,0 +1,657 @@
+from datetime import datetime
+
+
+class InventoryReport:
+    """
+    A utility class for fetching and processing inventory reports from GCS.
+
+    The 'InventoryReport' class provides logic to support logic to fetch
+    inventory reports, and process their content to obtain a final snapshot
+    of objects in the latest inventory reports.
+
+    High-Level Functionality:
+    ------------------------
+    1. Fetching Inventory Reports:
+       - The class offers methods to fetch inventory report configurations and
+         metadata from GCS.
+       - It validates the inventory report information provided by the user.
+       - Inventory report configurations include options for parsing CSV format
+         and specifying the bucket and destination path.
+
+    2. Parsing and Processing Inventory Report Content:
+       - The class processes the raw content of inventory reports to extract
+         object details such as name, size, etc.
+       - It supports listing objects using a snapshot option or filtering
+         based on a user-defined prefix.
+       - The class handles CSV parsing, removes header (if specified), and
+         fetches required object metadata.
+
+    3. Constructing the Final Snapshot:
+       - If the user wishes to use the snapshot to do listing directly, the
+         snapshot will contain the relevant object details and subdirectory
+         prefixes, filtered by the prefix.
+
+       - If the user wishes to use the snapshot as a starting point for async
+         listing, the snapshot will only contain a list of object names,
+         filtered by the prefix.
+
+    Note:
+    -----
+    - The class should only be internally used in the 'GCSFileSystem' as an
+      optional configuration during listing.
+
+    Example Usage:
+    --------------
+    # Should already be instanted in 'core.py'
+    gcs_file_system = GCSFileSystem(...)
+
+    # User defines inventory report information
+    inventory_report_info = {
+        "use_snapshot_listing": True,
+        "location": "us-east1",
+        "id": "inventory_report_id"
+    }
+
+    # User defines a prefix for filtering objects
+    prefix = "prefix/"
+
+    # Fetch the snapshot based on inventory reports
+    items, prefixes = await InventoryReport.fetch_snapshot(
+    gcs_file_system, inventory_report_info, prefix)
+    """
+
+    # HTTP endpoint of the Storage Insights Service.
+    BASE_URL = "https://storageinsights.googleapis.com/v1"
+
+    @classmethod
+    async def fetch_snapshot(cls, gcs_file_system, inventory_report_info, prefix):
+        """
+        Main entry point of the 'InventoryReport' class.
+        Fetches the latest snapshot of objects based on inventory report configuration.
+
+        Parameters:
+            gcs_file_system (GCSFileSystem): An instance of the 'GCSFileSystem'
+            class (see 'core.py').
+            inventory_report_info (dict): A client-configured dictionary
+            containing inventory report information.
+            prefix (str): Listing prefix specified by the client.
+
+        Returns:
+            tuple: A tuple containing two lists: the 'items' list representing
+            object details for the snapshot, and the 'prefixes' list containing
+            subdirectory prefixes.
+
+            Note: when 'use_snapshot_listing' in 'inventory_report_info' is set
+            to False, the 'prefixes' list will be empty, and the 'items' list
+            will contain only the object names.
+        """
+        # Validate the inventory report info that the user passes in.
+        cls._validate_inventory_report_info(inventory_report_info)
+
+        # Parse the inventory report info.
+        use_snapshot_listing = inventory_report_info.get("use_snapshot_listing")
+        inventory_report_location = inventory_report_info.get("location")
+        inventory_report_id = inventory_report_info.get("id")
+
+        # Fetch the inventory report configuration.
+        raw_inventory_report_config = await cls._fetch_raw_inventory_report_config(
+            gcs_file_system=gcs_file_system,
+            location=inventory_report_location,
+            id=inventory_report_id,
+        )
+
+        # Parse the inventory report configuration.
+        inventory_report_config = cls._parse_raw_inventory_report_config(
+            raw_inventory_report_config=raw_inventory_report_config,
+            use_snapshot_listing=use_snapshot_listing,
+        )
+
+        # Use the config to fetch all inventory report medadata.
+        unsorted_inventory_report_metadata = await cls._fetch_inventory_report_metadata(
+            gcs_file_system=gcs_file_system,
+            inventory_report_config=inventory_report_config,
+        )
+
+        # Sort the metadata based on reverse created time order.
+        inventory_report_metadata = cls._sort_inventory_report_medatada(
+            unsorted_inventory_report_metadata=unsorted_inventory_report_metadata
+        )
+
+        # Download the most recent inventory reports in raw form.
+        bucket = inventory_report_config.bucket
+        inventory_report_content = await cls._download_inventory_report_content(
+            gcs_file_system=gcs_file_system,
+            inventory_report_metadata=inventory_report_metadata,
+            bucket=bucket,
+        )
+
+        # Parse the raw inventory reports into snapshot objects.
+        objects = cls._parse_inventory_report_content(
+            gcs_file_system=gcs_file_system,
+            inventory_report_content=inventory_report_content,
+            inventory_report_config=inventory_report_config,
+            use_snapshot_listing=use_snapshot_listing,
+            bucket=bucket,
+        )
+
+        # Construct the final snapshot based on the fetched objects.
+        snapshot = cls._construct_final_snapshot(
+            objects=objects, prefix=prefix, use_snapshot_listing=use_snapshot_listing
+        )
+
+        # Return the final snapshot.
+        return snapshot
+
+    def _validate_inventory_report_info(inventory_report_info):
+        """
+        Validates the inventory report information dictionary that user
+        passes in.
+
+        Parameters:
+            inventory_report_info (dict): A dictionary containing the inventory
+            report information with the following keys:
+                - "use_snapshot_listing" (bool): A flag indicating whether
+                  to use snapshot listing in the inventory report.
+                - "location" (str): The location of the inventory report in GCS.
+                - "id" (str): The ID of the inventory report in GCS.
+
+        Raises:
+            ValueError: If any required key (use_snapshot_listing, location, id)
+            is missing from the inventory_report_info dictionary.
+        """
+        if "use_snapshot_listing" not in inventory_report_info:
+            raise ValueError("Use snapshot listing is not configured.")
+        if "location" not in inventory_report_info:
+            raise ValueError("Inventory report location is not configured.")
+        if "id" not in inventory_report_info:
+            raise ValueError("Inventory report id is not configured.")
+
+    async def _fetch_raw_inventory_report_config(gcs_file_system, location, id):
+        """
+        Fetches the raw inventory report configuration from GCS based on the
+        specified location and ID.
+
+        Parameters:
+            gcs_file_system (GCSFileSystem): An instance of the 'GCSFileSystem'
+            class (see 'core.py').
+            location (str): The location of the inventory report in GCS.
+            id (str): The ID of the inventory report in GCS.
+
+        Returns:
+            dict: A dictionary containing the raw inventory report
+            configuration retrieved from GCS.
+
+        Raises:
+            Exception: If there is an error while fetching the inventory
+            report configuration.
+        """
+        project = gcs_file_system.project
+        url = "{}/projects/{}/locations/{}/reportConfigs/{}"
+        url = url.format(InventoryReport.BASE_URL, project, location, id)
+        try:
+            raw_inventory_report_config = await gcs_file_system._call(
+                "GET", url, json_out=True
+            )
+            return raw_inventory_report_config
+        except Exception as e:
+            raise ValueError(
+                f"Error encountered when fetching inventory report config: {e}."
+            )
+
+    def _parse_raw_inventory_report_config(
+        raw_inventory_report_config, use_snapshot_listing
+    ):
+        """
+        Parses the raw inventory report configuration and validates its properties.
+
+        Parameters:
+            raw_inventory_report_config (dict): A dictionary containing the raw
+            inventory report configuration retrieved from GCS.
+            use_snapshot_listing (bool): A flag indicating whether to use snapshot
+            listing in the inventory report.
+
+        Returns:
+            InventoryReportConfig: An instance of the InventoryReportConfig
+            class representing the parsed inventory report configuration.
+
+        Raises:
+            ValueError: If the current date is outside the start and
+            end range specified in the inventory report config.
+            ValueError: If the "name" field is not present in the metadata
+            fields of the report config.
+            ValueError: If "size" field is not present in the metadata
+            fields and use_snapshot_listing is True.
+        """
+        # Parse the report config.
+        frequency_options = raw_inventory_report_config.get("frequencyOptions")
+        start_date = InventoryReport._convert_obj_to_date(
+            frequency_options.get("startDate")
+        )
+        end_date = InventoryReport._convert_obj_to_date(
+            frequency_options.get("endDate")
+        )
+        object_metadata_report_options = raw_inventory_report_config.get(
+            "objectMetadataReportOptions"
+        )
+        storage_destination_options = object_metadata_report_options.get(
+            "storageDestinationOptions"
+        )
+
+        # Save relevant report config properties.
+        csv_options = raw_inventory_report_config.get("csvOptions")
+        bucket = storage_destination_options.get("bucket")
+        destination_path = storage_destination_options.get("destinationPath")
+        metadata_fields = object_metadata_report_options.get("metadataFields")
+
+        # Validate date, making sure the current date is within the start and end range.
+        today = datetime.now()
+        if today < start_date or today > end_date:
+            raise ValueError(
+                f"Current date {today} is outside the range \
+                {start_date} and {end_date} specified by the inventory report config."
+            )
+
+        # Validate object name exists in the metadata fields.
+        # Note that the size field is mandated to be included in the
+        # config when the client sets up the inventory report.
+        obj_name_idx = metadata_fields.index("name")
+
+        # If the user wants to do listing based on the snapshot, also
+        # validate the report contains size metadata for each object.
+        if use_snapshot_listing:
+            try:
+                metadata_fields.index("size")
+            except ValueError:
+                raise ValueError(
+                    "If you want to use the snapshot for listing, the object size \
+                        metadata has to be included in the inventory report."
+                )
+
+        # Finally, construct and return the inventory report config.
+        inventory_report_config = InventoryReportConfig(
+            csv_options=csv_options,
+            bucket=bucket,
+            destination_path=destination_path,
+            metadata_fields=metadata_fields,
+            obj_name_idx=obj_name_idx,
+        )
+
+        return inventory_report_config
+
+    async def _fetch_inventory_report_metadata(
+        gcs_file_system, inventory_report_config
+    ):
+        """
+        Fetches all inventory report metadata from GCS based on the specified
+        inventory report config.
+
+        Parameters:
+            gcs_file_system (GCSFileSystem): An instance of the 'GCSFileSystem'
+            class (see 'core.py').
+            inventory_report_config (InventoryReportConfig): An instance of
+            the InventoryReportConfig class representing the inventory report
+            configuration.
+
+        Returns:
+            list: A list containing dictionaries representing the metadata of
+            objects from the inventory reports.
+
+        Raises:
+            ValueError: If the fetched inventory reports are empty.
+        """
+        # There might be multiple inventory reports in the bucket.
+        inventory_report_metadata = []
+
+        # Extract out bucket and destination path of the inventory reports.
+        bucket = inventory_report_config.bucket
+        destination_path = inventory_report_config.destination_path
+
+        # Fetch the first page.
+        page = await gcs_file_system._call(
+            "GET", "b/{}/o", bucket, prefix=destination_path, json_out=True
+        )
+
+        inventory_report_metadata.extend(page.get("items", []))
+        next_page_token = page.get("nextPageToken", None)
+
+        # Keep fetching new pages as long as next page token exists.
+        # Note that the iteration in the while loop should most likely
+        # be minimal. For reference, a million objects is split up into
+        # two reports, and if the report is generated daily, then in a year,
+        # there will be roughly ~700 reports generated, which will still be
+        # fetched in a single page.
+        while next_page_token is not None:
+            page = await gcs_file_system._call(
+                "GET",
+                "b/{}/o",
+                bucket,
+                prefix=destination_path,
+                json_out=True,
+                pageToken=next_page_token,
+            )
+
+            inventory_report_metadata.extend(page.get("items", []))
+            next_page_token = page.get("nextPageToken", None)
+
+        # If no reports are fetched, indicates there is an error.
+        if len(inventory_report_metadata) == 0:
+            raise ValueError(
+                "No inventory reports to fetch. Check if \
+                your inventory report is set up correctly."
+            )
+
+        return inventory_report_metadata
+
+    def _sort_inventory_report_metadata(unsorted_inventory_report_metadata):
+        """
+        Sorts the inventory report metadata based on the 'timeCreated' field
+        in reverse chronological order.
+
+        Parameters:
+            unsorted_inventory_report_metadata (list): A list of dictionaries
+            representing the metadata of objects from the inventory reports.
+
+        Returns:
+            list: A sorted list of dictionaries representing the inventory
+            report metadata, sorted in reverse chronological order based
+            on 'timeCreated'.
+        """
+        return sorted(
+            unsorted_inventory_report_metadata,
+            key=lambda ir: InventoryReport._convert_str_to_datetime(
+                ir.get("timeCreated")
+            ),
+            reverse=True,
+        )
+
+    async def _download_inventory_report_content(
+        gcs_file_system, inventory_report_metadata, bucket
+    ):
+        """
+        Downloads the most recent inventory report content from GCS based on
+        the inventory report metadata.
+
+        Parameters:
+            gcs_file_system (GCSFileSystem): An instance of the 'GCSFileSystem'
+            class (see 'core.py').
+            inventory_report_metadata (list): A list of dictionaries
+            representing the metadata of objects from the inventory reports.
+            bucket (str): The name of the GCS bucket containing
+            the inventory reports.
+
+        Returns:
+            list: A list containing the content of the most recent inventory
+            report as strings.
+        """
+        # Get the most recent inventory report date.
+        most_recent_inventory_report = inventory_report_metadata[0]
+        most_recent_date = InventoryReport._convert_str_to_datetime(
+            most_recent_inventory_report.get("timeCreated")
+        ).date()
+
+        inventory_report_content = []
+
+        # Run a for loop here, since there might be multiple inventory reports
+        # generated on the same day. For reference, 1 million objects will be
+        # split into only 2 inventory reports, so it is very rare that there
+        # will be many inventory reports on the same day. But including this
+        # logic for robustness.
+        for metadata in inventory_report_metadata:
+
+            inventory_report_date = InventoryReport._convert_str_to_datetime(
+                metadata["timeCreated"]
+            ).date()
+
+            if inventory_report_date == most_recent_date:
+
+                # Download the raw inventory report if the date matches.
+                # Header is not needed, we only need to process and store
+                # the content.
+                _header, encoded_content = await gcs_file_system._call(
+                    "GET", "b/{}/o/{}", bucket, metadata.get("name"), alt="media"
+                )
+
+                # Decode the binary content into string for the content.
+                decoded_content = encoded_content.decode()
+
+                inventory_report_content.append(decoded_content)
+
+        return inventory_report_content
+
+    def _parse_inventory_report_content(
+        gcs_file_system,
+        inventory_report_content,
+        inventory_report_config,
+        use_snapshot_listing,
+        bucket,
+    ):
+        """
+        Parses the raw inventory report content and extracts object details.
+
+        Parameters:
+            gcs_file_system (GCSFileSystem): An instance of the 'GCSFileSystem'
+            class (see 'core.py').
+            inventory_report_content (list): A list of strings containing the
+            raw content of the inventory report.
+            inventory_report_config (InventoryReportConfig): An instance of the
+            InventoryReportConfig class representing the inventory report
+            configuration.
+            use_snapshot_listing (bool): A flag indicating whether to use snapshot
+            listing in the inventory report.
+            bucket (str): The name of the GCS bucket containing the inventory
+            reports.
+
+        Returns:
+            list: A list of dictionaries representing object details parsed
+            from the inventory report content.
+        """
+        # Get the csv configuration for each inventory report.
+        csv_options = inventory_report_config.csv_options
+        record_separator = csv_options.get("recordSeparator", "\n")
+        delimiter = csv_options.get("delimiter", ",")
+        header_required = csv_options.get("headerRequired", False)
+
+        objects = []
+
+        for content in inventory_report_content:
+
+            # Split the content into lines based on the specified separator.
+            lines = content.split(record_separator)
+
+            # Remove the header, if present.
+            if header_required:
+                lines = lines[1:]
+
+            # Parse each line of the inventory report.
+            for line in lines:
+
+                obj = InventoryReport._parse_inventory_report_line(
+                    inventory_report_line=line,
+                    use_snapshot_listing=use_snapshot_listing,
+                    gcs_file_system=gcs_file_system,
+                    inventory_report_config=inventory_report_config,
+                    delimiter=delimiter,
+                    bucket=bucket,
+                )
+
+                objects.append(obj)
+
+        return objects
+
+    def _parse_inventory_report_line(
+        inventory_report_line,
+        use_snapshot_listing,
+        gcs_file_system,
+        inventory_report_config,
+        delimiter,
+        bucket,
+    ):
+        """
+        Parses a single line of the inventory report and extracts object details.
+
+        Parameters:
+            inventory_report_line (str): A string representing a single line of
+            the raw content from the inventory report.
+            use_snapshot_listing (bool): A flag indicating whether to use snapshot
+            listing in the inventory report.
+            gcs_file_system (GCSFileSystem): An instance of the 'GCSFileSystem'
+            class (see 'core.py').
+            inventory_report_config (InventoryReportConfig): An instance of the
+            InventoryReportConfig class representing the inventory report
+            configuration.
+            delimiter (str): The delimiter used in the inventory report content
+            to separate fields.
+            bucket (str): The name of the GCS bucket containing the inventory
+            reports.
+
+        Returns:
+            dict: A dictionary representing object details parsed from the
+            inventory report line.
+        """
+        obj_name_idx = inventory_report_config.obj_name_idx
+        metadata_fields = inventory_report_config.metadata_fields
+
+        # If the client wants to do listing from the snapshot, we need
+        # to fetch all the metadata for each object. Otherwise, we only
+        # need to fetch the name.
+        if use_snapshot_listing is True:
+            obj = gcs_file_system._process_object(
+                {
+                    key: value
+                    for key, value in zip(
+                        metadata_fields, inventory_report_line.strip().split(delimiter)
+                    )
+                },
+                bucket,
+            )
+        else:
+            obj = {"name": inventory_report_line.strip().split(delimiter)[obj_name_idx]}
+
+        return obj
+
+    def _construct_final_snapshot(objects, prefix, use_snapshot_listing):
+        """
+        Constructs the final snapshot based on the retrieved objects and prefix.
+
+        Parameters:
+            objects (list): A list of dictionaries representing object details
+            from the inventory report.
+            prefix (str): A prefix used to filter objects in the snapshot based
+            on their names.
+            use_snapshot_listing (bool): A flag indicating whether to use snapshot
+            listing in the inventory report.
+
+        Returns:
+            tuple: A tuple containing two lists: the 'items' list representing
+            object details for the snapshot, and the 'prefixes' list containing
+            subdirectory prefixes. If 'use_snapshot_listing' is set to False,
+            'prefix' will also be empty, and 'items' will contains the object
+            names in the snapshot.
+        """
+        if prefix is None:
+            prefix = ""
+
+        # Filter the prefix and returns the list if the user does not want to use
+        # the snapshot for listing.
+        if use_snapshot_listing is False:
+
+            return [obj for obj in objects if obj.get("name").startswith(prefix)], []
+
+        else:
+
+            # If the user wants to use the snapshot, generate both the items and
+            # prefixes manually.
+            items = []
+            prefixes = set()
+
+            for obj in objects:
+
+                # Fetch the name of the object.
+                obj_name = obj.get("name")
+
+                # If the object name doesn't start with the prefix, continue.
+                # In the case where prefix is empty, it will always return
+                # true (which is the expected behavior).
+                if not obj_name.startswith(prefix):
+                    continue
+
+                # Remove the prefix.
+                object_name_no_prefix = obj_name[len(prefix) :]
+
+                # Determine whether the object name is a directory.
+                first_delimiter_idx = object_name_no_prefix.find("/")
+
+                # If not, then append it to items.
+                if first_delimiter_idx == -1:
+                    items.append(obj)
+                    continue
+
+                # If it is, recompose the directory and add to the prefix set.
+                dir = object_name_no_prefix[:first_delimiter_idx]
+                obj_prefix = (
+                    prefix.rstrip("/")
+                    + ("" if prefix == "" else "/")
+                    + dir
+                    + ("" if dir == "" else "/")
+                )
+                prefixes.add(obj_prefix)
+
+        return items, list(prefixes)
+
+    @staticmethod
+    def _convert_obj_to_date(obj):
+        """
+        Converts a dictionary representing a date object to a datetime object.
+
+        Parameters:
+            obj (dict): A dictionary representing a date object with keys "day",
+            "month", and "year".
+
+        Returns:
+            datetime: A datetime object representing the converted date.
+        """
+        day = obj["day"]
+        month = obj["month"]
+        year = obj["year"]
+        return datetime(year, month, day)
+
+    @staticmethod
+    def _convert_str_to_datetime(str):
+        """
+        Converts an ISO-formatted date string to a datetime object.
+
+        Parameters:
+            date_string (str): An ISO-formatted date string with or without
+            timezone information (Z).
+
+        Returns:
+            datetime: A datetime object representing the converted date and time.
+        """
+        return datetime.fromisoformat(str.replace("Z", "+00:00"))
+
+
+class InventoryReportConfig(object):
+    """
+    Represents the configuration for fetching inventory reports.
+
+    Attributes:
+        csv_options (dict): A dictionary containing options for parsing CSV
+        format in the inventory reports.
+        bucket (str): The name of the GCS bucket from which to fetch the
+        inventory reports.
+        destination_path (str): The path within the GCS bucket where the
+        inventory reports are stored.
+        metadata_fields (list): A list of strings representing metadata
+        fields to be extracted from the inventory reports.
+        obj_name_idx (int): The index of the "name" field in the 'metadata_fields'
+        list, used to identify object names.
+    """
+
+    def __init__(
+        self, csv_options, bucket, destination_path, metadata_fields, obj_name_idx
+    ):
+        self.csv_options = csv_options
+        self.bucket = bucket
+        self.destination_path = destination_path
+        self.metadata_fields = metadata_fields
+        self.obj_name_idx = obj_name_idx


Test patch:
diff --git a/gcsfs/tests/derived/__init__.py b/gcsfs/tests/derived/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/gcsfs/tests/derived/gcsfs_fixtures.py b/gcsfs/tests/derived/gcsfs_fixtures.py
new file mode 100644
index 0000000..21ce243
--- /dev/null
+++ b/gcsfs/tests/derived/gcsfs_fixtures.py
@@ -0,0 +1,42 @@
+import fsspec
+import pytest
+from fsspec.tests.abstract import AbstractFixtures
+
+from gcsfs.core import GCSFileSystem
+from gcsfs.tests.conftest import allfiles
+from gcsfs.tests.settings import TEST_BUCKET
+
+
+class GcsfsFixtures(AbstractFixtures):
+    @pytest.fixture(scope="class")
+    def fs(self, docker_gcs):
+        GCSFileSystem.clear_instance_cache()
+        gcs = fsspec.filesystem("gcs", endpoint_url=docker_gcs)
+        try:
+            # ensure we're empty.
+            try:
+                gcs.rm(TEST_BUCKET, recursive=True)
+            except FileNotFoundError:
+                pass
+            try:
+                gcs.mkdir(TEST_BUCKET)
+            except Exception:
+                pass
+
+            gcs.pipe({TEST_BUCKET + "/" + k: v for k, v in allfiles.items()})
+            gcs.invalidate_cache()
+            yield gcs
+        finally:
+            try:
+                gcs.rm(gcs.find(TEST_BUCKET))
+                gcs.rm(TEST_BUCKET)
+            except:  # noqa: E722
+                pass
+
+    @pytest.fixture
+    def fs_path(self):
+        return TEST_BUCKET
+
+    @pytest.fixture
+    def supports_empty_directories(self):
+        return False
diff --git a/gcsfs/tests/derived/gcsfs_test.py b/gcsfs/tests/derived/gcsfs_test.py
new file mode 100644
index 0000000..2e3f8d4
--- /dev/null
+++ b/gcsfs/tests/derived/gcsfs_test.py
@@ -0,0 +1,15 @@
+import fsspec.tests.abstract as abstract
+
+from gcsfs.tests.derived.gcsfs_fixtures import GcsfsFixtures
+
+
+class TestGcsfsCopy(abstract.AbstractCopyTests, GcsfsFixtures):
+    pass
+
+
+class TestGcsfsGet(abstract.AbstractGetTests, GcsfsFixtures):
+    pass
+
+
+class TestGcsfsPut(abstract.AbstractPutTests, GcsfsFixtures):
+    pass
diff --git a/gcsfs/tests/test_core.py b/gcsfs/tests/test_core.py
index 51956e8..69ff0e3 100644
--- a/gcsfs/tests/test_core.py
+++ b/gcsfs/tests/test_core.py
@@ -274,7 +274,7 @@ def test_gcs_glob(gcs):
     fn = TEST_BUCKET + "/nested/file1"
     assert fn not in gcs.glob(TEST_BUCKET + "/")
     assert fn not in gcs.glob(TEST_BUCKET + "/*")
-    assert fn in gcs.glob(TEST_BUCKET + "/nested/")
+    assert fn not in gcs.glob(TEST_BUCKET + "/nested/")
     assert fn in gcs.glob(TEST_BUCKET + "/nested/*")
     assert fn in gcs.glob(TEST_BUCKET + "/nested/file*")
     assert fn in gcs.glob(TEST_BUCKET + "/*/*")
diff --git a/gcsfs/tests/test_inventory_report.py b/gcsfs/tests/test_inventory_report.py
new file mode 100644
index 0000000..c01e3d6
--- /dev/null
+++ b/gcsfs/tests/test_inventory_report.py
@@ -0,0 +1,763 @@
+import asyncio
+from datetime import datetime, timedelta
+from unittest import mock
+
+import pytest
+
+from gcsfs.core import GCSFileSystem
+from gcsfs.inventory_report import InventoryReport, InventoryReportConfig
+
+
+class TestInventoryReport(object):
+    """
+    Unit tests for the inventory report logic, see 'inventory_report.py'.
+
+    The test cases follow the same ordering as the methods in `inventory.report.py`.
+    Each method is covered by either one or more parametrized test cases. Some
+    methods include a setup method just above them.
+    """
+
+    @pytest.mark.parametrize(
+        "inventory_report_info, expected_error",
+        [
+            # Check whether missing inventory report info will raise exception.
+            (
+                {"location": "us-west", "id": "123"},
+                "Use snapshot listing is not configured.",
+            ),
+            (
+                {"use_snapshot_listing": True, "id": "123"},
+                "Inventory report location is not configured.",
+            ),
+            (
+                {"use_snapshot_listing": True, "location": "us-west"},
+                "Inventory report id is not configured.",
+            ),
+            # Check complete inventory report info will not raise exception.
+            ({"use_snapshot_listing": True, "location": "us-west", "id": "123"}, None),
+        ],
+    )
+    def test_validate_inventory_report_info(
+        self, inventory_report_info, expected_error
+    ):
+        if expected_error is not None:
+            with pytest.raises(ValueError) as e_info:
+                InventoryReport._validate_inventory_report_info(
+                    inventory_report_info=inventory_report_info
+                )
+                assert str(e_info.value) == expected_error
+        else:
+            # If no error is expected, we simply call the function
+            # to ensure no exception is raised.
+            InventoryReport._validate_inventory_report_info(
+                inventory_report_info=inventory_report_info
+            )
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "location, id, exception, expected_result",
+        [
+            # Test no error fetching proceeds normally.
+            ("us-west", "id1", None, {"config": "config1"}),
+            # Test if the exception is caught successfully.
+            ("us-west", "id2", Exception("fetch error"), None),
+        ],
+    )
+    async def test_fetch_raw_inventory_report_config(
+        self, location, id, exception, expected_result
+    ):
+
+        # Mocking the gcs_file_system.
+        gcs_file_system = mock.MagicMock()
+        gcs_file_system.project = "project"
+
+        # Mocking gcs_file_system._call.
+        if exception is not None:
+            gcs_file_system._call = mock.MagicMock(side_effect=exception)
+        else:
+            return_value = asyncio.Future()
+            return_value.set_result(expected_result)
+            gcs_file_system._call = mock.MagicMock(return_value=return_value)
+
+        if exception is not None:
+            with pytest.raises(Exception) as e_info:
+                await InventoryReport._fetch_raw_inventory_report_config(
+                    gcs_file_system=gcs_file_system, location=location, id=id
+                )
+                assert str(e_info.value) == str(exception)
+        else:
+            result = await InventoryReport._fetch_raw_inventory_report_config(
+                gcs_file_system=gcs_file_system, location=location, id=id
+            )
+            gcs_file_system._call.assert_called_once_with(
+                "GET", mock.ANY, json_out=True
+            )
+            assert result == expected_result
+
+    def test_parse_raw_inventory_report_config_invalid_date(self):
+
+        today = datetime.today().date()
+
+        # Get tomorrow's date.
+        tomorrow = today + timedelta(days=1)
+
+        # Get the date a week later.
+        next_week = today + timedelta(days=7)
+
+        raw_inventory_report_config = {
+            "frequencyOptions": {
+                "startDate": {
+                    "day": tomorrow.day,
+                    "month": tomorrow.month,
+                    "year": tomorrow.year,
+                },
+                "endDate": {
+                    "day": next_week.day,
+                    "month": next_week.month,
+                    "year": next_week.year,
+                },
+            },
+            "objectMetadataReportOptions": mock.MagicMock(),
+            "csvOptions": mock.MagicMock(),
+        }
+
+        # If the current date is outside the ranges in the inventory report
+        # an exception should be raised.
+        with pytest.raises(ValueError):
+            InventoryReport._parse_raw_inventory_report_config(
+                raw_inventory_report_config=raw_inventory_report_config,
+                use_snapshot_listing=mock.MagicMock(),
+            )
+
+    def test_parse_raw_inventory_report_config_missing_metadata_fields(self):
+
+        raw_inventory_report_config = {
+            "frequencyOptions": mock.MagicMock(),
+            "objectMetadataReportOptions": {
+                "metadataFields": ["project", "bucket", "name"],
+                "storageDestinationOptions": mock.MagicMock(),
+            },
+            "csvOptions": mock.MagicMock(),
+        }
+
+        # When the user wants to use snapshot listing, but object size is not
+        # included in the inventory reports, an exception should be raised.
+        with pytest.raises(ValueError):
+            InventoryReport._parse_raw_inventory_report_config(
+                raw_inventory_report_config=raw_inventory_report_config,
+                use_snapshot_listing=True,
+            )
+
+    def test_parse_raw_inventory_report_config_returns_correct_config(self):
+
+        bucket = "bucket"
+        destination_path = "path/to/inventory-report"
+        metadata_fields = ["project", "bucket", "name", "size"]
+        obj_name_idx = metadata_fields.index("name")
+        today = datetime.today().date()
+        yesterday = today - timedelta(days=1)
+        tomorrow = today + timedelta(days=1)
+        use_snapshot_listing = False
+
+        csv_options = {
+            "recordSeparator": "\n",
+            "delimiter": ",",
+            "headerRequired": False,
+        }
+
+        raw_inventory_report_config = {
+            "frequencyOptions": {
+                "startDate": {
+                    "day": yesterday.day,
+                    "month": yesterday.month,
+                    "year": yesterday.year,
+                },
+                "endDate": {
+                    "day": tomorrow.day,
+                    "month": tomorrow.month,
+                    "year": tomorrow.year,
+                },
+            },
+            "objectMetadataReportOptions": {
+                "metadataFields": metadata_fields,
+                "storageDestinationOptions": {
+                    "bucket": bucket,
+                    "destinationPath": destination_path,
+                },
+            },
+            "csvOptions": csv_options,
+        }
+
+        try:
+            inventory_report_config = (
+                InventoryReport._parse_raw_inventory_report_config(
+                    raw_inventory_report_config=raw_inventory_report_config,
+                    use_snapshot_listing=use_snapshot_listing,
+                )
+            )
+
+            assert isinstance(inventory_report_config, InventoryReportConfig)
+
+            assert inventory_report_config.csv_options == csv_options
+            assert inventory_report_config.bucket == bucket
+            assert inventory_report_config.destination_path == destination_path
+            assert inventory_report_config.metadata_fields == metadata_fields
+            assert inventory_report_config.obj_name_idx == obj_name_idx
+
+        except Exception as e:
+            pytest.fail(f"Unexpected exception: {e}.")
+
+    @pytest.mark.asyncio
+    async def test_fetch_inventory_report_metadata_no_reports(self):
+
+        # Create a mock for GCSFileSystem.
+        gcs_file_system = mock.MagicMock(spec=GCSFileSystem)
+
+        # Mock the _call method to return a page with two items
+        # and then a page with one item and without next page token.
+        gcs_file_system._call.side_effect = [{"items": [], "nextPageToken": None}]
+
+        # Create a mock for InventoryReportConfig.
+        inventory_report_config = mock.MagicMock(spec=InventoryReportConfig)
+        inventory_report_config.bucket = "bucket_name"
+        inventory_report_config.destination_path = "destination_path"
+
+        # If no inventory report metadata is fetched, an exception should be raised.
+        match = "No inventory reports to fetch. Check if \
+                your inventory report is set up correctly."
+        with pytest.raises(ValueError, match=match):
+            await InventoryReport._fetch_inventory_report_metadata(
+                gcs_file_system=gcs_file_system,
+                inventory_report_config=inventory_report_config,
+            )
+
+    @pytest.mark.asyncio
+    async def test_fetch_inventory_report_metadata_multiple_calls(self):
+
+        # Create a mock for GCSFileSystem.
+        gcs_file_system = mock.MagicMock(spec=GCSFileSystem)
+
+        # Mock the _call method to return a page with two items
+        # and then a page with one item and without next page token.
+        gcs_file_system._call.side_effect = [
+            {"items": ["item1", "item2"], "nextPageToken": "token1"},
+            {"items": ["item3"], "nextPageToken": None},
+        ]
+
+        # Create a mock for InventoryReportConfig.
+        inventory_report_config = mock.MagicMock(spec=InventoryReportConfig)
+        inventory_report_config.bucket = "bucket_name"
+        inventory_report_config.destination_path = "destination_path"
+
+        result = await InventoryReport._fetch_inventory_report_metadata(
+            gcs_file_system=gcs_file_system,
+            inventory_report_config=inventory_report_config,
+        )
+
+        # Check that _call was called with the right arguments.
+        calls = [
+            mock.call(
+                "GET", "b/{}/o", "bucket_name", prefix="destination_path", json_out=True
+            ),
+            mock.call(
+                "GET",
+                "b/{}/o",
+                "bucket_name",
+                prefix="destination_path",
+                pageToken="token1",
+                json_out=True,
+            ),
+        ]
+        gcs_file_system._call.assert_has_calls(calls)
+
+        # Check that the function correctly processed the response
+        # and returned the right result.
+        assert result == ["item1", "item2", "item3"]
+
+    @pytest.mark.parametrize(
+        "unsorted_inventory_report_metadata, expected",
+        [
+            (
+                # Input.
+                [
+                    {"timeCreated": "2023-08-01T12:00:00Z"},
+                    {"timeCreated": "2023-08-02T12:00:00Z"},
+                    {"timeCreated": "2023-08-03T12:00:00Z"},
+                ],
+                # Expected output.
+                [
+                    {"timeCreated": "2023-08-03T12:00:00Z"},
+                    {"timeCreated": "2023-08-02T12:00:00Z"},
+                    {"timeCreated": "2023-08-01T12:00:00Z"},
+                ],
+            ),
+            (
+                # Input.
+                [
+                    {"timeCreated": "2023-08-01T12:00:00Z"},
+                    {"timeCreated": "2023-07-31T12:00:00Z"},
+                    {"timeCreated": "2023-08-02T12:00:00Z"},
+                ],
+                # Expected output.
+                [
+                    {"timeCreated": "2023-08-02T12:00:00Z"},
+                    {"timeCreated": "2023-08-01T12:00:00Z"},
+                    {"timeCreated": "2023-07-31T12:00:00Z"},
+                ],
+            ),
+        ],
+    )
+    def test_sort_inventory_report_metadata(
+        self, unsorted_inventory_report_metadata, expected
+    ):
+        result = InventoryReport._sort_inventory_report_metadata(
+            unsorted_inventory_report_metadata=unsorted_inventory_report_metadata
+        )
+        assert result == expected
+
+    @pytest.fixture(
+        params=[
+            # Unique most recent day, same datetime.
+            (
+                [
+                    {"name": "report1", "timeCreated": "2023-08-02T12:00:00.000Z"},
+                    {"name": "report2", "timeCreated": "2023-08-01T12:00:00.000Z"},
+                ],
+                # Expected results.
+                ["report1"],
+            ),
+            # Multiple most recent day, same datetime.
+            (
+                [
+                    {"name": "report1", "timeCreated": "2023-08-02T12:00:00.000Z"},
+                    {"name": "report2", "timeCreated": "2023-08-02T12:00:00.000Z"},
+                    {"name": "report3", "timeCreated": "2023-08-01T12:00:00.000Z"},
+                ],
+                # Expected results.
+                ["report1", "report2"],
+            ),
+            # Multiple most recent day, different datetimes (same day, different hour).
+            (
+                [
+                    {"name": "report1", "timeCreated": "2023-08-02T12:00:00.000Z"},
+                    {"name": "report2", "timeCreated": "2023-08-02T11:00:00.000Z"},
+                    {"name": "report3", "timeCreated": "2023-08-01T12:00:00.000Z"},
+                ],
+                # Expected results.
+                ["report1", "report2"],
+            ),
+        ]
+    )
+    def download_inventory_report_content_setup(self, request):
+        bucket = "bucket"
+        gcs_file_system = mock.MagicMock()
+        inventory_report_metadata, expected_reports = request.param
+
+        # We are accessing the third argument as the return value,
+        # since it is the object name in the function.
+        # We are also encoding the content, since the actual method call needs
+        # to decode the content.
+        async_side_effect = mock.AsyncMock(
+            side_effect=lambda *args, **kwargs: ("_header", args[3].encode())
+        )
+        gcs_file_system._call = async_side_effect
+        return gcs_file_system, inventory_report_metadata, bucket, expected_reports
+
+    @pytest.mark.asyncio
+    async def test_download_inventory_report_content(
+        self, download_inventory_report_content_setup
+    ):
+
+        (
+            gcs_file_system,
+            inventory_report_metadata,
+            bucket,
+            expected_reports,
+        ) = download_inventory_report_content_setup
+
+        result = await InventoryReport._download_inventory_report_content(
+            gcs_file_system=gcs_file_system,
+            inventory_report_metadata=inventory_report_metadata,
+            bucket=bucket,
+        )
+
+        # Verify the mocked downloaded reports match (ordering does not matter).
+        assert sorted(result) == sorted(expected_reports)
+
+    @pytest.mark.parametrize(
+        "inventory_report_line, use_snapshot_listing, \
+        inventory_report_config_attrs, delimiter, bucket, expected",
+        [
+            # Test case 1: use snapshot listing with specific metadata
+            # fields and delimiter.
+            (
+                "object1,value1,value2",
+                True,
+                {"obj_name_idx": 0, "metadata_fields": ["name", "field1", "field2"]},
+                ",",
+                "bucket",
+                {"name": "object1", "field1": "value1", "field2": "value2"},
+            ),
+            # Test case 2: do not use snapshot listing and only fetch the name.
+            (
+                "object1,value1,value2",
+                False,
+                {"obj_name_idx": 0, "metadata_fields": ["name", "field1", "field2"]},
+                ",",
+                "bucket",
+                {"name": "object1"},
+            ),
+        ],
+    )
+    def test_parse_inventory_report_line(
+        self,
+        inventory_report_line,
+        use_snapshot_listing,
+        inventory_report_config_attrs,
+        delimiter,
+        bucket,
+        expected,
+    ):
+
+        # Mock InventoryReportConfig.
+        inventory_report_config = mock.MagicMock(spec=InventoryReportConfig)
+        inventory_report_config.obj_name_idx = inventory_report_config_attrs.get(
+            "obj_name_idx"
+        )
+        inventory_report_config.metadata_fields = inventory_report_config_attrs.get(
+            "metadata_fields"
+        )
+
+        # Mock GCSFileSystem.
+        gcs_file_system = mock.MagicMock(spec=GCSFileSystem)
+        gcs_file_system._process_object = mock.Mock(side_effect=lambda obj, bucket: obj)
+
+        result = InventoryReport._parse_inventory_report_line(
+            inventory_report_line=inventory_report_line,
+            use_snapshot_listing=use_snapshot_listing,
+            gcs_file_system=gcs_file_system,
+            inventory_report_config=inventory_report_config,
+            delimiter=delimiter,
+            bucket=bucket,
+        )
+
+        assert result == expected
+
+    @pytest.fixture(
+        params=[
+            # One file, one lines.
+            (["header \n line1"], {"recordSeparator": "\n", "headerRequired": True}),
+            (["line1"], {"recordSeparator": "\n", "headerRequired": False}),
+            (
+                ["header \r\n line1"],
+                {"recordSeparator": "\r\n", "headerRequired": True},
+            ),
+            (["line1"], {"recordSeparator": "\r\n", "headerRequired": False}),
+            # One file, multiple lines.
+            (
+                ["header \n line1 \n line2 \n line3"],
+                {"recordSeparator": "\n", "headerRequired": True},
+            ),
+            (
+                ["line1 \n line2 \n line3"],
+                {"recordSeparator": "\n", "headerRequired": False},
+            ),
+            (
+                ["header \r\n line1 \r\n line2 \r\n line3"],
+                {"recordSeparator": "\r\n", "headerRequired": True},
+            ),
+            (
+                ["line1 \r\n line2 \r\n line3"],
+                {"recordSeparator": "\r\n", "headerRequired": False},
+            ),
+            # Multiple files.
+            (
+                ["line1", "line2 \n line3"],
+                {"recordSeparator": "\n", "headerRequired": False},
+            ),
+            (
+                ["header \n line1", "header \n line2 \n line3"],
+                {"recordSeparator": "\n", "headerRequired": True},
+            ),
+        ]
+    )
+    def parse_inventory_report_content_setup(self, request):
+
+        # Mock the necessary parameters.
+        gcs_file_system = mock.MagicMock()
+        bucket = mock.MagicMock()
+        use_snapshot_listing = mock.MagicMock()
+
+        # Parse the content and config data.
+        inventory_report_content = request.param[0]
+        inventory_report_config = request.param[1]
+        record_separator = inventory_report_config["recordSeparator"]
+        header_required = inventory_report_config["headerRequired"]
+
+        # Construct custom inventory report config.
+        inventory_report_config = mock.MagicMock(spec=InventoryReportConfig)
+        inventory_report_config.csv_options = {
+            "recordSeparator": record_separator,
+            "headerRequired": header_required,
+        }
+
+        # Stub parse_inventory_report_line method.
+        InventoryReport._parse_inventory_report_line = mock.MagicMock(
+            side_effect="parsed_inventory_report_line"
+        )
+
+        return (
+            gcs_file_system,
+            inventory_report_content,
+            inventory_report_config,
+            bucket,
+            use_snapshot_listing,
+        )
+
+    def test_parse_inventory_reports(self, parse_inventory_report_content_setup):
+
+        (
+            gcs_file_system,
+            inventory_report_content,
+            inventory_report_config,
+            bucket,
+            use_snapshot_listing,
+        ) = parse_inventory_report_content_setup
+
+        record_separator = inventory_report_config.csv_options["recordSeparator"]
+        header_required = inventory_report_config.csv_options["headerRequired"]
+
+        # Number of inventory reports.
+        num_inventory_reports = len(inventory_report_content)
+
+        # Tota, number of object metadata lines.
+        total_lines_in_reports = sum(
+            content.count(record_separator) + 1 for content in inventory_report_content
+        )
+
+        # Remove the header line for each line if header is present.
+        total_lines_in_reports -= num_inventory_reports * 1 if header_required else 0
+
+        result = InventoryReport._parse_inventory_report_content(
+            gcs_file_system=gcs_file_system,
+            inventory_report_content=inventory_report_content,
+            inventory_report_config=inventory_report_config,
+            use_snapshot_listing=use_snapshot_listing,
+            bucket=bucket,
+        )
+
+        # Assert that the number of objects returned is correct.
+        assert len(result) == total_lines_in_reports
+
+        # Assert parse_inventory_report_line was called the correct
+        # number of times.
+        assert (
+            InventoryReport._parse_inventory_report_line.call_count
+            == total_lines_in_reports
+        )
+
+    @pytest.mark.parametrize(
+        "use_snapshot_listing, prefix, mock_objects, expected_result",
+        [
+            # Not using snapshot, no prefix, directory, all matched.
+            (
+                False,
+                None,
+                [{"name": "prefix/object1"}, {"name": "prefix/object2"}],
+                ([{"name": "prefix/object1"}, {"name": "prefix/object2"}], []),
+            ),
+            # Not using snapshot, no prefix, no directory, all matched.
+            (
+                False,
+                None,
+                [{"name": "object1"}, {"name": "object2"}],
+                ([{"name": "object1"}, {"name": "object2"}], []),
+            ),
+            # Not using snapshot, prefix, directory, all matched.
+            (
+                False,
+                "prefix",
+                [{"name": "prefix/object1"}, {"name": "prefix/object2"}],
+                ([{"name": "prefix/object1"}, {"name": "prefix/object2"}], []),
+            ),
+            # Not using snapshot, prefix, directory, some matched.
+            (
+                False,
+                "prefix",
+                [{"name": "prefix/object1"}, {"name": "object2"}],
+                ([{"name": "prefix/object1"}], []),
+            ),
+            # Not using snapshot, prefix, directory, none matched.
+            (False, "prefix", [{"name": "a/object1"}, {"name": "b/object2"}], ([], [])),
+            # Not using snapshot, prefix, no directory, all matched.
+            (
+                False,
+                "object",
+                [{"name": "object1"}, {"name": "object2"}],
+                ([{"name": "object1"}, {"name": "object2"}], []),
+            ),
+            # Not using snapshot, prefix, no directory, some matched.
+            (
+                False,
+                "object",
+                [{"name": "object1"}, {"name": "obj2"}],
+                ([{"name": "object1"}], []),
+            ),
+            # Not using snapshot, prefix, no directory, none matched.
+            (False, "object", [{"name": "obj1"}, {"name": "obj2"}], ([], [])),
+            # Using snapshot, no prefix, no directory.
+            (
+                True,
+                None,
+                [{"name": "object1"}, {"name": "object2"}],
+                ([{"name": "object1"}, {"name": "object2"}], []),
+            ),
+            # Using snapshot, no prefix, a single directory.
+            (
+                True,
+                None,
+                [{"name": "object1"}, {"name": "dir/object2"}],
+                ([{"name": "object1"}], ["dir/"]),
+            ),
+            # Using snapshot, no prefix, multiple directories.
+            (
+                True,
+                None,
+                [
+                    {"name": "object1"},
+                    {"name": "dir1/object2"},
+                    {"name": "dir2/object3"},
+                ],
+                ([{"name": "object1"}], ["dir1/", "dir2/"]),
+            ),
+            # Using snapshot, no prefix, same directory multiple times.
+            (
+                True,
+                None,
+                [
+                    {"name": "object1"},
+                    {"name": "dir1/object2"},
+                    {"name": "dir1/object3"},
+                ],
+                ([{"name": "object1"}], ["dir1/"]),
+            ),
+            # Using snapshot, prefix, no directory.
+            (
+                True,
+                "object",
+                [{"name": "object1"}, {"name": "object2"}],
+                ([{"name": "object1"}, {"name": "object2"}], []),
+            ),
+            # Using snapshot, prefix, a single directory.
+            (
+                True,
+                "dir1/",
+                [{"name": "dir1/dir2/object1"}, {"name": "dir1/object2"}],
+                ([{"name": "dir1/object2"}], ["dir1/dir2/"]),
+            ),
+            # Using snapshot, prefix, multiple directories.
+            (
+                True,
+                "dir1/",
+                [
+                    {"name": "dir1/dir2/object1"},
+                    {"name": "dir1/dir3/object2"},
+                    {"name": "dir1/object3"},
+                ],
+                ([{"name": "dir1/object3"}], ["dir1/dir2/", "dir1/dir3/"]),
+            ),
+            # Using snapshot, prefix, same directory multiple times.
+            (
+                True,
+                "dir1/",
+                [
+                    {"name": "dir1/dir2/object1"},
+                    {"name": "dir1/dir2/object2"},
+                    {"name": "dir1/object3"},
+                ],
+                ([{"name": "dir1/object3"}], ["dir1/dir2/"]),
+            ),
+            # Sanity check from the examples given by the JSON API.
+            # https://cloud.google.com/storage/docs/json_api/v1/objects/list
+            (
+                True,
+                None,
+                [
+                    {"name": "a/b"},
+                    {"name": "a/c"},
+                    {"name": "d"},
+                    {"name": "e"},
+                    {"name": "e/f"},
+                    {"name": "e/g/h"},
+                ],
+                ([{"name": "d"}, {"name": "e"}], ["a/", "e/"]),
+            ),
+            (
+                True,
+                "e/",
+                [
+                    {"name": "a/b"},
+                    {"name": "a/c"},
+                    {"name": "d"},
+                    {"name": "e"},
+                    {"name": "e/f"},
+                    {"name": "e/g/h"},
+                ],
+                ([{"name": "e/f"}], ["e/g/"]),
+            ),
+            (
+                True,
+                "e",
+                [
+                    {"name": "a/b"},
+                    {"name": "a/c"},
+                    {"name": "d"},
+                    {"name": "e"},
+                    {"name": "e/f"},
+                    {"name": "e/g/h"},
+                ],
+                ([{"name": "e"}], ["e/"]),
+            ),
+        ],
+    )
+    def test_construct_final_snapshot(
+        self, use_snapshot_listing, prefix, mock_objects, expected_result
+    ):
+
+        # Construct the final snapshot.
+        result = InventoryReport._construct_final_snapshot(
+            objects=mock_objects,
+            prefix=prefix,
+            use_snapshot_listing=use_snapshot_listing,
+        )
+
+        # Assert the expected outcomes.
+        items, prefixes = result
+        expected_items, expected_prefixes = expected_result
+        assert items == expected_items
+        assert sorted(prefixes) == sorted(expected_prefixes)
+
+
+# Test fields of the inventory report config is correctly stored.
+class TestInventoryReportConfig:
+    def test_inventory_report_config_creation(self):
+
+        csv_options = {}
+        bucket = "bucket"
+        destination_path = ""
+        metadata_fields = []
+        obj_name_idx = 0
+
+        inventory_report_config = InventoryReportConfig(
+            csv_options=csv_options,
+            bucket=bucket,
+            destination_path=destination_path,
+            metadata_fields=metadata_fields,
+            obj_name_idx=obj_name_idx,
+        )
+
+        assert inventory_report_config.csv_options == csv_options
+        assert inventory_report_config.bucket == bucket
+        assert inventory_report_config.destination_path == destination_path
+        assert inventory_report_config.metadata_fields == metadata_fields
+        assert inventory_report_config.obj_name_idx == obj_name_idx
diff --git a/gcsfs/tests/test_inventory_report_listing.py b/gcsfs/tests/test_inventory_report_listing.py
new file mode 100644
index 0000000..abbc008
--- /dev/null
+++ b/gcsfs/tests/test_inventory_report_listing.py
@@ -0,0 +1,30 @@
+import gcsfs.checkers
+import gcsfs.tests.settings
+from gcsfs.inventory_report import InventoryReport
+
+TEST_BUCKET = gcsfs.tests.settings.TEST_BUCKET
+
+
+# Basic integration test to ensure listing returns the correct result.
+def test_ls_base(monkeypatch, gcs):
+
+    # First get results from original listing.
+    items = gcs.ls(TEST_BUCKET)
+
+    async def mock_fetch_snapshot(*args, **kwargs):
+        return [{"name": item} for item in items], []
+
+    # Patch the fetch_snapshot method with the replacement.
+    monkeypatch.setattr(InventoryReport, "fetch_snapshot", mock_fetch_snapshot)
+
+    inventory_report_info = {
+        "location": "location",
+        "id": "id",
+        "use_snapshot_listing": False,
+    }
+
+    # Then get results from listing with inventory report.
+    actual_items = gcs.ls(TEST_BUCKET, inventory_report_info=inventory_report_info)
+
+    # Check equality.
+    assert actual_items == items