# pylint: disable=unused-argument,protected-access,invalid-name from datetime import datetime import pytest import smart_open from db_schema.schemas.slz import Report from delphi_slz_admin.services.config import ReprocessingConfig from delphi_slz_admin.services.entities import TimeslotDateTimeRange from delphi_slz_admin.services.reprocessing import ExplorationReprocessingService @pytest.mark.parametrize( 'uow_code, context, data_source_id, expected', [ ( 'gras-20200427-sme-dim_track_participant-v1', 'dim_track_participant', 9, 'dim_track_participant' ), ('amazonmusicunlimited-20200912-theorchard-activity-v1', 'ORED_ROW_NA', 5, 'ORED_ROW_NA'), ('apple-20200912-sme-amArtistDemographics-v1_0', '4632::85625474', 1, '85625474'), ( 'youtubereporting-20200101-sme-content_owner_asset_traffic_source-a2', 'content_owner_asset_traffic_source_a2', 7, 'content_owner_asset_traffic_source' ), ] ) def test_get_filename_part( db, clean_db, content_status, uow_code, context, data_source_id, expected ): report_type = uow_code.split('-')[3] report = db.query(Report).filter( Report.report_name == report_type, Report.data_source_id == data_source_id, ).first() content_status.unit_of_work.unit_of_work_code = uow_code content_status.context = context content_status.unit_of_work.report = report db.flush() service = ExplorationReprocessingService() actual = service._get_filename_part(content_status) assert actual == expected @pytest.mark.parametrize( 'uow_code, path_to_file, file_part, timeslot, initial_fnames, expexted_fnames', [( 'gras-20200427-sme-dim_track_participant-v1', 'gras/dim_track_participant/v1/report_date=2020-04-27/report_licensor=sme', 'dim_track_participant', None, ['dim_track_participant_1', 'dim_track_participant_20200427'], ['dim_track_participant_20200427'], ), ( 'amazonmusicunlimited-20200912-theorchard-activity-v1', 'amazonmusicunlimited/activity/v1/report_date=2020-09-12/report_licensor=theorchard', 'ORED_ROW_NA', None, ['ORED_ROW_NA_1', 'ORED_ROW_NA_20200912'], ['ORED_ROW_NA_20200912'], ), ( 'amazonmusicunlimited-20200912-theorchard-activity-v1', 'amazonmusicunlimited/activity/v1/report_date=2020-09-12/report_licensor=theorchard', 'ORED_ROW_NA', TimeslotDateTimeRange(lower=datetime(2020, 9, 12, 12), upper=datetime(2020, 9, 12, 14)), ['ORED_ROW_NA_20200912_100000_120000', 'ORED_ROW_NA_20200912_120000_140000'], ['ORED_ROW_NA_20200912_120000_140000'], ), ( 'apple-20200912-sme-amArtistDemographics-v1_0', 'apple/artistdemographics/v1_0/report_date=2020-09-12/report_licensor=sme', '85625474', None, ['85625474_1_1(1)', '85625474_1_1(2)'], ['85625474_1_1(1)', '85625474_1_1(2)'], )] ) def test_find_keys( s3, uow_code, path_to_file, file_part, timeslot, initial_fnames, expexted_fnames ): bucket = 'bucket_archive' # prepare several_test_files for part in ['test', 'foo', 'bar', *initial_fnames]: initial_s3_path = f's3://{bucket}/{path_to_file}/filename_{part}.txt' with smart_open.open(initial_s3_path, 'wb') as fout: fout.write(b'some test data') service = ExplorationReprocessingService() expected = ','.join([ f's3://{bucket}/{path_to_file}/filename_{fname}.txt' for fname in expexted_fnames ]) actual = service._find_keys( bucket=bucket, uow_code=uow_code, filename_part=file_part, timeslot=timeslot, ) assert actual == expected @pytest.mark.parametrize( 's3_keys, uow_code, file_part, timeslot, expected', [ ([ 's3://decompr/report_date=2021-04-05/report_licensor=sme/BR_FiltrStore_20210405.json', 's3://decompr/report_date=2021-04-05/report_licensor=sme/BR_20210405.json', ], 'appreciationengine-20210405-sme-membersextended-v1', 'BR', None, [ 's3://decompr/report_date=2021-04-05/report_licensor=sme/BR_20210405.json', ]), ([ 's3://decompr/report_date=2021-04-05/report_licensor=sme/BR_FiltrStore_20210405.json', 's3://decompr/report_date=2021-04-05/report_licensor=sme/BR_20210405.json', ], 'appreciationengine-20210405-sme-membersextended-v1', 'BR_FiltrStore', None, [ 's3://decompr/report_date=2021-04-05/report_licensor=sme/BR_FiltrStore_20210405.json', ]), ([ 's3://decompr/report_date=2021-04-05/report_licensor=sme/CA_20210405.json', 's3://decompr/report_date=2021-04-05/report_licensor=sme/US_RCA_20210405.json', 's3://decompr/report_date=2021-04-05/report_licensor=sme/US_ProvidentRCA_20210405.json', ], 'appreciationengine-20210405-sme-membersextended-v1', 'CA', None, [ 's3://decompr/report_date=2021-04-05/report_licensor=sme/CA_20210405.json', ]), ([ 's3://decompr/report_date=2021-04-05/report_licensor=sme/US_20210405.json', 's3://decompr/report_date=2021-04-05/report_licensor=sme/US_RCA_20210405.json', 's3://decompr/report_date=2021-04-05/report_licensor=sme/US_ProvidentRCA_20210405.json', 's3://decompr/report_date=2021-04-05/report_licensor=sme/US_Legacy_20210405.json', ], 'appreciationengine-20210405-sme-membersextended-v1', 'US', None, [ 's3://decompr/report_date=2021-04-05/report_licensor=sme/US_20210405.json', ]), ([ 's3://decompr/report_date=2021-04-05/report_licensor=sme/US_RCA_20210405.json', ], 'appreciationengine-20210405-sme-membersextended-v1', 'US_RCA', None, [ 's3://decompr/report_date=2021-04-05/report_licensor=sme/US_RCA_20210405.json', ]), ([ 's3://decompr/report_date=2021-04-05/report_licensor=sme/' 'US_RCA_20210405_120000_140000.json', 's3://decompr/report_date=2021-04-05/report_licensor=sme/' 'US_RCA_20210405_100000_120000.json', 's3://decompr/report_date=2021-04-05/report_licensor=sme/' 'US_RCA_20210405.json', ], 'appreciationengine-20210405-sme-membersextended-v1', 'US_RCA', TimeslotDateTimeRange(lower=datetime(2020, 9, 12, 12), upper=datetime(2020, 9, 12, 14)), [ 's3://decompr/report_date=2021-04-05/report_licensor=sme/' 'US_RCA_20210405_120000_140000.json', ]), ([ 's3://decompr/report_date=2021-04-05/report_licensor=sme/' 'CA_20210405_120000_140000.json', 's3://decompr/report_date=2021-04-05/report_licensor=sme/' 'US_RCA_20210405_120000_140000.json', 's3://decompr/report_date=2021-04-05/report_licensor=sme/' 'US_RCA_20210405_100000_120000.json', 's3://decompr/report_date=2021-04-05/report_licensor=sme/' 'CA_20210405.json', ], 'appreciationengine-20210405-sme-membersextended-v1', 'CA', TimeslotDateTimeRange(lower=datetime(2020, 9, 12, 12), upper=datetime(2020, 9, 12, 14)), [ 's3://decompr/report_date=2021-04-05/report_licensor=sme/' 'CA_20210405_120000_140000.json', ]), ] ) def test_regexp_key_check(s3_keys, uow_code, file_part, timeslot, expected): service = ExplorationReprocessingService() actual = service._regexp_key_check( s3_keys=s3_keys, uow_code=uow_code, filename_part=file_part, timeslot=timeslot, ) assert actual == expected def test_reprocess_cs_ok(content_status, sqs, sqs_queue_arn, s3): config = ReprocessingConfig( slz_bucket_archive='bucket_archive', slz_bucket_decompressed='bucket_decompressed', sqs_exp_split_notification_queue=sqs_queue_arn['arn-1'] ) key = 'gras/dim_track_participant/v1/report_date=2020-04-27/report_licensor=sme' paths = [ f's3://{config.slz_bucket_archive}/{key}/{content_status.context}.txt.gz', f's3://{config.slz_bucket_decompressed}/{key}/{content_status.context}.txt' ] for path in paths: with smart_open.open(path, 'wb') as fout: fout.write(b'some test data') service = ExplorationReprocessingService() result = service.reprocess_cs(content_statuses=[ content_status, ], config=config) assert not result assert 'Messages' in sqs.receive_message(QueueUrl=sqs_queue_arn['arn-1']) def test_reprocess_cs_no_files(content_status, sqs, sqs_queue_arn, s3): config = ReprocessingConfig( slz_bucket_archive='bucket_archive', slz_bucket_decompressed='bucket_decompressed', sqs_exp_split_notification_queue=sqs_queue_arn['arn-1'] ) service = ExplorationReprocessingService() result = service.reprocess_cs(content_statuses=[ content_status, ], config=config) assert 'File is absent in decompressed bucket' in result.values() assert 'Messages' not in sqs.receive_message(QueueUrl=sqs_queue_arn['arn-1'])