"""Test generate_upload_data module.""" from datetime import datetime, timezone from typing import Any from unittest.mock import MagicMock, call, patch import freezegun import pytest import pytest_mock from botocore.exceptions import ClientError from requests.exceptions import HTTPError from sqlalchemy.exc import SQLAlchemyError from assets import config from assets.connectors import s3 as s3_connector from assets.constants import ( asset_upload as asset_upload_constants, asset_upload_types as asset_upload_type_constants, error as error_constants, product as product_constants, ) from assets.exceptions import ( AssetStatusNotFound, AssetUploadError, AssetUploadNotFound, InvalidAssetUploadType, S3MultipartUploadFailed, TrackDoesNotBelongToProduct, WrongAssetStatus, ) from assets.logic import generate_upload_data from assets.models import asset_upload, asset_upload_type, token @pytest.fixture def fixture_username() -> str: """Fixture to get a name representing a user.""" return "anyone" @pytest.fixture def fixture_duration() -> int: """Fixture to duration.""" return 1000 @pytest.fixture def fixture_num_files() -> int: """Fixture to num_files.""" return 100 @pytest.fixture def fixture_credential() -> dict[str, Any]: """Fixture to get an AWS credential dictionary.""" return {"token": "a_token"} @pytest.fixture def fixture_stereo_asset_upload_type_id() -> int: """Fixture to get stereo asset upload type id.""" return 1 @pytest.fixture def fixture_static_artwork_asset_upload_type_id() -> int: """Fixture to get static artwork asset upload type id.""" return 2 def test__generate_filename() -> None: """Test the generated filename does not contain hyphens.""" result = generate_upload_data._generate_filename() assert "-" not in result @patch( "assets.models.ows_product.get_product_by_id", side_effect=HTTPError("error", response=MagicMock(status_code=500)), ) def test__get_additional_fields_for_asset_upload_product_fetch_error( mocked_get_product_by_id: MagicMock, ) -> None: """Test _get_additional_fields_for_asset_upload product fetch error.""" with pytest.raises(HTTPError) as exc: generate_upload_data._get_additional_fields_for_asset_upload( 123, asset_upload_type_constants.STEREO ) mocked_get_product_by_id.assert_called_once_with(123) assert str(exc.value) == "error" @patch( "assets.models.ows_product.get_product_by_id", return_value={ "status": product_constants.PRODUCT_STATUS_LABEL_PROCESSING, "context_type": "digital", "upc": 123, }, ) def test__get_additional_fields_for_asset_upload_label_processing( mocked_get_product_by_id: MagicMock, ) -> None: """Test _get_additional_fields_for_asset_upload label processing.""" assert generate_upload_data._get_additional_fields_for_asset_upload( 123, asset_upload_type_constants.STEREO ) == { "upc": 123, "is_correction": False, } mocked_get_product_by_id.assert_called_once_with(123) @patch( "assets.models.ows_product.get_product_by_id", return_value={ "status": product_constants.PRODUCT_STATUS_IN_CONTENT, "context_type": "digital", "upc": 123, }, ) def test__get_additional_fields_for_asset_upload_digital_in_content( mocked_get_product_by_id: MagicMock, ) -> None: """Test _get_additional_fields_for_asset_upload digital in_content.""" assert generate_upload_data._get_additional_fields_for_asset_upload( 123, asset_upload_type_constants.STEREO ) == { "upc": 123, "is_correction": True, } mocked_get_product_by_id.assert_called_once_with(123) @patch( "assets.models.ows_product.get_product_by_id", return_value={ "status": product_constants.PRODUCT_STATUS_IN_CONTENT, "context_type": "digital", "upc": 123, }, ) def test__get_additional_fields_for_asset_upload_digital_in_content_atmos( mocked_get_product_by_id: MagicMock, ) -> None: """Test _get_additional_fields_for_asset_upload digital in_content with atmos type.""" assert generate_upload_data._get_additional_fields_for_asset_upload( 123, asset_upload_type_constants.ATMOS ) == { "upc": 123, "is_correction": False, } mocked_get_product_by_id.assert_called_once_with(123) @patch( "assets.models.ows_product.get_product_by_id", return_value={ "status": "in_content", "context_type": "physical", "upc": 123, }, ) def test__get_additional_fields_for_asset_upload_physical_in_content( mocked_get_product_by_id: MagicMock, ) -> None: """Test _get_additional_fields_for_asset_upload physical in_content.""" assert generate_upload_data._get_additional_fields_for_asset_upload( 123, asset_upload_type_constants.STATIC_ARTWORK ) == { "upc": 123, "is_correction": False, } mocked_get_product_by_id.assert_called_once_with(123) @patch( "assets.models.ows_product.get_product_by_id", side_effect=HTTPError( "Product is not in a state that allows asset upload", response=MagicMock(status_code=400), ), ) def test__get_additional_fields_for_asset_upload_transfer_to_content( mocked_get_product_by_id: MagicMock, ) -> None: """Test _get_additional_fields_for_asset_upload transfer_to_content.""" with pytest.raises(HTTPError) as exc: generate_upload_data._get_additional_fields_for_asset_upload( 123, asset_upload_type_constants.STEREO ) mocked_get_product_by_id.assert_called_once_with(123) assert str(exc.value) == "Product is not in a state that allows asset upload" @patch( "assets.models.asset_upload_type.resolve_asset_upload_type_id", return_value=1, ) @patch( "assets.logic.generate_upload_data._get_additional_fields_for_asset_upload", side_effect=HTTPError(response=MagicMock(status_code=500)), ) def test_create_asset_upload__get_additional_fields_for_asset_upload_error( mocked__get_additional_fields_for_asset_upload: MagicMock, mocked_resolve_asset_upload_type_id: MagicMock, ) -> None: """Test create_asset_upload with _get_additional_fields_for_asset_upload error.""" with pytest.raises(HTTPError) as exc: generate_upload_data.create_asset_upload( product_id=123, track_unique_id=456, original_filename="a_filename.wav", user_id="alw:123", asset_upload_type=asset_upload_type_constants.STEREO, ) mocked__get_additional_fields_for_asset_upload.assert_called_once_with( 123, asset_upload_type_constants.STEREO ) mocked_resolve_asset_upload_type_id.assert_called_once_with( asset_upload_type_constants.STEREO ) assert str(exc.value) == "" @patch( "assets.models.ows_track.get_track_by_id", side_effect=HTTPError("ows-track error", response=MagicMock(status_code=500)), ) @patch( "assets.models.asset_upload_type.resolve_asset_upload_type_id", return_value=1, ) @patch( "assets.logic.generate_upload_data._get_additional_fields_for_asset_upload", return_value={"upc": 123, "is_correction": False}, ) def test_create_asset_upload_get_track_by_id_error( mocked__get_additional_fields_for_asset_upload: MagicMock, mocked_resolve_asset_upload_type_id: MagicMock, mocked_get_track_by_id: MagicMock, ) -> None: """Test create_asset_upload with get_track_by_id error.""" with pytest.raises(HTTPError) as exc: generate_upload_data.create_asset_upload( product_id=123, track_unique_id=456, original_filename="a_filename.wav", user_id="alw:123", asset_upload_type=asset_upload_type_constants.STEREO, ) mocked__get_additional_fields_for_asset_upload.assert_called_once_with( 123, asset_upload_type_constants.STEREO ) mocked_resolve_asset_upload_type_id.assert_called_once_with( asset_upload_type_constants.STEREO ) mocked_get_track_by_id.assert_called_once_with(456) assert str(exc.value) == "ows-track error" @patch( "assets.models.ows_track.get_track_by_id", return_value={"product_id": 987}, ) @patch( "assets.models.asset_upload_type.resolve_asset_upload_type_id", return_value=1, ) @patch( "assets.logic.generate_upload_data._get_additional_fields_for_asset_upload", return_value={"upc": 123, "is_correction": False}, ) def test_create_asset_upload_track_not_on_product( mocked__get_additional_fields_for_asset_upload: MagicMock, mocked_resolve_asset_upload_type_id: MagicMock, mocked_get_track_by_id: MagicMock, ) -> None: """Test create_asset_upload with track not on product.""" with pytest.raises(TrackDoesNotBelongToProduct) as exc: generate_upload_data.create_asset_upload( product_id=123, track_unique_id=456, original_filename="a_filename.wav", user_id="alw:123", asset_upload_type=asset_upload_type_constants.STEREO, ) mocked__get_additional_fields_for_asset_upload.assert_called_once_with( 123, asset_upload_type_constants.STEREO ) mocked_resolve_asset_upload_type_id.assert_called_once_with( asset_upload_type_constants.STEREO ) mocked_get_track_by_id.assert_called_once_with(456) assert exc.value.description == "Track does not belong to the product" def test_create_asset_upload_image_type_with_track_unique_id_exception() -> None: """Test create_asset_upload raises when image upload type is paired with track_unique_id.""" with pytest.raises(InvalidAssetUploadType): generate_upload_data.create_asset_upload( product_id=123, track_unique_id=456, original_filename="a_filename.tiff", user_id="alw:123", asset_upload_type=asset_upload_type_constants.STATIC_ARTWORK, ) def test_create_asset_upload_audio_type_without_track_unique_id_exception() -> None: """Test create_asset_upload raises when audio upload type is provided without track_unique_id.""" with pytest.raises(InvalidAssetUploadType): generate_upload_data.create_asset_upload( product_id=123, track_unique_id=0, original_filename="a_filename.wav", user_id="alw:123", asset_upload_type=asset_upload_type_constants.STEREO, ) @pytest.mark.parametrize( "mock_asset_upload_type", [ asset_upload_type_constants.STEREO, None, ], ) @patch( "assets.models.asset_upload_type.resolve_asset_upload_type_id", return_value=1, ) @patch( "assets.models.asset_upload.create_asset_upload_with_context", return_value=None, ) @patch("assets.connectors.s3.get_s3_client") @patch( "assets.logic.generate_upload_data._generate_filename", return_value="cf5dbe84_4a88_43f3_b90a_186289d748cc", ) @patch( "assets.models.ows_track.get_track_by_id", return_value={"product_id": 123}, ) @patch( "assets.logic.generate_upload_data._get_additional_fields_for_asset_upload", # this should not be patched return_value={"upc": 123, "is_correction": False}, ) def test_create_asset_upload_success_with_track_unique_id( mocked__get_additional_fields_for_asset_upload: MagicMock, mocked_get_track_by_id: MagicMock, mocked__generate_filename: MagicMock, mocked_get_s3_client: MagicMock, mocked_create_asset_upload_with_context: MagicMock, mocked_resolve_asset_upload_type_id: MagicMock, mock_asset_upload_type: str | None, fixture_stereo_asset_upload_type_id: int, ) -> None: """Test create_asset_upload success with track_unique_id.""" mocked_get_s3_client.return_value.create_multipart_upload.return_value = { "UploadId": "some_s3_multipart_upload_id" } filename = "cf5dbe84_4a88_43f3_b90a_186289d748cc" result = generate_upload_data.create_asset_upload( product_id=123, track_unique_id=456, original_filename="a_filename.wav", user_id="alw:123", asset_upload_type=mock_asset_upload_type, ) mocked__get_additional_fields_for_asset_upload.assert_called_once_with( 123, asset_upload_type_constants.STEREO ) mocked_get_track_by_id.assert_called_once_with(456) mocked__generate_filename.assert_called_once() mocked_get_s3_client.return_value.create_multipart_upload.assert_called_once_with( Bucket=config.RAW_ASSETS_BUCKET_NAME, Key=filename, ContentType="application/octet-stream", ) mocked_create_asset_upload_with_context.assert_called_once_with( user_id="alw:123", asset_upload_type_id=fixture_stereo_asset_upload_type_id, token="some_s3_multipart_upload_id", filename=filename, product_id=123, upc=123, track_unique_id=456, is_correction=False, original_filename="a_filename.wav", ) mocked_resolve_asset_upload_type_id.assert_called_once_with( asset_upload_type_constants.STEREO, ) assert result == filename @pytest.mark.parametrize( "mock_asset_upload_type", [ asset_upload_type_constants.STATIC_ARTWORK, None, ], ) @patch( "assets.models.asset_upload.create_asset_upload_with_context", return_value={}, ) @patch("assets.connectors.s3.get_s3_client") @patch( "assets.logic.generate_upload_data._generate_filename", return_value="cf5dbe84_4a88_43f3_b90a_186289d748cc", ) @patch("assets.models.ows_track.get_track_by_id") @patch( "assets.logic.generate_upload_data._get_additional_fields_for_asset_upload", return_value={"upc": 123, "is_correction": False}, ) @patch( "assets.models.asset_upload_type.resolve_asset_upload_type_id", return_value=2, ) def test_create_asset_upload_success_without_track_unique_id( mocked_resolve_asset_upload_type_id: MagicMock, mocked__get_additional_fields_for_asset_upload: MagicMock, mocked_get_track_by_id: MagicMock, mocked__generate_filename: MagicMock, mocked_get_s3_client: MagicMock, mocked_create_asset_upload_with_context: MagicMock, mock_asset_upload_type: str | None, fixture_static_artwork_asset_upload_type_id: int, ) -> None: """Test create_asset_upload success without track_unique_id.""" mocked_get_s3_client.return_value.create_multipart_upload.return_value = { "UploadId": "some_s3_multipart_upload_id" } filename = "cf5dbe84_4a88_43f3_b90a_186289d748cc" result = generate_upload_data.create_asset_upload( product_id=123, track_unique_id=0, original_filename="a_filename.wav", user_id="alw:123", asset_upload_type=mock_asset_upload_type, ) mocked__get_additional_fields_for_asset_upload.assert_called_once_with( 123, asset_upload_type_constants.STATIC_ARTWORK ) mocked_get_track_by_id.assert_not_called() mocked__generate_filename.assert_called_once() mocked_get_s3_client.return_value.create_multipart_upload.assert_called_once_with( Bucket=config.RAW_ASSETS_BUCKET_NAME, Key=filename, ContentType="application/octet-stream", ) mocked_create_asset_upload_with_context.assert_called_once_with( user_id="alw:123", asset_upload_type_id=fixture_static_artwork_asset_upload_type_id, token="some_s3_multipart_upload_id", filename=filename, product_id=123, upc=123, track_unique_id=0, is_correction=False, original_filename="a_filename.wav", ) mocked_resolve_asset_upload_type_id.assert_called_once_with( asset_upload_type_constants.STATIC_ARTWORK ) assert result == filename @patch( "assets.models.asset_upload.create_asset_upload_with_context", side_effect=SQLAlchemyError("Query error"), ) @patch( "assets.connectors.s3.get_s3_client", return_value=MagicMock( create_multipart_upload=MagicMock( return_value={"UploadId": "some_s3_multipart_upload_id"}, ), ), ) @patch( "assets.logic.generate_upload_data._generate_filename", return_value="cf5dbe84_4a88_43f3_b90a_186289d748cc", ) @patch( "assets.models.ows_track.get_track_by_id", return_value={"product_id": 123}, ) @patch( "assets.logic.generate_upload_data._get_additional_fields_for_asset_upload", # this should not be patched return_value={"upc": 123, "is_correction": False}, ) @patch( "assets.models.asset_upload_type.resolve_asset_upload_type_id", return_value=1, ) def test_create_asset_upload_create_asset_upload_with_context_error( mocked_resolve_asset_upload_type_id: MagicMock, mocked__get_additional_fields_for_asset_upload: MagicMock, mocked_get_track_by_id: MagicMock, mocked__generate_filename: MagicMock, mocked_get_s3_client: MagicMock, mocked_create_asset_upload_with_context: MagicMock, fixture_stereo_asset_upload_type_id: int, ) -> None: """Test create_asset_upload create_asset_upload_with_context error.""" with pytest.raises(SQLAlchemyError) as exc: generate_upload_data.create_asset_upload( product_id=123, track_unique_id=456, original_filename="a_filename.wav", user_id="alw:123", asset_upload_type=asset_upload_type_constants.STEREO, ) mocked__get_additional_fields_for_asset_upload.assert_called_once_with( 123, asset_upload_type_constants.STEREO ) mocked_get_track_by_id.assert_called_once_with(456) mocked__generate_filename.assert_called_once() mocked_get_s3_client.return_value.create_multipart_upload.assert_called_once_with( Bucket=config.RAW_ASSETS_BUCKET_NAME, Key="cf5dbe84_4a88_43f3_b90a_186289d748cc", ContentType="application/octet-stream", ) mocked_create_asset_upload_with_context.assert_called_once_with( user_id="alw:123", asset_upload_type_id=fixture_stereo_asset_upload_type_id, token="some_s3_multipart_upload_id", filename="cf5dbe84_4a88_43f3_b90a_186289d748cc", product_id=123, upc=123, track_unique_id=456, is_correction=False, original_filename="a_filename.wav", ) mocked_resolve_asset_upload_type_id.assert_called_once_with( asset_upload_type_constants.STEREO ) assert str(exc.value) == "Query error" @patch( "assets.models.asset_upload.get_asset_upload", side_effect=AssetUploadNotFound(error_constants.ERROR_ASSET_UPLOAD_NOT_FOUND), ) def test_get_presigned_urls_for_asset_upload_get_asset_upload_error( mocked_get_asset_upload: MagicMock, ) -> None: """Test get_presigned_urls_for_asset_upload get_asset_upload error.""" with pytest.raises(AssetUploadNotFound) as exc: generate_upload_data.get_presigned_urls_for_asset_upload( part_numbers=[1, 2, 3], filename="a_filename", user_id="alw:123", ) assert exc.value.description == error_constants.ERROR_ASSET_UPLOAD_NOT_FOUND mocked_get_asset_upload.assert_called_once_with( "a_filename", user_id="alw:123", api_version=2, ) @patch( "assets.models.asset_status.get_last_asset_status", side_effect=[AssetStatusNotFound(error_constants.ERROR_ASSET_STATUS_NOT_FOUND)], ) @patch("assets.models.asset_upload.get_asset_upload") def test_get_presigned_urls_for_asset_upload_get_last_asset_status_error( mocked_get_asset_upload: MagicMock, mocked_get_last_asset_status: MagicMock ) -> None: """Test get_presigned_urls_for_asset_upload get_last_asset_status error.""" with pytest.raises(AssetStatusNotFound) as exc: generate_upload_data.get_presigned_urls_for_asset_upload( part_numbers=[1, 2, 3], filename="a_filename", user_id="alw:123", ) mocked_get_asset_upload.assert_called_once_with( "a_filename", user_id="alw:123", api_version=2, ) mocked_get_last_asset_status.assert_called_once_with( mocked_get_asset_upload.return_value["id"], ) assert exc.value.description == error_constants.ERROR_ASSET_STATUS_NOT_FOUND @patch( "assets.models.asset_status.get_last_asset_status", return_value={"status": "asdf"}, ) @patch("assets.models.asset_upload.get_asset_upload") def test_get_presigned_urls_for_asset_upload_invalid_status( mocked_get_asset_upload: MagicMock, mocked_get_last_asset_status: MagicMock ) -> None: """Test get_presigned_urls_for_asset_upload invalid status.""" with pytest.raises(WrongAssetStatus) as exc: generate_upload_data.get_presigned_urls_for_asset_upload( part_numbers=[1, 2, 3], filename="a_filename", user_id="alw:123", ) mocked_get_asset_upload.assert_called_once_with( "a_filename", user_id="alw:123", api_version=2, ) mocked_get_last_asset_status.assert_called_once_with( mocked_get_asset_upload.return_value["id"], ) assert ( exc.value.description == "asset_upload not in state that allows generation of presigned urls for uploads." ) @patch( "assets.models.asset_status.get_last_asset_status", return_value={"status": "uploading"}, ) @patch("assets.models.asset_upload.get_asset_upload") @patch("assets.connectors.s3.get_s3_client") def test_get_presigned_urls_for_asset_upload_success( mocked_get_s3_client: MagicMock, mocked_get_asset_upload: MagicMock, mocked_get_last_asset_status: MagicMock, ) -> None: """Test get_presigned_urls_for_asset_upload success.""" mocked_get_s3_client.return_value.generate_presigned_url.side_effect = [ "presigned_url_1", "presigned_url_2", "presigned_url_3", ] result = generate_upload_data.get_presigned_urls_for_asset_upload( part_numbers=[1, 2, 3], filename="a_filename", user_id="alw:123", ) mocked_get_asset_upload.assert_called_once_with( "a_filename", user_id="alw:123", api_version=2, ) mocked_get_last_asset_status.assert_called_once_with( mocked_get_asset_upload.return_value["id"], ) mocked_get_s3_client.assert_called_with(use_accelerate_endpoint=True) mocked_get_s3_client.return_value.generate_presigned_url.assert_has_calls( [ call( ClientMethod="upload_part", ExpiresIn=config.ASSET_UPLOAD_PRESIGNED_URL_EXPIRES_IN_SECONDS, Params={ "Bucket": config.RAW_ASSETS_BUCKET_NAME, "Key": "a_filename", "UploadId": mocked_get_asset_upload.return_value["token"], "PartNumber": 1, }, ), call( ClientMethod="upload_part", ExpiresIn=config.ASSET_UPLOAD_PRESIGNED_URL_EXPIRES_IN_SECONDS, Params={ "Bucket": config.RAW_ASSETS_BUCKET_NAME, "Key": "a_filename", "UploadId": mocked_get_asset_upload.return_value["token"], "PartNumber": 2, }, ), call( ClientMethod="upload_part", ExpiresIn=config.ASSET_UPLOAD_PRESIGNED_URL_EXPIRES_IN_SECONDS, Params={ "Bucket": config.RAW_ASSETS_BUCKET_NAME, "Key": "a_filename", "UploadId": mocked_get_asset_upload.return_value["token"], "PartNumber": 3, }, ), ] ) assert result == { "part_number_to_presigned_url": { 1: "presigned_url_1", 2: "presigned_url_2", 3: "presigned_url_3", } } @patch("assets.models.asset_status.create_asset_status") @patch( "assets.models.asset_upload.get_asset_upload", side_effect=AssetUploadNotFound(error_constants.ERROR_ASSET_UPLOAD_NOT_FOUND), ) def test_complete_multipart_upload_get_asset_upload_error( mocked_get_asset_upload: MagicMock, mocked_create_asset_status: MagicMock, mocker: pytest_mock.MockerFixture, ) -> None: """Test complete_multipart_upload get_asset_upload error.""" mock_complete_multipart_upload = mocker.patch.object( s3_connector.get_s3_client(), "complete_multipart_upload" ) with pytest.raises(AssetUploadNotFound) as exc: generate_upload_data.complete_multipart_upload( filename="a_filename", parts=[ {"part_number": 1, "etag": "etag_1"}, ], user_id="alw:123", ) assert exc.value.description == error_constants.ERROR_ASSET_UPLOAD_NOT_FOUND mocked_get_asset_upload.assert_called_once_with( "a_filename", user_id="alw:123", api_version=2, ) mock_complete_multipart_upload.assert_not_called() mocked_create_asset_status.assert_not_called() @patch("assets.models.asset_status.create_asset_status") @patch( "assets.models.asset_status.get_last_asset_status", side_effect=[AssetStatusNotFound(error_constants.ERROR_ASSET_STATUS_NOT_FOUND)], ) @patch("assets.models.asset_upload.get_asset_upload") def test_complete_multipart_upload_get_last_asset_status_error( mocked_get_asset_upload: MagicMock, mocked_get_last_asset_status: MagicMock, mocked_create_asset_status: MagicMock, ) -> None: """Test complete_multipart_upload get_last_asset_status error.""" with pytest.raises(AssetStatusNotFound) as exc: generate_upload_data.complete_multipart_upload( filename="a_filename", parts=[ {"part_number": 1, "etag": "etag_1"}, ], user_id="alw:123", ) mocked_get_asset_upload.assert_called_once_with( "a_filename", user_id="alw:123", api_version=2, ) mocked_get_last_asset_status.assert_called_once_with( mocked_get_asset_upload.return_value["id"], ) mocked_create_asset_status.assert_not_called() assert exc.value.description == error_constants.ERROR_ASSET_STATUS_NOT_FOUND @patch("assets.models.asset_status.create_asset_status") @patch( "assets.models.asset_status.get_last_asset_status", return_value={"status": "asdf"}, ) @patch("assets.models.asset_upload.get_asset_upload") def test_complete_multipart_upload_invalid_status( mocked_get_asset_upload: MagicMock, mocked_get_last_asset_status: MagicMock, mocked_create_asset_status: MagicMock, ) -> None: """Test complete_multipart_upload invalid status.""" with pytest.raises(WrongAssetStatus) as exc: generate_upload_data.complete_multipart_upload( filename="a_filename", parts=[ {"part_number": 1, "etag": "etag_1"}, ], user_id="alw:123", ) mocked_get_asset_upload.assert_called_once_with( "a_filename", user_id="alw:123", api_version=2, ) mocked_get_last_asset_status.assert_called_once_with( mocked_get_asset_upload.return_value["id"], ) mocked_create_asset_status.assert_not_called() assert ( exc.value.description == "asset_upload not in state that allows completion of upload." ) @pytest.mark.parametrize( "error_code", [ "EntityTooSmall", "InvalidPart", "InvalidPartOrder", "NoSuchUpload", ], ) @patch("assets.models.asset_status.create_asset_status") @patch( "assets.models.asset_status.get_last_asset_status", return_value={"status": "uploading"}, ) @patch("assets.models.asset_upload.get_asset_upload") def test_complete_multipart_upload_s3_user_error( mocked_get_asset_upload: MagicMock, mocked_get_last_asset_status: MagicMock, mocked_create_asset_status: MagicMock, mocker: pytest_mock.MockerFixture, error_code: str, ) -> None: """Test complete_multipart_upload s3 user error.""" mock_complete_multipart_upload = mocker.patch.object( s3_connector.get_s3_client(), "complete_multipart_upload", side_effect=ClientError( {"Error": {"Code": error_code}}, "complete_multipart_upload" ), ) with pytest.raises(S3MultipartUploadFailed) as exc: generate_upload_data.complete_multipart_upload( filename="a_filename", parts=[ {"part_number": 1, "etag": "etag_1"}, ], user_id="alw:123", ) assert exc.value.description.startswith("An error occurred") mocked_get_asset_upload.assert_called_once_with( "a_filename", user_id="alw:123", api_version=2, ) mocked_get_last_asset_status.assert_called_once_with( mocked_get_asset_upload.return_value["id"], ) mock_complete_multipart_upload.assert_called_once_with( Bucket=config.RAW_ASSETS_BUCKET_NAME, Key="a_filename", UploadId=mocked_get_asset_upload.return_value["token"], MultipartUpload={"Parts": [{"PartNumber": 1, "ETag": "etag_1"}]}, ) mocked_create_asset_status.assert_not_called() @pytest.mark.parametrize( "error, raised_exception", [ (Exception, Exception), ( ClientError({"Error": {"Code": "SomeError"}}, "complete_multipart_upload"), ClientError, ), ], ) @patch("assets.models.asset_status.create_asset_status") @patch( "assets.models.asset_status.get_last_asset_status", return_value={"status": "uploading"}, ) @patch("assets.models.asset_upload.get_asset_upload") def test_complete_multipart_upload_raised_errors( mocked_get_asset_upload: MagicMock, mocked_get_last_asset_status: MagicMock, mocked_create_asset_status: MagicMock, mocker: pytest_mock.MockerFixture, error: Any, raised_exception: Any, ) -> None: """Test complete_multipart_upload s3 raised error.""" mock_complete_multipart_upload = mocker.patch.object( s3_connector.get_s3_client(), "complete_multipart_upload", side_effect=error, ) with pytest.raises(raised_exception): generate_upload_data.complete_multipart_upload( filename="a_filename", parts=[ {"part_number": 1, "etag": "etag_1"}, ], user_id="alw:123", ) mocked_get_asset_upload.assert_called_once_with( "a_filename", user_id="alw:123", api_version=2, ) mocked_get_last_asset_status.assert_called_once_with( mocked_get_asset_upload.return_value["id"], ) mock_complete_multipart_upload.assert_called_once_with( Bucket=config.RAW_ASSETS_BUCKET_NAME, Key="a_filename", UploadId=mocked_get_asset_upload.return_value["token"], MultipartUpload={"Parts": [{"PartNumber": 1, "ETag": "etag_1"}]}, ) mocked_create_asset_status.assert_not_called() @freezegun.freeze_time("2024-02-27 21:58:08.183726") @patch("assets.models.asset_status.create_asset_status") @patch( "assets.models.asset_status.get_last_asset_status", return_value={"status": "uploading"}, ) @patch("assets.models.asset_upload.get_asset_upload") def test_complete_multipart_upload_success( mocked_get_asset_upload: MagicMock, mocked_get_last_asset_status: MagicMock, mocked_create_asset_status: MagicMock, mocker: pytest_mock.MockerFixture, ) -> None: """Test complete_multipart_upload success.""" mock_complete_multipart_upload = mocker.patch.object( s3_connector.get_s3_client(), "complete_multipart_upload" ) generate_upload_data.complete_multipart_upload( filename="a_filename", parts=[ {"part_number": 1, "etag": "etag_1"}, ], user_id="alw:123", ) mocked_get_asset_upload.assert_called_once_with( "a_filename", user_id="alw:123", api_version=2, ) mocked_get_last_asset_status.assert_called_once_with( mocked_get_asset_upload.return_value["id"], ) mock_complete_multipart_upload.assert_called_once_with( Bucket=config.RAW_ASSETS_BUCKET_NAME, Key="a_filename", UploadId=mocked_get_asset_upload.return_value["token"], MultipartUpload={"Parts": [{"PartNumber": 1, "ETag": "etag_1"}]}, ) mocked_create_asset_status.assert_called_once_with( asset_upload_id=mocked_get_asset_upload.return_value["id"], status="upload_complete", status_time=datetime(2024, 2, 27, 21, 58, 8, 183726, tzinfo=timezone.utc), ) def test_get_upload_permission_no_token( monkeypatch: pytest.MonkeyPatch, fixture_username: str, fixture_duration: int ) -> None: """Test get_upload_permission fails without token.""" monkeypatch.setattr( token, "get_s3_token", MagicMock( side_effect=ClientError( {"Error": {"Code": "StsError", "Message": "internal server error"}}, "sts", ) ), ) with pytest.raises(ClientError) as exc: generate_upload_data.get_upload_permission( fixture_username, fixture_duration, "", ) assert "An error occurred (StsError)" in str(exc.value) def test_get_upload_permission_no_asset_upload( monkeypatch: pytest.MonkeyPatch, fixture_credential: dict[str, Any], fixture_username: str, fixture_duration: int, fixture_stereo_asset_upload_type_id: int, ) -> None: """Test get_upload_permission fails when unable to log request.""" monkeypatch.setattr( token, "get_s3_token", MagicMock(return_value=fixture_credential) ) monkeypatch.setattr( asset_upload, "create_asset_upload", MagicMock( side_effect=AssetUploadError("Failed to create an asset upload record") ), ) monkeypatch.setattr( asset_upload_type, "resolve_asset_upload_type_id", MagicMock(return_value=fixture_stereo_asset_upload_type_id), ) with pytest.raises(AssetUploadError) as exc: generate_upload_data.get_upload_permission( fixture_username, fixture_duration, asset_upload_constants.ASSET_TYPE_AUDIO ) assert str(exc.value) == "Failed to create an asset upload record" def test_get_upload_permission( monkeypatch: pytest.MonkeyPatch, fixture_credential: dict[str, Any], fixture_username: str, fixture_duration: int, fixture_stereo_asset_upload_type_id: int, ) -> None: """Test get_upload_permission succeeds.""" bucket = "a_bucket" filename = "a_filename" monkeypatch.setattr(config, "RAW_ASSETS_BUCKET_NAME", bucket) monkeypatch.setattr( generate_upload_data, "_generate_filename", MagicMock(return_value=filename) ) token_mock = MagicMock(return_value=fixture_credential) monkeypatch.setattr(token, "get_s3_token", token_mock) asset_upload_mock = MagicMock(return_value={"something"}) monkeypatch.setattr(asset_upload, "create_asset_upload", asset_upload_mock) asset_upload_type_mock = MagicMock(return_value=fixture_stereo_asset_upload_type_id) monkeypatch.setattr( asset_upload_type, "resolve_asset_upload_type_id", asset_upload_type_mock ) expected_au_calls = [ call( fixture_username, filename, fixture_credential.get("token"), fixture_stereo_asset_upload_type_id, 2, ) ] expected_aut_calls = [call(asset_upload_type_constants.STEREO)] result = generate_upload_data.get_upload_permission( fixture_username, fixture_duration, asset_upload_constants.ASSET_TYPE_AUDIO ) assert result["bucket"] == bucket assert result["filename"] == filename assert result["credentials"] == fixture_credential token_mock.assert_called_with(filename, fixture_duration) asset_upload_mock.assert_has_calls(expected_au_calls, any_order=True) asset_upload_type_mock.assert_has_calls(expected_aut_calls, any_order=True) def test_get_entity_upload_permission_for_vendor( monkeypatch: pytest.MonkeyPatch, fixture_credential: dict[str, Any], fixture_duration: int, fixture_vendor_entity: str, fixture_vendor_id: int, ) -> None: """Test get_upload_permission_for_entity succeeds for vendor entity.""" bucket = "a_bucket" filename = "a_filename" logo_path = "images/vendor/logo" icon_path = "images/vendor/icon" monkeypatch.setattr(config, "ASSET_STORAGE_BUCKET_NAME", bucket) monkeypatch.setattr( generate_upload_data, "_generate_filename", MagicMock(return_value=filename) ) token_mock = MagicMock(return_value=fixture_credential) monkeypatch.setattr(token, "get_s3_entity_token", token_mock) result = generate_upload_data.get_entity_upload_permission( fixture_duration, fixture_vendor_entity, fixture_vendor_id ) assert result["bucket"] == bucket assert result["path"]["logo"] == logo_path assert result["path"]["icon"] == icon_path assert result["credentials"] == fixture_credential def test_get_entity_upload_permission_for_artist( monkeypatch: pytest.MonkeyPatch, fixture_credential: dict[str, Any], fixture_vendor_id: int, fixture_duration: int, fixture_artist_entity: str, ) -> None: """Test get_upload_permission_for_entity succeeds for artist entity.""" bucket = "a_bucket" filename = "a_filename" photos_path = "images/artist/photos" thumb_path = "images/artist/thumb" web_path = "images/artist/web" monkeypatch.setattr(config, "ASSET_STORAGE_BUCKET_NAME", bucket) monkeypatch.setattr( generate_upload_data, "_generate_filename", MagicMock(return_value=filename) ) token_mock = MagicMock(return_value=fixture_credential) monkeypatch.setattr(token, "get_s3_entity_token", token_mock) result = generate_upload_data.get_entity_upload_permission( fixture_duration, fixture_artist_entity, fixture_vendor_id ) assert result["bucket"] == bucket assert result["path"]["photos"] == photos_path assert result["path"]["thumb"] == thumb_path assert result["path"]["web"] == web_path assert result["credentials"] == fixture_credential def test_get_entity_upload_permission_failure( monkeypatch: pytest.MonkeyPatch, fixture_vendor_id: int, fixture_duration: int, fixture_artist_entity: str, ) -> None: """Test get_upload_permission_for_entity succeeds for artist entity.""" bucket = "a_bucket" filename = "a_filename" monkeypatch.setattr(config, "ASSET_STORAGE_BUCKET_NAME", bucket) monkeypatch.setattr( generate_upload_data, "_generate_filename", MagicMock(return_value=filename) ) token_mock = MagicMock( side_effect=ClientError( {"Error": {"Code": "StsError", "Message": "internal server error"}}, "sts", ) ) monkeypatch.setattr(token, "get_s3_entity_token", token_mock) with pytest.raises(ClientError) as exc: generate_upload_data.get_entity_upload_permission( fixture_duration, fixture_artist_entity, fixture_vendor_id ) assert "An error occurred (StsError)" in str(exc.value)