Fixes #1530: Replace deprecated utcnow and utcfromtimestamp with timezone-aware datetime

storages/backends/azure_storage.py

@@ -1,6 +1,6 @@
+import datetime
 import mimetypes
 import warnings
-from datetime import datetime
 from datetime import timedelta
 from tempfile import SpooledTemporaryFile
 from urllib.parse import urlparse
@@ -123,7 +123,7 @@ def __init__(self, **settings):
         self._service_client = None
         self._client = None
         self._user_delegation_key = None
-        self._user_delegation_key_expiry = datetime.utcnow()
+        self._user_delegation_key_expiry = datetime.datetime.now(datetime.timezone.utc)
         if self.connection_string and (not self.account_name or not self.account_key):
             parsed = parse_connection_string(
                 self.connection_string, case_sensitive_keys=True
@@ -212,7 +212,7 @@ def get_user_delegation_key(self, expiry):
             self._user_delegation_key is None
             or expiry > self._user_delegation_key_expiry
         ):
-            now = datetime.utcnow()
+            now = datetime.datetime.now(datetime.timezone.utc)
             key_expiry_time = now + timedelta(days=7)
             self._user_delegation_key = self.service_client.get_user_delegation_key(
                 key_start_time=now, key_expiry_time=key_expiry_time
@@ -291,7 +291,7 @@ def _save(self, name, content):
 
     def _expire_at(self, expire):
         # azure expects time in UTC
-        return datetime.utcnow() + timedelta(seconds=expire)
+        return datetime.datetime.now(datetime.timezone.utc) + timedelta(seconds=expire)
 
     def url(self, name, expire=None, parameters=None, mode="r"):
         name = self._get_valid_path(name)

storages/backends/s3.py

@@ -7,6 +7,7 @@
 import warnings
 from datetime import datetime
 from datetime import timedelta
+from datetime import timezone
 from urllib.parse import urlencode
 
 from django.contrib.staticfiles.storage import ManifestFilesMixin
@@ -681,7 +682,7 @@ def url(self, name, parameters=None, expire=None, http_method=None):
             )
 
             if self.querystring_auth and self.cloudfront_signer:
-                expiration = datetime.utcnow() + timedelta(seconds=expire)
+                expiration = datetime.now(timezone.utc) + timedelta(seconds=expire)
                 return self.cloudfront_signer.generate_presigned_url(
                     url, date_less_than=expiration
                 )

tests/test_azure.py

@@ -159,9 +159,9 @@ def test_url_expire(self, generate_blob_sas_mocked):
             datetime.datetime(2016, 11, 6, 4), datetime.timezone.utc
         )
 
-        with mock.patch("storages.backends.azure_storage.datetime") as d_mocked:
+        with mock.patch("storages.backends.azure_storage.datetime") as datetime_mocked:
             # Implicit read permission
-            d_mocked.utcnow.return_value = fixed_time
+            datetime_mocked.datetime.now.return_value = fixed_time
             self.assertEqual(
                 self.storage.url("some blob", 100),
                 "https://ret_foo.blob.core.windows.net/test/some%20blob",
@@ -179,7 +179,7 @@ def test_url_expire(self, generate_blob_sas_mocked):
             self.assertEqual(str(called_kwargs["permission"]), "r")
 
             # Explicit write permission
-            d_mocked.utcnow.return_value = fixed_time
+            datetime_mocked.datetime.now.return_value = fixed_time
             self.assertEqual(
                 self.storage.url("some blob", expire=100, mode="w"),
                 "https://ret_foo.blob.core.windows.net/test/some%20blob",
@@ -218,8 +218,8 @@ def test_url_expire_user_delegation_key(self, generate_blob_sas_mocked):
         fixed_time = make_aware(
             datetime.datetime(2016, 11, 6, 4), datetime.timezone.utc
         )
-        with mock.patch("storages.backends.azure_storage.datetime") as d_mocked:
-            d_mocked.utcnow.return_value = fixed_time
+        with mock.patch("storages.backends.azure_storage.datetime") as datetime_mocked:
+            datetime_mocked.datetime.now.return_value = fixed_time
             service_client.get_user_delegation_key.return_value = "user delegation key"
             self.assertEqual(
                 self.storage.url("some blob", 100),
@@ -435,3 +435,13 @@ def test_client_settings(self, bsc):
             bsc.assert_called_once_with(
                 "https://test.blob.core.windows.net", credential=None, api_version="1.3"
             )
+
+    def test_user_delegation_key_expiry_is_aware(self):
+        """
+        Ensure _user_delegation_key_expiry is a UTC-aware datetime.
+        """
+        storage = azure_storage.AzureStorage()
+        self.assertIsNotNone(storage._user_delegation_key_expiry.tzinfo)
+        self.assertEqual(
+            storage._user_delegation_key_expiry.utcoffset().total_seconds(), 0
+        )

tests/test_s3.py

@@ -728,7 +728,7 @@ def test_url_protocol(self):
             self.assertEqual(storage.url_protocol, "https:")
 
     @mock.patch("storages.backends.s3.datetime")
-    def test_storage_url_custom_domain_signed_urls(self, dt):
+    def test_storage_url_custom_domain_signed_urls(self, datetime_mocked):
         key_id = "test-key"
         filename = "file.txt"
         pem = dedent(
@@ -766,7 +766,9 @@ def test_storage_url_custom_domain_signed_urls(self, dt):
             self.assertEqual(self.storage.url(filename), url)
 
             self.storage.querystring_auth = True
-            dt.utcnow.return_value = datetime.datetime.utcfromtimestamp(0)
+            datetime_mocked.now.return_value = datetime.datetime.fromtimestamp(
+                0, datetime.timezone.utc
+            )
             self.assertEqual(self.storage.url(filename), signed_url)
 
     def test_generated_url_is_encoded(self):