[Storage] Migrate migrated test proxy to new util (Azure#25636)

Author: vincenttran-msft

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

@@ -1526,10 +1526,7 @@ def test_create_append_blob_with_immutability_policy(self, **kwargs):
         versioned_storage_account_name = kwargs.pop("versioned_storage_account_name")
         versioned_storage_account_key = kwargs.pop("versioned_storage_account_key")
         storage_resource_group_name = kwargs.pop("storage_resource_group_name")
-
         variables = kwargs.pop("variables", {})
-        expiry_time = datetime.utcnow() + timedelta(seconds=10)
-        expiry_time_string = variables.setdefault("expiry_time", expiry_time.isoformat())
 
         bsc = BlobServiceClient(self.account_url(versioned_storage_account_name, "blob"), versioned_storage_account_key, max_block_size=4 * 1024)
         self._setup(bsc)
@@ -1547,8 +1544,8 @@ def test_create_append_blob_with_immutability_policy(self, **kwargs):
         blob_name = self.get_resource_name('vlwblob')
         blob = bsc.get_blob_client(container_name, blob_name)
 
-        expires = datetime.strptime(expiry_time_string, "%Y-%m-%dT%H:%M:%S.%f")
-        immutability_policy = ImmutabilityPolicy(expiry_time=expires, policy_mode=BlobImmutabilityPolicyMode.Unlocked)
+        expiry_time = self.get_datetime_variable(variables, 'expiry_time', datetime.utcnow() + timedelta(seconds=10))
+        immutability_policy = ImmutabilityPolicy(expiry_time=expiry_time, policy_mode=BlobImmutabilityPolicyMode.Unlocked)
         blob.create_append_blob(immutability_policy=immutability_policy,
                                 legal_hold=True)
 

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

@@ -1520,10 +1520,7 @@ async def test_create_append_blob_with_immutability_policy(self, **kwargs):
         versioned_storage_account_name = kwargs.pop("versioned_storage_account_name")
         versioned_storage_account_key = kwargs.pop("versioned_storage_account_key")
         storage_resource_group_name = kwargs.pop("storage_resource_group_name")
-
         variables = kwargs.pop("variables", {})
-        expiry_time = datetime.utcnow() + timedelta(seconds=10)
-        expiry_time_string = variables.setdefault("expiry_time", expiry_time.isoformat())
 
         bsc = BlobServiceClient(self.account_url(versioned_storage_account_name, "blob"), versioned_storage_account_key, max_block_size=4 * 1024)
         await self._setup(bsc)
@@ -1541,8 +1538,8 @@ async def test_create_append_blob_with_immutability_policy(self, **kwargs):
         blob_name = self.get_resource_name('vlwblob')
         blob = bsc.get_blob_client(container_name, blob_name)
 
-        expires = datetime.strptime(expiry_time_string, "%Y-%m-%dT%H:%M:%S.%f")
-        immutability_policy = ImmutabilityPolicy(expiry_time=expires, policy_mode=BlobImmutabilityPolicyMode.Unlocked)
+        expiry_time = self.get_datetime_variable(variables, 'expiry_time', datetime.utcnow() + timedelta(seconds=10))
+        immutability_policy = ImmutabilityPolicy(expiry_time=expiry_time, policy_mode=BlobImmutabilityPolicyMode.Unlocked)
         await blob.create_append_blob(immutability_policy=immutability_policy,
                                       legal_hold=True)
 

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

@@ -87,10 +87,6 @@ def assertBlobEqual(self, container_name, blob_name, expected_data):
         actual_data = blob.download_blob()
         assert actual_data.readall() == expected_data
 
-    def _get_datetime_variable(self, variables, name, dt):
-        dt_string = variables.setdefault(name, dt.isoformat())
-        return datetime.strptime(dt_string, "%Y-%m-%dT%H:%M:%S.%f")
-
     #--Test cases for block blobs --------------------------------------------
     @BlobPreparer()
     @recorded_by_proxy
@@ -244,9 +240,9 @@ def test_upload_blob_from_url_if_match_condition(self, **kwargs):
         # Act
         self._setup(storage_account_name, storage_account_key)
         source_blob = self._create_blob()
-        early_test_datetime = self._get_datetime_variable(
+        early_test_datetime = self.get_datetime_variable(
             variables, "early_test_dt", (datetime.utcnow() - timedelta(minutes=15)))
-        late_test_datetime = self._get_datetime_variable(
+        late_test_datetime = self.get_datetime_variable(
             variables, "late_test_dt", (datetime.utcnow() + timedelta(minutes=15)))
         sas = self.generate_sas(
             generate_blob_sas,
@@ -574,7 +570,7 @@ def test_put_block_with_immutability_policy(self, **kwargs):
         blob.stage_block('3', b'CCC')
 
         # Act
-        expiry_time = self._get_datetime_variable(variables, "expiry_time", datetime.utcnow() + timedelta(seconds=5))
+        expiry_time = self.get_datetime_variable(variables, "expiry_time", datetime.utcnow() + timedelta(seconds=5))
         block_list = [BlobBlock(block_id='1'), BlobBlock(block_id='2'), BlobBlock(block_id='3')]
         immutability_policy = ImmutabilityPolicy(expiry_time=expiry_time,
                                                  policy_mode=BlobImmutabilityPolicyMode.Unlocked)

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

@@ -104,10 +104,6 @@ async def assertBlobEqual(self, container_name, blob_name, expected_data):
         actual_data = await stream.readall()
         assert actual_data == expected_data
 
-    def _get_datetime_variable(self, variables, name, dt):
-        dt_string = variables.setdefault(name, dt.isoformat())
-        return datetime.strptime(dt_string, "%Y-%m-%dT%H:%M:%S.%f")
-
     # --Test cases for block blobs --------------------------------------------
     @BlobPreparer()
     @recorded_by_proxy_async
@@ -270,9 +266,9 @@ async def test_upload_blob_from_url_if_match_condition(self, **kwargs):
         # Act
         await self._setup(storage_account_name, storage_account_key)
         source_blob = await self._create_blob()
-        early_test_datetime = self._get_datetime_variable(
+        early_test_datetime = self.get_datetime_variable(
             variables, "early_test_dt", (datetime.utcnow() - timedelta(minutes=15)))
-        late_test_datetime = self._get_datetime_variable(
+        late_test_datetime = self.get_datetime_variable(
             variables, "late_test_dt", (datetime.utcnow() + timedelta(minutes=15)))
         sas = self.generate_sas(
             generate_blob_sas,
@@ -659,7 +655,7 @@ async def test_put_block_with_immutability_policy(self, **kwargs):
         await blob.stage_block('3', b'CCC')
 
         # Act
-        expiry_time = self._get_datetime_variable(variables, "expiry_time", datetime.utcnow() + timedelta(seconds=5))
+        expiry_time = self.get_datetime_variable(variables, "expiry_time", datetime.utcnow() + timedelta(seconds=5))
         block_list = [BlobBlock(block_id='1'), BlobBlock(block_id='2'), BlobBlock(block_id='3')]
         immutability_policy = ImmutabilityPolicy(expiry_time=expiry_time,
                                                  policy_mode=BlobImmutabilityPolicyMode.Unlocked)

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

@@ -117,10 +117,6 @@ def assertRangeEqual(self, container_name, blob_name, expected_data, offset, len
         actual_data = blob.download_blob(offset=offset, length=length)
         assert actual_data.readall() == expected_data
 
-    def _get_datetime_variable(self, variables, name, dt):
-        dt_string = variables.setdefault(name, dt.isoformat())
-        return datetime.strptime(dt_string, "%Y-%m-%dT%H:%M:%S.%f")
-
     # --Test cases for page blobs --------------------------------------------
     @BlobPreparer()
     @recorded_by_proxy
@@ -164,7 +160,7 @@ def test_create_blob_with_immutability_policy(self, **kwargs):
         blob = bsc.get_blob_client(container_name, blob_name)
 
         # Act
-        expiry_time = self._get_datetime_variable(variables, 'expiry_time', datetime.utcnow() + timedelta(seconds=5))
+        expiry_time = self.get_datetime_variable(variables, 'expiry_time', datetime.utcnow() + timedelta(seconds=5))
         immutability_policy = ImmutabilityPolicy(expiry_time=expiry_time,
                                                  policy_mode=BlobImmutabilityPolicyMode.Unlocked)
         resp = blob.create_page_blob(1024, immutability_policy=immutability_policy,

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

@@ -114,10 +114,6 @@ async def assertRangeEqual(self, container_name, blob_name, expected_data, offse
         actual_data = await stream.readall()
         assert actual_data == expected_data
 
-    def _get_datetime_variable(self, variables, name, dt):
-        dt_string = variables.setdefault(name, dt.isoformat())
-        return datetime.strptime(dt_string, "%Y-%m-%dT%H:%M:%S.%f")
-
     # --Test cases for page blobs --------------------------------------------
 
     @BlobPreparer()
@@ -193,7 +189,7 @@ async def test_create_blob_with_immutability_policy(self, **kwargs):
         blob = bsc.get_blob_client(container_name, blob_name)
 
         # Act
-        expiry_time = self._get_datetime_variable(variables, 'expiry_time', datetime.utcnow() + timedelta(seconds=5))
+        expiry_time = self.get_datetime_variable(variables, 'expiry_time', datetime.utcnow() + timedelta(seconds=5))
         immutability_policy = ImmutabilityPolicy(expiry_time=expiry_time,
                                                  policy_mode=BlobImmutabilityPolicyMode.Unlocked)
         resp = await blob.create_page_blob(1024,