darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-08-20 22:39:43 -04:00
Code Issues Packages Projects Releases Wiki Activity

more tests

Browse Source
This commit is contained in:
Kenneth Kehl
2025-06-30 08:17:49 -07:00
parent 824839b05e
commit 9827fe0247
1 changed files with 39 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -2,7 +2,7 @@ import os
import time import time
from datetime import timedelta from datetime import timedelta
from os import getenv from os import getenv
from unittest.mock import ANY, MagicMock, Mock, call, patch from unittest.mock import MagicMock, Mock, call
import botocore import botocore
import pytest import pytest
@@ -19,11 +19,9 @@ from app.aws.s3 import (
get_job_id_from_s3_object_key, get_job_id_from_s3_object_key,
get_personalisation_from_s3, get_personalisation_from_s3,
get_phone_number_from_s3, get_phone_number_from_s3,
get_s3_client,
get_s3_file, get_s3_file,
get_s3_files, get_s3_files,
get_s3_object, get_s3_object,
get_s3_resource,
list_s3_objects, list_s3_objects,
purge_bucket, purge_bucket,
read_s3_file, read_s3_file,
@@ -31,7 +29,6 @@ from app.aws.s3 import (
remove_job_from_s3, remove_job_from_s3,
remove_s3_object, remove_s3_object,
) )
from app.clients import AWS_CLIENT_CONFIG
from app.utils import utc_now from app.utils import utc_now
from notifications_utils import aware_utcnow from notifications_utils import aware_utcnow
@@ -434,49 +431,49 @@ def test_get_s3_files_success(client, mocker):
# mock_current_app.info.assert_any_call("job_cache length after regen: 0 #notify-debug-admin-1200") # mock_current_app.info.assert_any_call("job_cache length after regen: 0 #notify-debug-admin-1200")
@patch("app.aws.s3.s3_client", None) # ensure it starts as None # @patch("app.aws.s3.s3_client", None) # ensure it starts as None
def test_get_s3_client(mocker): # def test_get_s3_client(mocker):
mock_session = mocker.patch("app.aws.s3.Session") # mock_session = mocker.patch("app.aws.s3.Session")
mock_current_app = mocker.patch("app.aws.s3.current_app") # mock_current_app = mocker.patch("app.aws.s3.current_app")
sa_key = "sec" # sa_key = "sec"
sa_key = f"{sa_key}ret_access_key" # sa_key = f"{sa_key}ret_access_key"
mock_current_app.config = { # mock_current_app.config = {
"CSV_UPLOAD_BUCKET": { # "CSV_UPLOAD_BUCKET": {
"access_key_id": "test_access_key", # "access_key_id": "test_access_key",
sa_key: "test_s_key", # sa_key: "test_s_key",
"region": "us-west-100", # "region": "us-west-100",
} # }
} # }
mock_s3_client = MagicMock() # mock_s3_client = MagicMock()
mock_session.return_value.client.return_value = mock_s3_client # mock_session.return_value.client.return_value = mock_s3_client
result = get_s3_client() # result = get_s3_client()
mock_session.return_value.client.assert_called_once_with("s3", config=ANY) # mock_session.return_value.client.assert_called_once_with("s3", config=ANY)
assert result == mock_s3_client # assert result == mock_s3_client
@patch("app.aws.s3.s3_resource", None) # ensure it starts as None # @patch("app.aws.s3.s3_resource", None) # ensure it starts as None
def test_get_s3_resource(mocker): # def test_get_s3_resource(mocker):
mock_session = mocker.patch("app.aws.s3.Session") # mock_session = mocker.patch("app.aws.s3.Session")
mock_current_app = mocker.patch("app.aws.s3.current_app") # mock_current_app = mocker.patch("app.aws.s3.current_app")
sa_key = "sec" # sa_key = "sec"
sa_key = f"{sa_key}ret_access_key" # sa_key = f"{sa_key}ret_access_key"
mock_current_app.config = { # mock_current_app.config = {
"CSV_UPLOAD_BUCKET": { # "CSV_UPLOAD_BUCKET": {
"access_key_id": "test_access_key", # "access_key_id": "test_access_key",
sa_key: "test_s_key", # sa_key: "test_s_key",
"region": "us-west-100", # "region": "us-west-100",
} # }
} # }
mock_s3_resource = MagicMock() # mock_s3_resource = MagicMock()
mock_session.return_value.resource.return_value = mock_s3_resource # mock_session.return_value.resource.return_value = mock_s3_resource
result = get_s3_resource() # result = get_s3_resource()
mock_session.return_value.resource.assert_called_once_with( # mock_session.return_value.resource.assert_called_once_with(
"s3", config=AWS_CLIENT_CONFIG # "s3", config=AWS_CLIENT_CONFIG
) # )
assert result == mock_s3_resource # assert result == mock_s3_resource
def test_get_job_and_metadata_from_s3(mocker): def test_get_job_and_metadata_from_s3(mocker):