From c2f668736976867e4616bcee0d819697357fced1 Mon Sep 17 00:00:00 2001 From: JBSL Date: Thu, 20 Feb 2025 18:46:36 +0100 Subject: [PATCH] Refactored usages of utcnow() method for method now(datetime.UTC). The first method is being deprecated and displaying deprecation warnings --- botocore/auth.py | 6 +- botocore/crt/auth.py | 12 +- botocore/endpoint.py | 4 +- botocore/signers.py | 2 +- botocore/utils.py | 2 +- tests/__init__.py | 6 +- tests/functional/test_ec2.py | 2 +- tests/functional/test_lex.py | 2 +- tests/functional/test_retry.py | 12 +- tests/functional/test_s3express.py | 8 +- tests/functional/test_sts.py | 2 +- tests/unit/auth/test_signers.py | 12 +- tests/unit/test_credentials.py | 14 +- tests/unit/test_signers.py | 4 +- tests/unit/test_utils.py | 234 ++++++++++++++--------------- 15 files changed, 154 insertions(+), 168 deletions(-) diff --git a/botocore/auth.py b/botocore/auth.py index bacbd39dde..317717c9c5 100644 --- a/botocore/auth.py +++ b/botocore/auth.py @@ -422,7 +422,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. @@ -560,7 +560,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 = {} @@ -819,7 +819,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 = {} diff --git a/botocore/crt/auth.py b/botocore/crt/auth.py index 0d1a81def4..8be6021f2a 100644 --- a/botocore/crt/auth.py +++ b/botocore/crt/auth.py @@ -54,11 +54,7 @@ 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) # Use existing 'X-Amz-Content-SHA256' header if able existing_sha256 = self._get_existing_sha256(request) @@ -251,11 +247,7 @@ 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) # Use existing 'X-Amz-Content-SHA256' header if able existing_sha256 = self._get_existing_sha256(request) diff --git a/botocore/endpoint.py b/botocore/endpoint.py index 1c2cee068b..cc706e922e 100644 --- a/botocore/endpoint.py +++ b/botocore/endpoint.py @@ -150,7 +150,7 @@ 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" ) @@ -167,7 +167,7 @@ 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, diff --git a/botocore/signers.py b/botocore/signers.py index cf90fe4a5f..7ce28eb790 100644 --- a/botocore/signers.py +++ b/botocore/signers.py @@ -723,7 +723,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) diff --git a/botocore/utils.py b/botocore/utils.py index 30a513ea90..ce1e25d79a 100644 --- a/botocore/utils.py +++ b/botocore/utils.py @@ -667,7 +667,7 @@ def _evaluate_expiration(self, credentials): ) 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 ) diff --git a/tests/__init__.py b/tests/__init__.py index 68af4ac9dc..e2072b2374 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -574,12 +574,12 @@ 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() """ 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) @@ -587,7 +587,7 @@ def __init__(self, module, date=None): def __enter__(self, *args, **kwargs): mock = self.datetime_patcher.start() - mock.utcnow.return_value = self.date + mock.now.return_value = self.date def __exit__(self, *args, **kwargs): self.datetime_patcher.stop() diff --git a/tests/functional/test_ec2.py b/tests/functional/test_ec2.py index 75ec6dc8f4..9250dbce11 100644 --- a/tests/functional/test_ec2.py +++ b/tests/functional/test_ec2.py @@ -98,7 +98,7 @@ def setUp(self): mock.Mock(wraps=datetime.datetime), ) self.mocked_datetime = self.datetime_patch.start() - self.mocked_datetime.utcnow.return_value = self.now + self.mocked_datetime.now.return_value = self.now def tearDown(self): super().tearDown() diff --git a/tests/functional/test_lex.py b/tests/functional/test_lex.py index a62b0357dd..7dfc22c0b0 100644 --- a/tests/functional/test_lex.py +++ b/tests/functional/test_lex.py @@ -34,7 +34,7 @@ def test_unsigned_payload(self): timestamp = datetime(2017, 3, 22, 0, 0) with mock.patch('botocore.auth.datetime.datetime') as _datetime: - _datetime.utcnow.return_value = timestamp + _datetime.now.return_value = timestamp self.http_stubber.add_response(body=b'{}') with self.http_stubber: self.client.post_content(**params) diff --git a/tests/functional/test_retry.py b/tests/functional/test_retry.py index da7460313c..d49fdd3ddc 100644 --- a/tests/functional/test_retry.py +++ b/tests/functional/test_retry.py @@ -66,9 +66,9 @@ 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), @@ -101,12 +101,12 @@ def _retry_headers_test_cases(self): ], ] test_cases = list( - zip(responses, utcnow_side_effects, expected_headers) + 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, @@ -114,7 +114,7 @@ def _test_amz_sdk_request_header_with_test_case( 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 diff --git a/tests/functional/test_s3express.py b/tests/functional/test_s3express.py index 390721ee12..ebb89b0b60 100644 --- a/tests/functional/test_s3express.py +++ b/tests/functional/test_s3express.py @@ -108,7 +108,6 @@ 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 identity_cache = S3ExpressIdentityCache( default_s3_client, @@ -126,7 +125,6 @@ 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 bucket = 'my_bucket' identity_cache = S3ExpressIdentityCache( @@ -151,7 +149,6 @@ 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 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.return_value = DATE with ClientHTTPStubber(default_s3_client) as stubber: stubber.add_response() @@ -228,7 +225,6 @@ 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 with ClientHTTPStubber(default_s3_client) as stubber: @@ -250,7 +246,6 @@ 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 with ClientHTTPStubber(default_s3_client) as stubber: @@ -275,7 +270,6 @@ 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 with ClientHTTPStubber(default_s3_client) as stubber: diff --git a/tests/functional/test_sts.py b/tests/functional/test_sts.py index 42fac96711..903bc986ee 100644 --- a/tests/functional/test_sts.py +++ b/tests/functional/test_sts.py @@ -39,7 +39,7 @@ def setUp(self): def test_presigned_url_contains_no_content_type(self): timestamp = datetime(2017, 3, 22, 0, 0) with mock.patch('botocore.auth.datetime.datetime') as _datetime: - _datetime.utcnow.return_value = timestamp + _datetime.now.return_value = timestamp url = self.client.generate_presigned_url('get_caller_identity', {}) # There should be no 'content-type' in x-amz-signedheaders diff --git a/tests/unit/auth/test_signers.py b/tests/unit/auth/test_signers.py index c95d042b04..cfb6566218 100644 --- a/tests/unit/auth/test_signers.py +++ b/tests/unit/auth/test_signers.py @@ -30,7 +30,7 @@ def setUp(self): self.fixed_date = datetime.datetime(2014, 3, 10, 17, 2, 55, 0) self.datetime_patch = mock.patch('botocore.auth.datetime.datetime') self.datetime_mock = self.datetime_patch.start() - self.datetime_mock.utcnow.return_value = self.fixed_date + self.datetime_mock.now.return_value = self.fixed_date self.datetime_mock.strptime.return_value = self.fixed_date def tearDown(self): @@ -524,9 +524,9 @@ def test_thread_safe_timestamp(self): 'datetime', mock.Mock(wraps=datetime.datetime), ) as mock_datetime: - original_utcnow = datetime.datetime(2014, 1, 1, 0, 0) + original_now = datetime.datetime(2014, 1, 1, 0, 0) - mock_datetime.utcnow.return_value = original_utcnow + mock_datetime.now.return_value = original_now # Go through the add_auth process once. This will attach # a timestamp to the request at the beginning of auth. auth.add_auth(request) @@ -534,7 +534,7 @@ def test_thread_safe_timestamp(self): # Ensure the date is in the Authorization header self.assertIn('20140101', request.headers['Authorization']) # Now suppose the utc time becomes the next day all of a sudden - mock_datetime.utcnow.return_value = datetime.datetime( + mock_datetime.now.return_value = datetime.datetime( 2014, 1, 2, 0, 0 ) # Smaller methods like the canonical request and string_to_sign @@ -799,7 +799,7 @@ def setUp(self): mock.Mock(wraps=datetime.datetime), ) mocked_datetime = self.datetime_patcher.start() - mocked_datetime.utcnow.return_value = datetime.datetime( + mocked_datetime.now.return_value = datetime.datetime( 2014, 1, 1, 0, 0 ) @@ -1103,7 +1103,7 @@ def setUp(self): mock.Mock(wraps=datetime.datetime), ) mocked_datetime = self.datetime_patcher.start() - mocked_datetime.utcnow.return_value = datetime.datetime( + mocked_datetime.now.return_value = datetime.datetime( 2014, 1, 1, 0, 0 ) diff --git a/tests/unit/test_credentials.py b/tests/unit/test_credentials.py index 3017444a6f..50e084791a 100644 --- a/tests/unit/test_credentials.py +++ b/tests/unit/test_credentials.py @@ -16,7 +16,7 @@ import subprocess import tempfile import time -from datetime import datetime, timedelta +from datetime import datetime, timedelta, UTC from pathlib import Path import pytest @@ -128,7 +128,7 @@ def setUp(self): def test_refresh_needed(self): # The expiry time was set for 30 minutes ago, so if we - # say the current time is utcnow(), then we should need + # say the current time is now(), then we should need # a refresh. self.mock_time.return_value = datetime.now(tzlocal()) self.assertTrue(self.creds.refresh_needed()) @@ -315,7 +315,7 @@ def test_expiration_in_datetime_format(self): self.assertEqual(response, expected_response) def test_retrieves_from_cache(self): - date_in_future = datetime.utcnow() + timedelta(seconds=1000) + date_in_future = datetime.now(UTC) + timedelta(seconds=1000) utc_timestamp = date_in_future.isoformat() + 'Z' cache_key = '793d6e2f27667ab2da104824407e486bfec24a47' cache = { @@ -749,7 +749,7 @@ def test_no_cache(self): self.assertEqual(response, expected_response) def test_retrieves_from_cache(self): - date_in_future = datetime.utcnow() + timedelta(seconds=1000) + date_in_future = datetime.now(UTC) + timedelta(seconds=1000) utc_timestamp = date_in_future.isoformat() + 'Z' cache_key = '793d6e2f27667ab2da104824407e486bfec24a47' cache = { @@ -867,7 +867,7 @@ def test_assume_role_with_no_cache(self): mock_loader_cls.assert_called_with('/some/path/token.jwt') def test_assume_role_retrieves_from_cache(self): - date_in_future = datetime.utcnow() + timedelta(seconds=1000) + date_in_future = datetime.now(UTC) + timedelta(seconds=1000) utc_timestamp = date_in_future.isoformat() + 'Z' cache_key = 'c29461feeacfbed43017d20612606ff76abc073d' @@ -2037,7 +2037,7 @@ def test_assume_role_refresher_serializes_datetime(self): self.assertEqual(expiry_time, '2016-11-06T01:30:00UTC') def test_assume_role_retrieves_from_cache(self): - date_in_future = datetime.utcnow() + timedelta(seconds=1000) + date_in_future = datetime.now(UTC) + timedelta(seconds=1000) utc_timestamp = date_in_future.isoformat() + 'Z' self.fake_config['profiles']['development']['role_arn'] = 'myrole' @@ -2066,7 +2066,7 @@ def test_assume_role_retrieves_from_cache(self): self.assertEqual(creds.token, 'baz-cached') def test_chain_prefers_cache(self): - date_in_future = datetime.utcnow() + timedelta(seconds=1000) + date_in_future = datetime.now(UTC) + timedelta(seconds=1000) utc_timestamp = date_in_future.isoformat() + 'Z' # The profile we will be using has a cache entry, but the profile it diff --git a/tests/unit/test_signers.py b/tests/unit/test_signers.py index 199a1b09e4..ce3cd10706 100644 --- a/tests/unit/test_signers.py +++ b/tests/unit/test_signers.py @@ -696,7 +696,7 @@ def setUp(self): self.datetime_mock = self.datetime_patch.start() self.fixed_date = datetime.datetime(2014, 3, 10, 17, 2, 55, 0) self.fixed_delta = datetime.timedelta(seconds=3600) - self.datetime_mock.datetime.utcnow.return_value = self.fixed_date + self.datetime_mock.datetime.now.return_value = self.fixed_date self.datetime_mock.timedelta.return_value = self.fixed_delta def tearDown(self): @@ -1151,7 +1151,7 @@ def test_generate_db_auth_token(self): clock = datetime.datetime(2016, 11, 7, 17, 39, 33, tzinfo=tzutc()) with mock.patch('datetime.datetime') as dt: - dt.utcnow.return_value = clock + dt.now.return_value = clock result = generate_db_auth_token( self.client, hostname, port, username ) diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index 6143183418..0fe2595804 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -18,85 +18,85 @@ from sys import getrefcount import pytest -from dateutil.tz import tzoffset, tzutc +from dateutil.tz import tzoffset +from dateutil.tz import tzutc import botocore from botocore import xform_name -from botocore.awsrequest import AWSRequest, HeadersDict +from botocore.awsrequest import AWSRequest +from botocore.awsrequest import HeadersDict from botocore.compat import json from botocore.config import Config from botocore.endpoint_provider import RuleSetEndpoint -from botocore.exceptions import ( - ClientError, - ConfigNotFound, - ConnectionClosedError, - ConnectTimeoutError, - InvalidDNSNameError, - InvalidExpressionError, - InvalidIMDSEndpointError, - InvalidIMDSEndpointModeError, - MetadataRetrievalError, - ReadTimeoutError, - SSOTokenLoadError, - UnsupportedOutpostResourceError, - UnsupportedS3AccesspointConfigurationError, - UnsupportedS3ArnError, -) -from botocore.model import ( - DenormalizedStructureBuilder, - OperationModel, - ServiceModel, - ShapeResolver, -) +from botocore.exceptions import ClientError +from botocore.exceptions import ConfigNotFound +from botocore.exceptions import ConnectTimeoutError +from botocore.exceptions import ConnectionClosedError +from botocore.exceptions import InvalidDNSNameError +from botocore.exceptions import InvalidExpressionError +from botocore.exceptions import InvalidIMDSEndpointError +from botocore.exceptions import InvalidIMDSEndpointModeError +from botocore.exceptions import MetadataRetrievalError +from botocore.exceptions import ReadTimeoutError +from botocore.exceptions import SSOTokenLoadError +from botocore.exceptions import UnsupportedOutpostResourceError +from botocore.exceptions import UnsupportedS3AccesspointConfigurationError +from botocore.exceptions import UnsupportedS3ArnError +from botocore.model import DenormalizedStructureBuilder +from botocore.model import OperationModel +from botocore.model import ServiceModel +from botocore.model import ShapeResolver from botocore.regions import EndpointRulesetResolver from botocore.session import Session -from botocore.utils import ( - ArgumentGenerator, - ArnParser, - CachedProperty, - ContainerMetadataFetcher, - IMDSRegionProvider, - InstanceMetadataFetcher, - InstanceMetadataRegionFetcher, - InvalidArnException, - S3ArnParamHandler, - S3EndpointSetter, - S3RegionRedirectorv2, - SSOTokenLoader, - calculate_sha256, - calculate_tree_hash, - datetime2timestamp, - deep_merge, - determine_content_length, - ensure_boolean, - fix_s3_host, - get_encoding_from_headers, - get_service_module_name, - has_header, - instance_cache, - is_json_value_header, - is_s3_accelerate_url, - is_valid_endpoint_url, - is_valid_ipv6_endpoint_url, - is_valid_uri, - lowercase_dict, - lru_cache_weakref, - merge_dicts, - normalize_url_path, - parse_key_val_file, - parse_key_val_file_contents, - parse_timestamp, - parse_to_aware_datetime, - percent_encode, - percent_encode_sequence, - remove_dot_segments, - resolve_imds_endpoint_mode, - set_value_from_jmespath, - switch_host_s3_accelerate, - switch_to_virtual_host_style, - validate_jmespath_for_set, -) -from tests import FreezeTime, RawResponse, create_session, mock, unittest +from botocore.utils import ArgumentGenerator +from botocore.utils import ArnParser +from botocore.utils import CachedProperty +from botocore.utils import ContainerMetadataFetcher +from botocore.utils import IMDSRegionProvider +from botocore.utils import InstanceMetadataFetcher +from botocore.utils import InstanceMetadataRegionFetcher +from botocore.utils import InvalidArnException +from botocore.utils import S3ArnParamHandler +from botocore.utils import S3EndpointSetter +from botocore.utils import S3RegionRedirectorv2 +from botocore.utils import SSOTokenLoader +from botocore.utils import calculate_sha256 +from botocore.utils import calculate_tree_hash +from botocore.utils import datetime2timestamp +from botocore.utils import deep_merge +from botocore.utils import determine_content_length +from botocore.utils import ensure_boolean +from botocore.utils import fix_s3_host +from botocore.utils import get_encoding_from_headers +from botocore.utils import get_service_module_name +from botocore.utils import has_header +from botocore.utils import instance_cache +from botocore.utils import is_json_value_header +from botocore.utils import is_s3_accelerate_url +from botocore.utils import is_valid_endpoint_url +from botocore.utils import is_valid_ipv6_endpoint_url +from botocore.utils import is_valid_uri +from botocore.utils import lowercase_dict +from botocore.utils import lru_cache_weakref +from botocore.utils import merge_dicts +from botocore.utils import normalize_url_path +from botocore.utils import parse_key_val_file +from botocore.utils import parse_key_val_file_contents +from botocore.utils import parse_timestamp +from botocore.utils import parse_to_aware_datetime +from botocore.utils import percent_encode +from botocore.utils import percent_encode_sequence +from botocore.utils import remove_dot_segments +from botocore.utils import resolve_imds_endpoint_mode +from botocore.utils import set_value_from_jmespath +from botocore.utils import switch_host_s3_accelerate +from botocore.utils import switch_to_virtual_host_style +from botocore.utils import validate_jmespath_for_set +from tests import FreezeTime +from tests import RawResponse +from tests import create_session +from tests import mock +from tests import unittest DATE = datetime.datetime(2021, 12, 10, 00, 00, 00) DT_FORMAT = "%Y-%m-%dT%H:%M:%SZ" @@ -512,7 +512,7 @@ def test_parse_timestamp_fails_with_bad_tzinfo(self): mock_get_tzinfo_options = mock.MagicMock(return_value=(mock_tzinfo,)) with mock.patch( - 'botocore.utils.get_tzinfo_options', mock_get_tzinfo_options + 'botocore.utils.get_tzinfo_options', mock_get_tzinfo_options ): with self.assertRaises(RuntimeError): parse_timestamp(0) @@ -1908,7 +1908,7 @@ def test_redirects_on_illegal_location_constraint_from_opt_in_region(self): self.assertEqual(redirect_response, 0) def test_no_redirect_on_illegal_location_constraint_from_bad_location_constraint( - self, + self, ): request_dict = { 'url': 'https://us-west-2.amazonaws.com/foo', @@ -2234,7 +2234,7 @@ def test_outpost_arn_with_slash(self): def test_outpost_arn_errors_for_missing_fields(self): params = { 'Bucket': 'arn:aws:s3-outposts:us-west-2:123456789012:outpost/' - 'op-01234567890123456/accesspoint' + 'op-01234567890123456/accesspoint' } with self.assertRaises(UnsupportedOutpostResourceError): self.arn_handler.handle_arn(params, self.model, {}) @@ -2242,7 +2242,7 @@ def test_outpost_arn_errors_for_missing_fields(self): def test_outpost_arn_errors_for_empty_fields(self): params = { 'Bucket': 'arn:aws:s3-outposts:us-west-2:123456789012:outpost/' - '/accesspoint/myaccesspoint' + '/accesspoint/myaccesspoint' } with self.assertRaises(UnsupportedOutpostResourceError): self.arn_handler.handle_arn(params, self.model, {}) @@ -2292,7 +2292,7 @@ def get_endpoint_setter(self, **kwargs): return S3EndpointSetter(**setter_kwargs) def get_s3_request( - self, bucket=None, key=None, scheme='https://', querystring=None + self, bucket=None, key=None, scheme='https://', querystring=None ): url = scheme + 's3.us-west-2.amazonaws.com/' if bucket: @@ -2314,10 +2314,10 @@ def get_s3_outpost_request(self, **s3_request_kwargs): return request def get_s3_accesspoint_request( - self, - accesspoint_name=None, - accesspoint_context=None, - **s3_request_kwargs, + self, + accesspoint_name=None, + accesspoint_context=None, + **s3_request_kwargs, ): if not accesspoint_name: accesspoint_name = self.accesspoint_name @@ -3144,7 +3144,7 @@ def test_v1_disabled_by_config(self): def _get_datetime(self, dt=None, offset=None, offset_func=operator.add): if dt is None: - dt = datetime.datetime.utcnow() + dt = datetime.datetime.now(datetime.UTC) if offset is not None: dt = offset_func(dt, offset) @@ -3286,9 +3286,9 @@ def tearDown(self): self.environ_patch.stop() def assert_does_provide_expected_value( - self, - fetcher_region=None, - expected_result=None, + self, + fetcher_region=None, + expected_result=None, ): fake_session = mock.Mock(spec=Session) fake_fetcher = mock.Mock(spec=InstanceMetadataRegionFetcher) @@ -3453,16 +3453,16 @@ def test_can_load_token_exists_sso_session_name(self): @pytest.mark.parametrize( 'header_name, headers, expected', ( - ('test_header', {'test_header': 'foo'}, True), - ('Test_Header', {'test_header': 'foo'}, True), - ('test_header', {'Test_Header': 'foo'}, True), - ('missing_header', {'Test_Header': 'foo'}, False), - (None, {'Test_Header': 'foo'}, False), - ('test_header', HeadersDict({'test_header': 'foo'}), True), - ('Test_Header', HeadersDict({'test_header': 'foo'}), True), - ('test_header', HeadersDict({'Test_Header': 'foo'}), True), - ('missing_header', HeadersDict({'Test_Header': 'foo'}), False), - (None, HeadersDict({'Test_Header': 'foo'}), False), + ('test_header', {'test_header': 'foo'}, True), + ('Test_Header', {'test_header': 'foo'}, True), + ('test_header', {'Test_Header': 'foo'}, True), + ('missing_header', {'Test_Header': 'foo'}, False), + (None, {'Test_Header': 'foo'}, False), + ('test_header', HeadersDict({'test_header': 'foo'}), True), + ('Test_Header', HeadersDict({'test_header': 'foo'}), True), + ('test_header', HeadersDict({'Test_Header': 'foo'}), True), + ('missing_header', HeadersDict({'Test_Header': 'foo'}), False), + (None, HeadersDict({'Test_Header': 'foo'}), False), ), ) def test_has_header(header_name, headers, expected): @@ -3522,24 +3522,24 @@ def seek(self, *args, **kwargs): @pytest.mark.parametrize( 'url, expected', ( - ('https://s3-accelerate.amazonaws.com', True), - ('https://s3-accelerate.amazonaws.com/', True), - ('https://s3-accelerate.amazonaws.com/key', True), - ('http://s3-accelerate.amazonaws.com/key', True), - ('https://s3-accelerate.foo.amazonaws.com/key', False), - # bucket prefixes are not allowed - ('https://bucket.s3-accelerate.amazonaws.com/key', False), - # S3 accelerate can be combined with dualstack - ('https://s3-accelerate.dualstack.amazonaws.com/key', True), - ('https://bucket.s3-accelerate.dualstack.amazonaws.com/key', False), - ('https://s3-accelerate.dualstack.dualstack.amazonaws.com/key', False), - ('https://s3-accelerate.dualstack.foo.amazonaws.com/key', False), - ('https://dualstack.s3-accelerate.amazonaws.com/key', False), - # assorted other ways for URLs to not be valid for s3-accelerate - ('ftp://s3-accelerate.dualstack.foo.amazonaws.com/key', False), - ('https://s3-accelerate.dualstack.foo.c2s.ic.gov/key', False), - # None-valued url is accepted - (None, False), + ('https://s3-accelerate.amazonaws.com', True), + ('https://s3-accelerate.amazonaws.com/', True), + ('https://s3-accelerate.amazonaws.com/key', True), + ('http://s3-accelerate.amazonaws.com/key', True), + ('https://s3-accelerate.foo.amazonaws.com/key', False), + # bucket prefixes are not allowed + ('https://bucket.s3-accelerate.amazonaws.com/key', False), + # S3 accelerate can be combined with dualstack + ('https://s3-accelerate.dualstack.amazonaws.com/key', True), + ('https://bucket.s3-accelerate.dualstack.amazonaws.com/key', False), + ('https://s3-accelerate.dualstack.dualstack.amazonaws.com/key', False), + ('https://s3-accelerate.dualstack.foo.amazonaws.com/key', False), + ('https://dualstack.s3-accelerate.amazonaws.com/key', False), + # assorted other ways for URLs to not be valid for s3-accelerate + ('ftp://s3-accelerate.dualstack.foo.amazonaws.com/key', False), + ('https://s3-accelerate.dualstack.foo.c2s.ic.gov/key', False), + # None-valued url is accepted + (None, False), ), ) def test_is_s3_accelerate_url(url, expected): @@ -3549,11 +3549,11 @@ def test_is_s3_accelerate_url(url, expected): @pytest.mark.parametrize( 'headers, default, expected', ( - ({}, 'ISO-8859-1', None), - ({'Content-Type': 'text/html; charset=utf-8'}, 'default', 'utf-8'), - ({'Content-Type': 'text/html; charset="utf-8"'}, 'default', 'utf-8'), - ({'Content-Type': 'text/html'}, 'ascii', 'ascii'), - ({'Content-Type': 'application/json'}, 'ISO-8859-1', None), + ({}, 'ISO-8859-1', None), + ({'Content-Type': 'text/html; charset=utf-8'}, 'default', 'utf-8'), + ({'Content-Type': 'text/html; charset="utf-8"'}, 'default', 'utf-8'), + ({'Content-Type': 'text/html'}, 'ascii', 'ascii'), + ({'Content-Type': 'application/json'}, 'ISO-8859-1', None), ), ) def test_get_encoding_from_headers(headers, default, expected):