notifications-api/tests/notifications_utils/test_s3.py

from urllib.parse import parse_qs

import botocore
import pytest

from notifications_utils.s3 import S3ObjectNotFound, s3download, s3upload

contents = "some file data"
region = "eu-west-1"
bucket = "some_bucket"
location = "some_file_location"
content_type = "binary/octet-stream"


def test_s3upload_save_file_to_bucket(mocker):

    mock_s3_resource = mocker.Mock()
    mocked = mocker.patch(
        "notifications_utils.s3.get_s3_resource", return_value=mock_s3_resource
    )
    s3upload(
        filedata=contents, region=region, bucket_name=bucket, file_location=location
    )
    mocked_put = mocked.return_value.Object.return_value.put
    mocked_put.assert_called_once_with(
        Body=contents,
        ServerSideEncryption="AES256",
        ContentType=content_type,
    )


def test_s3upload_save_file_to_bucket_with_contenttype(mocker):
    content_type = "image/png"

    mock_s3_resource = mocker.Mock()
    mocked = mocker.patch(
        "notifications_utils.s3.get_s3_resource", return_value=mock_s3_resource
    )
    s3upload(
        filedata=contents,
        region=region,
        bucket_name=bucket,
        file_location=location,
        content_type=content_type,
    )
    mocked_put = mocked.return_value.Object.return_value.put
    mocked_put.assert_called_once_with(
        Body=contents,
        ServerSideEncryption="AES256",
        ContentType=content_type,
    )


def test_s3upload_raises_exception(app, mocker):

    mock_s3_resource = mocker.Mock()
    mocked = mocker.patch(
        "notifications_utils.s3.get_s3_resource", return_value=mock_s3_resource
    )
    response = {"Error": {"Code": 500}}
    exception = botocore.exceptions.ClientError(response, "Bad exception")
    mocked.return_value.Object.return_value.put.side_effect = exception
    with pytest.raises(botocore.exceptions.ClientError):
        s3upload(
            filedata=contents,
            region=region,
            bucket_name=bucket,
            file_location="location",
        )


def test_s3upload_save_file_to_bucket_with_urlencoded_tags(mocker):

    mock_s3_resource = mocker.Mock()
    mocked = mocker.patch(
        "notifications_utils.s3.get_s3_resource", return_value=mock_s3_resource
    )

    s3upload(
        filedata=contents,
        region=region,
        bucket_name=bucket,
        file_location=location,
        tags={"a": "1/2", "b": "x y"},
    )
    mocked_put = mocked.return_value.Object.return_value.put

    # make sure tags were a urlencoded query string
    encoded_tags = mocked_put.call_args[1]["Tagging"]
    assert parse_qs(encoded_tags) == {"a": ["1/2"], "b": ["x y"]}


def test_s3upload_save_file_to_bucket_with_metadata(mocker):

    mock_s3_resource = mocker.Mock()
    mocked = mocker.patch(
        "notifications_utils.s3.get_s3_resource", return_value=mock_s3_resource
    )

    s3upload(
        filedata=contents,
        region=region,
        bucket_name=bucket,
        file_location=location,
        metadata={"status": "valid", "pages": "5"},
    )
    mocked_put = mocked.return_value.Object.return_value.put

    metadata = mocked_put.call_args[1]["Metadata"]
    assert metadata == {"status": "valid", "pages": "5"}


def test_s3download_gets_file(mocker):

    mock_s3_resource = mocker.Mock()
    mocked = mocker.patch(
        "notifications_utils.s3.get_s3_resource", return_value=mock_s3_resource
    )

    mocked_object = mocked.return_value.Object
    mocked_object.return_value.get.return_value = {"Body": mocker.Mock()}
    s3download("bucket", "location.file")
    mocked_object.assert_called_once_with("bucket", "location.file")


def test_s3download_raises_on_error(mocker):

    mock_s3_resource = mocker.Mock()
    mocked = mocker.patch(
        "notifications_utils.s3.get_s3_resource", return_value=mock_s3_resource
    )

    mocked.return_value.Object.side_effect = botocore.exceptions.ClientError(
        {"Error": {"Code": 404}},
        "Bad exception",
    )

    with pytest.raises(S3ObjectNotFound):
        s3download("bucket", "location.file")