remove datetime.datetime.utcnow

botocore/auth.py

@@ -416,7 +416,7 @@ def signature(self, string_to_sign, request):
     def add_auth(self, request):
         if self.credentials is None:
             raise NoCredentialsError()
-        datetime_now = datetime.datetime.utcnow()
+        datetime_now = datetime.datetime.now(datetime.UTC)
         request.context['timestamp'] = datetime_now.strftime(SIGV4_TIMESTAMP)
         # This could be a retry.  Make sure the previous
         # authorization header is removed first.
@@ -554,7 +554,7 @@ class S3ExpressPostAuth(S3ExpressAuth):
     REQUIRES_IDENTITY_CACHE = True
 
     def add_auth(self, request):
-        datetime_now = datetime.datetime.utcnow()
+        datetime_now = datetime.datetime.now(datetime.UTC)
         request.context['timestamp'] = datetime_now.strftime(SIGV4_TIMESTAMP)
 
         fields = {}
@@ -813,7 +813,7 @@ class S3SigV4PostAuth(SigV4Auth):
     """
 
     def add_auth(self, request):
-        datetime_now = datetime.datetime.utcnow()
+        datetime_now = datetime.datetime.now(datetime.UTC)
         request.context['timestamp'] = datetime_now.strftime(SIGV4_TIMESTAMP)
 
         fields = {}

botocore/crt/auth.py

@@ -54,10 +54,8 @@ def add_auth(self, request):
         if self.credentials is None:
             raise NoCredentialsError()
 
-        # Use utcnow() because that's what gets mocked by tests, but set
-        # timezone because CRT assumes naive datetime is local time.
-        datetime_now = datetime.datetime.utcnow().replace(
-            tzinfo=datetime.timezone.utc
+        datetime_now = datetime.datetime.now(datetime.UTC).replace(
+            tzinfo=datetime.UTC
         )
 
         # Use existing 'X-Amz-Content-SHA256' header if able
@@ -251,10 +249,8 @@ def add_auth(self, request):
         if self.credentials is None:
             raise NoCredentialsError()
 
-        # Use utcnow() because that's what gets mocked by tests, but set
-        # timezone because CRT assumes naive datetime is local time.
-        datetime_now = datetime.datetime.utcnow().replace(
-            tzinfo=datetime.timezone.utc
+        datetime_now = datetime.datetime.now(datetime.UTC).replace(
+            tzinfo=datetime.UTC
         )
 
         # Use existing 'X-Amz-Content-SHA256' header if able

botocore/endpoint.py

@@ -152,10 +152,10 @@ def prepare_request(self, request):
     def _calculate_ttl(
         self, response_received_timestamp, date_header, read_timeout
     ):
-        local_timestamp = datetime.datetime.utcnow()
+        local_timestamp = datetime.datetime.now(datetime.UTC)
         date_conversion = datetime.datetime.strptime(
             date_header, "%a, %d %b %Y %H:%M:%S %Z"
-        )
+        ).replace(tzinfo=datetime.UTC)
         estimated_skew = date_conversion - response_received_timestamp
         ttl = (
             local_timestamp
@@ -169,7 +169,9 @@ def _set_ttl(self, retries_context, read_timeout, success_response):
         has_streaming_input = retries_context.get('has_streaming_input')
         if response_date_header and not has_streaming_input:
             try:
-                response_received_timestamp = datetime.datetime.utcnow()
+                response_received_timestamp = datetime.datetime.now(
+                    datetime.UTC
+                )
                 retries_context['ttl'] = self._calculate_ttl(
                     response_received_timestamp,
                     response_date_header,
@@ -298,9 +300,9 @@ def _do_get_response(self, request, operation_model, context):
         )
 
         http_response_record_dict = response_dict.copy()
-        http_response_record_dict[
-            'streaming'
-        ] = operation_model.has_streaming_output
+        http_response_record_dict['streaming'] = (
+            operation_model.has_streaming_output
+        )
         history_recorder.record('HTTP_RESPONSE', http_response_record_dict)
 
         protocol = operation_model.metadata['protocol']

botocore/signers.py

@@ -607,7 +607,7 @@ def generate_presigned_post(
         policy = {}
 
         # Create an expiration date for the policy
-        datetime_now = datetime.datetime.utcnow()
+        datetime_now = datetime.datetime.now(datetime.UTC)
         expire_date = datetime_now + datetime.timedelta(seconds=expires_in)
         policy['expiration'] = expire_date.strftime(botocore.auth.ISO8601)
 

botocore/utils.py

@@ -667,13 +667,13 @@ def _evaluate_expiration(self, credentials):
         try:
             expiration = datetime.datetime.strptime(
                 expiration, "%Y-%m-%dT%H:%M:%SZ"
-            )
+            ).replace(tzinfo=datetime.UTC)
             refresh_interval = self._config.get(
                 "ec2_credential_refresh_window", 60 * 10
             )
             jitter = random.randint(120, 600)  # Between 2 to 10 minutes
             refresh_interval_with_jitter = refresh_interval + jitter
-            current_time = datetime.datetime.utcnow()
+            current_time = datetime.datetime.now(datetime.UTC)
             refresh_offset = datetime.timedelta(
                 seconds=refresh_interval_with_jitter
             )

tests/__init__.py

@@ -559,6 +559,22 @@ def verify_stubs(self):
             stub.assert_no_pending_responses()
 
 
+def now_patcher_factory(date):
+    """
+    Factory for creating a datetime.datetime.now() patcher.
+
+    :type date: datetime.datetime
+    :param date: datetime object specifying the default output for now()
+    """
+
+    def _now_patcher(tz=None):
+        if tz is None:
+            return date
+        return date.astimezone(tz)
+
+    return _now_patcher
+
+
 class FreezeTime(ContextDecorator):
     """
     Context manager for mocking out datetime in arbitrary modules when creating
