code: def test_default_file_metadata_provider_many_files_partitioned(
      propagate_logs,
      caplog,
      fs,
      data_path,
      endpoint_url,
      write_partitioned_df,
      assert_base_partitioned_ds,
  ):
      if endpoint_url is None:
          storage_options = {}
      else:
          storage_options = dict(
              client_kwargs=dict(endpoint_url=endpoint_url)
          )
  
      partition_keys = ["one"]
      partition_path_encoder = PathPartitionEncoder.of(
          base_dir=data_path,
          field_names=partition_keys,
          filesystem=fs,
      )
      paths = []
      dfs = []
      num_dfs = FILE_SIZE_FETCH_PARALLELIZATION_THRESHOLD
      for i in range(num_dfs):
          df = pd.DataFrame(
              {
                  "one": [1, 1, 1, 3, 3, 3],
                  "two": list(range(6 * i, 6 * (i + 1))),
              }
          )
          df_paths = write_partitioned_df(
              df,
              partition_keys,
              partition_path_encoder,
              partial(
                  df_to_csv,
                  storage_options=storage_options,
                  index=False,
              ),
              file_name_suffix=i,
          )
          dfs.append(df)
          paths.extend(df_paths)
      paths, fs = _resolve_paths_and_filesystem(paths, fs)
      partitioning = partition_path_encoder.scheme
  
      meta_provider = DefaultFileMetadataProvider()
      if isinstance(fs, LocalFileSystem):
          patcher = patch(
              "ray.data.datasource.file_meta_provider._get_file_infos_serial",
              wraps=_get_file_infos_serial,
          )
      else:
          patcher = patch(
              "ray.data.datasource.file_meta_provider._get_file_infos_common_path_prefix",
              wraps=_get_file_infos_common_path_prefix,
          )
      with caplog.at_level(logging.WARNING), patcher as mock_get:
          file_paths, file_sizes = map(
              list,
              zip(*meta_provider.expand_paths(paths, fs, partitioning)),
          )
      if isinstance(fs, LocalFileSystem):
          mock_get.assert_called_once_with(paths, fs, False)
      else:
          mock_get.assert_called_once_with(
              paths, _unwrap_protocol(partitioning.base_dir), fs, False
          )
      assert len(caplog.text) == 0
      assert file_paths == paths
      expected_file_sizes = _get_file_sizes_bytes(paths, fs)
      assert file_sizes == expected_file_sizes
  
imports: import logging
  import os
  import posixpath
  import urllib.parse
  from functools import partial
  from unittest.mock import patch
  import pandas
  import pyarrow
  import pyarrow.parquet
  import pytest
  from pyarrow.fs import LocalFileSystem
  from pytest_lazyfixture import lazy_fixture
  from ray.data.datasource import BaseFileMetadataProvider, DefaultFileMetadataProvider, DefaultParquetMetadataProvider, FastFileMetadataProvider, FileMetadataProvider, ParquetMetadataProvider
  from ray.data.datasource.file_based_datasource import FILE_SIZE_FETCH_PARALLELIZATION_THRESHOLD
  from ray.data.datasource.file_meta_provider import _get_file_infos_common_path_prefix, _get_file_infos_parallel, _get_file_infos_serial
  from ray.data.datasource.path_util import _resolve_paths_and_filesystem, _unwrap_protocol
  from ray.data.tests.conftest import *
  from ray.data.tests.test_partitioning import PathPartitionEncoder
  from ray.tests.conftest import *
label_desc: Use image node_id 9 for calling a global function with example usage: test_default_file_metadata_provider_many_files_partitioned(propagate_logs, caplog, fs, data_path, endpoint_url, write_partitioned_df, assert_base_partitioned_ds) without return types