"""Test for asset_upload model.""" import copy import datetime from typing import Any, TypedDict from unittest.mock import MagicMock, patch import freezegun import pytest from flexmock import flexmock from pytest_mock import MockerFixture from sqlalchemy.exc import SQLAlchemyError from werkzeug.exceptions import HTTPException from assets.connectors import mysql from assets.constants import asset_upload_types, error as error_constants from assets.exceptions import ( AssetUploadNotFound, DuplicateAssetUploadFilename, MissingContext, NoAssetsFound, ) from assets.models import asset_upload from tests.unit.models import au_operations @pytest.fixture def valid_asset_upload_data() -> dict[str, Any]: """Get valid data for create_asset_upload.""" return { "user_id": "oa:123", "asset_upload_type_id": 1, "filename": "unique_filename", "token": "sts_s3_session_token", } @pytest.fixture def valid_asset_upload_bulk_data() -> dict[str, str | list[str]]: """Get valid data for create_asset_upload.""" return { "user_id": "oa:123", "filename_list": ["filename_1", "filename_2"], "token": "sts_s3_session_token", } @pytest.fixture def valid_asset_upload_data_for_copy() -> dict[str, Any]: """Get valid data for create_asset_upload_record_for_copy.""" return { "user_id": "oa:123", "asset_upload_type_id": 1, "filename": "filename", "token": None, "product_id": 123, "upc": "UPC_123", } @pytest.fixture def existing_asset_upload() -> dict[str, Any]: """Get search params for existing asset_upload record.""" return { "filename": "unique_filename_1", "user_id": "oa:123456", "token": "session_token_1", } @pytest.fixture def update_asset_upload_data(existing_asset_upload: dict[str, Any]) -> dict[str, Any]: """Get search params for existing asset_upload record.""" existing_asset_upload.update( { "product_id": 12345, "track_unique_id": 321, "upc": 123, "original_filename": "original_filename.jpg", "is_correction": True, } ) return existing_asset_upload @pytest.fixture def update_asset_multi_upload_data( update_asset_upload_data: dict[str, Any], ) -> dict[str, Any]: """Params for updating fixture data that has multiple assets per track.""" update_asset_upload_data.update( { "filename": "unique_filename_17", "is_correction": False, "track_unique_id": TRACK_ID_4, "product_id": PRODUCT_ID_5, "api_version": 2, } ) return update_asset_upload_data @pytest.fixture def get_audio_asset_uploads_by_track_and_product_ids_data() -> list[dict[str, Any]]: """Get expected entries for get_asset_uploads_by_track_and_product_ids.""" expected_data = [ { "id": ASSET_UPLOAD_PK_22, "bucket_name": "some_bucket", "filename": "unique_filename_1", "api_version": 1, "user_id": "oa:123456", "asset_upload_type_id": 1, "token": "session_token_1", "original_filename": "original_filename_1", "track_unique_id": TRACK_ID_1, "product_id": PRODUCT_ID_1, "upc": None, "deleted": False, "is_correction": True, "updated_timestamp": datetime.datetime(2017, 1, 6, 0, 0), }, { "id": ASSET_UPLOAD_PK_7, "bucket_name": "some_bucket", "filename": "unique_filename_1_1", "api_version": 1, "user_id": "oa:123456", "asset_upload_type_id": 1, "token": "session_token_1", "original_filename": "original_filename_1_1", "track_unique_id": TRACK_ID_1, "product_id": PRODUCT_ID_1, "deleted": False, "is_correction": False, "upc": None, "updated_timestamp": datetime.datetime(2017, 1, 2, 0, 0), }, { "id": ASSET_UPLOAD_PK_1, "bucket_name": "some_bucket", "filename": "unique_filename_1", "api_version": 1, "user_id": "oa:123456", "asset_upload_type_id": 1, "token": "session_token_1", "original_filename": "original_filename_1", "track_unique_id": TRACK_ID_1, "product_id": PRODUCT_ID_1, "deleted": False, "is_correction": True, "upc": None, "updated_timestamp": datetime.datetime(2017, 1, 6, 0, 0), }, ] return expected_data @pytest.fixture def get_image_asset_uploads_by_track_and_product_ids_data() -> list[dict[str, Any]]: """Get expected entries for get_asset_uploads_by_track_and_product_ids.""" expected_data = [ { "id": ASSET_UPLOAD_PK_12, "bucket_name": "some_bucket", "filename": "unique_filename_12", "api_version": 1, "user_id": "oa:654321", "asset_upload_type_id": 2, "token": "session_token_3", "original_filename": "original_filename_3", "track_unique_id": 0, "product_id": PRODUCT_ID_1, "deleted": False, "is_correction": False, "upc": None, "updated_timestamp": datetime.datetime(2017, 1, 1, 0, 0), } ] return expected_data ASSET_UPLOAD_PK_0 = 1000 ASSET_UPLOAD_PK_1 = 1001 ASSET_UPLOAD_PK_2 = 1002 ASSET_UPLOAD_PK_3 = 1003 ASSET_UPLOAD_PK_4 = 1004 ASSET_UPLOAD_PK_5 = 1005 ASSET_UPLOAD_PK_6 = 1006 ASSET_UPLOAD_PK_7 = 1007 ASSET_UPLOAD_PK_8 = 1008 ASSET_UPLOAD_PK_9 = 1009 ASSET_UPLOAD_PK_10 = 1010 ASSET_UPLOAD_PK_11 = 1011 ASSET_UPLOAD_PK_12 = 1012 ASSET_UPLOAD_PK_13 = 1013 ASSET_UPLOAD_PK_14 = 1014 ASSET_UPLOAD_PK_15 = 1015 ASSET_UPLOAD_PK_16 = 1016 ASSET_UPLOAD_PK_17 = 1017 ASSET_UPLOAD_PK_18 = 1018 ASSET_UPLOAD_PK_19 = 1019 ASSET_UPLOAD_PK_20 = 1020 ASSET_UPLOAD_PK_21 = 1021 ASSET_UPLOAD_PK_22 = 1022 ASSET_UPLOAD_PK_23 = 1023 ASSET_UPLOAD_PK_24 = 1024 ASSET_UPLOAD_PK_25 = 1025 ASSET_UPLOAD_PK_26 = 1026 ASSET_UPLOAD_PK_27 = 1027 ASSET_UPLOAD_PK_28 = 1028 ASSET_UPLOAD_PK_29 = 1029 ASSET_UPLOAD_PK_30 = 1030 ASSET_UPLOAD_PK_31 = 1031 ASSET_UPLOAD_PK_32 = 1032 ASSET_UPLOAD_PK_33 = 1033 ASSET_UPLOAD_PK_34 = 1034 ASSET_UPLOAD_PK_35 = 1035 ASSET_UPLOAD_PK_36 = 1036 ASSET_UPLOAD_PK_37 = 1037 ASSET_UPLOAD_NOT_EXISTS = 1100 PRODUCT_ID_1 = 1001 PRODUCT_ID_2 = 1002 PRODUCT_ID_3 = 1003 PRODUCT_ID_4 = 1004 PRODUCT_ID_5 = 1005 PRODUCT_ID_6 = 1006 PRODUCT_ID_7 = 1007 PRODUCT_ID_8 = 1008 PRODUCT_ID_9 = 1009 PRODUCT_ID_10 = 1010 PRODUCT_ID_11 = 1011 PRODUCT_ID_NOT_EXISTS = 2001 TRACK_ID_1 = 501 TRACK_ID_2 = 502 TRACK_ID_3 = 503 TRACK_ID_4 = 504 TRACK_ID_5 = 505 TRACK_ID_6 = 506 TRACK_ID_7 = 507 TRACK_ID_8 = 508 TRACK_ID_NOT_EXISTS = 600 STEREO_ASSET_UPLOAD_TYPE_ID = 1 ATMOS_ASSET_UPLOAD_TYPE_ID = 3 @pytest.fixture def data() -> list[dict[str, Any]]: """Get data to seed asset_upload table.""" asset_upload_to_duplicate = { "id": ASSET_UPLOAD_PK_1, "bucket_name": "some_bucket", "asset_upload_type_id": 1, "filename": "unique_filename_1", "api_version": 1, "user_id": "oa:123456", "token": "session_token_1".encode(), "original_filename": "original_filename_1", "track_unique_id": TRACK_ID_1, "product_id": PRODUCT_ID_1, "deleted": False, "is_correction": True, "updated_timestamp": datetime.datetime(2017, 1, 6, 0, 0), } return [ { "id": ASSET_UPLOAD_PK_0, "bucket_name": "some_bucket", "asset_upload_type_id": 1, "filename": "unique_filename_0", "api_version": 1, "user_id": "oa:123456", "token": "session_token_0".encode(), "original_filename": "original_filename_1", "track_unique_id": TRACK_ID_1, "product_id": PRODUCT_ID_1, "deleted": True, "updated_timestamp": datetime.datetime(2017, 1, 1, 0, 0), }, asset_upload_to_duplicate, # this should not be returned because it is not the highest non-deleted id { **asset_upload_to_duplicate, "id": ASSET_UPLOAD_PK_22, }, # this should be returned because is the highest non-deleted id { **asset_upload_to_duplicate, "id": ASSET_UPLOAD_PK_23, "deleted": True, }, # even though this has the highest id, it should not be returned because it is deleted { "id": ASSET_UPLOAD_PK_2, "bucket_name": "some_bucket", "asset_upload_type_id": 1, "filename": "unique_filename_2", "api_version": 1, "user_id": "oa:654321", "token": "session_token_2".encode(), "original_filename": "original_filename_2", "track_unique_id": TRACK_ID_2, "product_id": PRODUCT_ID_1, "deleted": False, "updated_timestamp": datetime.datetime(2017, 1, 1, 0, 0), }, { "id": ASSET_UPLOAD_PK_3, "bucket_name": "some_bucket", "asset_upload_type_id": 2, "filename": "unique_filename_3", "api_version": 1, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_3", "track_unique_id": 0, "product_id": PRODUCT_ID_2, "deleted": False, "updated_timestamp": datetime.datetime(2017, 1, 1, 0, 0), }, { "id": ASSET_UPLOAD_PK_4, "bucket_name": "some_bucket", "asset_upload_type_id": 1, "filename": "unique_filename_4", "api_version": 1, "user_id": "oa:123456", "token": "session_token_4".encode(), "original_filename": "original_filename_1", "track_unique_id": TRACK_ID_1, "product_id": PRODUCT_ID_1, "deleted": True, "updated_timestamp": datetime.datetime(2017, 1, 1, 0, 0), }, { "id": ASSET_UPLOAD_PK_5, "bucket_name": "some_bucket", "asset_upload_type_id": 2, "filename": "unique_filename_5", "api_version": 1, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_3", "track_unique_id": 0, "product_id": PRODUCT_ID_2, "deleted": False, "updated_timestamp": datetime.datetime(2017, 1, 1, 0, 0), }, { "id": ASSET_UPLOAD_PK_6, "bucket_name": "some_bucket", "asset_upload_type_id": 2, "filename": "unique_filename_7", "api_version": 1, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_4", "track_unique_id": 0, "product_id": PRODUCT_ID_2, "deleted": True, "updated_timestamp": datetime.datetime(2017, 1, 1, 0, 0), }, { "id": ASSET_UPLOAD_PK_7, "bucket_name": "some_bucket", "asset_upload_type_id": 1, "filename": "unique_filename_1_1", "api_version": 1, "user_id": "oa:123456", "token": "session_token_1".encode(), "original_filename": "original_filename_1_1", "track_unique_id": TRACK_ID_1, "product_id": PRODUCT_ID_1, "deleted": False, "updated_timestamp": datetime.datetime(2017, 1, 2, 0, 0), }, { "id": ASSET_UPLOAD_PK_8, "bucket_name": "some_bucket", "asset_upload_type_id": 1, "filename": "unique_filename_8_8", "api_version": 2, "user_id": "oa:123456", "token": "session_token_1".encode(), "original_filename": "original_filename_1_1", "track_unique_id": TRACK_ID_2, "product_id": PRODUCT_ID_1, "deleted": False, "updated_timestamp": datetime.datetime(2017, 1, 2, 0, 0), }, { "id": ASSET_UPLOAD_PK_9, "bucket_name": "some_bucket", "asset_upload_type_id": 1, "filename": "unique_filename_8_8", "api_version": 1, "user_id": "oa:123456", "token": "session_token_1".encode(), "original_filename": "original_filename_1_1", "track_unique_id": TRACK_ID_2, "product_id": PRODUCT_ID_1, "deleted": False, "updated_timestamp": datetime.datetime(2017, 1, 2, 0, 0), }, { "id": ASSET_UPLOAD_PK_10, "bucket_name": "some_bucket", "asset_upload_type_id": 1, "filename": "unique_filename_10", "api_version": 2, "user_id": "oa:123456", "token": "session_token_10".encode(), "original_filename": "original_filename_10", "track_unique_id": TRACK_ID_1, "product_id": PRODUCT_ID_1, "deleted": False, "is_correction": False, "updated_timestamp": datetime.datetime(2017, 1, 6, 0, 0), }, { "id": ASSET_UPLOAD_PK_11, "bucket_name": "some_bucket", "asset_upload_type_id": 1, "filename": "unique_filename_11", "api_version": 2, "user_id": "oa:123456", "token": "session_token_11".encode(), "original_filename": "original_filename_11", "track_unique_id": TRACK_ID_1, "product_id": PRODUCT_ID_1, "deleted": False, "is_correction": True, "updated_timestamp": datetime.datetime(2017, 1, 6, 0, 0), }, { "id": ASSET_UPLOAD_PK_12, "bucket_name": "some_bucket", "asset_upload_type_id": 2, "filename": "unique_filename_12", "api_version": 1, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_3", "track_unique_id": 0, "product_id": PRODUCT_ID_1, "deleted": False, "is_correction": False, "updated_timestamp": datetime.datetime(2017, 1, 3, 0, 0), }, { "id": ASSET_UPLOAD_PK_13, "bucket_name": "some_bucket", "asset_upload_type_id": 2, "filename": "unique_filename_12", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_3", "track_unique_id": 0, "product_id": PRODUCT_ID_1, "deleted": False, "is_correction": False, "updated_timestamp": datetime.datetime(2017, 1, 3, 0, 0), }, { "id": ASSET_UPLOAD_PK_14, "bucket_name": "some_bucket", "asset_upload_type_id": 2, "filename": "unique_filename_13", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_3", "track_unique_id": 0, "product_id": PRODUCT_ID_3, "deleted": True, "is_correction": False, "updated_timestamp": datetime.datetime(2017, 1, 3, 0, 0), }, { "id": ASSET_UPLOAD_PK_15, "bucket_name": "some_bucket", "asset_upload_type_id": 2, "filename": "unique_filename_14", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_3", "track_unique_id": 0, "product_id": PRODUCT_ID_3, "deleted": False, "is_correction": True, "updated_timestamp": datetime.datetime(2017, 1, 3, 0, 0), }, { "id": ASSET_UPLOAD_PK_16, "bucket_name": "some_bucket", "asset_upload_type_id": 2, "filename": "unique_filename_16", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_3", "track_unique_id": 0, "product_id": PRODUCT_ID_4, "deleted": True, "is_correction": True, "updated_timestamp": datetime.datetime(2017, 1, 3, 0, 0), }, { "id": ASSET_UPLOAD_PK_17, "bucket_name": "some_bucket", "asset_upload_type_id": 2, "filename": "unique_filename_16", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_3", "track_unique_id": 0, "product_id": PRODUCT_ID_4, "deleted": False, "is_correction": False, "updated_timestamp": datetime.datetime(2017, 1, 3, 0, 0), }, { "id": ASSET_UPLOAD_PK_18, "bucket_name": "some_bucket", "asset_upload_type_id": 2, "filename": "unique_filename_17", "api_version": 2, "user_id": "oa:123456", "token": "session_token_1".encode(), "original_filename": "original_filename_1", "track_unique_id": 0, "product_id": 0, "deleted": False, "updated_timestamp": datetime.datetime(2017, 1, 1, 0, 0), }, { "id": ASSET_UPLOAD_PK_19, "bucket_name": "some_bucket", "asset_upload_type_id": 1, "filename": "unique_filename_18", "api_version": 2, "user_id": "oa:123456", "token": "session_token_0".encode(), "original_filename": "original_filename_1", "track_unique_id": TRACK_ID_4, "product_id": PRODUCT_ID_5, "deleted": False, "updated_timestamp": datetime.datetime(2017, 1, 1, 0, 0), }, { "id": ASSET_UPLOAD_PK_20, "bucket_name": "some_bucket", "asset_upload_type_id": 1, "filename": "unique_filename_19", "api_version": 2, "user_id": "oa:123456", "token": "session_token_0".encode(), "original_filename": "original_filename_1", "track_unique_id": TRACK_ID_4, "product_id": PRODUCT_ID_5, "deleted": False, "updated_timestamp": datetime.datetime(2017, 1, 1, 0, 0), }, { "id": ASSET_UPLOAD_PK_21, "bucket_name": "some_bucket", "asset_upload_type_id": 2, "filename": "unique_filename_14", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_3", "track_unique_id": 0, "product_id": PRODUCT_ID_6, "deleted": False, "is_correction": True, "updated_timestamp": datetime.datetime(2017, 1, 3, 0, 0), }, { "id": ASSET_UPLOAD_PK_24, "bucket_name": "some_bucket", "asset_upload_type_id": 1, # stereo "filename": "unique_filename_24", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_7_stereo", "track_unique_id": 0, "product_id": PRODUCT_ID_7, "deleted": False, "is_correction": True, "updated_timestamp": datetime.datetime(2017, 1, 3, 0, 0), }, { "id": ASSET_UPLOAD_PK_25, "bucket_name": "some_bucket", "asset_upload_type_id": 3, # atmos "filename": "unique_filename_25", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_7_atmos", "track_unique_id": 0, "product_id": PRODUCT_ID_7, "deleted": False, "is_correction": True, "updated_timestamp": datetime.datetime(2017, 1, 3, 0, 0), }, { "id": ASSET_UPLOAD_PK_26, "bucket_name": "some_bucket", "asset_upload_type_id": 1, "filename": "unique_filename_26", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_8_stereo", "track_unique_id": TRACK_ID_5, "product_id": PRODUCT_ID_8, "deleted": False, "is_correction": False, "updated_timestamp": datetime.datetime(2017, 1, 3, 0, 0), }, { "id": ASSET_UPLOAD_PK_27, "bucket_name": "some_bucket", "asset_upload_type_id": 3, # atmos "filename": "unique_filename_27", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_8_atmos", "track_unique_id": TRACK_ID_5, "product_id": PRODUCT_ID_8, "deleted": False, "is_correction": False, "updated_timestamp": datetime.datetime(2017, 1, 3, 0, 0), }, ] @pytest.fixture def data_with_multi_asset_upload_type_scenarios( data: list[dict[str, Any]], ) -> list[dict[str, Any]]: return data + [ { "id": ASSET_UPLOAD_PK_30, "bucket_name": "some_bucket", "asset_upload_type_id": ATMOS_ASSET_UPLOAD_TYPE_ID, "filename": "unique_filename_30", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_9_atmos_correction", "track_unique_id": TRACK_ID_6, "product_id": PRODUCT_ID_9, "deleted": False, "is_correction": True, "updated_timestamp": datetime.datetime(2017, 1, 4, 0, 0), }, { "id": ASSET_UPLOAD_PK_31, "bucket_name": "some_bucket", "asset_upload_type_id": ATMOS_ASSET_UPLOAD_TYPE_ID, "filename": "unique_filename_31", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_9_atmos_active", "track_unique_id": TRACK_ID_6, "product_id": PRODUCT_ID_9, "deleted": False, "is_correction": False, "updated_timestamp": datetime.datetime(2017, 1, 3, 0, 0), }, { "id": ASSET_UPLOAD_PK_32, "bucket_name": "some_bucket", "asset_upload_type_id": STEREO_ASSET_UPLOAD_TYPE_ID, "filename": "unique_filename_32", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_9_stereo_active", "track_unique_id": TRACK_ID_6, "product_id": PRODUCT_ID_9, "deleted": False, "is_correction": False, "updated_timestamp": datetime.datetime(2017, 1, 3, 0, 0), }, { "id": ASSET_UPLOAD_PK_33, "bucket_name": "some_bucket", "asset_upload_type_id": ATMOS_ASSET_UPLOAD_TYPE_ID, "filename": "unique_filename_33", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_10_atmos_correction", "track_unique_id": TRACK_ID_7, "product_id": PRODUCT_ID_10, "deleted": False, "is_correction": True, "updated_timestamp": datetime.datetime(2017, 1, 4, 0, 0), }, { "id": ASSET_UPLOAD_PK_34, "bucket_name": "some_bucket", "asset_upload_type_id": STEREO_ASSET_UPLOAD_TYPE_ID, "filename": "unique_filename_34", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_11_stereo_correction", "track_unique_id": TRACK_ID_8, "product_id": PRODUCT_ID_11, "deleted": False, "is_correction": True, "updated_timestamp": datetime.datetime(2017, 1, 4, 0, 0), }, { "id": ASSET_UPLOAD_PK_35, "bucket_name": "some_bucket", "asset_upload_type_id": STEREO_ASSET_UPLOAD_TYPE_ID, "filename": "unique_filename_35", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_11_stereo_active", "track_unique_id": TRACK_ID_8, "product_id": PRODUCT_ID_11, "deleted": False, "is_correction": False, "updated_timestamp": datetime.datetime(2017, 1, 3, 0, 0), }, { "id": ASSET_UPLOAD_PK_36, "bucket_name": "some_bucket", "asset_upload_type_id": ATMOS_ASSET_UPLOAD_TYPE_ID, "filename": "unique_filename_36", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_11_atmos_correction", "track_unique_id": TRACK_ID_8, "product_id": PRODUCT_ID_11, "deleted": False, "is_correction": True, "updated_timestamp": datetime.datetime(2017, 1, 4, 0, 0), }, { "id": ASSET_UPLOAD_PK_37, "bucket_name": "some_bucket", "asset_upload_type_id": ATMOS_ASSET_UPLOAD_TYPE_ID, "filename": "unique_filename_37", "api_version": 2, "user_id": "oa:654321", "token": "session_token_3".encode(), "original_filename": "original_filename_11_atmos_active", "track_unique_id": TRACK_ID_8, "product_id": PRODUCT_ID_11, "deleted": False, "is_correction": False, "updated_timestamp": datetime.datetime(2017, 1, 3, 0, 0), }, ] @pytest.fixture def db_fixture_with_multi_asset_upload_type( data_with_multi_asset_upload_type_scenarios: list[dict[str, Any]], asset_final_data: list[dict[str, Any]], asset_status_data: list[dict[str, Any]], ) -> None: au_operations.truncate_tables() au_operations.seed_asset_upload_types() au_operations.seed_asset_upload_table(data_with_multi_asset_upload_type_scenarios) au_operations.seed_asset_final_table(asset_final_data) au_operations.seed_asset_status_table(asset_status_data) @pytest.fixture def asset_final_data() -> list[dict[str, Any]]: """Fixture for asset finals list.""" return [ { "asset_upload_id": ASSET_UPLOAD_PK_13, "asset_type": "jpg", "asset_subtype": "cover", "bucket": "bucket", "filename": "qwe_123.jpg", "duration": 0, }, { "asset_upload_id": ASSET_UPLOAD_PK_13, "asset_type": "wav", "asset_subtype": "none", "bucket": "bucket", "filename": "filename.wav", "duration": 2543, }, { "asset_upload_id": ASSET_UPLOAD_PK_13, "asset_type": "mp3", "asset_subtype": "none", "bucket": "bucket", "filename": "filename.mp3", "duration": 221, }, { "asset_upload_id": ASSET_UPLOAD_PK_8, "asset_type": "wav", "asset_subtype": "none", "bucket": "bucket", "filename": "filename.wav", "duration": 2555, }, ] @pytest.fixture def asset_status_data() -> list[dict[str, Any]]: """Fixture for asset status list.""" return [ { "asset_upload_id": ASSET_UPLOAD_PK_13, "status": "upload_complete", "description": "test", "status_time": datetime.datetime(2017, 1, 6, 0, 0), "message": '{"function": "acknowledge", "status": "upload_complete"}', }, { "asset_upload_id": ASSET_UPLOAD_PK_13, "status": "acknowledge_complete", "description": "test", "status_time": datetime.datetime(2017, 1, 6, 0, 0), "message": '{"function": "acknowledge", "status": "acknowledge_complete"}', }, { "asset_upload_id": ASSET_UPLOAD_PK_13, "status": "encoding_route_complete", "description": "test", "status_time": datetime.datetime(2017, 1, 6, 0, 0), "message": '{"function": "encoding_route", "status": "encoding_route_complete"}', }, { "asset_upload_id": ASSET_UPLOAD_PK_13, "status": "validation_complete", "description": "test", "status_time": datetime.datetime(2017, 1, 6, 0, 0), "message": '{"function": "encoding-images-validation", "status": "validation_complete"}', }, { "asset_upload_id": ASSET_UPLOAD_PK_13, "status": "encoding_progressing", "description": "test", "status_time": datetime.datetime(2017, 1, 6, 0, 0), "message": '{"status": "PROCESSING"}', }, { "asset_upload_id": ASSET_UPLOAD_PK_13, "status": "encoding_completed", "description": "test", "status_time": datetime.datetime(2017, 1, 6, 0, 0), "message": '{"status": "COMPLETED"}', }, { "asset_upload_id": ASSET_UPLOAD_PK_8, "status": "upload_complete", "description": "test", "status_time": datetime.datetime(2017, 1, 6, 0, 0), "message": '{"function": "acknowledge", "status": "upload_complete"}', }, { "asset_upload_id": ASSET_UPLOAD_PK_8, "status": "acknowledge_complete", "description": "test", "status_time": datetime.datetime(2017, 1, 6, 0, 0), "message": '{"function": "acknowledge", "status": "acknowledge_complete"}', }, { "asset_upload_id": ASSET_UPLOAD_PK_8, "status": "encoding_route_complete", "description": "test", "status_time": datetime.datetime(2017, 1, 6, 0, 0), "message": '{"function": "encoding_route", "status": "encoding_route_complete"}', }, { "asset_upload_id": ASSET_UPLOAD_PK_8, "status": "validation_complete", "description": "test", "status_time": datetime.datetime(2017, 1, 6, 0, 0), "message": '{"function": "encoding-images-validation", "status": "validation_complete"}', }, { "asset_upload_id": ASSET_UPLOAD_PK_8, "status": "encoding_progressing", "description": "test", "status_time": datetime.datetime(2017, 1, 6, 0, 0), "message": '{"status": "PROCESSING"}', }, { "asset_upload_id": ASSET_UPLOAD_PK_8, "status": "encoding_completed", "description": "test", "status_time": datetime.datetime(2017, 1, 6, 0, 0), "message": '{"status": "COMPLETED"}', }, ] @pytest.fixture def db_fixture( data: list[dict[str, Any]], asset_final_data: list[dict[str, Any]], asset_status_data: list[dict[str, Any]], ) -> None: """Set up the asset received queue table.""" au_operations.truncate_tables() au_operations.seed_asset_upload_types() au_operations.seed_asset_upload_table(data) au_operations.seed_asset_final_table(asset_final_data) au_operations.seed_asset_status_table(asset_status_data) def test_create_new_asset_upload( db_fixture: None, valid_asset_upload_data: dict[str, Any] ) -> None: """Test successful create of new asset_upload.""" response = asset_upload.create_asset_upload(**valid_asset_upload_data) assert response["user_id"] == valid_asset_upload_data["user_id"] assert response["filename"] == valid_asset_upload_data["filename"] assert response["token"] == valid_asset_upload_data["token"] assert ( response["asset_upload_type_id"] == valid_asset_upload_data["asset_upload_type_id"] ) def test_create_new_asset_upload_sql_error( db_fixture: None, valid_asset_upload_data: dict[str, Any] ) -> None: """Test successful create of new asset_upload.""" error_message = "Query error" ( flexmock(mysql) .should_receive("au_db_session") .and_raise(SQLAlchemyError(error_message)) ) with pytest.raises(SQLAlchemyError) as exc: asset_upload.create_asset_upload(**valid_asset_upload_data) assert str(exc.value) == error_message @freezegun.freeze_time("2024-02-26 17:51:01.091325") def test_create_asset_upload_with_context_first_upload(db_fixture: None) -> None: """Test create_asset_upload_with_context first upload.""" result = asset_upload.create_asset_upload_with_context( user_id="alw:123", asset_upload_type_id=1, token="IC7AtIyaVwiAD6fBC3KENMDG7HxMtyAUymAu91.jQeDYNf5pSov312RQs8LoSasUF9FKRudVtWHZmYo5ccfI.xbrCR2RjyY4wBvtoSb9bNWT_1BR0CSJNFoO1gVH7Emy", # noqa: E501 filename="9e2158a8-474b-491b-9dcc-cc970a7fcb48", product_id=12345, upc=123456789012, track_unique_id=321, is_correction=False, original_filename="original_filename.wav", ) assert result == { "asset_status": { "asset_upload_id": ASSET_UPLOAD_PK_28, "description": None, "id": 13, "message": None, "status": "uploading", "status_time": datetime.datetime( 2024, 2, 26, 17, 51, 1, 91325, tzinfo=datetime.timezone.utc ), }, "asset_upload": { "api_version": 2, "asset_upload_type_id": 1, "deleted": False, "filename": "9e2158a8-474b-491b-9dcc-cc970a7fcb48", "id": ASSET_UPLOAD_PK_28, "is_correction": False, "original_filename": "original_filename.wav", "product_id": 12345, "token": "IC7AtIyaVwiAD6fBC3KENMDG7HxMtyAUymAu91.jQeDYNf5pSov312RQs8LoSasUF9FKRudVtWHZmYo5ccfI.xbrCR2RjyY4wBvtoSb9bNWT_1BR0CSJNFoO1gVH7Emy", # noqa: E501 "track_unique_id": 321, "bucket_name": "test-orcd-raw-assets", "upc": "123456789012", "user_id": "alw:123", }, "deleted_asset_uploads": [], } class _AssetUploadFields(TypedDict): user_id: str asset_upload_type_id: int token: str filename: str product_id: int upc: int track_unique_id: int is_correction: bool original_filename: str @freezegun.freeze_time("2024-02-26 17:51:01.091325") def test_create_asset_upload_with_context_replace_previous_upload( db_fixture: None, ) -> None: """Test create_asset_upload_with_context replace previous upload.""" asset_upload_fields: _AssetUploadFields = { "user_id": "alw:123", "asset_upload_type_id": 1, "token": "IC7AtIyaVwiAD6fBC3KENMDG7HxMtyAUymAu91.jQeDYNf5pSov312RQs8LoSasUF9FKRudVtWHZmYo5ccfI.xbrCR2RjyY4wBvtoSb9bNWT_1BR0CSJNFoO1gVH7Emy", # noqa: E501 "filename": "9e2158a8-474b-491b-9dcc-cc970a7fcb48", "product_id": 12345, "upc": 123456789012, "track_unique_id": 321, "is_correction": False, "original_filename": "original_filename.wav", } asset_upload.create_asset_upload_with_context(**asset_upload_fields) replacement_fields: _AssetUploadFields = { **asset_upload_fields, "user_id": "alw:456", "token": "new_token", "filename": "new_filename", "original_filename": "new_original_filename.wav", } replacement_upload_result = asset_upload.create_asset_upload_with_context( **replacement_fields ) assert replacement_upload_result == { "asset_upload": { "id": ASSET_UPLOAD_PK_29, "user_id": "alw:456", "asset_upload_type_id": 1, "token": "new_token", "filename": "new_filename", "api_version": 2, "product_id": 12345, "upc": "123456789012", "track_unique_id": 321, "bucket_name": "test-orcd-raw-assets", "original_filename": "new_original_filename.wav", "deleted": False, "is_correction": False, }, "asset_status": { "id": 14, "asset_upload_id": ASSET_UPLOAD_PK_29, "status": "uploading", "description": None, "status_time": datetime.datetime( 2024, 2, 26, 17, 51, 1, 91325, tzinfo=datetime.timezone.utc ), "message": None, }, "deleted_asset_uploads": [ { "id": ASSET_UPLOAD_PK_28, "user_id": "alw:123", "asset_upload_type_id": 1, "token": "IC7AtIyaVwiAD6fBC3KENMDG7HxMtyAUymAu91.jQeDYNf5pSov312RQs8LoSasUF9FKRudVtWHZmYo5ccfI.xbrCR2RjyY4wBvtoSb9bNWT_1BR0CSJNFoO1gVH7Emy", # noqa: E501 "filename": "9e2158a8-474b-491b-9dcc-cc970a7fcb48", "api_version": 2, "product_id": 12345, "upc": "123456789012", "track_unique_id": 321, "bucket_name": "test-orcd-raw-assets", "original_filename": "original_filename.wav", "deleted": True, "is_correction": False, } ], } def test_create_new_asset_upload_record_for_copy( db_fixture: None, valid_asset_upload_data_for_copy: dict[str, Any] ) -> None: """Test successful create of new asset_upload for copy.""" response = asset_upload.create_asset_upload_record_for_copy( valid_asset_upload_data_for_copy ) assert response["user_id"] == valid_asset_upload_data_for_copy["user_id"] assert response["filename"] == valid_asset_upload_data_for_copy["filename"] assert response["token"] == valid_asset_upload_data_for_copy["token"] assert response["product_id"] == valid_asset_upload_data_for_copy["product_id"] assert response["upc"] == valid_asset_upload_data_for_copy["upc"] def test_create_new_asset_upload_record_for_copy_sql_error( db_fixture: None, valid_asset_upload_data: dict[str, Any] ) -> None: """Test failure create of new asset_upload for copy.""" error_message = "Query error" ( flexmock(mysql) .should_receive("au_db_session") .and_raise(SQLAlchemyError(error_message)) ) with pytest.raises(SQLAlchemyError) as exc: asset_upload.create_asset_upload_record_for_copy(valid_asset_upload_data) assert str(exc.value) == error_message def test_get_asset_upload_success( db_fixture: None, existing_asset_upload: dict[str, Any] ) -> None: """Test successful get of existing asset_upload.""" response = asset_upload.get_asset_upload(**existing_asset_upload) assert response["id"] == ASSET_UPLOAD_PK_22 def test_get_asset_upload_no_user_id_no_token_success( db_fixture: None, existing_asset_upload: dict[str, Any] ) -> None: """Test successful get of existing asset_upload with no user_id filter.""" del existing_asset_upload["user_id"] del existing_asset_upload["token"] response = asset_upload.get_asset_upload(**existing_asset_upload) assert response["id"] == ASSET_UPLOAD_PK_22 @pytest.mark.parametrize("search_condition", ["filename", "user_id", "token"]) def test_get_asset_upload_failures( db_fixture: None, existing_asset_upload: dict[str, Any], search_condition: str ) -> None: """Test get_asset_upload fail if one of fields is wrong.""" existing_asset_upload[search_condition] = "some_wrong_value" with pytest.raises(AssetUploadNotFound) as exc: asset_upload.get_asset_upload(**existing_asset_upload) assert exc.value.description == error_constants.ERROR_ASSET_UPLOAD_NOT_FOUND def test_get_asset_upload_sql_error( db_fixture: None, existing_asset_upload: dict[str, Any] ) -> None: """Test SQLAlchemyError failure.""" error_message = "Query error" ( flexmock(mysql) .should_receive("au_db_session") .and_raise(SQLAlchemyError(error_message)) ) with pytest.raises(SQLAlchemyError) as exc: asset_upload.get_asset_upload(**existing_asset_upload) assert str(exc.value) == error_message def test_update_asset_lock_name() -> None: """Test generating lock name for asset_upload updates.""" lock_name = asset_upload._asset_upload_lock_name(123, 456, False) assert lock_name == "123:456:False" @patch("assets.models.asset_upload.mysql.db_lock") def test_update_asset_upload_success_setting_context( mocked_db_lock: MagicMock, db_fixture: None, mocker: MockerFixture ) -> None: """Test successful update of fields on existing asset_upload. Tests setting context like product_id and track_unique_id on an existing asset_upload that was created without context. """ au_db_session_mock = mocker.spy(mysql, "au_database_session") au_operations.truncate_tables() au_operations.seed_asset_upload_types() is_correction = True original_filename = "original_filename.wav" product_id = 12345 track_unique_id = 321 upc = 123 asset_upload.create_asset_upload( user_id="oa:654321", filename="some_filename", token="some_token", asset_upload_type_id=1, api_version=2, ) update_asset_upload_response_1 = asset_upload.update_asset_upload( filename="some_filename", is_correction=is_correction, original_filename=original_filename, product_id=product_id, track_unique_id=track_unique_id, upc=upc, api_version=2, ) create_asset_upload_response_2 = asset_upload.create_asset_upload( user_id="oa:654321", filename="another_uuid4", token="some_token", asset_upload_type_id=1, api_version=2, ) update_asset_upload_response_2 = asset_upload.update_asset_upload( filename="another_uuid4", is_correction=is_correction, original_filename=original_filename, product_id=product_id, track_unique_id=track_unique_id, upc=upc, api_version=2, ) assert update_asset_upload_response_2 == { "id": 2, "user_id": "oa:654321", "asset_upload_type_id": 1, "token": "some_token", "filename": "another_uuid4", "api_version": 2, "product_id": product_id, "upc": upc, "track_unique_id": track_unique_id, "bucket_name": None, "original_filename": original_filename, "deleted": False, "is_correction": is_correction, } mocked_db_lock.assert_called_with( au_db_session_mock.spy_return, f"{product_id}:{track_unique_id}:{is_correction}", 90, ) # verify previous upload was soft deleted with mysql.au_db_session() as session: assert ( session.query(asset_upload.AssetUpload) .filter( asset_upload.AssetUpload.asset_upload_id == update_asset_upload_response_1["id"], asset_upload.AssetUpload.deleted == 1, ) .one() ) # verify fields were set on new upload assert ( session.query(asset_upload.AssetUpload) .filter( asset_upload.AssetUpload.asset_upload_id == create_asset_upload_response_2["id"], asset_upload.AssetUpload.asset_upload_type_id == 1, asset_upload.AssetUpload.is_correction == is_correction, asset_upload.AssetUpload.original_filename == original_filename, asset_upload.AssetUpload.product_id == product_id, asset_upload.AssetUpload.track_unique_id == track_unique_id, asset_upload.AssetUpload.upc == upc, ) .one() ) @patch("assets.models.asset_upload.mysql.db_lock") def test_update_asset_upload_failure_setting_context( mocked_db_lock: MagicMock, db_fixture: None, mocker: MockerFixture ) -> None: """Test failed update of fields on existing asset_upload. Tests setting context like product_id and track_unique_id on an existing asset_upload that was created with context fails. """ au_db_session_mock = mocker.spy(mysql, "au_database_session") is_correction = True original_filename = "original_filename.wav" product_id = 12345 track_unique_id = 321 asset_upload_type_id = 1 upc = 123 create_asset_upload_with_context_response = ( asset_upload.create_asset_upload_with_context( user_id="oa:654321", asset_upload_type_id=asset_upload_type_id, token="some_token", filename="some_filename", product_id=product_id, track_unique_id=track_unique_id, upc=upc, is_correction=is_correction, original_filename=original_filename, ) ) with pytest.raises(MissingContext) as exc: asset_upload.update_asset_upload( filename="some_filename", is_correction=True, original_filename="original_filename.wav", product_id=12345, track_unique_id=321, upc=123, api_version=2, ) mocked_db_lock.assert_called_with( au_db_session_mock.spy_return, "12345:321:True", 90, ) assert exc.value.description == ( "asset_upload has context fields (product_id, track_unique_id," " etc.) set but they were also provided in the request." ) # verify no fields were updated with mysql.au_db_session() as session: assert ( session.query(asset_upload.AssetUpload) .filter( asset_upload.AssetUpload.asset_upload_id == create_asset_upload_with_context_response["asset_upload"]["id"], asset_upload.AssetUpload.asset_upload_type_id == asset_upload_type_id, asset_upload.AssetUpload.is_correction == is_correction, asset_upload.AssetUpload.original_filename == original_filename, asset_upload.AssetUpload.product_id == product_id, asset_upload.AssetUpload.track_unique_id == track_unique_id, asset_upload.AssetUpload.upc == upc, ) .one() ) @patch("assets.models.asset_upload.mysql.db_lock") def test_update_asset_upload_success_without_setting_context( mocked_db_lock: MagicMock, db_fixture: None, mocker: MockerFixture ) -> None: """Test successful passthrough of request without context. Tests that a request with no context information succeeds if the existing asset_upload was created with context like product_id, track_unique_id, etc. """ au_db_session_mock = mocker.spy(mysql, "au_database_session") au_operations.truncate_tables() au_operations.seed_asset_upload_types() is_correction = True original_filename = "original_filename.wav" product_id = 12345 track_unique_id = 321 asset_upload_type_id = 1 upc = 123 create_asset_upload_with_context_response = ( asset_upload.create_asset_upload_with_context( user_id="oa:654321", asset_upload_type_id=asset_upload_type_id, token="some_token", filename="some_filename", product_id=product_id, track_unique_id=track_unique_id, upc=upc, is_correction=is_correction, original_filename=original_filename, ) ) update_asset_upload_response = asset_upload.update_asset_upload( filename="some_filename", is_correction=False, original_filename=None, product_id=None, track_unique_id=None, upc=None, api_version=2, ) mocked_db_lock.assert_called_with( au_db_session_mock.spy_return, "some_filename", 90, ) assert update_asset_upload_response == {} # verify no fields were updated with mysql.au_db_session() as session: assert ( session.query(asset_upload.AssetUpload) .filter( asset_upload.AssetUpload.asset_upload_id == create_asset_upload_with_context_response["asset_upload"]["id"], asset_upload.AssetUpload.asset_upload_type_id == asset_upload_type_id, asset_upload.AssetUpload.is_correction == is_correction, asset_upload.AssetUpload.original_filename == original_filename, asset_upload.AssetUpload.product_id == product_id, asset_upload.AssetUpload.track_unique_id == track_unique_id, asset_upload.AssetUpload.upc == upc, ) .one() ) @patch("assets.models.asset_upload.mysql.db_lock") def test_update_asset_upload_failure_without_setting_context( mocked_db_lock: MagicMock, db_fixture: None, mocker: MockerFixture ) -> None: """Test failed passthrough of request without context. Tests that a request with no context information fails if the existing asset_upload was created without context like product_id, track_unique_id, etc. """ au_db_session_mock = mocker.spy(mysql, "au_database_session") au_operations.truncate_tables() au_operations.seed_asset_upload_types() create_asset_upload_response = asset_upload.create_asset_upload( user_id="oa:654321", filename="some_filename", token="some_token", asset_upload_type_id=1, api_version=2, ) with pytest.raises(MissingContext) as exc: asset_upload.update_asset_upload( filename="some_filename", is_correction=False, original_filename=None, product_id=None, track_unique_id=None, upc=None, api_version=2, ) assert exc.value.description == ( "asset_upload is missing context fields (product_id, " "track_unique_id, etc.) and none were provided in the request." ) mocked_db_lock.assert_called_with( au_db_session_mock.spy_return, "some_filename", 90, ) # verify fields were set on new upload with mysql.au_db_session() as session: assert ( session.query(asset_upload.AssetUpload) .filter( asset_upload.AssetUpload.asset_upload_id == create_asset_upload_response["id"], asset_upload.AssetUpload.user_id == "oa:654321", asset_upload.AssetUpload.asset_upload_type_id == 1, asset_upload.AssetUpload.filename == "some_filename", asset_upload.AssetUpload.token == b"some_token", asset_upload.AssetUpload.api_version == 2, ) .one() ) @pytest.mark.parametrize( ( "asset_upload_data_overrides", "expected_exception", "expected_message", ), [ ( {"filename": "unique_filename_123345"}, NoAssetsFound, "No asset_upload record to update", ), ( { "user_id": "wrong_value", "filename": "unique_filename_123345", }, NoAssetsFound, "No asset_upload record to update", ), ( { "token": "wrong_value", "filename": "unique_filename_123345", }, NoAssetsFound, "No asset_upload record to update", ), ( {"filename": "unique_filename_1"}, DuplicateAssetUploadFilename, ( "This filename is being used by multiple asset_upload " "records which is the result of copying a product. Once multiple " "asset_upload records exist for a filename, the " "ENV-assets-transcoding-v2 sfn can't be rerun for that filename since " "we don't know which asset_upload record to update. Use this script to " "recreate asset finals: " "https://github.com/theorchard/scripts-assets/tree/master/create-asset-finals" ), ), ], ) @patch("assets.models.asset_upload.mysql.db_lock") def test_update_asset_upload_quantity_failure( mocked_db_lock: MagicMock, db_fixture: None, asset_upload_data_overrides: dict[str, Any], update_asset_upload_data: dict[str, Any], expected_exception: type[HTTPException], expected_message: str, mocker: MockerFixture, ) -> None: """Test update_asset_upload quantity failure. Tests failure of update_asset_upload when there are multiple asset_upload records for the same filename or when no asset_upload record to update is found. """ au_db_session_mock = mocker.spy(mysql, "au_database_session") update_asset_upload_data = { **update_asset_upload_data, **asset_upload_data_overrides, } with pytest.raises(expected_exception) as exc: asset_upload.update_asset_upload(**update_asset_upload_data) mocked_db_lock.assert_called_with( au_db_session_mock.spy_return, "12345:321:True", 90, ) assert exc.value.description == expected_message def test_update_asset_upload_sql_error( db_fixture: None, update_asset_upload_data: dict[str, Any] ) -> None: """Test SQLAlchemyError failure.""" error_message = "Query error" ( flexmock(mysql) .should_receive("au_db_session") .and_raise(SQLAlchemyError(error_message)) ) with pytest.raises(SQLAlchemyError) as exc: asset_upload.update_asset_upload(**update_asset_upload_data) assert str(exc.value) == error_message def test_apply_asset_corrections_success(db_fixture: None) -> None: response = asset_upload.apply_asset_corrections(PRODUCT_ID_3) assert len(response) == 1 promoted = response[0] assert promoted["id"] == ASSET_UPLOAD_PK_15 assert promoted["product_id"] == PRODUCT_ID_3 assert not promoted["is_correction"] assert not promoted["deleted"] def test_apply_asset_corrections_error(db_fixture: None) -> None: """Test SQLAlchemyError failure.""" error_message = "Query error" ( flexmock(mysql) .should_receive("au_db_session") .and_raise(SQLAlchemyError(error_message)) ) with pytest.raises(SQLAlchemyError) as exc: asset_upload.apply_asset_corrections(PRODUCT_ID_3) assert str(exc.value) == error_message def test_apply_asset_corrections_no_rows_updated(db_fixture: None) -> None: """Test apply_asset_corrections when no rows are updated.""" response = asset_upload.apply_asset_corrections(PRODUCT_ID_4) assert not response def test_apply_asset_corrections_correction_with_no_previous_value( db_fixture: None, ) -> None: """Test apply_asset_corrections when there is no previous value to mark as deleted.""" # noqa response = asset_upload.apply_asset_corrections(PRODUCT_ID_6) assert response == [ { "id": 1021, "user_id": "oa:654321", "asset_upload_type_id": 2, "token": "session_token_3", "bucket_name": "some_bucket", "filename": "unique_filename_14", "api_version": 2, "product_id": 1006, "upc": None, "track_unique_id": 0, "original_filename": "original_filename_3", "deleted": False, "is_correction": False, } ] def test_get_asset_upload_by_id_success(db_fixture: None) -> None: """Test get_asset_upload_by_id success.""" response = asset_upload.get_asset_upload_by_id(ASSET_UPLOAD_PK_2) assert response["id"] == ASSET_UPLOAD_PK_2 def test_get_asset_upload_by_id_not_found(db_fixture: None) -> None: """Test get_asset_upload_by_id not found.""" with pytest.raises(AssetUploadNotFound) as exc: asset_upload.get_asset_upload_by_id(ASSET_UPLOAD_NOT_EXISTS) assert exc.value.description == error_constants.ERROR_ASSET_UPLOAD_NOT_FOUND def test_get_asset_upload_by_id_failure(db_fixture: None) -> None: """Test get_asset_upload_by_id db failure.""" error_message = "Query error" ( flexmock(mysql) .should_receive("au_db_session") .and_raise(SQLAlchemyError(error_message)), ) with pytest.raises(SQLAlchemyError) as exc: asset_upload.get_asset_upload_by_id(ASSET_UPLOAD_NOT_EXISTS) assert str(exc.value) == error_message def test_get_asset_uploads_by_product_id_success_with_fix(db_fixture: None) -> None: """Test for success of get_asset_uploads_by_product_id with flag enabled.""" response = asset_upload.get_asset_uploads_by_product_id(PRODUCT_ID_1) assert len(response) == 3 ids = [msg["id"] for msg in response] ids.sort() # This now will give the latest non-deleted asset_upload ordered by id, not last_updated column. assert ids == [ASSET_UPLOAD_PK_9, ASSET_UPLOAD_PK_12, ASSET_UPLOAD_PK_22] def test_get_asset_uploads_by_product_id_not_db_error(db_fixture: None) -> None: """Test for db failure of get_asset_uploads_by_product_id.""" error_message = "Query error" ( flexmock(mysql) .should_receive("au_db_session") .and_raise(SQLAlchemyError(error_message)), ) with pytest.raises(SQLAlchemyError) as exc: asset_upload.get_asset_uploads_by_product_id(PRODUCT_ID_1) assert str(exc.value) == error_message def test_get_all_asset_uploads_by_product_id_success(db_fixture: None) -> None: """Test for success of get_all_asset_uploads_by_product_id.""" # test function call response = asset_upload.get_asset_uploads(product_id=PRODUCT_ID_1) # checking assert len(response) == 6 assert response[0]["id"] == ASSET_UPLOAD_PK_1 assert response[1]["id"] == ASSET_UPLOAD_PK_2 assert response[2]["id"] == ASSET_UPLOAD_PK_7 assert response[3]["id"] == ASSET_UPLOAD_PK_9 assert response[4]["id"] == ASSET_UPLOAD_PK_12 assert response[5]["id"] == ASSET_UPLOAD_PK_22 def test_get_all_asset_uploads_by_product_id_not_db_error(db_fixture: None) -> None: """Test for db failure of get_all_asset_uploads_by_product_id.""" error_message = "Query error" # mocking ( flexmock(mysql) .should_receive("au_db_session") .and_raise(SQLAlchemyError(error_message)), ) # test function call with pytest.raises(SQLAlchemyError) as exc: asset_upload.get_asset_uploads(product_id=PRODUCT_ID_1) # checking assert str(exc.value) == error_message def test_get_asset_uploads_by_track_id(db_fixture: None) -> None: """Test for get_asset_upload_by_track_id success.""" response = asset_upload.get_asset_uploads(track_id=TRACK_ID_1) assert len(response) == 3 assert response[0]["id"] == ASSET_UPLOAD_PK_1 assert response[0]["track_unique_id"] == TRACK_ID_1 assert response[1]["id"] == ASSET_UPLOAD_PK_7 assert response[1]["track_unique_id"] == TRACK_ID_1 assert response[2]["id"] == ASSET_UPLOAD_PK_22 assert response[2]["track_unique_id"] == TRACK_ID_1 def test_get_asset_uploads_by_track_id_order_desc(db_fixture: None) -> None: """Test that results are in descending order by PK.""" result = asset_upload.get_asset_uploads( track_id=TRACK_ID_1, order=asset_upload.ORDER_DESC ) assert result[0]["id"] == ASSET_UPLOAD_PK_22 assert result[1]["id"] == ASSET_UPLOAD_PK_7 def test_get_asset_uploads_by_track_id_order_asc(db_fixture: None) -> None: """Test that results are in ascending order by PK.""" result = asset_upload.get_asset_uploads( track_id=TRACK_ID_1, order=asset_upload.ORDER_ASC ) assert result[0]["id"] == ASSET_UPLOAD_PK_1 assert result[1]["id"] == ASSET_UPLOAD_PK_7 def test_get_asset_uploads_by_track_id_limit(db_fixture: None) -> None: """Test that the result set is limited to the given size.""" result = asset_upload.get_asset_uploads(track_id=TRACK_ID_1, limit=1) assert len(result) == 1 def test_get_asset_uploads_by_track_id_not_found(db_fixture: None) -> None: """Test for get_asset_upload_by_track_id not found.""" with pytest.raises(AssetUploadNotFound) as exc: asset_upload.get_asset_uploads(track_id=TRACK_ID_NOT_EXISTS) assert exc.value.description == error_constants.ERROR_ASSET_UPLOAD_NOT_FOUND def test_get_asset_uploads_by_track_id_db_error(db_fixture: None) -> None: """Test for get_asset_upload_by_track_id db error.""" error_message = "Query error" ( flexmock(mysql) .should_receive("au_db_session") .and_raise(SQLAlchemyError(error_message)), ) with pytest.raises(SQLAlchemyError) as exc: asset_upload.get_asset_uploads(track_id=TRACK_ID_1) assert str(exc.value) == error_message def test_get_asset_upload_in_correction_mode_by_product_id_success( db_fixture: None, ) -> None: """Test for success of get_asset_upload_in_correction_mode_by_product_id.""" # test function call, API version 2 response = asset_upload.get_latest_asset_upload( product_id=PRODUCT_ID_1, is_correction_mode=True ) # checking assert response[0]["id"] == ASSET_UPLOAD_PK_13 def test_get_asset_upload_in_correction_mode_by_product_id_not_db_error( db_fixture: None, ) -> None: """Test for db failure of get_asset_upload_by_track_id.""" error_message = "Query error" # mocking ( flexmock(mysql) .should_receive("au_db_session") .and_raise(SQLAlchemyError(error_message)), ) # test function call with pytest.raises(SQLAlchemyError) as exc: asset_upload.get_latest_asset_upload(product_id=PRODUCT_ID_1) assert str(exc.value) == error_message def test_get_asset_upload_in_correction_mode_by_track_id_success( db_fixture: None, ) -> None: """Test for get_asset_upload_in_correction_mode_by_track_id success.""" response = asset_upload.get_latest_asset_upload( track_id=TRACK_ID_1, is_correction_mode=True ) assert response[0]["id"] == ASSET_UPLOAD_PK_11 assert response[0]["track_unique_id"] == TRACK_ID_1 def test_get_asset_upload_in_correction_mode_by_track_id_not_found( db_fixture: None, ) -> None: """Test for get_asset_upload_by_track_id not found.""" with pytest.raises(AssetUploadNotFound) as exc: asset_upload.get_latest_asset_upload(track_id=TRACK_ID_NOT_EXISTS) assert exc.value.description == error_constants.ERROR_ASSET_UPLOAD_NOT_FOUND def test_get_asset_upload_in_correction_mode_by_track_id_not_db_error( db_fixture: None, ) -> None: """Test for db failure of get_asset_upload_by_track_id.""" error_message = "Query error" # mocking ( flexmock(mysql) .should_receive("au_db_session") .and_raise(SQLAlchemyError(error_message)), ) # test function call with pytest.raises(SQLAlchemyError) as exc: asset_upload.get_latest_asset_upload(track_id=TRACK_ID_1) # checking assert str(exc.value) == error_message def test_mark_deleted_success(db_fixture: None) -> None: """Test for mark_asset_upload_deleted success.""" response = asset_upload.mark_asset_upload_deleted(ASSET_UPLOAD_PK_1) assert len(response) == 1 assert response[0]["deleted"] == 1 def test_mark_deleted_success_batch(db_fixture: None) -> None: """Test for mark_asset_upload_deleted success batch.""" ids_to_mark = [ASSET_UPLOAD_PK_1, ASSET_UPLOAD_PK_2] response = asset_upload.mark_asset_upload_deleted(ids_to_mark) assert len(response) == 2 ids = [item["id"] for item in response] assert ids == ids_to_mark assert response[0]["deleted"] == 1 assert response[1]["deleted"] == 1 def test_mark_deleted_not_found(db_fixture: None) -> None: """Test for mark_asset_upload_deleted not existing record.""" with pytest.raises(Exception) as exc: asset_upload.mark_asset_upload_deleted(ASSET_UPLOAD_NOT_EXISTS) assert str(exc.value) == "0 records marked, 1 expected" def test_mark_deleted_db_error(db_fixture: None) -> None: """Test for mark_asset_upload_deleted db failure.""" error_message = "Query error" ( flexmock(mysql) .should_receive("au_db_session") .and_raise(SQLAlchemyError(error_message)) ) with pytest.raises(SQLAlchemyError) as exc: asset_upload.mark_asset_upload_deleted(ASSET_UPLOAD_PK_1) assert str(exc.value) == error_message def test_get_asset_uploads_by_product_id_and_asset_types(db_fixture: None) -> None: """Test for get_asset_uploads_by_product_id_and_asset_types success.""" response = asset_upload.get_asset_uploads(product_id=PRODUCT_ID_2, track_id=0) assert len(response) == 2 assert response[0]["id"] == ASSET_UPLOAD_PK_3 assert response[1]["id"] == ASSET_UPLOAD_PK_5 def test_get_asset_uploads_by_product_id_and_asset_types_db_error( db_fixture: None, ) -> None: """Test for get_asset_uploads_by_product_id_and_asset_types db error.""" error_message = "Query error" ( flexmock(mysql) .should_receive("au_db_session") .and_raise(SQLAlchemyError(error_message)), ) with pytest.raises(SQLAlchemyError) as exc: asset_upload.get_asset_uploads(product_id=PRODUCT_ID_2, track_id=0) assert str(exc.value) == error_message def test_get_audio_asset_uploads_by_track_and_product_ids( db_fixture: None, get_audio_asset_uploads_by_track_and_product_ids_data: list[dict[str, Any]], ) -> None: """Test for get_audio_asset_uploads_by_track_and_product_ids success.""" # data preparation expected_messages = copy.copy(get_audio_asset_uploads_by_track_and_product_ids_data) for asset in expected_messages: del asset["updated_timestamp"] # test function call for an audio asset with an actual track id response = asset_upload.get_asset_uploads( track_id=TRACK_ID_1, product_id=PRODUCT_ID_1, order=asset_upload.ORDER_DESC ) # checking assert len(response) == 3 assert response == expected_messages def test_get_image_asset_uploads_by_track_and_product_ids( db_fixture: None, get_image_asset_uploads_by_track_and_product_ids_data: list[dict[str, Any]], ) -> None: """Test for get_image_asset_uploads_by_track_and_product_ids success.""" # data preparation expected_messages = copy.copy(get_image_asset_uploads_by_track_and_product_ids_data) for asset in expected_messages: del asset["updated_timestamp"] # test function call for an audio asset with an actual track id response = asset_upload.get_asset_uploads( track_id=0, product_id=PRODUCT_ID_1, order=asset_upload.ORDER_DESC ) # checking assert len(response) == 1 assert response == expected_messages def test_get_asset_uploads_by_track_and_product_ids_not_found(db_fixture: None) -> None: """Test for get_asset_uploads_by_track_and_product_ids not found.""" with pytest.raises(AssetUploadNotFound) as exc: asset_upload.get_asset_uploads( track_id=TRACK_ID_NOT_EXISTS, product_id=PRODUCT_ID_1, order=asset_upload.ORDER_DESC, ) assert exc.value.description == error_constants.ERROR_ASSET_UPLOAD_NOT_FOUND def test_get_asset_uploads_by_track_and_product_ids_db_error(db_fixture: None) -> None: """Test for get_asset_uploads_by_track_and_product_ids db error.""" error_message = "Query error" ( flexmock(mysql) .should_receive("au_db_session") .and_raise(SQLAlchemyError(error_message)), ) with pytest.raises(SQLAlchemyError) as exc: asset_upload.get_asset_uploads(track_id=TRACK_ID_1, product_id=PRODUCT_ID_1) assert str(exc.value) == error_message def test_get_asset_records_to_copy_success(db_fixture: None) -> None: """Test for get_asset_records_to_copy success.""" expected_result = { "asset_upload": { "bucket_name": "some_bucket", "filename": "unique_filename_12", "original_filename": "original_filename_3", "asset_upload_type_id": 2, }, "asset_finals": [ { "asset_type": "jpg", "asset_subtype": "cover", "bucket": "bucket", "filename": "qwe_123.jpg", "duration": 0, }, { "asset_type": "wav", "asset_subtype": "none", "bucket": "bucket", "filename": "filename.wav", "duration": 2543, }, { "asset_type": "mp3", "asset_subtype": "none", "bucket": "bucket", "filename": "filename.mp3", "duration": 221, }, ], "asset_status": [ { "status": "upload_complete", "description": "test", "status_time": "2020-03-17 06:12:59.072000", "message": '{"function": "acknowledge", "status": "upload_complete"}', }, { "status": "acknowledge_complete", "description": "test", "status_time": "2020-03-17 06:12:59.072000", "message": '{"function": "acknowledge", "status": "acknowledge_complete"}', }, { "status": "encoding_route_complete", "description": "test", "status_time": "2020-03-17 06:12:59.072000", "message": '{"function": "encoding_route", "status": "encoding_route_complete"}', }, { "status": "validation_complete", "description": "test", "status_time": "2020-03-17 06:12:59.072000", "message": '{"function": "encoding-images-validation", "status": "validation_complete"}', }, { "status": "encoding_progressing", "description": "test", "status_time": "2020-03-17 06:12:59.072000", "message": '{"status": "PROCESSING"}', }, { "status": "encoding_completed", "description": "test", "status_time": "2020-03-17 06:12:59.072000", "message": '{"status": "COMPLETED"}', }, ], } response = asset_upload.get_asset_records_to_copy( PRODUCT_ID_1, 0, asset_upload_types.STATIC_ARTWORK ) assert response is not None assert response["asset_upload"] == expected_result["asset_upload"] assert len(response["asset_finals"]) == len(expected_result["asset_finals"]) assert len(response["asset_status"]) == len(expected_result["asset_status"]) def test_get_asset_records_to_copy_success_v2_audio(db_fixture: None) -> None: """Test for get_asset_records_to_copy success for v2 audio.""" expected_result = { "asset_upload": { "asset_upload_type_id": 1, "bucket_name": "some_bucket", "filename": "unique_filename_8_8", "original_filename": "original_filename_1_1", }, "asset_finals": [ { "asset_type": "wav", "asset_subtype": "none", "bucket": "bucket", "filename": "filename.wav", "duration": 2555, } ], "asset_status": [ { "status": "upload_complete", "description": "test", "status_time": "2020-03-17 06:12:59.072000", "message": '{"function": "acknowledge", "status": "upload_complete"}', }, { "status": "acknowledge_complete", "description": "test", "status_time": "2020-03-17 06:12:59.072000", "message": '{"function": "acknowledge", "status": "acknowledge_complete"}', }, { "status": "encoding_route_complete", "description": "test", "status_time": "2020-03-17 06:12:59.072000", "message": '{"function": "encoding_route", "status": "encoding_route_complete"}', }, { "status": "validation_complete", "description": "test", "status_time": "2020-03-17 06:12:59.072000", "message": '{"function": "encoding-images-validation", "status": "validation_complete"}', }, { "status": "encoding_progressing", "description": "test", "status_time": "2020-03-17 06:12:59.072000", "message": '{"status": "PROCESSING"}', }, { "status": "encoding_completed", "description": "test", "status_time": "2020-03-17 06:12:59.072000", "message": '{"status": "COMPLETED"}', }, ], } response = asset_upload.get_asset_records_to_copy( PRODUCT_ID_1, TRACK_ID_2, asset_upload_types.STEREO ) assert response is not None assert response["asset_upload"] == expected_result["asset_upload"] assert len(response["asset_finals"]) == len(expected_result["asset_finals"]) assert len(response["asset_status"]) == len(expected_result["asset_status"]) def test_get_asset_records_to_copy_error(db_fixture: None) -> None: """Test for test_get_asset_records_to_copy db error.""" error_message = "Query error" ( flexmock(mysql) .should_receive("au_db_session") .and_raise(SQLAlchemyError(error_message)), ) with pytest.raises(SQLAlchemyError) as exc: asset_upload.get_asset_records_to_copy( PRODUCT_ID_1, 0, asset_upload_types.STEREO ) assert str(exc.value) == error_message def test_get_asset_records_to_copy_not_found(db_fixture: None) -> None: """Test for test_get_asset_records_to_copy no results.""" with pytest.raises(AssetUploadNotFound) as exc: asset_upload.get_asset_records_to_copy( PRODUCT_ID_NOT_EXISTS, 0, asset_upload_types.ATMOS ) assert exc.value.description == error_constants.ERROR_ASSET_UPLOAD_NOT_FOUND def test_apply_asset_corrections_with_multiple_upload_types_per_track( db_fixture: None, ) -> None: """Test that apply_asset_corrections returns one result per (track, upload_type).""" response = asset_upload.apply_asset_corrections(PRODUCT_ID_7) assert len(response) == 2 returned_type_ids = {item["asset_upload_type_id"] for item in response} assert returned_type_ids == {1, 3} # stereo and atmos for item in response: assert item["product_id"] == PRODUCT_ID_7 assert not item["is_correction"] assert not item["deleted"] def test_apply_asset_corrections_does_not_touch_unrelated_asset_upload_type_on_same_track( db_fixture_with_multi_asset_upload_type: None, ) -> None: response = asset_upload.apply_asset_corrections(PRODUCT_ID_9) assert len(response) == 1 assert response[0]["id"] == ASSET_UPLOAD_PK_30 assert response[0]["asset_upload_type_id"] == ATMOS_ASSET_UPLOAD_TYPE_ID assert not response[0]["is_correction"] assert not response[0]["deleted"] with mysql.au_db_session() as session: promoted_atmos = ( session.query(asset_upload.AssetUpload) .filter(asset_upload.AssetUpload.asset_upload_id == ASSET_UPLOAD_PK_30) .one() ) prior_atmos = ( session.query(asset_upload.AssetUpload) .filter(asset_upload.AssetUpload.asset_upload_id == ASSET_UPLOAD_PK_31) .one() ) unrelated_stereo = ( session.query(asset_upload.AssetUpload) .filter(asset_upload.AssetUpload.asset_upload_id == ASSET_UPLOAD_PK_32) .one() ) assert not promoted_atmos.is_correction assert not promoted_atmos.deleted assert not prior_atmos.is_correction assert prior_atmos.deleted assert not unrelated_stereo.is_correction assert not unrelated_stereo.deleted def test_apply_asset_corrections_promotes_two_asset_upload_types_with_priors_on_same_track( db_fixture_with_multi_asset_upload_type: None, ) -> None: response = asset_upload.apply_asset_corrections(PRODUCT_ID_11) assert len(response) == 2 promoted_by_type = {item["asset_upload_type_id"]: item for item in response} assert promoted_by_type[STEREO_ASSET_UPLOAD_TYPE_ID]["id"] == ASSET_UPLOAD_PK_34 assert promoted_by_type[ATMOS_ASSET_UPLOAD_TYPE_ID]["id"] == ASSET_UPLOAD_PK_36 with mysql.au_db_session() as session: rows_by_id = { row.asset_upload_id: row for row in session.query(asset_upload.AssetUpload) .filter( asset_upload.AssetUpload.asset_upload_id.in_( [ ASSET_UPLOAD_PK_34, ASSET_UPLOAD_PK_35, ASSET_UPLOAD_PK_36, ASSET_UPLOAD_PK_37, ] ) ) .all() } promoted_stereo = rows_by_id[ASSET_UPLOAD_PK_34] prior_stereo = rows_by_id[ASSET_UPLOAD_PK_35] promoted_atmos = rows_by_id[ASSET_UPLOAD_PK_36] prior_atmos = rows_by_id[ASSET_UPLOAD_PK_37] assert not promoted_stereo.is_correction assert not promoted_stereo.deleted assert prior_stereo.deleted assert not promoted_atmos.is_correction assert not promoted_atmos.deleted assert prior_atmos.deleted def test_apply_asset_corrections_promotes_correction_with_no_prior_active( db_fixture_with_multi_asset_upload_type: None, ) -> None: response = asset_upload.apply_asset_corrections(PRODUCT_ID_10) assert len(response) == 1 assert response[0]["id"] == ASSET_UPLOAD_PK_33 assert response[0]["asset_upload_type_id"] == ATMOS_ASSET_UPLOAD_TYPE_ID assert not response[0]["is_correction"] assert not response[0]["deleted"] with mysql.au_db_session() as session: promoted_atmos = ( session.query(asset_upload.AssetUpload) .filter(asset_upload.AssetUpload.asset_upload_id == ASSET_UPLOAD_PK_33) .one() ) unrelated_atmos_active = ( session.query(asset_upload.AssetUpload) .filter(asset_upload.AssetUpload.asset_upload_id == ASSET_UPLOAD_PK_31) .one() ) assert not promoted_atmos.is_correction assert not promoted_atmos.deleted assert not unrelated_atmos_active.is_correction assert not unrelated_atmos_active.deleted def test_get_latest_asset_upload_filtered_by_upload_type(db_fixture: None) -> None: """Test that get_latest_asset_upload respects the asset_upload_type_id filter.""" response = asset_upload.get_latest_asset_upload( product_id=PRODUCT_ID_7, is_correction_mode=True, asset_upload_type_id=3, # atmos ) assert len(response) == 1 assert response[0]["id"] == ASSET_UPLOAD_PK_25 assert response[0]["asset_upload_type_id"] == 3 def test_get_asset_uploads_by_product_id_returns_separate_entries_per_upload_type( db_fixture: None, ) -> None: """Test that get_asset_uploads_by_product_id groups by (track, upload_type).""" response = asset_upload.get_asset_uploads_by_product_id(PRODUCT_ID_7, api_version=2) assert len(response) == 2 returned_type_ids = {item["asset_upload_type_id"] for item in response} assert returned_type_ids == {1, 3} def test_mark_previous_asset_deleted_only_deletes_matching_atmos_upload_type( db_fixture: None, ) -> None: """Test that mark_previous_asset_deleted only deletes records of the given asset_upload_type_id.""" with mysql.au_db_session() as session: deleted = asset_upload.mark_previous_asset_deleted( api_version=2, product_id=PRODUCT_ID_8, track_unique_id=TRACK_ID_5, is_correction=False, asset_upload_type_id=3, # atmos only new_asset_upload_id=ASSET_UPLOAD_NOT_EXISTS, session=session, ) assert len(deleted) == 1 assert deleted[0]["id"] == ASSET_UPLOAD_PK_27 assert deleted[0]["asset_upload_type_id"] == 3 assert deleted[0]["deleted"] with mysql.au_db_session() as session: stereo_record = ( session.query(asset_upload.AssetUpload) .filter(asset_upload.AssetUpload.asset_upload_id == ASSET_UPLOAD_PK_26) .one() ) atmos_record = ( session.query(asset_upload.AssetUpload) .filter(asset_upload.AssetUpload.asset_upload_id == ASSET_UPLOAD_PK_27) .one() ) assert not stereo_record.deleted assert atmos_record.deleted def test_mark_previous_asset_deleted_atmos_does_not_delete_stereo( db_fixture: None, ) -> None: """Test that mark_previous_asset_deleted does not delete stereo with new atmos record.""" with mysql.au_db_session() as session: deleted = asset_upload.mark_previous_asset_deleted( api_version=2, product_id=PRODUCT_ID_5, track_unique_id=TRACK_ID_4, is_correction=False, asset_upload_type_id=3, # atmos only new_asset_upload_id=ASSET_UPLOAD_NOT_EXISTS, session=session, ) assert len(deleted) == 0 with mysql.au_db_session() as session: stereo_record_1 = ( session.query(asset_upload.AssetUpload) .filter(asset_upload.AssetUpload.asset_upload_id == ASSET_UPLOAD_PK_19) .one() ) stereo_record_2 = ( session.query(asset_upload.AssetUpload) .filter(asset_upload.AssetUpload.asset_upload_id == ASSET_UPLOAD_PK_20) .one() ) assert not stereo_record_1.deleted assert not stereo_record_2.deleted def test_get_asset_records_to_copy_when_ignore_flag_true( db_fixture: None, ) -> None: """Test get_asset_records_to_copy returns None when ignore_if_not_exist is True.""" response = asset_upload.get_asset_records_to_copy( PRODUCT_ID_NOT_EXISTS, 0, asset_upload_types.ATMOS, True, ) assert response is None