"""Unit tests for Meta Daily workflow tasks.""" from unittest.mock import MagicMock, patch from botocore.exceptions import ClientError as BotocoreClientError import pytest from feed_ingestion.flows.meta_daily import tasks from feed_ingestion.tasks import STOP_RESPONSE _DATE = '2026-04-17' _DATE_COMPACT = '20260417' # --------------------------------------------------------------------------- # bootstrap # --------------------------------------------------------------------------- @pytest.mark.parametrize( 'report,expected_table,expected_sme_file,expected_orchard_file', [ ( 'consumption', 'staging_raw_meta_consumption', 'sony_consumption_export_20260417.txt', 'orchard_consumption_export_20260417.txt', ), ( 'production', 'staging_raw_meta_production', 'sony_production_export_20260417.txt', 'orchard_production_export_20260417.txt', ), ], ) @patch('feed_ingestion.flows.meta_daily.tasks.garcon_feed_status') def test_bootstrap_both_sources_pending( mock_status, report, expected_table, expected_sme_file, expected_orchard_file, ): """Both licensor sources are pending when neither has been ingested.""" mock_status.get_overall_status.return_value = None mock_status.STATUS_INGESTED = 'INGESTED' result = tasks.bootstrap(MagicMock(), _DATE, report, reload='False') assert result['date'] == _DATE assert result['report'] == report assert result['report_feed_name'] == f'meta_daily_{report}' assert result['staging_raw_table'] == expected_table assert result['archive_s3_path'] == ( f's3://dev-cucumbers/MetaDaily/archives/{_DATE}/' ) pending = result['pending_sources'] assert len(pending) == 2 sme_source = next(s for s in pending if s['licensor'] == 'sme') assert sme_source['file_name'] == expected_sme_file assert sme_source['licensor_feed'] == f'meta_daily_{report}_sme' assert sme_source['archive_key'] == ( f'MetaDaily/archives/{_DATE}/{expected_sme_file}' ) assert 'source_bucket' in sme_source orchard_source = next(s for s in pending if s['licensor'] == 'theorchard') assert orchard_source['file_name'] == expected_orchard_file assert orchard_source['licensor_feed'] == ( f'meta_daily_{report}_theorchard' ) @patch('feed_ingestion.flows.meta_daily.tasks.garcon_feed_status') def test_bootstrap_sme_already_ingested(mock_status): """SME source is excluded from pending when already ingested.""" mock_status.STATUS_INGESTED = 'INGESTED' def get_overall(feed_name, date): if 'sme' in feed_name: return 'INGESTED' return None mock_status.get_overall_status.side_effect = get_overall result = tasks.bootstrap(MagicMock(), _DATE, 'consumption', reload='False') pending = result['pending_sources'] assert len(pending) == 1 assert pending[0]['licensor'] == 'theorchard' @patch('feed_ingestion.flows.meta_daily.tasks.garcon_feed_status') def test_bootstrap_all_sources_already_ingested(mock_status): """Returns empty pending_sources when all licensor statuses are done.""" mock_status.STATUS_INGESTED = 'INGESTED' def get_overall(feed_name, date): if feed_name == 'meta_daily_consumption': return None # overall not yet set return 'INGESTED' # per-licensor all done mock_status.get_overall_status.side_effect = get_overall result = tasks.bootstrap(MagicMock(), _DATE, 'consumption', reload='False') assert result['pending_sources'] == [] @patch('feed_ingestion.flows.meta_daily.tasks.garcon_feed_status') def test_bootstrap_returns_stop_when_overall_ingested(mock_status): """Returns STOP_RESPONSE when the overall feed status is INGESTED.""" mock_status.STATUS_INGESTED = 'INGESTED' mock_status.get_overall_status.return_value = 'INGESTED' result = tasks.bootstrap(MagicMock(), _DATE, 'consumption', reload='False') assert result == STOP_RESPONSE @patch('feed_ingestion.flows.meta_daily.tasks.garcon_feed_status') def test_bootstrap_reload_deletes_all_statuses(mock_status): """Reload=True deletes per-licensor and overall statuses.""" mock_status.STATUS_INGESTED = 'INGESTED' mock_status.get_overall_status.return_value = None tasks.bootstrap(MagicMock(), _DATE, 'consumption', reload='True') delete_calls = [ c.args[0] for c in mock_status.delete_status.call_args_list ] assert 'meta_daily_consumption_sme' in delete_calls assert 'meta_daily_consumption_theorchard' in delete_calls assert 'meta_daily_consumption' in delete_calls # --------------------------------------------------------------------------- # grab_available_files # --------------------------------------------------------------------------- def test_grab_available_files_empty_pending(): """Returns empty available_sources without stop when pending is empty.""" result = tasks.grab_available_files(MagicMock(), pending_sources=[]) assert result == {'available_sources': []} @patch('feed_ingestion.flows.meta_daily.tasks.boto3') def test_grab_available_files_file_found(mock_boto3): """Returns available_sources with file_size when file found in drop S3.""" file_name = 'sony_consumption_export_20260417.txt' mock_client = MagicMock() mock_boto3.client.return_value = mock_client mock_client.head_object.return_value = {'ContentLength': 12345} pending = [ { 'licensor': 'sme', 'licensor_feed': 'meta_daily_consumption_sme', 'source_bucket': 'sme-ca-prod-partners', 'source_key': ( f'facebook/in/sme/daily/v1/{_DATE_COMPACT}/{file_name}' ), 'archive_key': f'MetaDaily/archives/{_DATE}/{file_name}', 'file_name': file_name, } ] result = tasks.grab_available_files(MagicMock(), pending) mock_client.copy.assert_called_once() assert len(result['available_sources']) == 1 assert result['available_sources'][0]['file_name'] == file_name assert result['available_sources'][0]['file_size'] == 12345 assert 'stop' not in result @patch('feed_ingestion.flows.meta_daily.tasks.boto3') def test_grab_available_files_file_not_found(mock_boto3): """Returns stop=True when all pending files are missing from drop S3.""" file_name = 'sony_consumption_export_20260417.txt' mock_client = MagicMock() mock_boto3.client.return_value = mock_client mock_client.copy.side_effect = BotocoreClientError( {'Error': {'Code': '404', 'Message': 'Not Found'}}, 'CopyObject' ) pending = [ { 'licensor': 'sme', 'licensor_feed': 'meta_daily_consumption_sme', 'source_bucket': 'sme-ca-prod-partners', 'source_key': ( f'facebook/in/sme/daily/v1/{_DATE_COMPACT}/{file_name}' ), 'archive_key': f'MetaDaily/archives/{_DATE}/{file_name}', 'file_name': file_name, } ] result = tasks.grab_available_files(MagicMock(), pending) assert result['stop'] is True assert result['available_sources'] == [] @patch('feed_ingestion.flows.meta_daily.tasks.boto3') def test_grab_available_files_partial_availability(mock_boto3): """Loads available files even when only one of two sources is present.""" sme_file = 'sony_consumption_export_20260417.txt' orchard_file = 'orchard_consumption_export_20260417.txt' mock_client = MagicMock() mock_boto3.client.return_value = mock_client mock_client.head_object.return_value = {'ContentLength': 100} def copy_side_effect(CopySource, Bucket, Key): if 'orchard' in CopySource['Key']: raise BotocoreClientError( {'Error': {'Code': '404', 'Message': 'Not Found'}}, 'CopyObject', ) mock_client.copy.side_effect = copy_side_effect pending = [ { 'licensor': 'sme', 'licensor_feed': 'meta_daily_consumption_sme', 'source_bucket': 'sme-ca-prod-partners', 'source_key': ( f'facebook/in/sme/daily/v1/{_DATE_COMPACT}/{sme_file}' ), 'archive_key': f'MetaDaily/archives/{_DATE}/{sme_file}', 'file_name': sme_file, }, { 'licensor': 'theorchard', 'licensor_feed': 'meta_daily_consumption_theorchard', 'source_bucket': 'sme-ca-prod-partners', 'source_key': ( f'facebook/in/orchard/daily/v1/{_DATE_COMPACT}/{orchard_file}' ), 'archive_key': f'MetaDaily/archives/{_DATE}/{orchard_file}', 'file_name': orchard_file, }, ] result = tasks.grab_available_files(MagicMock(), pending) assert len(result['available_sources']) == 1 assert result['available_sources'][0]['licensor'] == 'sme' assert 'stop' not in result # --------------------------------------------------------------------------- # load_staging_raw # --------------------------------------------------------------------------- def test_load_staging_raw_empty_sources(): """Returns immediately when available_sources is empty.""" result = tasks.load_staging_raw( MagicMock(), available_sources=[], staging_raw_table='t', date=_DATE, report='consumption', archive_s3_path='s3://b/p/', ) assert result is None @patch('feed_ingestion.flows.meta_daily.tasks.set_overall_status_enhanced') @patch('feed_ingestion.flows.meta_daily.tasks.garcon_feed_status') @patch('snowflake_connector.etl_connector.SnowflakeSQLExecutor') @patch('feed_ingestion.flows.helpers.get_sf_config') @patch('feed_ingestion.flows.meta_daily.stage_loader.MetaDailySL') def test_load_staging_raw_single_source( mock_sl_cls, mock_get_sf, mock_sf_exec, mock_status, mock_set_status ): """Loads one source file, cleans rows, and sets per-licensor status.""" mock_status.get_overall_status.return_value = None mock_status.STATUS_INGESTED = 'INGESTED' mock_executor = MagicMock() mock_sf_exec.return_value.__enter__ = MagicMock(return_value=mock_executor) mock_sf_exec.return_value.__exit__ = MagicMock(return_value=False) mock_sl = MagicMock() mock_sl_cls.return_value = mock_sl file_name = 'sony_consumption_export_20260417.txt' available = [ { 'licensor': 'sme', 'licensor_feed': 'meta_daily_consumption_sme', 'file_name': file_name, 'file_size': 12345, } ] tasks.load_staging_raw( MagicMock(), available, staging_raw_table='staging_raw_meta_consumption', date=_DATE, report='consumption', archive_s3_path='s3://dev-cucumbers/MetaDaily/archives/2026-04-17/', ) mock_sl.create_stage.assert_called_once() mock_sl.clean_staging_raw_table.assert_called_once_with( 'staging_raw_meta_consumption', _DATE_COMPACT, licensor='sme' ) mock_sl.load_staging_raw_table.assert_called_once() call_kwargs = mock_sl.load_staging_raw_table.call_args assert call_kwargs[1]['licensor'] == 'sme' assert call_kwargs[1]['activity_date'] == _DATE_COMPACT mock_sl.drop_stage.assert_called_once() mock_set_status.assert_called_once() @patch('feed_ingestion.flows.meta_daily.tasks.set_overall_status_enhanced') @patch('feed_ingestion.flows.meta_daily.tasks.garcon_feed_status') @patch('snowflake_connector.etl_connector.SnowflakeSQLExecutor') @patch('feed_ingestion.flows.helpers.get_sf_config') @patch('feed_ingestion.flows.meta_daily.stage_loader.MetaDailySL') def test_load_staging_raw_skips_already_ingested( mock_sl_cls, mock_get_sf, mock_sf_exec, mock_status, mock_set_status ): """Skips a source whose per-licensor status is already INGESTED.""" mock_status.STATUS_INGESTED = 'INGESTED' mock_status.get_overall_status.return_value = 'INGESTED' mock_executor = MagicMock() mock_sf_exec.return_value.__enter__ = MagicMock(return_value=mock_executor) mock_sf_exec.return_value.__exit__ = MagicMock(return_value=False) mock_sl = MagicMock() mock_sl_cls.return_value = mock_sl available = [ { 'licensor': 'sme', 'licensor_feed': 'meta_daily_consumption_sme', 'file_name': 'sony_consumption_export_20260417.txt', 'file_size': 12345, } ] tasks.load_staging_raw( MagicMock(), available, staging_raw_table='staging_raw_meta_consumption', date=_DATE, report='consumption', archive_s3_path='s3://dev-cucumbers/MetaDaily/archives/2026-04-17/', ) mock_sl.clean_staging_raw_table.assert_not_called() mock_sl.load_staging_raw_table.assert_not_called() mock_set_status.assert_not_called() # --------------------------------------------------------------------------- # set_overall_status_if_complete # --------------------------------------------------------------------------- @patch('feed_ingestion.flows.meta_daily.tasks.set_overall_status_enhanced') @patch('feed_ingestion.flows.meta_daily.tasks.garcon_feed_status') def test_set_overall_status_all_ingested(mock_status, mock_set_status): """Sets overall INGESTED when all licensor sources are done.""" mock_status.STATUS_INGESTED = 'INGESTED' mock_status.get_overall_status.return_value = 'INGESTED' tasks.set_overall_status_if_complete( MagicMock(), 'consumption', 'meta_daily_consumption', _DATE ) assert mock_set_status.call_count == 1 call_args = mock_set_status.call_args[0] assert call_args[0] == 'meta_daily_consumption' assert call_args[1] == _DATE assert call_args[2] == 'INGESTED' @patch('feed_ingestion.flows.meta_daily.tasks.set_overall_status_enhanced') @patch('feed_ingestion.flows.meta_daily.tasks.garcon_feed_status') def test_set_overall_status_not_complete(mock_status, mock_set_status): """Does not set overall status when a licensor is still pending.""" mock_status.STATUS_INGESTED = 'INGESTED' def get_overall(feed_name, date): if 'sme' in feed_name: return 'INGESTED' return None mock_status.get_overall_status.side_effect = get_overall tasks.set_overall_status_if_complete( MagicMock(), 'consumption', 'meta_daily_consumption', _DATE ) mock_set_status.assert_not_called()