@@ -568,20 +584,20 @@ class FreezeTime(ContextDecorator):
     :param module: reference to imported module to patch (e.g. botocore.auth.datetime)
 
     :type date: datetime.datetime
-    :param date: datetime object specifying the output for utcnow()
+    :param date: datetime object specifying the output for now(datetime.UTC)
     """
 
     def __init__(self, module, date=None):
         if date is None:
-            date = datetime.datetime.utcnow()
+            date = datetime.datetime.now(datetime.UTC)
         self.date = date
         self.datetime_patcher = mock.patch.object(
             module, 'datetime', mock.Mock(wraps=datetime.datetime)
         )
 
     def __enter__(self, *args, **kwargs):
         mock = self.datetime_patcher.start()
-        mock.utcnow.return_value = self.date
+        mock.now.side_effect = now_patcher_factory(self.date)
 
     def __exit__(self, *args, **kwargs):
         self.datetime_patcher.stop()

tests/functional/test_ec2.py

@@ -15,7 +15,13 @@
 import botocore.session
 from botocore.compat import parse_qs, urlparse
 from botocore.stub import ANY, Stubber
-from tests import BaseSessionTest, ClientHTTPStubber, mock, unittest
+from tests import (
+    BaseSessionTest,
+    ClientHTTPStubber,
+    mock,
+    unittest,
+    now_patcher_factory,
+)
 
 
 class TestIdempotencyToken(unittest.TestCase):
@@ -91,14 +97,14 @@ def setUp(self):
             '<snapshotId>%s</snapshotId>\n'
             '</CopySnapshotResponse>\n'
         )
-        self.now = datetime.datetime(2011, 9, 9, 23, 36)
+        self.now = datetime.datetime(2011, 9, 9, 23, 36, tzinfo=datetime.UTC)
         self.datetime_patch = mock.patch.object(
             botocore.auth.datetime,
             'datetime',
             mock.Mock(wraps=datetime.datetime),
         )
         self.mocked_datetime = self.datetime_patch.start()
-        self.mocked_datetime.utcnow.return_value = self.now
+        self.mocked_datetime.now.side_effect = now_patcher_factory(self.now)
 
     def tearDown(self):
         super().tearDown()

tests/functional/test_lex.py

@@ -10,9 +10,9 @@
 # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
-from datetime import datetime
+from datetime import datetime, UTC
 
-from tests import BaseSessionTest, ClientHTTPStubber, mock
+from tests import BaseSessionTest, ClientHTTPStubber, mock, now_patcher_factory
 
 
 class TestLex(BaseSessionTest):
@@ -31,10 +31,10 @@ def test_unsigned_payload(self):
             'inputStream': b'',
         }
 
-        timestamp = datetime(2017, 3, 22, 0, 0)
+        timestamp = datetime(2017, 3, 22, 0, 0, tzinfo=UTC)
 
         with mock.patch('botocore.auth.datetime.datetime') as _datetime:
-            _datetime.utcnow.return_value = timestamp
+            _datetime.now.side_effect = now_patcher_factory(timestamp)
             self.http_stubber.add_response(body=b'{}')
             with self.http_stubber:
                 self.client.post_content(**params)

tests/functional/test_retry.py

@@ -17,7 +17,7 @@
 import botocore.endpoint
 from botocore.config import Config
 from botocore.exceptions import ClientError
-from tests import BaseSessionTest, ClientHTTPStubber, mock
+from tests import BaseSessionTest, ClientHTTPStubber, mock, now_patcher_factory
 
 RETRY_MODES = ('legacy', 'standard', 'adaptive')
 
@@ -66,26 +66,30 @@ def _retry_headers_test_cases(self):
         ]
 
         # The first, third and seventh datetime values of each
-        # utcnow_side_effects list are side_effect values for when
-        # utcnow is called in SigV4 signing.
-        utcnow_side_effects = [
+        # now_side_effects list are side_effect values for when
+        # now is called in SigV4 signing.
+        now_side_effects = [
             [
-                datetime.datetime(2019, 6, 1, 0, 0, 0, 0),
-                datetime.datetime(2019, 6, 1, 0, 0, 0, 0),
-                datetime.datetime(2019, 6, 1, 0, 0, 1, 0),
-                datetime.datetime(2019, 6, 1, 0, 0, 0, 0),
-                datetime.datetime(2019, 6, 1, 0, 0, 1, 0),
-                datetime.datetime(2019, 6, 1, 0, 0, 2, 0),
-                datetime.datetime(2019, 6, 1, 0, 0, 0, 0),
+                datetime.datetime(2019, 6, 1, 0, 0, 0, 0, tzinfo=datetime.UTC),
+                datetime.datetime(2019, 6, 1, 0, 0, 0, 0, tzinfo=datetime.UTC),
+                datetime.datetime(2019, 6, 1, 0, 0, 1, 0, tzinfo=datetime.UTC),
+                datetime.datetime(2019, 6, 1, 0, 0, 0, 0, tzinfo=datetime.UTC),
+                datetime.datetime(2019, 6, 1, 0, 0, 1, 0, tzinfo=datetime.UTC),
+                datetime.datetime(2019, 6, 1, 0, 0, 2, 0, tzinfo=datetime.UTC),
+                datetime.datetime(2019, 6, 1, 0, 0, 0, 0, tzinfo=datetime.UTC),
             ],
             [
-                datetime.datetime(2020, 6, 1, 0, 0, 0, 0),
-                datetime.datetime(2019, 6, 1, 0, 0, 5, 0),
-                datetime.datetime(2019, 6, 1, 0, 0, 6, 0),
-                datetime.datetime(2019, 6, 1, 0, 0, 0, 0),
-                datetime.datetime(2019, 6, 1, 0, 0, 11, 0),
-                datetime.datetime(2019, 6, 1, 0, 0, 12, 0),
-                datetime.datetime(2019, 6, 1, 0, 0, 0, 0),
+                datetime.datetime(2020, 6, 1, 0, 0, 0, 0, tzinfo=datetime.UTC),
+                datetime.datetime(2019, 6, 1, 0, 0, 5, 0, tzinfo=datetime.UTC),
+                datetime.datetime(2019, 6, 1, 0, 0, 6, 0, tzinfo=datetime.UTC),
+                datetime.datetime(2019, 6, 1, 0, 0, 0, 0, tzinfo=datetime.UTC),
+                datetime.datetime(
+                    2019, 6, 1, 0, 0, 11, 0, tzinfo=datetime.UTC
+                ),
+                datetime.datetime(
+                    2019, 6, 1, 0, 0, 12, 0, tzinfo=datetime.UTC
+                ),
+                datetime.datetime(2019, 6, 1, 0, 0, 0, 0, tzinfo=datetime.UTC),
             ],
         ]
         expected_headers = [
@@ -100,21 +104,20 @@ def _retry_headers_test_cases(self):
                 b'ttl=20190601T001020Z; attempt=3; max=3',
             ],
         ]
-        test_cases = list(
-            zip(responses, utcnow_side_effects, expected_headers)
-        )
+        test_cases = list(zip(responses, now_side_effects, expected_headers))
+
         return test_cases
 
     def _test_amz_sdk_request_header_with_test_case(
-        self, responses, utcnow_side_effects, expected_headers, client_config
+        self, responses, now_side_effects, expected_headers, client_config
     ):
         datetime_patcher = mock.patch.object(
             botocore.endpoint.datetime,
             'datetime',
             mock.Mock(wraps=datetime.datetime),
         )
         mocked_datetime = datetime_patcher.start()
-        mocked_datetime.utcnow.side_effect = utcnow_side_effects
+        mocked_datetime.now.side_effect = now_side_effects
 
         client = self.session.create_client(
             'dynamodb', self.region, config=client_config

tests/functional/test_s3.py

@@ -39,7 +39,7 @@
     unittest,
 )
 
-DATE = datetime.datetime(2021, 8, 27, 0, 0, 0)
+DATE = datetime.datetime(2021, 8, 27, 0, 0, 0, tzinfo=datetime.UTC)
 
 
 class TestS3BucketValidation(unittest.TestCase):

tests/functional/test_s3express.py

@@ -20,7 +20,7 @@
 from botocore.awsrequest import AWSRequest
 from botocore.credentials import Credentials, RefreshableCredentials
 from botocore.utils import S3ExpressIdentityCache
-from tests import ClientHTTPStubber, mock
+from tests import ClientHTTPStubber, mock, now_patcher_factory
 
 ACCESS_KEY = "AKIDEXAMPLE"
 SECRET_KEY = "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY"
@@ -107,8 +107,7 @@ def test_s3_express_auth_headers(self):
 
 class TestS3ExpressIdentityCache:
     def test_default_s3_express_cache(self, default_s3_client, mock_datetime):
-        mock_datetime.now.return_value = DATE
-        mock_datetime.utcnow.return_value = DATE
+        mock_datetime.now.side_effect = now_patcher_factory(DATE)
 
         identity_cache = S3ExpressIdentityCache(
             default_s3_client,
@@ -125,8 +124,7 @@ def test_default_s3_express_cache(self, default_s3_client, mock_datetime):
     def test_s3_express_cache_one_network_call(
         self, default_s3_client, mock_datetime
     ):
-        mock_datetime.now.return_value = DATE
-        mock_datetime.utcnow.return_value = DATE
+        mock_datetime.now.side_effect = now_patcher_factory(DATE)
         bucket = 'my_bucket'
 
         identity_cache = S3ExpressIdentityCache(
@@ -150,8 +148,7 @@ def test_s3_express_cache_one_network_call(
     def test_s3_express_cache_multiple_buckets(
         self, default_s3_client, mock_datetime
     ):
-        mock_datetime.now.return_value = DATE
-        mock_datetime.utcnow.return_value = DATE
+        mock_datetime.now.side_effect = now_patcher_factory(DATE)
         bucket = 'my_bucket'
         other_bucket = 'other_bucket'
 
@@ -204,7 +201,7 @@ def _call_get_object(self, client):
         )
 
     def test_create_bucket(self, default_s3_client, mock_datetime):
-        mock_datetime.utcnow.return_value = DATE
+        mock_datetime.now.side_effect = now_patcher_factory(DATE)
 
         with ClientHTTPStubber(default_s3_client) as stubber:
             stubber.add_response()
@@ -228,8 +225,7 @@ def test_create_bucket(self, default_s3_client, mock_datetime):
         self._assert_standard_sigv4_signature(stubber.requests[0].headers)
 
     def test_get_object(self, default_s3_client, mock_datetime):
-        mock_datetime.utcnow.return_value = DATE
-        mock_datetime.now.return_value = DATE
+        mock_datetime.now.side_effect = now_patcher_factory(DATE)
 
         with ClientHTTPStubber(default_s3_client) as stubber:
             stubber.add_response(body=CREATE_SESSION_RESPONSE)
@@ -250,8 +246,7 @@ def test_get_object(self, default_s3_client, mock_datetime):
     def test_cache_with_multiple_requests(
         self, default_s3_client, mock_datetime
     ):
-        mock_datetime.utcnow.return_value = DATE
-        mock_datetime.now.return_value = DATE
+        mock_datetime.now.side_effect = now_patcher_factory(DATE)
 
         with ClientHTTPStubber(default_s3_client) as stubber:
             stubber.add_response(body=CREATE_SESSION_RESPONSE)
@@ -275,8 +270,7 @@ def test_cache_with_multiple_requests(
     def test_delete_objects_injects_correct_checksum(
         self, default_s3_client, mock_datetime
     ):
-        mock_datetime.utcnow.return_value = DATE
-        mock_datetime.now.return_value = DATE
+        mock_datetime.now.side_effect = now_patcher_factory(DATE)
 
         with ClientHTTPStubber(default_s3_client) as stubber:
             stubber.add_response(body=CREATE_SESSION_RESPONSE)