"""Tests for asset_final model.""" import datetime from typing import Any import pytest from flexmock import flexmock from sqlalchemy.exc import SQLAlchemyError from syrupy.assertion import SnapshotAssertion from assets.connectors import mysql from assets.constants import error as error_const from assets.exceptions import ( AssetFinalExists, AssetFinalNotFound, BitsPerSampleNotFound, ) from assets.models import asset_final from tests.unit.models import au_operations ASSET_UPLOAD_ID_NOT_EXISTS = 1001 ASSET_UPLOAD_ID_1 = 1 ASSET_UPLOAD_ID_2 = 2 ASSET_UPLOAD_ID_3 = 3 @pytest.fixture def valid_asset_final_data() -> dict[str, Any]: """Fixture for valid data for create_asset_final.""" return { "asset_upload_id": ASSET_UPLOAD_ID_3, "asset_type": "flac", "asset_subtype": "none", "bucket": "bucket", "filename": "lossless.flac", "duration": 5, "channels": 2, "codec": "flac", "sample_rate": 44100, "bit_rate": 141121323, "bit_depth": 16, } @pytest.fixture def asset_final_data() -> list[dict[str, Any]]: """Fixture for asset finals list.""" return [ { "asset_upload_id": ASSET_UPLOAD_ID_1, "asset_type": "jpg", "asset_subtype": "cover", "bucket": "bucket", "filename": "qwe_123.jpg", "duration": 0, "channels": None, "codec": None, "sample_rate": None, "bit_rate": None, "bit_depth": None, }, { "asset_upload_id": ASSET_UPLOAD_ID_2, "asset_type": "wav", "asset_subtype": "none", "bucket": "bucket", "filename": "filename.wav", "duration": 2543, "channels": 2, "codec": "pcm", "sample_rate": 44100, "bit_rate": 141121323, "bit_depth": 16, }, { "asset_upload_id": ASSET_UPLOAD_ID_2, "asset_type": "mp3", "asset_subtype": "none", "bucket": "bucket", "filename": "filename.mp3", "duration": 221, "channels": 2, "codec": "mp3", "sample_rate": 44100, "bit_rate": 192000, "bit_depth": None, }, ] @pytest.fixture def db_fixture( asset_final_data: list[dict[str, Any]], asset_upload_data: list[dict[str, Any]] ) -> None: """Set up the asset_final table.""" au_operations.truncate_tables() au_operations.seed_asset_upload_table(asset_upload_data) au_operations.seed_asset_final_table(asset_final_data) def test_create_asset_final( db_fixture: None, valid_asset_final_data: dict[str, Any] ) -> None: """Test creating a new asset final.""" asset_final.create_asset_final(**valid_asset_final_data) assert True def test_create_asset_final_no_asset_type( db_fixture: None, valid_asset_final_data: dict[str, Any] ) -> None: """Test creating a new asset final with None asset_type.""" valid_asset_final_data["asset_type"] = None asset_final.create_asset_final(**valid_asset_final_data) assert True def test_create_asset_final_no_asset_subtype( db_fixture: None, valid_asset_final_data: dict[str, Any] ) -> None: """Test creating a new asset final with None asset_type.""" valid_asset_final_data["asset_subtype"] = None asset_final.create_asset_final(**valid_asset_final_data) assert True def test_create_asset_final_error( db_fixture: None, valid_asset_final_data: dict[str, Any] ) -> None: """Test creating a new asset final with sql error.""" error_message = "Query error" ( flexmock(mysql) .should_receive("au_db_session") .and_raise(SQLAlchemyError(error_message)) ) with pytest.raises(SQLAlchemyError) as exc: asset_final.create_asset_final(**valid_asset_final_data) assert str(exc.value) == error_message def test_get_asset_final( db_fixture: None, asset_final_data: list[dict[str, Any]] ) -> None: """Test getting asset final.""" response = asset_final.get_asset_final(asset_final_data[1]["filename"]) assert response["id"] == 2 assert response["asset_upload_id"] == asset_final_data[1]["asset_upload_id"] assert response["asset_type"] == asset_final_data[1]["asset_type"] assert response["asset_subtype"] == asset_final_data[1]["asset_subtype"] assert response["bucket"] == asset_final_data[1]["bucket"] assert response["filename"] == asset_final_data[1]["filename"] assert response["duration"] == asset_final_data[1]["duration"] def test_get_asset_final_not_found(db_fixture: None) -> None: """Test getting final asset for non-existent asset.""" with pytest.raises(AssetFinalNotFound) as exc: asset_final.get_asset_final("filename") assert exc.value.description == error_const.ERROR_ASSET_FINAL_NOT_FOUND def test_get_asset_final_error() -> None: """Test getting asset final with 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_final.get_asset_final("filename") assert str(exc.value) == error_message def test_get_asset_final_by_asset_upload_id(db_fixture: None) -> None: """Test get_asset_final_by_asset_upload_id success.""" response = asset_final.get_asset_final_by_asset_upload_ids(ASSET_UPLOAD_ID_2) filenames = [asset["filename"] for asset in response[ASSET_UPLOAD_ID_2]] assert sorted(filenames) == ["filename.mp3", "filename.wav"] def test_get_asset_final_by_asset_upload_ids_batch( db_fixture: None, asset_final_data: list[dict[str, Any]] ) -> None: """Test get_asset_final_by_asset_upload_id success.""" response = asset_final.get_asset_final_by_asset_upload_ids( [ASSET_UPLOAD_ID_1, ASSET_UPLOAD_ID_2] ) assert response[ASSET_UPLOAD_ID_1][0]["filename"] == asset_final_data[0]["filename"] assert sorted( [asset["filename"] for asset in response[ASSET_UPLOAD_ID_2]] ) == sorted([asset_final_data[1]["filename"], asset_final_data[2]["filename"]]) def test_get_asset_final_by_asset_upload_id_not_found(db_fixture: None) -> None: """Test getting final asset for non-existent asset.""" with pytest.raises(AssetFinalNotFound) as exc: asset_final.get_asset_final_by_asset_upload_ids(ASSET_UPLOAD_ID_NOT_EXISTS) assert exc.value.description == error_const.ERROR_ASSET_FINAL_NOT_FOUND def test_get_asset_final_by_asset_upload_id_error() -> None: """Test get_asset_final_by_asset_upload_id 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_final.get_asset_final_by_asset_upload_ids(ASSET_UPLOAD_ID_2) assert str(exc.value) == error_message def test_get_asset_final_by_asset_upload_id_and_subtype( db_fixture: None, asset_final_data: list[dict[str, Any]] ) -> None: """Test get_asset_final_by_asset_upload_id_and_subtype success.""" response = asset_final.get_asset_final_by_asset_upload_id_and_subtype( ASSET_UPLOAD_ID_1, asset_final_data[0]["asset_subtype"] ) assert response["asset_upload_id"] == ASSET_UPLOAD_ID_1 assert response["asset_type"] == asset_final_data[0]["asset_type"] assert response["asset_subtype"] == asset_final_data[0]["asset_subtype"] assert response["bucket"] == asset_final_data[0]["bucket"] assert response["filename"] == asset_final_data[0]["filename"] assert response["duration"] == asset_final_data[0]["duration"] def test_get_asset_final_by_asset_upload_id_and_subtype_not_found( db_fixture: None, ) -> None: """Test get_asset_final_by_asset_upload_id_and_subtype for non-existent asset.""" with pytest.raises(AssetFinalNotFound) as exc: asset_final.get_asset_final_by_asset_upload_id_and_subtype( ASSET_UPLOAD_ID_1, "something" ) assert exc.value.description == error_const.ERROR_ASSET_FINAL_NOT_FOUND def test_get_asset_final_by_asset_upload_id_and_subtype_error() -> None: """Test get_asset_final_by_asset_upload_id_and_subtype with 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_final.get_asset_final_by_asset_upload_id_and_subtype( ASSET_UPLOAD_ID_1, "cover" ) assert str(exc.value) == error_message @pytest.fixture def get_product_assets_db_fixture() -> None: """DB fixture for get_product_assets.""" au_operations.truncate_tables() au_operations.seed_asset_upload_table( [ { "id": 1, "user_id": "alw:123", "token": b"123", "filename": "123", "product_id": 123, "track_unique_id": 0, "api_version": 1, "deleted": False, "is_correction": False, "updated_timestamp": datetime.date(2019, 1, 1), }, { "id": 2, "user_id": "alw:123", "token": b"123", "filename": "123", "product_id": 123, "track_unique_id": 0, "api_version": 2, "deleted": True, "is_correction": False, "updated_timestamp": datetime.date(2019, 1, 1), }, { "id": 3, "user_id": "alw:123", "token": b"123", "filename": "123", "product_id": 123, "track_unique_id": 0, "api_version": 2, "deleted": False, "is_correction": True, "updated_timestamp": datetime.date(2019, 1, 1), }, { "id": 4, "user_id": "alw:123", "token": b"123", "filename": "123", "product_id": 123, "track_unique_id": 0, "api_version": 2, "deleted": False, "is_correction": False, "updated_timestamp": datetime.date(2019, 1, 1), }, { "id": 5, "user_id": "alw:123", "token": b"123", "filename": "123", "product_id": 123, "track_unique_id": 1, "api_version": 2, "deleted": False, "is_correction": False, "updated_timestamp": datetime.date(2019, 1, 1), }, { "id": 6, "user_id": "alw:123", "token": b"123", "filename": "123", "product_id": 123, "track_unique_id": 2, "api_version": 2, "deleted": False, "is_correction": False, "updated_timestamp": datetime.date(2019, 1, 1), }, { "id": 7, "user_id": "alw:123", "token": b"123", "filename": "123", "product_id": 123, "track_unique_id": 8, "api_version": 2, "deleted": False, "is_correction": False, "updated_timestamp": datetime.date(2019, 1, 6), }, { "id": 8, "user_id": "alw:123", "token": b"123", "filename": "123", "product_id": 123, "track_unique_id": 3, "api_version": 2, "deleted": False, "is_correction": False, "updated_timestamp": datetime.date(2019, 1, 6), }, { "id": 9, "user_id": "alw:123", "token": b"123", "filename": "123", "product_id": 123, "track_unique_id": 4, "api_version": 2, "deleted": False, "is_correction": False, "updated_timestamp": datetime.date(2019, 1, 6), }, { "id": 10, "user_id": "alw:123", "token": b"123", "filename": "123", "product_id": 123, "track_unique_id": 4, "api_version": 2, "deleted": False, "is_correction": True, "updated_timestamp": datetime.date(2019, 1, 6), }, { "id": 11, "user_id": "alw:123", "token": b"123", "filename": "123", "product_id": 123, "track_unique_id": 4, "api_version": 2, "deleted": True, "is_correction": False, "updated_timestamp": datetime.date(2019, 1, 6), }, ] ) au_operations.seed_asset_final_table( [ { "asset_upload_id": 2, "asset_type": "TIF", "bucket": "WRONG", "filename": "WRONG", }, { "asset_upload_id": 3, "asset_type": "TIF", "bucket": "WRONG", "filename": "WRONG", }, { "asset_upload_id": 4, "asset_type": "TIF", "bucket": "BUCKET", "filename": "123.tif", }, { "asset_upload_id": 5, "asset_type": "WAV", "bucket": "BUCKET", "filename": "123-1.wav", }, { "asset_upload_id": 6, "asset_type": "WAV", "bucket": "BUCKET", "filename": "123-2.wav", }, { "asset_upload_id": 8, "asset_type": "FLAC", "bucket": "BUCKET", "filename": "123.flac", }, { "asset_upload_id": 9, "asset_type": "MP3_192", "bucket": "BUCKET", "filename": "123-192.mp3", }, { "asset_upload_id": 10, "asset_type": "FLAC", "bucket": "WRONG", "filename": "WRONG", }, { "asset_upload_id": 11, "asset_type": "FLAC", "bucket": "WRONG", "filename": "WRONG", }, ] ) @pytest.fixture def get_product_assets_with_bit_depth_db_fixture() -> None: """DB fixture for get_product_assets.""" au_operations.truncate_tables() au_operations.seed_asset_upload_table( [ { "id": 1, "user_id": "alw:123", "token": b"123", "filename": "123", "product_id": 123, "track_unique_id": 0, "api_version": 1, "deleted": 0, "is_correction": 0, "updated_timestamp": datetime.date(2019, 1, 1), }, { "id": 2, "user_id": "alw:123", "token": b"123", "filename": "123", "product_id": 123, "track_unique_id": 1, "api_version": 2, "deleted": 0, "is_correction": 0, "updated_timestamp": datetime.date(2019, 1, 1), }, { "id": 3, "user_id": "alw:123", "token": b"123", "filename": "123", "product_id": 123, "track_unique_id": 2, "api_version": 2, "deleted": 0, "is_correction": 0, "updated_timestamp": datetime.date(2019, 1, 1), }, { "id": 4, "user_id": "alw:123", "token": b"123", "filename": "123", "product_id": 123, "track_unique_id": 3, "api_version": 2, "deleted": 0, "is_correction": 0, "updated_timestamp": datetime.date(2019, 1, 6), }, ] ) au_operations.seed_asset_final_table( [ { "asset_upload_id": 1, "asset_type": "TIF", "bucket": "WRONG", "filename": "WRONG", "bit_depth": 0, }, { "asset_upload_id": 2, "asset_type": "WAV", "bucket": "BUCKET", "filename": "123-1.wav", "bit_depth": 24, }, { "asset_upload_id": 3, "asset_type": "WAV", "bucket": "BUCKET", "filename": "123-1.wav", "bit_depth": 24, }, { "asset_upload_id": 4, "asset_type": "WAV", "bucket": "BUCKET", "filename": "123-2.wav", "bit_depth": 24, }, ] ) @pytest.mark.parametrize( "filters", [ ([]), (["FLAC"]), (["flac"]), (["MP3_192"]), (["WAV"]), (["TIF"]), (["FLAC", "WAV"]), ], ) def test_get_product_asset( filters: list[str], get_product_assets_db_fixture: None, snapshot: SnapshotAssertion ) -> None: """Test get_product_asset.""" product_id = 123 result = asset_final.get_product_asset([product_id], filters) assert result == snapshot def test_get_product_asset_upload_and_final_not_found( get_product_assets_db_fixture: None, ) -> None: """Test get_product_asset when no assets are found.""" product_id = 456 with pytest.raises(AssetFinalNotFound) as exc: asset_final.get_product_asset([product_id]) assert exc.value.description == error_const.ERROR_ASSET_FINAL_NOT_FOUND def test_get_bit_depths_for_product( get_product_assets_with_bit_depth_db_fixture: None, ) -> None: """Test get_bit_depths_for_products.""" product_id = 123 result = asset_final.get_bit_depths_for_products(product_id) assert result == [24] def test_get_bit_depths_not_found_for_product( get_product_assets_with_bit_depth_db_fixture: None, ) -> None: """Test get_bit_depths_for_products.""" product_id = 123456 with pytest.raises(BitsPerSampleNotFound) as exc: asset_final.get_bit_depths_for_products(product_id) assert exc.value.description == error_const.NO_BITS_PER_SAMPLE_FOUND def test_get_asset_final_by_id( db_fixture: None, asset_final_data: list[dict[str, Any]] ) -> None: """Test getting asset final.""" response = asset_final.get_asset_final_by_id(2) assert response["id"] == 2 assert response["asset_upload_id"] == asset_final_data[1]["asset_upload_id"] assert response["asset_type"] == asset_final_data[1]["asset_type"] assert response["asset_subtype"] == asset_final_data[1]["asset_subtype"] assert response["bucket"] == asset_final_data[1]["bucket"] assert response["filename"] == asset_final_data[1]["filename"] assert response["duration"] == asset_final_data[1]["duration"] def test_get_asset_final_by_id_not_found(db_fixture: None) -> None: """Test getting final asset for non-existent asset.""" with pytest.raises(AssetFinalNotFound) as exc: asset_final.get_asset_final_by_id(1000) assert exc.value.description == error_const.ERROR_ASSET_FINAL_NOT_FOUND def test_get_asset_final_by_id_error() -> None: """Test getting asset final with 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_final.get_asset_final_by_id(1000) assert str(exc.value) == error_message def test_create_asset_final_conflict() -> None: """Test getting asset final with SQLAlchemyError failure.""" error_message = "Asset final already exists" ( flexmock(mysql) .should_receive("au_db_session") .and_raise(AssetFinalExists(error_message)) ) with pytest.raises(AssetFinalExists) as exc: asset_final.create_asset_final( 333, "FLAC", "", "foo", "bar", 10, 2, "FLAC", 1, 1, 1 ) assert exc.value.description == error_message