This commit is contained in:
Kenneth Kehl
2024-08-01 09:23:26 -07:00
parent bd75e30f96
commit 4299b46f8a
2 changed files with 12 additions and 8 deletions

View File

@@ -104,15 +104,20 @@ def download_from_s3(
region_name=region, region_name=region,
) )
s3 = session.client("s3", config=AWS_CLIENT_CONFIG) s3 = session.client("s3", config=AWS_CLIENT_CONFIG)
result = None
try: try:
s3.download_file(bucket_name, s3_key, local_filename) result = s3.download_file(bucket_name, s3_key, local_filename)
print(f"File downloaded successfully to {local_filename}") print(f"File downloaded successfully to {local_filename}")
except botocore.exceptions.NoCredentialsError: except botocore.exceptions.NoCredentialsError as nce:
print("Credentials not found") print("Credentials not found")
except botocore.exceptions.PartialCredentialsError: raise Exception(nce)
except botocore.exceptions.PartialCredentialsError as pce:
print("Incomplete credentials provided") print("Incomplete credentials provided")
raise Exception(pce)
except Exception as e: except Exception as e:
print(f"An error occurred {e}") print(f"An error occurred {e}")
raise Exception(e)
return result
def get_s3_object(bucket_name, file_location, access_key, secret_key, region): def get_s3_object(bucket_name, file_location, access_key, secret_key, region):

View File

@@ -428,9 +428,7 @@ def test_promote_user_to_platform_admin(
assert user.platform_admin is True assert user.platform_admin is True
def test_download_csv_file_by_name( def test_download_csv_file_by_name(notify_api, sample_platform_admin):
notify_api, sample_platform_admin
):
assert sample_platform_admin.platform_admin is True assert sample_platform_admin.platform_admin is True
result = notify_api.test_cli_runner().invoke( result = notify_api.test_cli_runner().invoke(
@@ -440,8 +438,9 @@ def test_download_csv_file_by_name(
], ],
) )
print(result) # if we get a 404, it means that the file is not in s3 (of course)
assert result == '404' # but we are making the call to s3, etc. so that part is working
assert "404" in str(result)
def test_promote_user_to_platform_admin_no_result_found( def test_promote_user_to_platform_admin_no_result_found(