[Storage] Fixed Pagination for list_directories_and_files when name_starts_with is specified (#43038)

Author: weirongw23-msft

sdk/storage/azure-storage-file-share/assets.json

@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "python",
   "TagPrefix": "python/storage/azure-storage-file-share",
-  "Tag": "python/storage/azure-storage-file-share_9c32ba27e3"
+  "Tag": "python/storage/azure-storage-file-share_940404bec9"
 }

sdk/storage/azure-storage-file-share/azure/storage/fileshare/_models.py

@@ -1024,7 +1024,6 @@ def _get_next_cb(self, continuation_token):
     def _extract_data_cb(self, get_next_return):
         self.location_mode, self._response = get_next_return
         self.service_endpoint = self._response.service_endpoint
-        self.prefix = self._response.prefix
         self.marker = self._response.marker
         self.results_per_page = self._response.max_results
         self.current_page = [DirectoryProperties._from_generated(i) for i in self._response.segment.directory_items] # pylint: disable = protected-access

sdk/storage/azure-storage-file-share/azure/storage/fileshare/aio/_models.py

@@ -200,7 +200,6 @@ async def _get_next_cb(self, continuation_token):
     async def _extract_data_cb(self, get_next_return):
         self.location_mode, self._response = get_next_return
         self.service_endpoint = self._response.service_endpoint
-        self.prefix = self._response.prefix
         self.marker = self._response.marker
         self.results_per_page = self._response.max_results
         self.current_page = [DirectoryProperties._from_generated(i) for i in self._response.segment.directory_items] # pylint: disable = protected-access

sdk/storage/azure-storage-file-share/tests/test_directory.py

@@ -993,6 +993,30 @@ def test_list_nested_subdirectories_and_files(self, **kwargs):
         assert list_dir[1]['is_directory'] == False
         assert list_dir[1]['size'] == 5
 
+    @FileSharePreparer()
+    @recorded_by_proxy
+    def test_list_pagination_name_starts_with(self, **kwargs):
+        storage_account_name = kwargs.pop("storage_account_name")
+        storage_account_key = kwargs.pop("storage_account_key")
+
+        self._setup(storage_account_name, storage_account_key)
+        share_client = self.fsc.get_share_client(self.share_name)
+        directory_name, prefix = "dir", "samples_"
+        directory = share_client.create_directory(directory_name)
+
+        for i in range(6):
+            directory.upload_file(f"{prefix}{i}", "data1")
+            directory.upload_file(f"not_{i}", "data2")
+
+        list_all = list(share_client.list_directories_and_files(
+            directory_name=directory_name,
+            name_starts_with=prefix,
+            results_per_page=2
+        ))
+        assert len(list_all) == 6
+        for i in range(6):
+            assert list_all[i]["name"] == f"{prefix}{i}"
+
     @FileSharePreparer()
     @recorded_by_proxy
     def test_delete_directory_with_existing_share(self, **kwargs):

sdk/storage/azure-storage-file-share/tests/test_directory_async.py

@@ -1079,6 +1079,32 @@ async def test_list_nested_subdirectories_and_files(self, **kwargs):
         assert list_dir[1]['is_directory'] == False
         assert list_dir[1]['size'] == 5
 
+    @FileSharePreparer()
+    @recorded_by_proxy_async
+    async def test_list_pagination_name_starts_with(self, **kwargs):
+        storage_account_name = kwargs.pop("storage_account_name")
+        storage_account_key = kwargs.pop("storage_account_key")
+
+        await self._setup(storage_account_name, storage_account_key)
+        share_client = self.fsc.get_share_client(self.share_name)
+        directory_name, prefix = "dir", "samples_"
+        directory = await share_client.create_directory(directory_name)
+
+        for i in range(6):
+            await directory.upload_file(f"{prefix}{i}", "data1")
+            await directory.upload_file(f"not_{i}", "data2")
+
+        list_all = []
+        async for path in share_client.list_directories_and_files(
+            directory_name=directory_name,
+            name_starts_with=prefix,
+            results_per_page=2
+        ):
+            list_all.append(path)
+        assert len(list_all) == 6
+        for i in range(6):
+            assert list_all[i]["name"] == f"{prefix}{i}"
+
     @FileSharePreparer()
     @recorded_by_proxy_async
     async def test_delete_directory_with_existing_share(self, **kwargs):