"""Test for asset model.""" from unittest.mock import patch import pytest from sqlalchemy.exc import SQLAlchemyError from assets import config from assets.constants import error from assets.models.legacy import asset from tests.models.legacy import dd_operations @pytest.fixture def asset_data(): """Get data to seed asset table.""" return [ { 'asset_id': 1, 'upc': 191018729677, 'asset_type_id': 1, 'cd': 1, 'track_id': 1 }, { 'asset_id': 2, 'upc': 191018729678, 'asset_type_id': 1, 'cd': 0, 'track_id': 0 }, { 'asset_id': 3, 'upc': 191018729679, 'asset_type_id': 1, 'cd': 0, 'track_id': 0 } ] @pytest.fixture def asset_location_data(): """Get data to seed asset location table.""" return [ { 'asset_location_id': 1, 'asset_id': 1, 'storage_drive_id': 1, 'status': 'Y' }, { 'asset_location_id': 2, 'asset_id': 1, 'storage_drive_id': 1, 'status': 'N' }, { 'asset_location_id': 3, 'asset_id': 1, 'storage_drive_id': 2, 'status': 'Y' }, { 'asset_location_id': 4, 'asset_id': 2, 'storage_drive_id': 2, 'status': 'N' }, { 'asset_location_id': 5, 'asset_id': 2, 'storage_drive_id': 3, 'status': 'N' }, { 'asset_location_id': 6, 'asset_id': 2, 'storage_drive_id': 3, 'status': 'N' }, { 'asset_location_id': 7, 'asset_id': 3, 'storage_drive_id': 1, 'status': 'L' } ] @pytest.fixture def asset_location_detail_data(): """Get data to seed asset location detail table.""" return [ { 'asset_location_detail_id': 1, 'asset_location_id': 1, 'bits_per_sample': 64, 'filename': '191018729677_1_1.ext', 'duration': 10.10 }, { 'asset_location_detail_id': 2, 'asset_location_id': 2, 'bits_per_sample': 64, 'filename': 'file2.ext', 'duration': 20.20 }, { 'asset_location_detail_id': 3, 'asset_location_id': 3, 'bits_per_sample': 64, 'filename': 'file3.ext', 'duration': 30.30 }, { 'asset_location_detail_id': 4, 'asset_location_id': 4, 'bits_per_sample': 64, 'filename': 'file4.ext', 'duration': 40.40 }, { 'asset_location_detail_id': 5, 'asset_location_id': 5, 'bits_per_sample': 64, 'filename': 'file5.ext', 'duration': 50.50 }, { 'asset_location_detail_id': 6, 'asset_location_id': 6, 'bits_per_sample': 16, 'filename': 'file6.ext', 'duration': 60.60 }, { 'asset_location_detail_id': 7, 'asset_location_id': 7, 'bits_per_sample': 16, 'filename': '191018729679_1_1.ext', 'duration': 70.70 } ] @pytest.fixture def asset_types_data(): """Get data to seed asset_type table.""" return [ { 'asset': 'WAV', 'file_type': 'directory', 'extension': '.wav', 'initial_folder': 'folder', 'folder_structure': '{upc(0,6)}', 'parent_asset_type_id': None } ] @pytest.fixture def storage_data(): """Get data to seed storage table.""" return [ { 'storage_id': 1, 'physical_location_id': 1, 'local_ip': 168437923 } ] @pytest.fixture def storage_drive_data(): """Get data to seed storage drive table.""" return [ { 'storage_drive_id': 1, 'storage_id': 1 }, { 'storage_drive_id': 2, 'storage_id': 2 } ] @pytest.fixture def storage_drive_asset_folder_data(): """Get data to seed storage drive asset folder table.""" return [ { 'storage_drive_asset_folder_id': 1, 'storage_drive_id': 1, 'asset_type_id': 1, 'initial_folder': 'wav/' }, { 'storage_drive_asset_folder_id': 2, 'storage_drive_id': 1, 'asset_type_id': 2, 'initial_folder': 'tif/' }, { 'storage_drive_asset_folder_id': 3, 'storage_drive_id': 2, 'asset_type_id': 1, 'initial_folder': 'wav/' } ] @pytest.fixture def db_fixture( asset_data, asset_location_data, asset_location_detail_data, asset_types_data, storage_data, storage_drive_data, storage_drive_asset_folder_data): """Set up DD tables.""" dd_operations.create_tables() dd_operations.seed_asset_table(asset_data) dd_operations.seed_asset_location_table(asset_location_data) dd_operations.seed_asset_location_detail_table(asset_location_detail_data) dd_operations.seed_asset_type_table(asset_types_data) dd_operations.seed_storage_table(storage_data) dd_operations.seed_storage_drive_table(storage_drive_data) dd_operations.seed_storage_drive_asset_folder_table( storage_drive_asset_folder_data) def test_get_number_distinct_bitrates_for_upc(db_fixture): """Test get_number_distinct_bitrates_for_upc with valid UPC.""" upc = 191018729677 # Considering .wav asset asset_type_id = 1 # Considering colo as physical location physical_location_id = 1 response = asset.get_number_distinct_bitrates_for_upc( upc, asset_type_id, physical_location_id) assert response.status == 200 assert response.message == 1 def test_get_number_distinct_bitrates_with_invalid_location(db_fixture): """Test get_number_distinct_bitrates_for_upc with valid UPC.""" upc = 191018729677 # Considering .wav asset asset_type_id = 1 # Considering colo as physical location physical_location_id = 2 response = asset.get_number_distinct_bitrates_for_upc( upc, asset_type_id, physical_location_id) assert response.status == 404 def test_get_number_distinct_bitrates_for_invalid_upc(db_fixture): """Test get_number_distinct_bitrates_for_upc with invalid UPC.""" upc = 'invalid_upc' # Considering .wav asset asset_type_id = 1 # Considering colo as physical location physical_location_id = 1 response = asset.get_number_distinct_bitrates_for_upc( upc, asset_type_id, physical_location_id) assert response.status == 404 assert response.errors.get('message') == \ 'Value for bits per sample not found for provided product id.' def test_get_number_distinct_bitrates_for_upc_with_no_upc(): """Test get_number_distinct_bitrates_for_upc with no UPC.""" upc = '' # Considering .wav asset asset_type_id = 1 # Considering colo as physical location physical_location_id = 1 response = asset.get_number_distinct_bitrates_for_upc( upc, asset_type_id, physical_location_id) assert response.status == 400 @patch('assets.connectors.mysql.dd_db_session', side_effect=SQLAlchemyError()) def test_get_number_distinct_bitrates_for_upc_with_exception(monkeypatch): """Test get_number_distinct_bitrates_for_upc with Exception.""" upc = '@' # Considering .wav asset asset_type_id = 1 # Considering colo as physical location physical_location_id = 1 response = asset.get_number_distinct_bitrates_for_upc( upc, asset_type_id, physical_location_id) assert response.status == 500 def test_get_bitrates_for_upc(db_fixture): """Test get_bitrates_for_upc with valid UPC.""" upc = 191018729677 expected_response = [64] # Considering .wav asset asset_type_id = 1 # Considering colo as physical location physical_location_id = 1 response = asset.get_bitrates_for_upc( upc, asset_type_id, physical_location_id) assert response.status == 200 assert response.message == expected_response def test_get_bitrates_for_upc_with_invalid_location(db_fixture): """Test get_bitrates_for_upc with valid UPC.""" upc = 191018729677 # Considering .wav asset asset_type_id = 1 # Considering colo as physical location physical_location_id = 2 response = asset.get_bitrates_for_upc( upc, asset_type_id, physical_location_id) assert response.status == 404 def test_get_bitrates_for_upc_for_invalid_upc(db_fixture): """Test get_number_distinct_bitrates_for_upc with invalid UPC.""" upc = 'invalid_upc' # Considering .wav asset asset_type_id = 1 # Considering colo as physical location physical_location_id = 1 response = asset.get_bitrates_for_upc( upc, asset_type_id, physical_location_id) assert response.status == 404 assert response.errors.get('message') == error.NO_BITS_PER_SAMPLE_FOUND def test_get_bitrates_for_upc_with_no_upc(): """Test get_bitrates_for_upc with no UPC.""" upc = '' # Considering .wav asset asset_type_id = 1 # Considering colo as physical location physical_location_id = 1 response = asset.get_bitrates_for_upc( upc, asset_type_id, physical_location_id) assert response.status == 400 assert response.errors.get('message') == error.ERROR_UPC_NOT_FOUND @patch('assets.connectors.mysql.dd_db_session', side_effect=SQLAlchemyError()) def test_get_bitrates_for_upc_with_exception(monkeypatch): """Test get_bitrates_for_upc with Exception.""" upc = '@' # Considering .wav asset asset_type_id = 1 # Considering colo as physical location physical_location_id = 1 response = asset.get_bitrates_for_upc( upc, asset_type_id, physical_location_id) assert response.status == 500 def test_get_bits_per_sample_data_for_tracks(db_fixture): """Test get_bits_per_sample_data_for_tracks with valid UPC.""" upc = 191018729677 expected_response = [ {'asset_type_id': 1, 'filename': '191018729677_1_1.ext', 'bits_per_sample': 64}, {'asset_type_id': 1, 'filename': 'file2.ext', 'bits_per_sample': 64}] # Considering .wav asset asset_type_list = [1, 353] # Considering colo as physical location physical_location_id = 1 response = asset.get_bits_per_sample_data_for_tracks( upc, physical_location_id, asset_type_list) assert response.status == 200 assert response.message == expected_response def test_get_bits_per_sample_data_for_tracks_with_invalid_location(db_fixture): """Test get_bits_per_sample_data_for_tracks with valid UPC.""" upc = 191018729677 # Considering .wav asset asset_type_list = [1] # Considering colo as physical location physical_location_id = 1000 response = asset.get_bits_per_sample_data_for_tracks( upc, physical_location_id, asset_type_list) assert response.status == 404 def test_get_bits_per_sample_data_for_tracks_for_invalid_upc(db_fixture): """Test get_number_distinct_bitrates_for_upc with invalid UPC.""" upc = 'invalid_upc' # Considering .wav asset asset_type_list = [1] # Considering colo as physical location physical_location_id = 1 response = asset.get_bits_per_sample_data_for_tracks( upc, physical_location_id, asset_type_list) assert response.status == 404 assert response.errors.get('message') == error.NO_BITS_PER_SAMPLE_FOUND def test_get_bits_per_sample_data_for_tracks_with_no_upc(): """Test get_bits_per_sample_data_for_tracks with no UPC.""" upc = '' # Considering .wav asset asset_type_list = [1] # Considering colo as physical location physical_location_id = 1 response = asset.get_bits_per_sample_data_for_tracks( upc, physical_location_id, asset_type_list) assert response.status == 400 assert response.errors.get('message') == error.ERROR_UPC_NOT_FOUND @patch('assets.connectors.mysql.dd_db_session', side_effect=SQLAlchemyError()) def test_get_bits_per_sample_data_for_tracks_with_exception(monkeypatch): """Test get_bits_per_sample_data_for_tracks with Exception.""" upc = '@' # Considering .wav asset asset_type_list = [1] # Considering colo as physical location physical_location_id = 1 response = asset.get_bits_per_sample_data_for_tracks( upc, physical_location_id, asset_type_list) assert response.status == 500 def test_get_streaming_info( db_fixture, asset_data, asset_types_data, storage_data, storage_drive_asset_folder_data, asset_location_detail_data): """Test getting asset streaming info.""" expected_message = { 'upc': asset_data[0]['upc'], 'cd': asset_data[0]['cd'], 'track_id': asset_data[0]['track_id'], 'local_ip': storage_data[0]['local_ip'], 'storage_id': storage_data[0]['storage_id'], 'extension': asset_types_data[0]['extension'], 'folder_structure': asset_types_data[0]['folder_structure'], 'path_structure_type': asset_types_data[0]['file_type'], 'initial_folder': storage_drive_asset_folder_data[0]['initial_folder'], 'filename': asset_location_detail_data[0]['filename'], 'duration': asset_location_detail_data[0]['duration'] } search_condition = '191018729677_1_1%' response = asset.get_streaming_info( expected_message['upc'], 1, config.SERVER_LOCATION_AVL, search_condition) assert response assert response.message == expected_message def test_get_streaming_info_label_copy( db_fixture, asset_data, asset_types_data, storage_data, storage_drive_asset_folder_data, asset_location_detail_data): """Test getting asset streaming info for asset in label copy.""" expected_message = { 'upc': asset_data[2]['upc'], 'cd': asset_data[2]['cd'], 'track_id': asset_data[2]['track_id'], 'local_ip': storage_data[0]['local_ip'], 'storage_id': storage_data[0]['storage_id'], 'extension': asset_types_data[0]['extension'], 'folder_structure': asset_types_data[0]['folder_structure'], 'path_structure_type': asset_types_data[0]['file_type'], 'initial_folder': storage_drive_asset_folder_data[0]['initial_folder'], 'filename': asset_location_detail_data[6]['filename'], 'duration': asset_location_detail_data[6]['duration'] } search_condition = '191018729679_1_1%' response = asset.get_streaming_info( expected_message['upc'], 1, config.SERVER_LOCATION_AVL, search_condition) assert response assert response.message == expected_message def test_get_streaming_info_not_found(db_fixture): """Test getting asset streaming info for non-existed asset.""" search_condition = '191018729677_0_0%' response = asset.get_streaming_info(191018729678, 1, 1, search_condition) assert response.status == 404 @patch('assets.connectors.mysql.dd_db_session', side_effect=SQLAlchemyError()) def test_get_streaming_info_error(monkeypatch): """Test getting asset streaming info with error.""" search_condition = '191018729678_0_0%' response = asset.get_streaming_info(191018729678, 1, 1, search_condition) assert response.status == 500