From 373866df44dd0f9007fc833f52bb2449def49aca Mon Sep 17 00:00:00 2001 From: Julien Boulanger Date: Thu, 4 Jan 2024 12:08:17 +0100 Subject: [PATCH] chore: reformat --- tests/storage/s3_storage_test.py | 185 +++++++++++++++---------------- 1 file changed, 91 insertions(+), 94 deletions(-) diff --git a/tests/storage/s3_storage_test.py b/tests/storage/s3_storage_test.py index 0f57462f..e27c1700 100644 --- a/tests/storage/s3_storage_test.py +++ b/tests/storage/s3_storage_test.py @@ -29,17 +29,13 @@ class S3StorageTest(unittest.TestCase): original_call = None def setUp(self): - print("setting up botocore mock") self.original_call = botocore.utils.FileWebIdentityTokenLoader.__call__ def tearDown(self): - print("tearing down botocore mock") botocore.utils.FileWebIdentityTokenLoader.__call__ = self.original_call def test_legacy_provider_region_replacement(self): - assert ( - S3BaseStorage._region_from_provider_name("s3_us_west_oregon") == "us-west-2" - ) + assert S3BaseStorage._region_from_provider_name("s3_us_west_oregon") == "us-west-2" def test_credentials_from_metadata(self): @@ -151,7 +147,7 @@ def test_credentials_from_everything(self): }) credentials = S3BaseStorage._consolidate_credentials(config) - self.assertEqual("key-from-file", credentials.access_key_id) + self.assertEqual('key-from-file', credentials.access_key_id) del(os.environ['AWS_ACCESS_KEY_ID']) del(os.environ['AWS_SECRET_ACCESS_KEY']) @@ -171,16 +167,16 @@ def test_credentials_with_default_region(self): self.assertIsNone(os.environ.get('AWS_SECRET_ACCESS_KEY', None)) config = AttributeDict({ - "api_profile": "default", - "region": "default", - "storage_provider": "s3_us_west_oregon", - "key_file": credentials_file.name, - "concurrent_transfers": "1", + 'api_profile': 'default', + 'region': 'default', + 'storage_provider': 's3_us_west_oregon', + 'key_file': credentials_file.name, + 'concurrent_transfers': '1' }) credentials = S3BaseStorage._consolidate_credentials(config) - self.assertEqual("key-from-file", credentials.access_key_id) - self.assertEqual("us-west-2", credentials.region) + self.assertEqual('key-from-file', credentials.access_key_id) + self.assertEqual('us-west-2', credentials.region) def test_credentials_with_default_region_and_s3_compatible_storage(self): credentials_file_content = """ @@ -197,101 +193,107 @@ def test_credentials_with_default_region_and_s3_compatible_storage(self): self.assertIsNone(os.environ.get('AWS_SECRET_ACCESS_KEY', None)) config = AttributeDict({ - "api_profile": "default", - "region": "default", - "storage_provider": "s3_compatible", - "key_file": credentials_file.name, - "concurrent_transfers": "1", + 'api_profile': 'default', + 'region': 'default', + 'storage_provider': 's3_compatible', + 'key_file': credentials_file.name, + 'concurrent_transfers': '1' }) credentials = S3BaseStorage._consolidate_credentials(config) - self.assertEqual("key-from-file", credentials.access_key_id) + self.assertEqual('key-from-file', credentials.access_key_id) # default AWS region - self.assertEqual("us-east-1", credentials.region) + self.assertEqual('us-east-1', credentials.region) def test_make_s3_url(self): with patch('botocore.httpsession.URLLib3Session', return_value=_make_instance_metadata_mock()): with tempfile.NamedTemporaryFile() as empty_file: config = AttributeDict({ - "storage_provider": "s3_us_west_oregon", - "region": "default", - "key_file": empty_file.name, - "api_profile": None, - "kms_id": None, - "transfer_max_bandwidth": None, - "bucket_name": "whatever-bucket", - "secure": "True", - "host": None, - "port": None, - "concurrent_transfers": "1", + 'storage_provider': 's3_us_west_oregon', + 'region': 'default', + 'key_file': empty_file.name, + 'api_profile': None, + 'kms_id': None, + 'transfer_max_bandwidth': None, + 'bucket_name': 'whatever-bucket', + 'secure': 'True', + 'host': None, + 'port': None, + 'concurrent_transfers': '1' }) s3_storage = S3BaseStorage(config) # there are no extra connection args when connecting to regular S3 - self.assertEqual(dict(), s3_storage.connection_extra_args) + self.assertEqual( + dict(), + s3_storage.connection_extra_args + ) def test_make_s3_url_without_secure(self): with patch('botocore.httpsession.URLLib3Session', return_value=_make_instance_metadata_mock()): with tempfile.NamedTemporaryFile() as empty_file: config = AttributeDict({ - "storage_provider": "s3_us_west_oregon", - "region": "default", - "key_file": empty_file.name, - "api_profile": None, - "kms_id": None, - "transfer_max_bandwidth": None, - "bucket_name": "whatever-bucket", - "secure": "False", - "host": None, - "port": None, - "concurrent_transfers": "1", + 'storage_provider': 's3_us_west_oregon', + 'region': 'default', + 'key_file': empty_file.name, + 'api_profile': None, + 'kms_id': None, + 'transfer_max_bandwidth': None, + 'bucket_name': 'whatever-bucket', + 'secure': 'False', + 'host': None, + 'port': None, + 'concurrent_transfers': '1' }) s3_storage = S3BaseStorage(config) # again, no extra connection args when connecting to regular S3 # we can't even disable HTTPS - self.assertEqual(dict(), s3_storage.connection_extra_args) + self.assertEqual( + dict(), + s3_storage.connection_extra_args + ) def test_make_s3_compatible_url(self): with patch('botocore.httpsession.URLLib3Session', return_value=_make_instance_metadata_mock()): with tempfile.NamedTemporaryFile() as empty_file: config = AttributeDict({ - "storage_provider": "s3_compatible", - "region": "default", - "key_file": empty_file.name, - "api_profile": None, - "kms_id": None, - "transfer_max_bandwidth": None, - "bucket_name": "whatever-bucket", - "secure": "True", - "host": "s3.example.com", - "port": "443", - "concurrent_transfers": "1", + 'storage_provider': 's3_compatible', + 'region': 'default', + 'key_file': empty_file.name, + 'api_profile': None, + 'kms_id': None, + 'transfer_max_bandwidth': None, + 'bucket_name': 'whatever-bucket', + 'secure': 'True', + 'host': 's3.example.com', + 'port': '443', + 'concurrent_transfers': '1' }) s3_storage = S3BaseStorage(config) self.assertEqual( - "https://s3.example.com:443", - s3_storage.connection_extra_args["endpoint_url"], + 'https://s3.example.com:443', + s3_storage.connection_extra_args['endpoint_url'] ) def test_make_s3_compatible_url_without_secure(self): with patch('botocore.httpsession.URLLib3Session', return_value=_make_instance_metadata_mock()): with tempfile.NamedTemporaryFile() as empty_file: config = AttributeDict({ - "storage_provider": "s3_compatible", - "region": "default", - "key_file": empty_file.name, - "api_profile": None, - "kms_id": None, - "transfer_max_bandwidth": None, - "bucket_name": "whatever-bucket", - "secure": "False", - "host": "s3.example.com", - "port": "8080", - "concurrent_transfers": "1", + 'storage_provider': 's3_compatible', + 'region': 'default', + 'key_file': empty_file.name, + 'api_profile': None, + 'kms_id': None, + 'transfer_max_bandwidth': None, + 'bucket_name': 'whatever-bucket', + 'secure': 'False', + 'host': 's3.example.com', + 'port': '8080', + 'concurrent_transfers': '1' }) s3_storage = S3BaseStorage(config) self.assertEqual( - "http://s3.example.com:8080", - s3_storage.connection_extra_args["endpoint_url"], + 'http://s3.example.com:8080', + s3_storage.connection_extra_args['endpoint_url'] ) def test_assume_role_authentication(self): @@ -318,17 +320,17 @@ def test_assume_role_authentication(self): mock_file_content = 'eyJh...' mock_call = mock_open(read_data=mock_file_content) config = AttributeDict({ - "storage_provider": "s3_us_west_oregon", - "region": "default", - "key_file": "", - "api_profile": None, - "kms_id": None, - "transfer_max_bandwidth": None, - "bucket_name": "whatever-bucket", - "secure": "True", - "host": None, - "port": None, - "concurrent_transfers": "1" + 'storage_provider': 's3_us_west_oregon', + 'region': 'default', + 'key_file': '', + 'api_profile': None, + 'kms_id': None, + 'transfer_max_bandwidth': None, + 'bucket_name': 'whatever-bucket', + 'secure': 'True', + 'host': None, + 'port': None, + 'concurrent_transfers': '1' }) # Replace the open function with the mock @@ -344,30 +346,25 @@ def _make_instance_metadata_mock(): mock_response.status_code = 200 in_one_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) mock_response.text = json.dumps({ - "AccessKeyId": "key-from-instance-metadata", - "SecretAccessKey": "secret-from-instance-metadata", - "Token": "token-from-metadata", - "Expiration": in_one_hour.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] - + "Z", # -3 to remove microseconds + "AccessKeyId": 'key-from-instance-metadata', + "SecretAccessKey": 'secret-from-instance-metadata', + "Token": 'token-from-metadata', + "Expiration": in_one_hour.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z' # -3 to remove microseconds }) mock_send = MagicMock(return_value=mock_response) mock_session = MagicMock() mock_session.send = mock_send return mock_session - def _make_assume_role_with_web_identity_mock(): # mock a call to the AssumeRoleWithWebIdentity endpoint mock_response = MagicMock() mock_response.status_code = 200 in_one_hour = datetime.datetime.utcnow() + datetime.timedelta(hours=1) mock_response.text = json.dumps({ - "Credentials": { - "AccessKeyId": "key-from-assume-role", - "SecretAccessKey": "secret-from-assume-role", - "SessionToken": "token-from-assume-role", - "Expiration": in_one_hour.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] - + "Z", # -3 to remove microseconds - } + "AccessKeyId": 'key-from-assume-role', + "SecretAccessKey": 'secret-from-assume-role', + "SessionToken": 'token-from-assume-role', + "Expiration": in_one_hour.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z' # -3 to remove microseconds }) return MagicMock(return_value=mock_response)