[Storage] Added Support for UseDevelopmentStorage=true; for Connection String Client Constructors

sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client.py

@@ -393,6 +393,20 @@ def _format_shared_key_credential(
     return credential
 
 
+def _parse_development_storage(service: str) -> Tuple[
+    str,
+    Optional[str],
+    Optional[Union[str, Dict[str, str], AzureNamedKeyCredential, AzureSasCredential, TokenCredential]],
+]:
+    devstore_protocol = "http"
+    devstore_endpoint = "127.0.0.1"
+    devstore_account_name = "devstoreaccount1"
+    devstore_account_key = "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="
+    port_numbers = {"blob": 10000, "dfs": 10000, "queue": 10001}
+    primary = f"{devstore_protocol}://{devstore_endpoint}:{port_numbers[service]}/{devstore_account_name}"
+    return primary, None, devstore_account_key
+
+
 def parse_connection_str(
     conn_str: str,
     credential: Optional[Union[str, Dict[str, str], AzureNamedKeyCredential, AzureSasCredential, TokenCredential]],
@@ -407,6 +421,8 @@ def parse_connection_str(
     if any(len(tup) != 2 for tup in conn_settings_list):
         raise ValueError("Connection string is either blank or malformed.")
     conn_settings = dict((key.upper(), val) for key, val in conn_settings_list)
+    if conn_settings.get('USEDEVELOPMENTSTORAGE') == 'true':
+        return _parse_development_storage(service)
     endpoints = _SERVICE_PARAMS[service]
     primary = None
     secondary = None

sdk/storage/azure-storage-blob/tests/test_blob_client.py

@@ -101,6 +101,29 @@ def test_create_service_with_connection_string(self, **kwargs):
             self.validate_standard_account_endpoints(service, service_type[1], storage_account_name, storage_account_key)
             assert service.scheme == 'https'
 
+    @BlobPreparer()
+    def test_create_service_use_development_storage(self, **kwargs):
+        devstore_account_name = "devstoreaccount1"
+        devstore_account_key = "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="
+        devstore_endpoint = "127.0.0.1"
+        devstore_blob_port = "10000"
+
+        for service_type in SERVICES.items():
+            # Act
+            service = service_type[0].from_connection_string(
+                "UseDevelopmentStorage=true;",
+                container_name="test",
+                blob_name="test"
+            )
+
+            # Assert
+            assert service is not None
+            assert service.scheme == "http"
+            assert service.account_name == devstore_account_name
+            assert service.credential.account_name == devstore_account_name
+            assert service.credential.account_key == devstore_account_key
+            assert f"{devstore_endpoint}:{devstore_blob_port}/{devstore_account_name}" in service.url
+
     @BlobPreparer()
     def test_create_service_with_sas(self, **kwargs):
         storage_account_name = kwargs.pop("storage_account_name")