"""Test handler.""" import os from unittest.mock import ANY, call, MagicMock, patch from ddex_ingester_common.constants import \ validation_rule_error_codes from ddex_ingester_common.constants.catalog_ingestion import REJECT from ddex_ingester_common.constants.send_acknowledgement import \ SME_ERROR_STATUS, SME_SUCCESS_STATUS from ddex_ingester_common.constants.xmltodict import DDEX_LIST_FIELDS from ddex_ingester_common.models.purged_ddex.product import Product from ddex_ingester_common.schemas.ddex_schema import DDEXSchema from ddex_ingester_common.schemas.purged_ddex_schema import PurgedDDEXSchema from ddex_ingester_common.schemas.s3_schema import S3Schema from ddex_ingester_common.validation.rules import RuleResult from freezegun import freeze_time from index import _get_ingestion_dates, get_earliest_start_time, handler import pytest import xmltodict DATA_DIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'data', ) @pytest.fixture def patched_modules(): """Patch the modules used in the handler.""" with ( patch('index.read_ddex') as mock_read_ddex, patch('index.prep_s3_context') as mock_prep_s3_context, patch( 'index.validate_and_save_results') as mock_validate_and_save_results, patch('index.send_acknowledgement') as mock_send_acknowledgement, patch('index.get_ids_from_ddex') as mock_get_ids_from_ddex, patch('index.S3Schema') as mock_s3_schema, ): yield { 'read_ddex': mock_read_ddex, 'prep_s3_context': mock_prep_s3_context, 'validate_and_save_results': mock_validate_and_save_results, 'send_acknowledgement': mock_send_acknowledgement, 'get_ids_from_ddex': mock_get_ids_from_ddex, 'S3Schema': mock_s3_schema, } def parse_xml_to_dict(xml): """Parse given XML file using xmltodict.parse().""" with open(os.path.join(DATA_DIR, xml), encoding='utf8') as fd: return xmltodict.parse(fd.read(), force_list=DDEX_LIST_FIELDS) @patch('index.run_rds_query') @patch('config.secrets_manager_client') @patch('index.uuid.uuid4', return_value='mock_uuid') def test_handler(mock_uuid, mock_secrets, mock_run_rds_query, event): """Test the main handler.""" output = handler(event, None) assert output == event @patch('index.uuid.uuid4', return_value='uuid') @patch('config.secrets_manager_client') def test_handler_none_xml_event(mock_secrets, mock_uuid, event): """Test the main handler with a none xml event key.""" event['detail']['object']['key'] = 'not-ddex.flac' assert not handler(event, None) @patch('index.uuid.uuid4', return_value='uuid') @patch('config.secrets_manager_client') def test_handler_xml_in_resources_path(mock_secrets, mock_uuid, event): """Test the main handler with an xml inside resources path.""" event['detail']['object']['key'] = ( 'sme_ddex/resources/extended-metadata.xml' ) assert not handler(event, None) @pytest.mark.parametrize( 'start_time_attribute, is_pre_order_deal, expected_result', [ ('start_date', False, '1999-01-14'), ('start_date', True, '2000-05-01'), ('start_date_time', False, '1999-01-12T00:00:00'), ('start_date_time', True, '2013-01-13T23:11:11'), ] ) def test_get_earliest_start_time(start_time_attribute, is_pre_order_deal, expected_result): """Test earliest start time calculations.""" xml = parse_xml_to_dict('deal_terms_template.xml') deal_list = { 'ReleaseDeal': [ {'DealReleaseReference': ['R1'], 'Deal': [ {'DealTerms': {'IsPreOrderDeal': 'false', 'ValidityPeriod': { 'StartDate': '2013-01-14'}}}]}, {'DealReleaseReference': ['R2'], 'Deal': [ {'DealTerms': {'IsPreOrderDeal': 'false', 'ValidityPeriod': {'StartDate': '2013-01-15'}}}, {'DealTerms': {'IsPreOrderDeal': 'false', 'ValidityPeriod': {'StartDate': '2025-02-13'}}}, {'DealTerms': {'IsPreOrderDeal': 'false', 'ValidityPeriod': {'StartDate': '2013-01-17'}}}, {'DealTerms': {'IsPreOrderDeal': 'false', 'ValidityPeriod': { 'StartDate': '1999-01-14'}}}]}, {'DealReleaseReference': ['R3'], 'Deal': [ {'DealTerms': {'IsPreOrderDeal': 'false', 'ValidityPeriod': {'StartDate': '2013-01-15'}}}, {'DealTerms': {'IsPreOrderDeal': 'false', 'ValidityPeriod': { 'StartDateTime': '2025-02-13T00:00:00Z'}}}, {'DealTerms': {'IsPreOrderDeal': 'false', 'ValidityPeriod': { 'StartDateTime': '2013-01-17T00:00:00'}}}, {'DealTerms': {'IsPreOrderDeal': 'false', 'ValidityPeriod': { 'StartDateTime': '1999-01-12T00:00:00'}}}]}, {'DealReleaseReference': ['R4'], 'Deal': [ {'DealTerms': {'IsPreOrderDeal': 'true', 'ValidityPeriod': { 'StartDate': '2013-01-14'}}}]}, {'DealReleaseReference': ['R5'], 'Deal': [ {'DealTerms': {'IsPreOrderDeal': 'true', 'ValidityPeriod': {'StartDate': '2013-01-15'}}}, {'DealTerms': {'IsPreOrderDeal': 'true', 'ValidityPeriod': {'StartDate': '2055-02-13'}}}, {'DealTerms': {'IsPreOrderDeal': 'true', 'ValidityPeriod': {'StartDate': '2073-01-17'}}}, {'DealTerms': {'IsPreOrderDeal': 'true', 'ValidityPeriod': { 'StartDate': '2000-05-01'}}}]}, {'DealReleaseReference': ['R6'], 'Deal': [ {'DealTerms': {'IsPreOrderDeal': 'true', 'ValidityPeriod': {'StartDate': '2013-01-15'}}}, {'DealTerms': {'IsPreOrderDeal': 'true', 'ValidityPeriod': { 'StartDateTime': '2023-01-14T00:00:00'}}}, {'DealTerms': {'IsPreOrderDeal': 'true', 'ValidityPeriod': { 'StartDateTime': '2013-01-13T23:11:11Z'}}}, {'DealTerms': {'IsPreOrderDeal': 'true', 'ValidityPeriod': { 'StartDateTime': '2013-01-13T23:11:12'}}}, {'DealTerms': {'IsPreOrderDeal': 'true', 'ValidityPeriod': { 'StartDateTime': '2014-01-14T00:00:00'}}}]}, ]} xml['ernm:NewReleaseMessage']['DealList'] = deal_list data = DDEXSchema().load(xml) earliest_start = get_earliest_start_time(data, start_time_attribute, is_pre_order_deal) assert earliest_start == expected_result @pytest.mark.parametrize( 'start_time_attribute, is_pre_order_deal, expected_result', [ ('start_date', False, '1980-01-15'), ('start_date_time', False, '1999-02-14T00:00:00'), ] ) def test_get_earliest_start_time_none_pre_order_deal(start_time_attribute, is_pre_order_deal, expected_result): """Test earliest start time calculations with None IsPreOrderDeal.""" xml = parse_xml_to_dict('deal_terms_template.xml') deal_list = { 'ReleaseDeal': [ {'DealReleaseReference': ['R1'], 'Deal': [ {'DealTerms': {'IsPreOrderDeal': None, 'ValidityPeriod': {'StartDate': '1980-01-15'}}}, {'DealTerms': {'IsPreOrderDeal': 'false', 'ValidityPeriod': { 'StartDate': '2000-01-14'}}}]}, {'DealReleaseReference': ['R2'], 'Deal': [ {'DealTerms': {'IsPreOrderDeal': 'false', 'ValidityPeriod': { 'StartDateTime': '2013-01-17T00:00:00'}}}, {'DealTerms': {'ValidityPeriod': { 'StartDateTime': '1999-02-14T00:00:00'}}}]}, ]} xml['ernm:NewReleaseMessage']['DealList'] = deal_list data = DDEXSchema().load(xml) earliest_start = get_earliest_start_time(data, start_time_attribute, is_pre_order_deal) assert earliest_start == expected_result @patch('index.TemplateLookup') @patch('index.send_email') @patch('config.REPORT_ERRORS_SME_ANALYTICS_DUMMY_EMAIL', '["test@mail.com"]') def test_send_email_is_called(send_email, TemplateLookup, patched_modules): """Test in case of error send_email_is_called.""" test_event = {'detail': { 'requestParameters': { 'bucketName': 'test_bucket', 'key': 'sme_analytics_ddex/XXXX1123_20250227225937426.xml' } }} patched_modules['read_ddex'].side_effect = Exception( 'failed to read s3 file') patched_modules['get_ids_from_ddex'].return_value = ( 'test_upc', 'test_grid', 'test_catalog_number') handler(test_event, None) send_email.assert_called_with( ['test@mail.com'], ANY, 'Failure to ingest SME Analytics Product test_grid test_upc, s3 location: sme_analytics_ddex/XXXX1123_20250227225937426.xml', []) @patch('index.TemplateLookup') @patch('index.send_email') @patch('index.boto3') @patch('config.REPORT_ERRORS_SME_ANALYTICS_DUMMY_EMAIL', '["test@mail.com"]') def test_send_email_is_called_with_wrong_file_name(boto3, send_email, TemplateLookup, patched_modules): """Test in case of error send_email_is_called.""" test_event = {'detail': { 'requestParameters': { 'bucketName': 'test_bucket', 'key': 'sme_analytics_ddex/wrong_name.xml' } }} doc = MagicMock() extended_doc = MagicMock() patched_modules['read_ddex'] = doc, extended_doc handler(test_event, None) send_email.assert_called_with( ['test@mail.com'], ANY, 'Failure to ingest SME Analytics Product s3 location: sme_analytics_ddex/wrong_name.xml', []) @patch('index.run_rds_query') @patch('index._get_ingestion_dates') def test_send_send_success_acknowledgement_called( _get_ingestion_dates, run_rds_query, patched_modules): """Test send_acknowledgement_called after processing.""" test_event = {'detail': { 'requestParameters': { 'bucketName': 'test_bucket', 'key': 'sme_analytics_ddex/XXXX1123_20250227225937426.xml' } }} patched_modules['S3Schema'].dump().return_value = MagicMock() patched_modules['S3Schema'].load().return_value = MagicMock() patched_modules['get_ids_from_ddex'].return_value = ( 'test_upc', 'test_grid', 'test_catalog_number') doc = MagicMock() doc.is_purged_release = False patched_modules['read_ddex'].return_value = doc handler(test_event, None) patched_modules['send_acknowledgement'].assert_called_with('test_grid', SME_SUCCESS_STATUS) @patch('index.run_rds_query') @patch('index.send_error_notification') @patch('index._get_ingestion_dates') def test_validation_failed_scenario( _get_ingestion_dates, send_error_notification, run_rds_query, patched_modules): """Test send_acknowledgement_called after processing.""" patched_modules['S3Schema'].dump().return_value = MagicMock() patched_modules['S3Schema'].load().return_value = MagicMock() test_event = {'detail': { 'requestParameters': { 'bucketName': 'test_bucket', 'key': 'sme_analytics_ddex/XXXX1123_20250227225937426.xml' } }} patched_modules['get_ids_from_ddex'].return_value = ( 'test_upc', 'test_grid', 'test_catalog_number') doc = MagicMock() doc.is_purged_release = False patched_modules['validate_and_save_results'].return_value = [ RuleResult( validation_rule_error_codes.MISSING_TRACK_NAME, REJECT, 'Track Name is missing' )] patched_modules['read_ddex'].return_value = doc handler(test_event, None) patched_modules['send_acknowledgement'].assert_called_with('test_grid', SME_ERROR_STATUS) send_error_notification.assert_called_with( 'test_bucket', 'sme_analytics_ddex/XXXX1123_20250227225937426.xml', 'SME_ANALYTICS_PROVIDER', None, ['Track Name is missing']) @patch('index._store_ddex_delivery') @freeze_time('2012-01-14') @pytest.mark.parametrize('is_confidential', [True, False]) def test_multiple_ddex_deliveries_created( mock_store_ddex_delivery, patched_modules, s3_preorder_tracks_ddex, is_confidential): """Test store_ddex_delivery with duplicated and None dates.""" bucket = 'test_bucket' key = 'sme_analytics_ddex/XXXX1123_20250227225937426.xml' test_event = {'detail': { 'requestParameters': { 'bucketName': bucket, 'key': key } }} patched_modules['S3Schema'].dump.return_value = MagicMock() s3_data = S3Schema().load(S3Schema().dump(s3_preorder_tracks_ddex)) s3_data.deals[0].release_references = ['R0'] s3_data.deals[0].deal_terms[0].start_date = None s3_data.deals[0].deal_terms[0].start_date_time = '2020-03-27T12:12:12' s3_data.tracks[0].original_release_date = '2025-01-01' s3_data.tracks[1].original_release_date = '2025-01-03' s3_data.tracks[2].original_release_date = '2025-01-05' s3_data.tracks[3].original_release_date = '2025-01-03' s3_data.tracks[4].original_release_date = '2025-01-07' s3_data.tracks[5].original_release_date = None s3_data.tracks[6].original_release_date = '2025-01-12' patched_modules['S3Schema']().load.return_value = s3_data doc = MagicMock() doc.is_purged_release = False doc.product.is_confidential = is_confidential patched_modules['get_ids_from_ddex'].return_value = ( 'test_upc', 'test_grid', 'test_catalog_number') patched_modules['read_ddex'].return_value = doc handler(test_event, None) mock_store_ddex_delivery.assert_has_calls([ call(5, bucket, key, is_confidential, True, None, None, None, None, 'DELIVERED', 'test_upc', 'test_grid', '2020-03-27 12:12:12'), call(5, bucket, key, is_confidential, True, None, None, None, None, 'DELIVERED', 'test_upc', 'test_grid', '2025-01-01 00:00:00'), call(5, bucket, key, is_confidential, True, None, None, None, None, 'DELIVERED', 'test_upc', 'test_grid', '2025-01-03 00:00:00'), call(5, bucket, key, is_confidential, True, None, None, None, None, 'DELIVERED', 'test_upc', 'test_grid', '2025-01-05 00:00:00'), call(5, bucket, key, is_confidential, True, None, None, None, None, 'DELIVERED', 'test_upc', 'test_grid', '2025-01-07 00:00:00'), call(5, bucket, key, is_confidential, True, None, None, None, None, 'DELIVERED', 'test_upc', 'test_grid', '2025-01-12 00:00:00') ], any_order=False) @patch('index._store_ddex_delivery') @pytest.mark.parametrize( # '2021-11-01' '2021-11-06' '2021-11-10' '2021-11-15' '2021-11-25' '2021-11-27' '2021-11-29' - track releases 'now, r0_start_date, r0_start_date_time, original_release_dates_inserted', [ ('2021-11-05', '2021-11-04', None, ['2021-11-04 00:00:00', '2021-11-06 00:00:00', '2021-11-10 00:00:00', '2021-11-15 00:00:00', '2021-11-25 00:00:00', '2021-11-27 00:00:00', '2021-11-29 00:00:00']), ('2021-11-17', '2021-11-04', None, ['2021-11-04 00:00:00', '2021-11-25 00:00:00', '2021-11-27 00:00:00', '2021-11-29 00:00:00']), ('2021-11-08', '2021-11-10', '2021-11-09T10:10:10', ['2021-11-09 10:10:10', '2021-11-10 00:00:00', '2021-11-15 00:00:00', '2021-11-25 00:00:00', '2021-11-27 00:00:00', '2021-11-29 00:00:00']), ('2021-11-09 10:10:11', '2021-11-10', '2021-11-09T10:10:10', ['2021-11-09 10:10:10', '2021-11-10 00:00:00', '2021-11-15 00:00:00', '2021-11-25 00:00:00', '2021-11-27 00:00:00', '2021-11-29 00:00:00']), ('2021-11-09 10:10:09', '2021-11-10', '2021-11-09T10:10:10', ['2021-11-09 10:10:10', '2021-11-10 00:00:00', '2021-11-15 00:00:00', '2021-11-25 00:00:00', '2021-11-27 00:00:00', '2021-11-29 00:00:00']), ('2021-11-14', '2021-11-14', '2021-11-13T23:10:10', ['2021-11-13 23:10:10', '2021-11-15 00:00:00', '2021-11-25 00:00:00', '2021-11-27 00:00:00', '2021-11-29 00:00:00']), ('2050-07-05', '2021-11-04', None, ['2021-11-04 00:00:00']), ('2000-01-01', '2021-11-14', None, ['2021-11-14 00:00:00', '2021-11-15 00:00:00', '2021-11-25 00:00:00', '2021-11-27 00:00:00', '2021-11-29 00:00:00']) ] ) def test_multiple_ddex_deliveries_created_for_confidential( mock_store_ddex_delivery, patched_modules, s3_preorder_tracks_ddex, now, r0_start_date, r0_start_date_time, original_release_dates_inserted): """Test store_ddex_delivery for confidential product.""" bucket = 'test_bucket' key = 'sme_analytics_ddex/XXXX1123_20250227225937426.xml' test_event = {'detail': { 'requestParameters': { 'bucketName': bucket, 'key': key } }} patched_modules['S3Schema'].dump.return_value = MagicMock() s3_data = S3Schema().load(S3Schema().dump(s3_preorder_tracks_ddex)) s3_data.deals[0].release_references = ['R0'] s3_data.deals[0].deal_terms[0].start_date = r0_start_date s3_data.deals[0].deal_terms[0].start_date_time = r0_start_date_time s3_data.tracks[0].original_release_date = '2021-11-01' s3_data.tracks[1].original_release_date = '2021-11-06' s3_data.tracks[2].original_release_date = '2021-11-10' s3_data.tracks[3].original_release_date = '2021-11-15' s3_data.tracks[4].original_release_date = '2021-11-25' s3_data.tracks[5].original_release_date = '2021-11-27' s3_data.tracks[6].original_release_date = '2021-11-29' patched_modules['S3Schema']().load.return_value = s3_data doc = MagicMock() doc.is_purged_release = False doc.product.is_confidential = True patched_modules['get_ids_from_ddex'].return_value = ( 'test_upc', 'test_grid', 'test_catalog_number') patched_modules['read_ddex'].return_value = doc with freeze_time(now): handler(test_event, None) dates_inserted = [call.args[12] for call in mock_store_ddex_delivery.call_args_list] assert original_release_dates_inserted == dates_inserted @patch('index._store_ddex_delivery') @patch('index.read_ddex') @patch('index.is_ccm_som_livre_grps_ddex_ingestion_enabled') @patch('index.validate_and_save_results') @patch('index.send_acknowledgement') @patch('index.get_ids_from_ddex') @freeze_time('2025-05-10') def test_multiple_ddex_deliveries_created_on_real_confidential_file( mock_get_ids_from_ddex, mock_send_acknowledgement, mock_validate_and_save_results, is_ccm_som_livre_grps_ddex_ingestion_enabled, mock_read_ddex, mock_store_ddex_delivery, ddex_confidential_example): """Test store_ddex_delivery for confidential product using real xml.""" bucket = 'test_bucket' key = 'sme_analytics_ddex/XXXX1123_20250227225937426.xml' test_event = {'detail': { 'requestParameters': { 'bucketName': bucket, 'key': key } }} is_ccm_som_livre_grps_ddex_ingestion_enabled.return_value = False doc = DDEXSchema().load(ddex_confidential_example) doc.is_purged_release = False doc.product.is_confidential = True doc.project = MagicMock() mock_get_ids_from_ddex.return_value = ( 'test_upc', 'test_grid', 'test_catalog_number') mock_read_ddex.return_value = doc handler(test_event, None) mock_store_ddex_delivery.assert_has_calls([ call(5, bucket, key, True, False, '2025-04-30T00:00:00', None, None, None, 'DELIVERED', 'test_upc', 'test_grid', '2025-04-30 00:00:00'), call(5, bucket, key, True, False, '2025-04-30T00:00:00', None, None, None, 'DELIVERED', 'test_upc', 'test_grid', '2025-05-16 00:00:00', ), call(5, bucket, key, True, False, '2025-04-30T00:00:00', None, None, None, 'DELIVERED', 'test_upc', 'test_grid', '2025-06-27 00:00:00', ) ], any_order=False) @patch('index._store_purged_ddex_delivery') @patch('index.read_ddex') @patch('index.send_acknowledgement') @freeze_time('2025-05-10') def test_purged_ddex_created( mock_send_acknowledgement, mock_read_ddex, mock_store_ddex_delivery): """Test store_ddex_delivery for confidential product using real xml.""" bucket = 'test_bucket' key = 'sme_analytics_ddex/XXXX1123_20250227225937426.xml' test_event = {'detail': { 'requestParameters': { 'bucketName': bucket, 'key': key } }} doc = PurgedDDEXSchema() doc.is_purged_release = True product = Product() product.upc = 'test_upc_123' doc.product = product mock_read_ddex.return_value = doc handler(test_event, None) mock_store_ddex_delivery.assert_called_with( 5, 'test_bucket', 'sme_analytics_ddex/XXXX1123_20250227225937426.xml', 'XXXX1123_20250227225937426', 'test_upc_123' ) mock_send_acknowledgement.assert_called_with('XXXX1123_20250227225937426', SME_SUCCESS_STATUS) @pytest.mark.parametrize('is_confidential, result', [ (True, ['2050-01-01 00:00:00']), (False, ['2021-11-10 00:00:00', '2021-11-15 00:00:00', '2021-11-25 00:00:00', '2021-11-27 00:00:00', '2021-11-29 00:00:00', '2050-01-01 00:00:00']) ]) def test_get_ingestion_dates(is_confidential, result, s3_preorder_tracks_ddex): """Test date filtered by deals['R0'].start_date only for confidential.""" s3_data = S3Schema().load(S3Schema().dump(s3_preorder_tracks_ddex)) s3_data.deals[0].release_references = ['R0'] s3_data.deals[0].deal_terms[0].start_date = '2050-01-01' s3_data.deals[0].deal_terms[0].start_date_time = None s3_data.tracks[0].original_release_date = '2021-11-01' s3_data.tracks[1].original_release_date = '2021-11-06' s3_data.tracks[2].original_release_date = '2021-11-10' s3_data.tracks[3].original_release_date = '2021-11-15' s3_data.tracks[4].original_release_date = '2021-11-25' s3_data.tracks[5].original_release_date = '2021-11-27' s3_data.tracks[6].original_release_date = '2021-11-29' with freeze_time('2021-11-15 10:00:00'): result_dates = _get_ingestion_dates(s3_data, is_confidential) result_dates_str = [date.strftime('%Y-%m-%d %H:%M:%S') for date in result_dates] assert result_dates_str == result @pytest.mark.parametrize('is_confidential, result', [ (True, ['2025-01-01 00:00:00']), (False, ['2021-11-29 00:00:00', '2025-01-01 00:00:00']) ]) def test_all_release_dates_are_before_now_and_r0(is_confidential, result, s3_preorder_tracks_ddex): """Test that for non confidential artificial event is created.""" s3_data = S3Schema().load(S3Schema().dump(s3_preorder_tracks_ddex)) s3_data.deals[0].release_references = ['R0'] s3_data.deals[0].deal_terms[0].start_date = '2025-01-01' s3_data.deals[0].deal_terms[0].start_date_time = None s3_data.tracks[0].original_release_date = '2021-11-01' s3_data.tracks[1].original_release_date = '2021-11-06' s3_data.tracks[2].original_release_date = '2021-11-10' s3_data.tracks[3].original_release_date = '2021-11-15' s3_data.tracks[4].original_release_date = '2021-11-25' s3_data.tracks[5].original_release_date = '2021-11-27' s3_data.tracks[6].original_release_date = '2021-11-29' with freeze_time('2023-01-01 12:30:00'): result_dates = _get_ingestion_dates(s3_data, is_confidential) result_dates_str = [date.strftime('%Y-%m-%d %H:%M:%S') for date in result_dates] assert result_dates_str == result @pytest.mark.parametrize('is_confidential, result', [ # dates >= r0.date (True, ['2030-11-21 00:00:00', '2030-11-25 00:00:00', '2030-11-27 00:00:00', '2030-11-29 00:00:00']), # dates disregarding r0.date (False, ['2030-11-01 00:00:00', '2030-11-06 00:00:00', '2030-11-10 00:00:00', '2030-11-15 00:00:00', '2030-11-21 00:00:00', '2030-11-25 00:00:00', '2030-11-27 00:00:00', '2030-11-29 00:00:00']) ]) def test_all_release_dates_are_in_future(is_confidential, result, s3_preorder_tracks_ddex): """Test _get_ingestion_dates with all dates in the future.""" s3_data = S3Schema().load(S3Schema().dump(s3_preorder_tracks_ddex)) s3_data.deals[0].release_references = ['R0'] s3_data.deals[0].deal_terms[0].start_date = '2030-11-21' s3_data.deals[0].deal_terms[0].start_date_time = None s3_data.tracks[0].original_release_date = '2030-11-01' s3_data.tracks[1].original_release_date = '2030-11-06' s3_data.tracks[2].original_release_date = '2030-11-10' s3_data.tracks[3].original_release_date = '2030-11-15' s3_data.tracks[4].original_release_date = '2030-11-25' s3_data.tracks[5].original_release_date = '2030-11-27' s3_data.tracks[6].original_release_date = '2030-11-29' with freeze_time('2029-01-01 12:30:00'): result_dates = _get_ingestion_dates(s3_data, is_confidential) result_dates_str = [date.strftime('%Y-%m-%d %H:%M:%S') for date in result_dates] assert result_dates_str == result @pytest.mark.parametrize('is_confidential, result', [ (True, ['2030-02-02 00:00:00', '2030-02-04 00:00:00', '2030-11-05 00:00:00']), (False, ['2030-02-01 00:00:00', '2030-02-04 00:00:00', '2030-11-05 00:00:00']) ]) def test_r0_in_past(is_confidential, result, s3_preorder_tracks_ddex): """Test _get_ingestion_dates with for r0 in past.""" s3_data = S3Schema().load(S3Schema().dump(s3_preorder_tracks_ddex)) s3_data.deals[0].release_references = ['R0'] s3_data.deals[0].deal_terms[0].start_date = '2030-02-02' s3_data.deals[0].deal_terms[0].start_date_time = None s3_data.tracks[0].original_release_date = '2030-02-01' s3_data.tracks[1].original_release_date = '2030-02-04' s3_data.tracks[2].original_release_date = '2030-11-05' with freeze_time('2030-02-03 12:30:00'): result_dates = _get_ingestion_dates(s3_data, is_confidential) result_dates_str = [date.strftime('%Y-%m-%d %H:%M:%S') for date in result_dates] assert result_dates_str == result @pytest.mark.parametrize('is_confidential , result', [ (True, None), (False, ['2030-02-01 00:00:00', '2030-02-04 00:00:00', '2030-11-05 00:00:00']) ]) def test_r0_is_missing(is_confidential, result, s3_preorder_tracks_ddex): """Test R0 deal is missing.""" s3_data = S3Schema().load(S3Schema().dump(s3_preorder_tracks_ddex)) s3_data.deals[0].release_references = ['R1'] s3_data.deals[0].deal_terms[0].start_date = '2030-02-02' s3_data.deals[0].deal_terms[0].start_date_time = None s3_data.tracks[0].original_release_date = '2030-02-01' s3_data.tracks[1].original_release_date = '2030-02-04' s3_data.tracks[2].original_release_date = '2030-11-05' with freeze_time('2030-02-03 12:30:00'): if is_confidential: with pytest.raises(Exception, match='R0 deal is missing for ' 'confidential product'): _get_ingestion_dates(s3_data, is_confidential) else: result_dates = _get_ingestion_dates(s3_data, is_confidential) result_dates_str = [date.strftime('%Y-%m-%d %H:%M:%S') for date in result_dates] assert result_dates_str == result