"""Unit tests for GfK Streaming tasks.""" import ftplib import io from unittest.mock import MagicMock, patch import zipfile import pytest from feed_ingestion.flows.gfk_streaming import tasks from feed_ingestion.flows.gfk_streaming.tasks import _normalize_csv_quotes _date = '2026-04-30' _filename = 'SONY_DAILY_TOPALL_DE_STREAM_ONLY20260430.zip' _archive_path = ( 'GfK_Streaming/archives/2026-04-30/' 'SONY_DAILY_TOPALL_DE_STREAM_ONLY20260430.zip' ) _processed_path = ( 'GfK_Streaming/processed/2026-04-30/' 'SONY_DAILY_TOPALL_DE_STREAM_ONLY20260430.csv' ) @pytest.fixture def mock_get_overall_status(): """Yield get_overall_status mock.""" with patch( 'feed_ingestion.flows.gfk_streaming.tasks.' 'garcon_feed_status.get_overall_status' ) as mock_obj: yield mock_obj @pytest.fixture def mock_set_overall_status(): """Yield set_overall_status mock.""" with patch( 'feed_ingestion.flows.gfk_streaming.tasks.' 'garcon_feed_status.set_overall_status' ) as mock_obj: yield mock_obj @pytest.fixture def mock_delete_status(): """Yield delete_status mock.""" with patch( 'feed_ingestion.flows.gfk_streaming.tasks.' 'garcon_feed_status.delete_status' ) as mock_obj: yield mock_obj @pytest.fixture def expected_bootstrap_response(): """Return expected response from bootstrap task.""" return { 'feed_name': 'gfk_streaming', 'date': '2026-04-30', 'filename': 'SONY_DAILY_TOPALL_DE_STREAM_ONLY20260430.zip', 'archive_path': ( 'GfK_Streaming/archives/2026-04-30/' 'SONY_DAILY_TOPALL_DE_STREAM_ONLY20260430.zip' ), 'processed_path': ( 'GfK_Streaming/processed/2026-04-30/' 'SONY_DAILY_TOPALL_DE_STREAM_ONLY20260430.csv' ), 'staging_raw_table': 'staging_raw_gfk_streaming', 's3_dir_path': ( 's3://dev-cucumbers/GfK_Streaming/processed/2026-04-30/' ), } class TestBootstrap: """Tests for the bootstrap task.""" @pytest.mark.parametrize( 'input_date,expected_date,expected_filename', [ ('2026-04-30', '2026-04-30', 'SONY_DAILY_TOPALL_DE_STREAM_ONLY20260430.zip'), ('2026-05-01', '2026-05-01', 'SONY_DAILY_TOPALL_DE_STREAM_ONLY20260501.zip'), ('2026-12-31', '2026-12-31', 'SONY_DAILY_TOPALL_DE_STREAM_ONLY20261231.zip'), ], ) def test_bootstrap_file_pattern( self, mock_get_overall_status, mock_delete_status, input_date, expected_date, expected_filename, ): """Test that bootstrap produces correct file names for given dates.""" result = tasks.bootstrap(MagicMock(), input_date, reload=None) assert result['date'] == expected_date assert result['filename'] == expected_filename def test_bootstrap_normal( self, expected_bootstrap_response, mock_get_overall_status, mock_delete_status, ): """Test bootstrap with normal execution.""" result = tasks.bootstrap(MagicMock(), _date, reload=None) assert result == expected_bootstrap_response mock_delete_status.assert_not_called() def test_bootstrap_reload( self, expected_bootstrap_response, mock_get_overall_status, mock_delete_status, ): """Test bootstrap with reload=True deletes the DynamoDB status.""" result = tasks.bootstrap(MagicMock(), _date, reload='True') assert result == expected_bootstrap_response mock_delete_status.assert_called_once_with( 'gfk_streaming', '2026-04-30' ) def test_bootstrap_already_ingested(self, mock_get_overall_status): """Test bootstrap returns stop when workflow is already ingested.""" mock_get_overall_status.return_value = 'INGESTED' result = tasks.bootstrap(MagicMock(), _date, reload=None) assert result == { 'stop': True, 'message': 'gfk_streaming is already ingested for 2026-04-30', } @pytest.fixture def mock_task_status(monkeypatch): """Bypass DynamoDB in check_status decorator.""" from feed_ingestion.util import task_status monkeypatch.setattr( task_status, 'is_completed_task', MagicMock(return_value=False) ) monkeypatch.setattr(task_status, 'mark_completed_task', MagicMock()) @pytest.fixture def mock_ftps_connection(): """Yield a mocked FTPS client returned by _get_ftps_connection.""" with patch( 'feed_ingestion.flows.gfk_streaming.tasks._get_ftps_connection' ) as mock_fn: mock_client = MagicMock() mock_fn.return_value = mock_client yield mock_client @pytest.fixture def mock_s3_client(): """Yield a mocked boto3 S3 client.""" with patch( 'feed_ingestion.flows.gfk_streaming.tasks.boto3.client' ) as mock_boto3: mock_client = MagicMock() mock_boto3.return_value = mock_client yield mock_client def _make_zip_with_csv(csv_content=b'col1,col2\nval1,val2\n'): """Build an in-memory ZIP containing a single CSV file.""" buf = io.BytesIO() with zipfile.ZipFile(buf, 'w') as zf: zf.writestr('data.csv', csv_content) buf.seek(0) return buf.read() @pytest.fixture def mock_retrbinary_writes_zip(tmp_path, monkeypatch): """Side-effect for retrbinary that writes a valid ZIP to the local path.""" zip_bytes = _make_zip_with_csv() def fake_retrbinary(cmd, callback, rest=None): callback(zip_bytes) return fake_retrbinary class TestFetchFromFtps: """Tests for the fetch_from_ftps task (FTPS implementation).""" def test_file_not_found_returns_stop( self, mock_task_status, mock_ftps_connection, mock_set_overall_status, ): """Test that a missing FTPS file returns a stop dict.""" mock_ftps_connection.size.side_effect = ftplib.error_perm( '550 Not found' ) result = tasks.fetch_from_ftps( MagicMock(), 'gfk_streaming', _date, _filename, _archive_path, _processed_path, ) assert result == {'stop': True, 'message': f'Missing file {_filename}'} mock_set_overall_status.assert_called_once_with( 'gfk_streaming', _date, 'NOT_AVAILABLE' ) def test_file_not_found_quits_ftps( self, mock_task_status, mock_ftps_connection, mock_set_overall_status, ): """Test that FTPS connection is closed when file is not found.""" mock_ftps_connection.size.side_effect = ftplib.error_perm( '550 Not found' ) tasks.fetch_from_ftps( MagicMock(), 'gfk_streaming', _date, _filename, _archive_path, _processed_path, ) mock_ftps_connection.quit.assert_called_once() def test_successful_download_uploads_zip_to_archive( self, mock_task_status, mock_ftps_connection, mock_s3_client, mock_set_overall_status, mock_retrbinary_writes_zip, monkeypatch, ): """Test that the raw ZIP is uploaded to the S3 archive path.""" mock_ftps_connection.retrbinary.side_effect = ( mock_retrbinary_writes_zip ) monkeypatch.setattr( 'feed_ingestion.flows.gfk_streaming.tasks.os.remove', MagicMock() ) tasks.fetch_from_ftps( MagicMock(), 'gfk_streaming', _date, _filename, _archive_path, _processed_path, ) assert mock_s3_client.upload_file.call_count == 2 zip_call_args = mock_s3_client.upload_file.call_args_list[0][0] assert zip_call_args[1] == 'dev-cucumbers' assert zip_call_args[2] == _archive_path def test_successful_download_uploads_csv_to_processed_path( self, mock_task_status, mock_ftps_connection, mock_s3_client, mock_set_overall_status, mock_retrbinary_writes_zip, monkeypatch, ): """Test that the extracted CSV is uploaded to the S3 processed path.""" mock_ftps_connection.retrbinary.side_effect = ( mock_retrbinary_writes_zip ) monkeypatch.setattr( 'feed_ingestion.flows.gfk_streaming.tasks.os.remove', MagicMock() ) tasks.fetch_from_ftps( MagicMock(), 'gfk_streaming', _date, _filename, _archive_path, _processed_path, ) assert mock_s3_client.upload_file.call_count == 2 csv_call_args = mock_s3_client.upload_file.call_args_list[1][0] assert csv_call_args[1] == 'dev-cucumbers' assert csv_call_args[2] == _processed_path def test_successful_download_sets_downloaded_status( self, mock_task_status, mock_ftps_connection, mock_s3_client, mock_set_overall_status, mock_retrbinary_writes_zip, monkeypatch, ): """Test that DOWNLOADED status is set after successful processing.""" mock_ftps_connection.retrbinary.side_effect = ( mock_retrbinary_writes_zip ) monkeypatch.setattr( 'feed_ingestion.flows.gfk_streaming.tasks.os.remove', MagicMock() ) tasks.fetch_from_ftps( MagicMock(), 'gfk_streaming', _date, _filename, _archive_path, _processed_path, ) mock_set_overall_status.assert_called_once_with( 'gfk_streaming', _date, 'DOWNLOADED' ) def test_zip_without_csv_raises( self, mock_task_status, mock_ftps_connection, mock_s3_client, mock_set_overall_status, monkeypatch, ): """Test that a ZIP with no CSV file raises ValueError.""" empty_zip = io.BytesIO() with zipfile.ZipFile(empty_zip, 'w') as zf: zf.writestr('readme.txt', 'no csv here') empty_zip_bytes = empty_zip.getvalue() def fake_retrbinary(cmd, callback, rest=None): callback(empty_zip_bytes) mock_ftps_connection.retrbinary.side_effect = fake_retrbinary monkeypatch.setattr( 'feed_ingestion.flows.gfk_streaming.tasks.os.remove', MagicMock() ) with pytest.raises(ValueError, match='No CSV file found inside ZIP'): tasks.fetch_from_ftps( MagicMock(), 'gfk_streaming', _date, _filename, _archive_path, _processed_path, ) def test_retries_on_eof_error_then_succeeds( self, mock_task_status, mock_s3_client, mock_set_overall_status, mock_retrbinary_writes_zip, monkeypatch, ): """Test EOFError on attempt 1 is retried and succeeds on attempt 2.""" monkeypatch.setattr( 'feed_ingestion.flows.gfk_streaming.tasks.time.sleep', MagicMock(), ) monkeypatch.setattr( 'feed_ingestion.flows.gfk_streaming.tasks.os.remove', MagicMock() ) mock_client = MagicMock() call_count = {'n': 0} def retrbinary_fails_first(cmd, callback, rest=None): call_count['n'] += 1 if call_count['n'] == 1: raise EOFError mock_retrbinary_writes_zip(cmd, callback, rest=rest) mock_client.retrbinary.side_effect = retrbinary_fails_first with patch( 'feed_ingestion.flows.gfk_streaming.tasks._get_ftps_connection', return_value=mock_client, ): activity = MagicMock() tasks.fetch_from_ftps( activity, 'gfk_streaming', _date, _filename, _archive_path, _processed_path, ) assert call_count['n'] == 2 activity.logger.warning.assert_called_once() mock_set_overall_status.assert_called_once_with( 'gfk_streaming', _date, 'DOWNLOADED' ) class TestNormalizeCsvQuotes: """Tests for _normalize_csv_quotes.""" @pytest.mark.parametrize('input_line,expected_line', [ ( b'"Towards the sun (From \\"imask ad\\")"\n', b'"Towards the sun (From ""imask ad"")"\n', ), ( b'"Sony Music Entertainment \\";\n', b'"Sony Music Entertainment \\";\n', ), ( b'"no escapes here";\n', b'"no escapes here";\n', ), ( b'"trailing backslash at eol\\"\r\n', b'"trailing backslash at eol\\"\r\n', ), ]) def test_normalize_line(self, tmp_path, input_line, expected_line): """Test that embedded backslash-quotes become double-quotes.""" csv_file = tmp_path / 'test.csv' csv_file.write_bytes(input_line) _normalize_csv_quotes(str(csv_file)) assert csv_file.read_bytes() == expected_line class TestDeleteProcessedFile: """Tests for the delete_processed_file task.""" @pytest.fixture def mock_s3_client(self): """Yield a mocked boto3 S3 client.""" with patch( 'feed_ingestion.flows.gfk_streaming.tasks.boto3.client' ) as mock_boto3: mock_client = MagicMock() mock_boto3.return_value = mock_client yield mock_client def test_deletes_correct_s3_object(self, mock_s3_client): """Test that the processed CSV is deleted from the correct S3 path.""" tasks.delete_processed_file(MagicMock(), _processed_path) mock_s3_client.delete_object.assert_called_once_with( Bucket='dev-cucumbers', Key=_processed_path, ) @pytest.mark.parametrize('processed_path', [ ( 'GfK_Streaming/processed/2026-04-30/' 'SONY_DAILY_TOPALL_DE_STREAM_ONLY20260430.csv' ), ( 'GfK_Streaming/processed/2026-05-01/' 'SONY_DAILY_TOPALL_DE_STREAM_ONLY20260501.csv' ), ]) def test_deletes_correct_path_for_different_dates( self, mock_s3_client, processed_path ): """Test deletion is called with the exact path passed in.""" tasks.delete_processed_file(MagicMock(), processed_path) mock_s3_client.delete_object.assert_called_once_with( Bucket='dev-cucumbers', Key=processed_path, )