darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-08-24 16:23:44 -04:00
Code Issues Packages Projects Releases Wiki Activity

more tests

Browse Source
This commit is contained in:
Kenneth Kehl
2025-06-30 07:22:31 -07:00
parent ce7003edb3
commit 3c4ec961fe
2 changed files with 16 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
app/aws/s3.py
View File

@@ -23,11 +23,6 @@ NEW_FILE_LOCATION_STRUCTURE = "{}-service-notify/{}.csv"
ttl = 60 * 60 * 24 * 7 ttl = 60 * 60 * 24 * 7
# Global variable
s3_client = None
s3_resource = None
def get_service_id_from_key(key): def get_service_id_from_key(key):
key = key.replace("service-", "") key = key.replace("service-", "")
key = key.split("/") key = key.split("/")
@@ -86,17 +81,15 @@ def get_s3_client():
def get_s3_resource(): def get_s3_resource():
global s3_resource access_key = current_app.config["CSV_UPLOAD_BUCKET"]["access_key_id"]
if s3_resource is None: secret_key = current_app.config["CSV_UPLOAD_BUCKET"]["secret_access_key"]
access_key = current_app.config["CSV_UPLOAD_BUCKET"]["access_key_id"] region = current_app.config["CSV_UPLOAD_BUCKET"]["region"]
secret_key = current_app.config["CSV_UPLOAD_BUCKET"]["secret_access_key"] session = Session(
region = current_app.config["CSV_UPLOAD_BUCKET"]["region"] aws_access_key_id=access_key,
session = Session( aws_secret_access_key=secret_key,
aws_access_key_id=access_key, region_name=region,
aws_secret_access_key=secret_key, )
region_name=region, s3_resource = session.resource("s3", config=AWS_CLIENT_CONFIG)
)
s3_resource = session.resource("s3", config=AWS_CLIENT_CONFIG)
return s3_resource return s3_resource

39
tests/app/aws/test_s3.py
View File

@@ -7,6 +7,7 @@ import botocore
import pytest import pytest
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
from app.aws import s3
from app.aws.s3 import ( from app.aws.s3 import (
cleanup_old_s3_objects, cleanup_old_s3_objects,
download_from_s3, download_from_s3,
@@ -603,35 +604,9 @@ def test_get_s3_files_handles_exception(mocker):
) )
def test_get_s3_client_default_credentials(): def test_get_service_id_from_key_various_formats():
with patch.dict(os.environ, {}, clear=True): assert s3.get_service_id_from_key("service-123-notify/abc.csv") == "123"
with patch("app.aws.s3.Session") as mock_session: assert s3.get_service_id_from_key("service-xyz-notify/def/ghi.csv") == "xyz"
mock_client = MagicMock() assert s3.get_service_id_from_key("noservice-foo") == "noservice-foo".replace(
mock_session.return_value.client.return_value = mock_client "-notify", ""
client = get_s3_client() )
assert client is not None
assert mock_session.called
assert mock_session.return_value.client.called
mock_session.return_value.client.assert_called_once_with(
"s3",
config=ANY,
)
def test_get_s3_client_invalid_credentials():
with patch("app.aws.s3.Session") as mock_session:
mock_session.return_value.client.side_effect = botocore.exceptions.ClientError(
{
"Error": {
"Code": "InvalidClientTokenId",
"Message": "Invalid credentials",
}
},
"HeadBucket",
)
try:
get_s3_client()
assert 1 == 0
except botocore.exceptions.ClientError as e:
assert e.response["Error"]["Code"] == "InvalidClientTokenId"