"""Test for logic hfa.py.""" import pandas as pd import numpy as np import pytest from unittest.mock import AsyncMock, MagicMock, Mock, patch from src.logic import hfa from src.models import ows_carveouts_python, ows_contracts, ows_product, ows_royalties from src.utils import constants @pytest.fixture def mock_abacus_active_contract_df(mock_get_all_active_contract_mechanical_deductions_response): """Return a mock abacus_active_contract_df DataFrame.""" df = pd.DataFrame(mock_get_all_active_contract_mechanical_deductions_response) return df.astype({'account_id': 'int64', 'contract_id': 'int64', 'term_type': 'string'}) @pytest.fixture def mock_filtered_abacus_active_contract_df(): """Return a mock filtered_abacus_active_contract_df DataFrame.""" return pd.DataFrame( [ { 'track_id': 103, 'track_isrc': 'USGHI7654321', 'upc': '135789012346', 'vendor_id': 1, 'matched_term_type': 'track', 'mechanical_type': ['digital', 'physical'], 'contract_id': 3, }, ] ) @pytest.fixture def mock_empty_filtered_abacus_active_contract_df(): """Return a mock empty filtered_abacus_active_contract_df DataFrame.""" return pd.DataFrame( [], columns=[ 'track_id', 'track_isrc', 'upc', 'vendor_id', 'matched_term_type', 'mechanical_type', 'contract_id', ], ) @pytest.fixture def mock_hfa_filtered_eligible_digital_physical_releases_df(): """Return a Filtered HFA eligible digital/physical tracks DataFrame.""" return pd.DataFrame( [ {'track_id': 101, 'hfa_configuration_code': 'SP'}, {'track_id': 102, 'hfa_configuration_code': 'CD'}, {'track_id': 102, 'hfa_configuration_code': 'LP'}, {'track_id': 102, 'hfa_configuration_code': 'CS'}, {'track_id': 103, 'hfa_configuration_code': 'SP'}, ] ) @pytest.fixture def mock_hfa_filtered_eligible_ringtones_df(): """Return a Filtered HFA eligible ringtone tracks DataFrame.""" return pd.DataFrame( [ {'track_id': 501, 'hfa_configuration_code': 'RT'}, ] ) @pytest.fixture def mock_hfa_filtered_empty_df(): """Return a Filtered HFA empty Dataframe.""" return pd.DataFrame(columns=['track_id', 'hfa_configuration_code']) @pytest.mark.parametrize( 'input_val, expected', [ pytest.param('mechanical,public', ['mechanical', 'public'], id='comma_separated_string'), pytest.param(['mechanical', 'public'], ['mechanical', 'public'], id='already_list'), pytest.param('', [''], id='empty_string'), pytest.param(np.nan, [], id='numpy_nan'), pytest.param(float('nan'), [], id='float_nan'), pytest.param(None, [], id='None_type'), pytest.param(123, [], id='unexpected_int_type'), pytest.param({'type': 'mechanical'}, [], id='unexpected_dict_type'), pytest.param( 'mechanical , public , sync', ['mechanical', 'public', 'sync'], id='string_with_extra_spaces', ), ], ) def test_parse_mechanical_type_to_list(input_val, expected): """Test parse_mechanical_type_to_list returns expected list.""" assert hfa.parse_mechanical_type_to_list(input_val) == expected def test_fetch_and_load_abacus_active_contracts_success( monkeypatch, mock_abacus_active_contract_df, mock_get_all_active_contract_mechanical_deductions_response, ): """Test fetch_and_load_abacus_active_contracts and parse into DataFrame.""" monkeypatch.setattr( ows_royalties, 'get_all_active_contract_mechanical_deductions', MagicMock(return_value=mock_get_all_active_contract_mechanical_deductions_response), ) result = hfa.fetch_and_load_abacus_active_contracts() pd.testing.assert_frame_equal(result, mock_abacus_active_contract_df) assert list(result.columns) == [ 'account_id', 'contract_id', 'term_type', 'attachments', 'mechanical_type', ] assert result.dtypes['account_id'] == 'int64' assert result.dtypes['contract_id'] == 'int64' assert result.dtypes['term_type'] == 'string' def test_fetch_and_load_abacus_active_contracts_exception(monkeypatch): """Test fetch_and_load_abacus_active_contracts raises exception.""" monkeypatch.setattr( ows_royalties, 'get_all_active_contract_mechanical_deductions', Mock(side_effect=Exception('Unexpected error while fetching Abacus Active Contracts')) ) with pytest.raises(Exception, match='Unexpected error while fetching Abacus Active Contracts'): hfa.fetch_and_load_abacus_active_contracts() @pytest.mark.parametrize( 'abacus_contracts, raw_digital, raw_ringtones, expected_rows', [ pytest.param( pd.DataFrame( [ { 'contract_id': 1, 'account_id': 100, 'term_type': 'track', 'attachments': ['ISRC123'], 'mechanical_type': ['type1'], } ] ), pd.DataFrame([{'track_id': 10, 'track_isrc': 'ISRC123', 'upc': 111, 'vendor_id': 100}]), pd.DataFrame([]), [ { 'track_id': 10, 'track_isrc': 'ISRC123', 'upc': 111, 'vendor_id': 100, 'matched_term_type': 'track', 'mechanical_type': ['type1'], 'contract_id': 1, } ], id='match_by_isrc', ), pytest.param( pd.DataFrame( [ { 'contract_id': 2, 'account_id': 200, 'term_type': 'product', 'attachments': ['222'], 'mechanical_type': ['type2'], } ] ), pd.DataFrame([{'track_id': 20, 'track_isrc': 'ISRC999', 'upc': 222, 'vendor_id': 200}]), pd.DataFrame([]), [ { 'track_id': 20, 'track_isrc': 'ISRC999', 'upc': 222, 'vendor_id': 200, 'matched_term_type': 'product', 'mechanical_type': ['type2'], 'contract_id': 2, } ], id='match_by_upc', ), pytest.param( pd.DataFrame( [ { 'contract_id': 3, 'account_id': 300, 'term_type': 'label', 'attachments': ['300'], 'mechanical_type': ['type3'], } ] ), pd.DataFrame([]), pd.DataFrame([{'track_id': 30, 'track_isrc': 'ISRC000', 'upc': 333, 'vendor_id': 300}]), [ { 'track_id': 30, 'track_isrc': 'ISRC000', 'upc': 333, 'vendor_id': 300, 'matched_term_type': 'label', 'mechanical_type': ['type3'], 'contract_id': 3, } ], id='match_by_vendor_id', ), pytest.param( pd.DataFrame( [ { 'contract_id': 4, 'account_id': 400, 'term_type': 'track', 'attachments': ['ISRCX'], 'mechanical_type': ['type4'], } ] ), pd.DataFrame( [{'track_id': 40, 'track_isrc': 'NO_MATCH', 'upc': 444, 'vendor_id': 401}] ), pd.DataFrame([]), [], id='no_match', ), pytest.param( pd.DataFrame( [ { 'contract_id': 1, 'account_id': 500, 'term_type': 'label', 'attachments': ['500'], 'mechanical_type': ['label_type'], }, { 'contract_id': 2, 'account_id': 500, 'term_type': 'product', 'attachments': ['555555'], 'mechanical_type': ['product_type'], }, { 'contract_id': 3, 'account_id': 500, 'term_type': 'track', 'attachments': ['ISRC555'], 'mechanical_type': ['track_type'], }, ] ), pd.DataFrame( [{'track_id': 55, 'track_isrc': 'ISRC555', 'upc': 555555, 'vendor_id': 500}] ), pd.DataFrame([]), [ { 'track_id': 55, 'track_isrc': 'ISRC555', 'upc': 555555, 'vendor_id': 500, 'matched_term_type': 'track', 'mechanical_type': ['track_type'], 'contract_id': 3, } ], id='match_priority_fallback_track_over_product_and_label', ), pytest.param( pd.DataFrame( [ { 'contract_id': 10, 'account_id': 601, 'term_type': 'track', 'attachments': ['DUPLICATEISRC'], 'mechanical_type': ['wrong_vendor'], }, { 'contract_id': 11, 'account_id': 600, 'term_type': 'track', 'attachments': ['DUPLICATEISRC'], 'mechanical_type': ['correct_vendor'], }, ] ), pd.DataFrame( [{'track_id': 60, 'track_isrc': 'DUPLICATEISRC', 'upc': 123123, 'vendor_id': 600}] ), pd.DataFrame([]), [ { 'track_id': 60, 'track_isrc': 'DUPLICATEISRC', 'upc': 123123, 'vendor_id': 600, 'matched_term_type': 'track', 'mechanical_type': ['correct_vendor'], 'contract_id': 11, } ], id='multiple_contracts_same_isrc_different_vendor_only_correct_matches', ), pytest.param( pd.DataFrame( [ { 'contract_id': 12, 'account_id': 700, 'term_type': 'track', 'attachments': ['ISRC777'], 'mechanical_type': ['type_digital'], }, { 'contract_id': 13, 'account_id': 701, 'term_type': 'label', 'attachments': ['701'], 'mechanical_type': ['type_ringtone'], }, ] ), pd.DataFrame([{'track_id': 70, 'track_isrc': 'ISRC777', 'upc': 123, 'vendor_id': 700}]), pd.DataFrame([{'track_id': 71, 'track_isrc': 'X', 'upc': 456, 'vendor_id': 701}]), [ { 'track_id': 70, 'track_isrc': 'ISRC777', 'upc': 123, 'vendor_id': 700, 'matched_term_type': 'track', 'mechanical_type': ['type_digital'], 'contract_id': 12, }, { 'track_id': 71, 'track_isrc': 'X', 'upc': 456, 'vendor_id': 701, 'matched_term_type': 'label', 'mechanical_type': ['type_ringtone'], 'contract_id': 13, }, ], id='match_mixed_digital_and_ringtone_inputs', ), ], ) def test_filter_abacus_contracts_by_priority( abacus_contracts, raw_digital, raw_ringtones, expected_rows ): """Test filter_abacus_contracts_by_priority returns expected dataframe.""" result = hfa.filter_abacus_contracts_by_priority( abacus_active_contract_df=abacus_contracts, raw_digital_physical_df=raw_digital, raw_ringtones_df=raw_ringtones, ) assert result.to_dict(orient='records') == expected_rows @pytest.mark.asyncio async def test_fetch_oa_active_contracts(monkeypatch, mock_active_contracts_rows, mock_active_contracts_df): """Test fetch_oa_active_contracts returns expected dataframe.""" vendor_ids = [1, 2, 3, 4, 5] monkeypatch.setattr( ows_contracts, 'get_oa_active_contracts', AsyncMock(return_value=mock_active_contracts_rows) ) contracts_df = await hfa.fetch_oa_active_contracts(vendor_ids) pd.testing.assert_frame_equal(contracts_df, mock_active_contracts_df) ows_contracts.get_oa_active_contracts.assert_awaited_once() assert list(contracts_df.columns) == [ 'vendor_id', 'physical_track', 'digital_track' ] @pytest.mark.asyncio async def test_fetch_oa_active_contracts_exception(monkeypatch): """Test fetch_oa_active_contracts re-raises on exception.""" vendor_ids = [1, 2, 3] monkeypatch.setattr( ows_contracts, 'get_oa_active_contracts', AsyncMock(side_effect=Exception('Unexpected Error while fetching OA Active Contracts')) ) with pytest.raises(Exception, match='Unexpected Error while fetching OA Active Contracts'): await hfa.fetch_oa_active_contracts(vendor_ids) @pytest.mark.asyncio async def test_fetch_carveouts(monkeypatch, mock_get_carveouts_row_1, mock_get_carveouts_row_2): """Test fetch_carveouts return expected dataframe.""" release_ids = [111, 222] carveouts_data = { 111: mock_get_carveouts_row_1, 222: mock_get_carveouts_row_2, } monkeypatch.setattr( ows_carveouts_python, 'get_carveouts', AsyncMock(return_value=carveouts_data) ) carveouts_df = await hfa.fetch_carveouts(release_ids) ows_carveouts_python.get_carveouts.assert_awaited_once() assert list(carveouts_df.columns) == [ 'release_id', 'has_us_carveout' ] @pytest.mark.asyncio async def test_fetch_carveouts_exception(monkeypatch): """Test fetch_carveouts re-raises on exception.""" release_ids = [111, 222] monkeypatch.setattr( ows_carveouts_python, 'get_carveouts', AsyncMock(side_effect=Exception('Unexpected error while fetching carveouts data')) ) with pytest.raises(Exception, match='Unexpected error while fetching carveouts data'): await hfa.fetch_carveouts(release_ids) def test_filter_eligible_hfa_ringtones( mock_hfa_eligible_ringtone_df, mock_active_contracts_df, mock_filtered_abacus_active_contract_df, mock_carveouts_df, mock_hfa_filtered_eligible_ringtones_df, ): """Test filter_eligible_hfa_ringtones returns expected dataframe.""" result = hfa.filter_eligible_hfa_ringtones( mock_hfa_eligible_ringtone_df, mock_active_contracts_df, mock_filtered_abacus_active_contract_df, mock_carveouts_df ) pd.testing.assert_frame_equal(result, mock_hfa_filtered_eligible_ringtones_df) def test_filter_eligible_hfa_ringtones_carveouts( mock_hfa_eligible_ringtone_df, mock_active_contracts_df, mock_filtered_abacus_active_contract_df, mock_has_us_carveouts_df ): """Test filter_eligible_hfa_ringtones returns empty dataframe due to carveouts.""" result = hfa.filter_eligible_hfa_ringtones( mock_hfa_eligible_ringtone_df, mock_active_contracts_df, mock_filtered_abacus_active_contract_df, mock_has_us_carveouts_df ) pd.testing.assert_frame_equal(result, pd.DataFrame(columns=['track_id', 'hfa_configuration_code'])) def test_filter_eligible_hfa_ringtones_empty( mock_active_contracts_df, mock_filtered_abacus_active_contract_df, mock_carveouts_df, mock_hfa_filtered_empty_df, ): """Test filter_eligible_hfa_ringtones returns empty dataframe.""" result = hfa.filter_eligible_hfa_ringtones( pd.DataFrame(), mock_active_contracts_df, mock_filtered_abacus_active_contract_df, mock_carveouts_df ) pd.testing.assert_frame_equal(result, mock_hfa_filtered_empty_df) def test_filter_eligible_hfa_digital_physical_tracks( mock_hfa_eligible_digital_physical_release_df, mock_active_contracts_df, mock_filtered_abacus_active_contract_df, mock_carveouts_df, mock_hfa_filtered_eligible_digital_physical_releases_df, ): """Test filter_eligible_hfa_digital_physical_tracks returns expected dataframe.""" result = hfa.filter_eligible_hfa_digital_physical_tracks( mock_hfa_eligible_digital_physical_release_df, mock_active_contracts_df, mock_filtered_abacus_active_contract_df, mock_carveouts_df ) pd.testing.assert_frame_equal( result.reset_index(drop=True), mock_hfa_filtered_eligible_digital_physical_releases_df.reset_index(drop=True), ) def test_filter_eligible_hfa_digital_physical_tracks_carveouts( mock_hfa_eligible_digital_physical_release_df, mock_active_contracts_df, mock_filtered_abacus_active_contract_df, mock_has_us_carveouts_df ): """Test filter_eligible_hfa_digital_physical_tracks returns empty dataframe due to carveouts.""" result = hfa.filter_eligible_hfa_digital_physical_tracks( mock_hfa_eligible_digital_physical_release_df, mock_active_contracts_df, mock_filtered_abacus_active_contract_df, mock_has_us_carveouts_df ) pd.testing.assert_frame_equal(result, pd.DataFrame(columns=['track_id', 'hfa_configuration_code'])) def test_filter_eligible_hfa_digital_physical_tracks_empty( mock_active_contracts_df, mock_filtered_abacus_active_contract_df, mock_hfa_filtered_empty_df, mock_carveouts_df ): """Test filter_eligible_hfa_digital_physical_tracks returns empty dataframe.""" result = hfa.filter_eligible_hfa_digital_physical_tracks( pd.DataFrame(), mock_active_contracts_df, mock_filtered_abacus_active_contract_df, mock_carveouts_df ) pd.testing.assert_frame_equal(result, mock_hfa_filtered_empty_df) @pytest.mark.asyncio async def test_get_hfa_pending_request( monkeypatch, mock_get_hfa_eligible_tracks_response, mock_active_contracts_df, mock_abacus_active_contract_df, mock_filtered_abacus_active_contract_df, mock_hfa_filtered_eligible_digital_physical_releases_df, mock_hfa_filtered_eligible_ringtones_df, mock_merge_tracks_df, mock_carveouts_df ): """Test get_hfa_pending_tracks returns expected result.""" monkeypatch.setattr( ows_product, 'get_hfa_eligible_tracks', MagicMock(return_value=mock_get_hfa_eligible_tracks_response), ) monkeypatch.setattr( hfa, 'fetch_and_load_abacus_active_contracts', MagicMock(return_value=mock_abacus_active_contract_df), ) monkeypatch.setattr( hfa, 'filter_abacus_contracts_by_priority', MagicMock(return_value=mock_filtered_abacus_active_contract_df), ) monkeypatch.setattr( hfa, 'fetch_oa_active_contracts', AsyncMock(return_value=mock_active_contracts_df), ) monkeypatch.setattr( hfa, 'fetch_carveouts', AsyncMock(return_value=mock_carveouts_df), ) monkeypatch.setattr( hfa, 'filter_eligible_hfa_digital_physical_tracks', MagicMock(return_value=mock_hfa_filtered_eligible_digital_physical_releases_df), ) monkeypatch.setattr( hfa, 'filter_eligible_hfa_ringtones', MagicMock(return_value=mock_hfa_filtered_eligible_ringtones_df), ) result = await hfa.get_pending_hfa_request() expected = [ {constants.FIELD_TRACK_ID: 101, constants.FIELD_HFA_CONF_CODE: 'SP'}, {constants.FIELD_TRACK_ID: 102, constants.FIELD_HFA_CONF_CODE: 'CD'}, {constants.FIELD_TRACK_ID: 102, constants.FIELD_HFA_CONF_CODE: 'LP'}, {constants.FIELD_TRACK_ID: 102, constants.FIELD_HFA_CONF_CODE: 'CS'}, {constants.FIELD_TRACK_ID: 103, constants.FIELD_HFA_CONF_CODE: 'SP'}, {constants.FIELD_TRACK_ID: 501, constants.FIELD_HFA_CONF_CODE: 'RT'} ] assert result == expected hfa.fetch_oa_active_contracts.assert_called_once_with( mock_merge_tracks_df['vendor_id'].dropna().unique().tolist() ) hfa.fetch_carveouts.assert_called_once_with( mock_merge_tracks_df['release_id'].dropna().unique().tolist() ) @pytest.mark.asyncio async def test_get_pending_hfa_request_exception(monkeypatch): """Test get_pending_hfa_request logs exception and re-raises it.""" monkeypatch.setattr( ows_product, 'get_hfa_eligible_tracks', MagicMock(side_effect=Exception('Unexpected error in get_pending_hfa_request')) ) with pytest.raises(Exception, match='Unexpected error in get_pending_hfa_request'): await hfa.get_pending_hfa_request() @pytest.mark.asyncio async def test_get_pending_hfa_request_no_eligible_tracks(monkeypatch): """Test get_pending_hfa_request returning empty response.""" monkeypatch.setattr( ows_product, 'get_hfa_eligible_tracks', MagicMock(return_value=[]) ) result = await hfa.get_pending_hfa_request() assert result == [] @patch('src.logic.hfa.logger.warning') @pytest.mark.asyncio async def test_get_pending_hfa_request_no_tracks( mock_logger_warning, monkeypatch, mock_get_hfa_eligible_tracks_response ): """Test get_pending_hfa_request has no pending tracks.""" monkeypatch.setattr( ows_product, 'get_hfa_eligible_tracks', MagicMock(return_value=mock_get_hfa_eligible_tracks_response), ) monkeypatch.setattr( hfa, 'fetch_and_load_abacus_active_contracts', MagicMock(return_value=pd.DataFrame()), ) monkeypatch.setattr( hfa, 'filter_abacus_contracts_by_priority', MagicMock(return_value=pd.DataFrame()), ) monkeypatch.setattr( hfa, 'fetch_oa_active_contracts', AsyncMock(return_value=pd.DataFrame()), ) monkeypatch.setattr( hfa, 'fetch_carveouts', AsyncMock(return_value=pd.DataFrame()), ) monkeypatch.setattr( hfa, 'filter_eligible_hfa_digital_physical_tracks', MagicMock(return_value=pd.DataFrame()), ) monkeypatch.setattr( hfa, 'filter_eligible_hfa_ringtones', MagicMock(return_value=pd.DataFrame()), ) result = await hfa.get_pending_hfa_request() assert result == [] mock_logger_warning.assert_called_once_with('No pending HFA requests found!')