"""Tests for backend.utils.logic.""" import datetime from oto import response from oto import status as response_code import pytest from backend.connectors import mysql from backend.constants import error from backend.constants import product as product_consts from backend.constants import track_field as tf from backend.models import ows_product from backend.models import track_audit from backend.models.track_query import TrackQuery from backend.utils import logic as logic_utils from backend.utils import product_utils from tests.testutils import constants as test_consts from tests.testutils import db from tests.testutils import mocks def test_get_product_data_and_validate_ownership( mocker, test_vendor_id, test_upc_code): """Verify product data is returned after ownership check.""" product_id = 1 mocks.ows_product_verify_product_ownership(mocker) mocks.ows_product_get_product_by_id( mocker, product_id, test_upc_code) res = logic_utils.get_product_data_and_validate_ownership( product_id, account_type='vendor', account_id=123) assert res.status == response_code.OK assert res.message['vendor_id'] == test_vendor_id assert res.message['product_id'] == product_id def test_get_product_data_and_validate_ownership_no_grass_headers( mocker, test_vendor_id, test_upc_code): """Verify product data is returned after bypassing ownership check.""" product_id = 1 mocks.ows_product_get_product_by_id( mocker, product_id, test_upc_code) res = logic_utils.get_product_data_and_validate_ownership( product_id, account_type=None, account_id=None) assert res.status == response_code.OK assert res.message['product_id'] == product_id def test_get_product_data_and_validate_ownership_bad_product_id( mocker): """Verify product data is returned after bypassing ownership check.""" res = logic_utils.get_product_data_and_validate_ownership( product_id='a', account_type=None, account_id=None) assert res.status == response_code.BAD_REQUEST assert res.errors['code'] == error.VALIDATION_ERROR_CODE def test_make_publishing_obligation_response_normalize_pub_obl(track_factory): """Verify us publishing obligation field value is normalized.""" test_track = track_factory(us_publishing_obligation='') results = logic_utils.make_publishing_obligation_response( [test_track.to_dict()]) track = results.message['items'][0] assert track[tf.US_PUBLISHING_OBLIGATION] is None def test_make_publishing_obligation_response_with_version(mocker, track_factory): """Verify publishing obligation field value with track version.""" version = 'test version' test_track = track_factory(us_publishing_obligation='', version=version) results = logic_utils.make_publishing_obligation_response( [test_track.to_dict()]) track = results.message['items'][0] assert track[tf.VERSION] == version def test_make_publishing_obligation_response_normalize_third_party( track_factory): """Verify third party publisher field value is normalized.""" test_track = track_factory( us_publishing_obligation='Composition', third_party_publisher='') results = logic_utils.make_publishing_obligation_response( [test_track.to_dict()]) track = results.message['items'][0] assert track[tf.THIRD_PARTY_PUBLISHER] is None def test_is_genre_classical(mocker): """Verify genre is classical.""" product_data = { 'genre_id': test_consts.TEST_CLASSICAL_GENRE_ID, 'subgenre_id': test_consts.TEST_CLASSICAL_SUBGENRE_ID} subgenres = [{ 'orchard_id': test_consts.TEST_CLASSICAL_SUBGENRE_ID, 'name': '20th/21st Century', 'genre_id': product_consts.CLASSICAL_GENRE_ID }] mocks.head_response(mocker) mocks.subgenre_query.get_genre_subgenres(mocker, subgenres) product_type = logic_utils.get_digital_product_genre_type(product_data) assert product_type['genre'] == product_consts.CLASSICAL_GENRE assert product_type['subgenre'] == '20th/21st Century' def test_is_genre_soundtrack(): """Verify genre is soundtrack.""" product_data = { 'genre_id': test_consts.TEST_SOUNDTRACK_GENRE_ID, 'subgenre_id': test_consts.TEST_SOUNDTRACK_SUBGENRE_ID} product_type = logic_utils.get_digital_product_genre_type(product_data) assert product_type['genre'] == product_consts.SOUNDTRACK_GENRE assert product_type['subgenre'] == 'Film Soundtracks' def test_is_genre_is_other(): """Verify genre is other.""" product_data = { 'genre_id': test_consts.TEST_GENRE_ID, 'subgenre_id': test_consts.TEST_SUBGENRE_ID} product_type = logic_utils.get_digital_product_genre_type(product_data) assert product_type['genre'] == product_consts.OTHER_GENRE assert product_type['subgenre'] == '' def test_corrected_product_uses_corrected_genre(mocker): """Verify the the corrected genre (classical) is used.""" product_data = { 'corrections': { 'items': [{ 'field_name': product_consts.GENRE_ID, 'key_value': product_consts.CLASSICAL_GENRE_ID }] }, product_consts.GENRE_ID: test_consts.TEST_GENRE_ID, product_consts.SUBGENRE_ID: test_consts.TEST_CLASSICAL_SUBGENRE_ID, product_consts.RELEASE_STATUS: product_consts.IS_ERR_CORR_MODE } subgenres = [{ 'orchard_id': test_consts.TEST_CLASSICAL_SUBGENRE_ID, 'name': '20th/21st Century', 'genre_id': product_consts.CLASSICAL_GENRE_ID }] mocks.head_response(mocker) mocks.subgenre_query.get_genre_subgenres(mocker, subgenres) genre_type = logic_utils.get_digital_product_genre_type(product_data) assert genre_type['genre'] == product_consts.CLASSICAL_GENRE assert genre_type['subgenre'] == '20th/21st Century' def test_corrected_product_uses_corrected_subgenre(): """Verify the the corrected subgenre (soundtrack) is used.""" product_data = { 'corrections': { 'items': [{ 'field_name': product_consts.RELEASE_SUBGENRE, 'key_value': [test_consts.TEST_SOUNDTRACK_SUBGENRE_ID] }] }, product_consts.GENRE_ID: test_consts.TEST_SOUNDTRACK_GENRE_ID, product_consts.SUBGENRE_ID: test_consts.TEST_SUBGENRE_ID, product_consts.RELEASE_STATUS: product_consts.IS_ERR_CORR_MODE } genre_type = logic_utils.get_digital_product_genre_type(product_data) assert genre_type['genre'] == product_consts.SOUNDTRACK_GENRE assert genre_type['subgenre'] == 'Film Soundtracks' def test_uncorrected_product_does_not_use_corrected_genre(): """Verify the the corrected genre (classical) is not used.""" product_data = { 'corrections': { 'items': [{ 'field_name': product_consts.GENRE_ID, 'key_value': product_consts.CLASSICAL_GENRE_ID }] }, product_consts.GENRE_ID: test_consts.TEST_GENRE_ID, product_consts.SUBGENRE_ID: test_consts.TEST_SUBGENRE_ID } genre_type = logic_utils.get_digital_product_genre_type(product_data) assert genre_type['genre'] == product_consts.OTHER_GENRE assert genre_type['subgenre'] == '' def test_verify_product_ownership(mocker, test_vendor_id): """Verify product ownership is successful using a product_id.""" @logic_utils.verify_product_ownership def wrapped_function(**kwargs): ows_product.verify_product_ownership.assert_called() assert kwargs['product_id'] == product_id product_id = 1 mocks.ows_product_verify_product_ownership(mocker) wrapped_function( product_id=1, account_type='vendor', account_id=test_vendor_id) def test_verify_product_ownership_no_grass_headers(mocker): """Should skip ownership check when no Grass Headers.""" @logic_utils.verify_product_ownership def wrapped_function(**kwargs): ows_product.verify_product_ownership.assert_not_called() assert kwargs['product_id'] == product_id product_id = 1 mocks.ows_product_verify_product_ownership(mocker) wrapped_function( product_id=1, account_type=None, account_id=None) def test_verify_products_ownership(mocker): """Should check whether each product belongs to given account.""" mock_side_effects = [ response.Response(), response.Response()] mocker.patch.object( logic_utils.ows_product, 'verify_product_ownership', side_effect=mock_side_effects) product_ownership_response = logic_utils.verify_products_ownership( product_ids=[1, 2], account_type='vendor', account_id=100) assert product_ownership_response.status == response_code.OK def test_verify_products_ownership_no_grass_headers(mocker): """Should skip ownership check when no Grass Headers.""" product_ownership_response = logic_utils.verify_products_ownership( product_ids=[1, 2], account_type=None, account_id=None) assert product_ownership_response.status == response_code.OK def test_verify_products_ownership_error_case(mocker): """Should return error response if any product not owned by account.""" mock_side_effects = [ response.Response(), response.create_error_response('error', 'wasted')] mocker.patch.object( logic_utils.ows_product, 'verify_product_ownership', side_effect=mock_side_effects) product_ownership_response = logic_utils.verify_products_ownership( [1, 2], account_type='vendor', account_id=100) assert product_ownership_response.status == response_code.BAD_REQUEST @db.test_schema def test_save_log_for_created_tracks(): """Should save log entries for created tracks to DB.""" date_created = datetime.datetime.utcnow() import_source_dest_list = [ { 'source': {'tuid': 100, 'product_id': 110}, 'destination': {'tuid': 3, 'product_id': 30} }, { 'source': {'tuid': 200, 'product_id': 220}, 'destination': {'tuid': 4, 'product_id': 40} } ] expected_log_entries = [ { 'record_id': 1, 'source_tuid': 100, 'destination_tuid': 3, 'source_product_id': 110, 'destination_product_id': 30, 'orchard_user_id': 'alw:123', 'action': 'import', 'created_date': date_created.strftime('%Y-%m-%d %H:%M'), 'audit_metadata': None }, { 'record_id': 2, 'source_tuid': 200, 'destination_tuid': 4, 'source_product_id': 220, 'destination_product_id': 40, 'orchard_user_id': 'alw:123', 'action': 'import', 'created_date': date_created.strftime('%Y-%m-%d %H:%M'), 'audit_metadata': None } ] log_response = logic_utils.add_tracks_log_entries( import_source_dest_list, orchard_user_id='alw:123', action='import') for entry in log_response.message: entry['created_date'] = ( entry['created_date'].strftime('%Y-%m-%d %H:%M')) with mysql.ows_track_db_session() as read_session: created_entries = read_session.query(track_audit.TrackAudit).all() created_entries = [entry.to_dict() for entry in created_entries] for entry in created_entries: entry['created_date'] = ( entry['created_date'].strftime('%Y-%m-%d %H:%M')) assert log_response assert isinstance(log_response, response.Response) assert log_response.message == expected_log_entries assert created_entries == expected_log_entries def test_is_in_correction_mode(): """Verify product is in correction mode if release status is ec.""" product_data = { product_consts.RELEASE_STATUS: product_consts.IS_ERR_CORR_MODE, product_consts.IS_ERR_CORR_ACTN_REQ_MODE: False} is_correction_mode = \ product_utils.is_in_correction_mode(product_data) assert is_correction_mode is True def test_is_digital_product_not_in_correction_mode(): """Verify product is not correction mode if release status not ec.""" product_data = { product_consts.RELEASE_STATUS: test_consts.TEST_RELEASE_STATUS, product_consts.IS_ERR_CORR_ACTN_REQ_MODE: False} is_correction_mode = \ product_utils.is_in_correction_mode(product_data) assert is_correction_mode is False def test_is_digital_product_correction_mode_when_comb_mode(): """Verify product is in correction mode if in combo mode.""" product_data = { product_consts.RELEASE_STATUS: test_consts.TEST_RELEASE_STATUS, product_consts.IS_ERR_CORR_ACTN_REQ_MODE: True} is_correction_mode = \ product_utils.is_in_correction_mode(product_data) assert is_correction_mode is True TEST_VERIFY_AND_UPDATE_FOCUS_TRACK_VALID = [ ( datetime.date.fromisoformat('2022-10-15'), datetime.date.fromisoformat('2022-10-20'), { 'focus_track_start_date': datetime.date.fromisoformat('2022-10-15'), 'focus_track_end_date': datetime.date.fromisoformat('2022-10-20') } ), ( datetime.date.fromisoformat('2022-10-26'), datetime.date.fromisoformat('2022-11-06'), { 'focus_track_start_date': datetime.date.fromisoformat('2022-10-26'), 'focus_track_end_date': datetime.date.fromisoformat('2022-11-06'), } ), ( datetime.date.fromisoformat('2022-11-20'), datetime.date.fromisoformat('2022-11-28'), { 'focus_track_start_date': datetime.date.fromisoformat('2022-11-20'), 'focus_track_end_date': datetime.date.fromisoformat('2022-11-28'), } ), ( datetime.date.fromisoformat('2022-10-15'), None, { 'focus_track_start_date': datetime.date.fromisoformat('2022-10-15'), 'focus_track_end_date': datetime.date.fromisoformat('2022-10-20'), } ), ( datetime.date.fromisoformat('2022-10-26'), None, { 'focus_track_start_date': datetime.date.fromisoformat('2022-10-26'), 'focus_track_end_date': datetime.date.fromisoformat('2022-11-13'), } ), ( datetime.date.fromisoformat('2022-11-20'), None, { 'focus_track_start_date': datetime.date.fromisoformat('2022-11-20'), 'focus_track_end_date': None, } ) ] @db.test_schema_no_seed @pytest.mark.parametrize( ( 'start_date', 'end_date', 'expected_data' ), TEST_VERIFY_AND_UPDATE_FOCUS_TRACK_VALID ) def test_verify_and_update_focus_track_dates_valid( start_date, end_date, expected_data, mocker, track_factory ): """Checks to see if a new start and end time overlaps with other focus tracks.""" tracks = [ track_factory( focus_track=1, focus_track__start_date=datetime.date.fromisoformat('2022-10-21'), focus_track__end_date=datetime.date.fromisoformat('2022-10-25')), track_factory( focus_track=1, focus_track__start_date=datetime.date.fromisoformat('2022-11-14'), focus_track__end_date=None), ] db.merge_model_objects(tracks) fake_data = { 'focus_track_start_date': start_date, 'focus_track_end_date': end_date } mocks.track_persister.get_all_focus_track_by_product_id(mocker, tracks) result = logic_utils.verify_and_update_focus_track_dates( tuid=201, product_id=1, data=fake_data) assert result == expected_data TEST_VERIFY_AND_UPDATE_FOCUS_TRACK_INVALID = [ ( datetime.date.fromisoformat('2022-10-15'), datetime.date.fromisoformat('2022-10-21') ), ( datetime.date.fromisoformat('2022-10-22'), datetime.date.fromisoformat('2022-10-24') ), ( datetime.date.fromisoformat('2022-10-24'), datetime.date.fromisoformat('2022-10-28') ), ( datetime.date.fromisoformat('2022-10-23'), datetime.date.fromisoformat('2022-11-16') ), ( datetime.date.fromisoformat('2022-10-15'), datetime.date.fromisoformat('2022-11-16') ) ] @db.test_schema_no_seed @pytest.mark.parametrize( ( 'start_date', 'end_date', ), TEST_VERIFY_AND_UPDATE_FOCUS_TRACK_INVALID ) def test_verify_and_update_focus_track_dates_invalid( start_date, end_date, mocker, track_factory ): """Checks to see if a new start and end time overlaps with other focus track.""" tracks = [ track_factory( focus_track=1, focus_track__start_date=datetime.date.fromisoformat('2022-10-21'), focus_track__end_date=datetime.date.fromisoformat('2022-10-25')), track_factory( focus_track=1, focus_track__start_date=datetime.date.fromisoformat('2022-11-14'), focus_track__end_date=datetime.date.fromisoformat('2022-11-19')), ] db.merge_model_objects(tracks) fake_data = { 'focus_track_start_date': start_date, 'focus_track_end_date': end_date } mocks.track_persister.get_all_focus_track_by_product_id(mocker, tracks) result = logic_utils.verify_and_update_focus_track_dates( tuid=201, product_id=1, data=fake_data) assert not result @db.test_schema_no_seed def test_verify_and_update_focus_track_dates_updates_end_date(mocker, track_factory): """Checks to see if a new start and end time overlaps with other focus track.""" tracks = [ track_factory( focus_track=1, focus_track__start_date=datetime.date.fromisoformat('2022-10-21'), focus_track__end_date=None), ] db.merge_model_objects(tracks) track = tracks[0].to_dict() start_date = datetime.date.fromisoformat('2022-10-27') end_date = None fake_data = { 'focus_track_start_date': start_date, 'focus_track_end_date': end_date } mocks.track_persister.get_all_focus_track_by_product_id(mocker, tracks) result = logic_utils.verify_and_update_focus_track_dates( tuid=201, product_id=1, data=fake_data) with mysql.db_session() as session: track_query = TrackQuery.get_by_tuid(track[tf.TUID], session) assert track_query.tuid == track[tf.TUID] assert track_query.focus_track_end_date == datetime.date.fromisoformat('2022-10-26') assert result['focus_track_end_date'] is None @db.test_schema_no_seed @pytest.mark.parametrize( ( 'req_data', 'expected' ), [ ( { 'focus_track_start_date': '2022-10-13', 'focus_track_end_date': '2022-11-12' }, True ), ( { 'focus_track_start_date': '2022-10-12', 'focus_track_end_date': '2022-11-13' }, True ), ( { 'focus_track_start_date': '2022-10-11', 'focus_track_end_date': '2022-11-13' }, True ), ( { 'focus_track_start_date': '2022-10-12', 'focus_track_end_date': '2022-11-12' }, False ) ] ) def test_focus_track_values_updated( req_data, expected): """Checks to see if passed in focus track data is not the same.""" cur_data = { 'focus_track_start_date': '2022-10-12', 'focus_track_end_date': '2022-11-12' } result = logic_utils.focus_track_values_updated( cur_data, req_data) assert result == expected