"""Unit tests for tasks of Proper Ingestion Workflow.""" from datetime import date as date_module from datetime import datetime from unittest import mock from unittest.mock import MagicMock from unittest.mock import Mock from unittest.mock import patch import boto3 from freezegun import freeze_time from garcon import activity from garcon_contrib.aws.utils import garcon_s3 from garcon_contrib.dynamo_feed_status import garcon_feed_status from garcon_contrib.ftp import garcon_ftp from garcon_contrib.mysql import garcon_mysql from garcon_contrib.snowflake import garcon_snowflake from jsonschema import exceptions as jsonschema_exceptions from moto import mock_aws import pandas as pd import pymysql import pytest from feed_ingestion.flows import helpers from feed_ingestion.flows.proper_incoming import config, json_schemas, tasks from feed_ingestion.util import task_status from feed_ingestion.util.aws import s3 as s3_feed_ingestion_utils @pytest.fixture(params=[ '2016-08-27', '2017-02-25', date_module.today().strftime('%Y-%m-%d')]) def context_date(request): """Fixture returns 3 different dates. Returns: str: Date in YYYY-MM-DD format. """ return request.param @pytest.fixture(params=config.proper_feeds.keys()) def feed_name(request): """Fixture returns feed name. Returns: str: Feed name. """ return 'proper_daily_{feed_type}'.format(feed_type=request.param) @pytest.fixture(params=[ config.FEED_TYPE_GOODSIN, config.FEED_TYPE_SALES, config.FEED_TYPE_SHORTAGES, config.FEED_TYPE_STOCK]) def proper_feed_type(request): """Fixture returns proper_feed_type acceptable values. Returns: str: Feed type. """ return request.param @pytest.fixture(params=[ (config.FEED_TYPE_GOODSIN, 'proper_daily_{}'.format(config.FEED_TYPE_GOODSIN), 'goodsin.csv'), (config.FEED_TYPE_SHORTAGES, 'proper_daily_{}'.format(config.FEED_TYPE_SHORTAGES), 'shortages.csv')]) def snowflake_feeds_attrs(request): """Fixture returns proper_feed_type acceptable values. Returns: tuple (str, str, str): Feed type, feed name, file name. """ return request.param @pytest.fixture def ftp_path(): """Fixture returns FTP path to file. Returns: str: FTP path. """ return '/ftp/path' @pytest.fixture def s3_path(): """Fixture returns S3 path to file. Returns: str: S3 path. """ return 's3://bucket/path' @pytest.fixture def ftp_file_name(): """Fixture returns file name on FTP. Returns: str: File name. """ return 'ftp_file.name' @pytest.fixture def s3_file_name(): """Fixture returns file name on S3. Returns: str: File name. """ return 's3_file.name' @pytest.fixture(params=[ 's3://bucket/ProperStock/archives/2016-08-27/GoodsIn_ESSN_20160827.csv', 's3://bucket/ProperStock/archives/2016-08-27/Shortages_ESSN_20160827.csv']) def file_to_ingest_to_snowflake(request): """Fixture returning S3 path to file to be ingested to Snowflake. Returns: str: Full S3 path. """ return request.param @pytest.fixture def file_to_validate(): """Fixture returns S3 path to file to mock StaticParam. Returns: str: Full S3 path. """ return ('s3://bucket/ProperStock/archives/2016-08-27/' 'Stock_ESSN_20160827.csv') @pytest.fixture def feed_type_stock(): """Fixture returns 'stock' feed_type. Returns: str: Feed type. """ return config.FEED_TYPE_STOCK @pytest.fixture def feed_type_sales(): """Fixture returns 'sales' feed_type. Returns: str: Feed type. """ return config.FEED_TYPE_SALES @pytest.fixture(params=[ (config.FEED_TYPE_GOODSIN, 's3://bucket/ProperIncoming/archives/2016-09-29/' 'GoodsIn_ESSN_2016-09-29.csv', ',\r\n'), (config.FEED_TYPE_SHORTAGES, 's3://bucket/ProperIncoming/archives/2016-09-29/' 'Shortages_ESSN_2016-09-29.csv', None)]) def snowflake_feed_file(request): """Fixture returns parameters for CSV -> Snowflake upload. Returns: tuple (str, str, str): feed_type, S3 CSV file path, right strip in CSV file if required. """ return request.param @pytest.fixture def fieldnames(feed_type_stock): """Fixture returns fieldnames of Stock_ESSN CSV file. (Retrieved from JSON Schema). Returns: list: Fieldnames of Stock_ESSN CSV file. """ return json_schemas.schemas['properties'][feed_type_stock][ 'sql_options']['csv_file_header']['items'] @pytest.fixture def columns_names(feed_type_stock): """Fixture returns columns names of proper_stock_essn table. (Retrieved from JSON Schema). Returns: list: Columns names of proper_stock_essn table. """ return json_schemas.schemas[ 'properties'][feed_type_stock]['sql_options']['columns_names'][ 'items'] @pytest.fixture def row_schema_stock(feed_type_stock): """Fixture returns row's schema of Stock_ESSN file. (In JSON Schema format). Returns: row_schema (dict): Row's schema of Stock_ESSN file. """ row_schema = json_schemas.schemas['properties'][feed_type_stock][ 'row_schema'] return row_schema @pytest.fixture def row_schema_sales(feed_type_sales): """Fixture returns row's schema of Sales_ESSN file. (In JSON Schema format). Returns: row_schema (dict): Row's schema of Stock_ESSN file. """ row_schema = json_schemas.schemas['properties'][feed_type_sales][ 'row_schema'] return row_schema @pytest.fixture( params=[ {'CatNumber': 'CD1322', 'OnHand': '789', 'Allocated': '1', 'Faulty': '0', 'Consignment': '402', 'Available': '386', 'LabelCode': 'E283', 'EAN': '0600116132222'}, {'CatNumber': '911361', 'OnHand': '315', 'Allocated': '7', 'Faulty': '0', 'Consignment': '271', 'Available': '37', 'LabelCode': 'E111', 'EAN': '4029759113614'}, {'CatNumber': '911361', 'OnHand': '315', 'Allocated': '7', 'Faulty': '0', 'Consignment': '271', 'Available': '-37', 'LabelCode': 'E111', 'EAN': '4029759113614'}, {'CatNumber': '911361', 'OnHand': '-1', 'Allocated': '7', 'Faulty': '0', 'Consignment': '271', 'Available': '-37', 'LabelCode': 'E111', 'EAN': '4029759113614'}]) def correct_row_stock(request): """Fixture returns correct rows of Stock_ESSN file. Returns: (dict): Correct row of Stock_ESSN file. """ return request.param @pytest.fixture( params=[ {'CatNumber': 5, 'OnHand': '789', 'Allocated': '1', 'Faulty': '0', 'Consignment': '402', 'Available': '386', 'LabelCode': 'E283', 'EAN': '0600116132222'}, {'CatNumber': '911361', 'OnHand': '315', 'Allocated': '7', 'Faulty': '0', 'Consignment': '271', 'Available': '37', 'LabelCode': 'E1115325', 'EAN': '4029759113614'}, {'CatNumber': '911361', 'OnHand': 'a315', 'Allocated': '7', 'Faulty': '0', 'Consignment': '271', 'Available': '37', 'LabelCode': 'E111', 'EAN': '4029759113614'}, {'CatNumber': '911361', 'OnHand': '315a', 'Allocated': '7', 'Faulty': '0', 'Consignment': '271', 'Available': '37', 'LabelCode': 'E111', 'EAN': '4029759113614'}, {'CatNumber': '911361', 'OnHand': '3a15', 'Allocated': '7', 'Faulty': '0', 'Consignment': '271', 'Available': '37', 'LabelCode': 'E111', 'EAN': '4029759113614'}, {'CatNumber': '911361', 'OnHand': '315', 'Allocated': '7', 'Faulty': '0', 'Consignment': '271', 'Available': '--37', 'LabelCode': 'E111', 'EAN': '4029759113614'}, {'CatNumber': '911361', 'OnHand': '315', 'Allocated': '7', 'Faulty': '0', 'Consignment': '271', 'Available': '3-7', 'LabelCode': 'E111', 'EAN': '4029759113614'}]) def incorrect_row_stock(request): """Fixture returns incorrect rows of Stock_ESSN file. Returns: (dict): Incorrect row of Stock_ESSN file. """ return request.param @pytest.fixture( params=[ {'CatNumber': 'CD1322', 'Barcode': '789', 'StoreCode': '1', 'StoreNumber': '520', 'StoreID': '40522', 'OrderNumber': '386', 'LineId': 'E283', 'InvoiceNumber': '0600116132222', 'InvoiceDate': '2016-02-02', 'Units': '4324', 'DealerPrice': '.70', 'InvoicePrice': '.20', 'DiscountPercent': '0.1', 'OrderType': 'ORDER', 'VatCode': '5353', 'CustomerRef': '0600116132222', 'ExchangeRate': '.01', 'CurrencyCode': '4F8', 'AccountName': 'wegFQ2R2VSSGWE', 'CountryCode': 'US'}]) def correct_row_sales(request): """Fixture returns correct row of Sales_ESSN file. Returns: (dict): Correct row of Sales_ESSN file. """ return request.param @pytest.fixture( params=[ {'CatNumber': 'CD1322', 'Barcode': '789', 'StoreCode': '1', 'StoreNumber': '520', 'StoreID': '40522', 'OrderNumber': '386', 'LineId': 'E283', 'InvoiceNumber': '0600116132222', 'InvoiceDate': '2016-02-02', 'Units': '4324', 'DealerPrice': '7.0.', 'InvoicePrice': '.20', 'DiscountPercent': '0.1', 'OrderType': 'ORDER', 'VatCode': '5353', 'CustomerRef': '0600116132222', 'ExchangeRate': '.01', 'CurrencyCode': '4F8', 'AccountName': 'wegFQ2R2VSSGWE', 'CountryCode': 'US'}]) def incorrect_row_sales(request): """Fixture returns incorrect row of Sales_ESSN file. Returns: (dict): Incorrect row of Sales_ESSN file. """ return request.param def test_bootstrap(context_date, proper_feed_type): """Test bootstrap task.""" activity = MagicMock() feed_name = 'proper_daily_{}'.format(proper_feed_type) response = tasks.bootstrap(activity, context_date, proper_feed_type) assert len(response) == 7 assert response['date'] == context_date assert response['proper_feed_type'] == proper_feed_type assert response['feed_name'] == feed_name s3_path = config.target_s3_path.format( s3_bucket=config.s3_bucket, date_YYYY_MM_DD=context_date) assert response['s3_path'] == s3_path ftp_file_name = config.proper_feeds.get(proper_feed_type).get( 'ftp_file_name').format(date_YYYY_MM_DD=context_date) assert response['ftp_file_name'] == ftp_file_name s3_file_name = config.proper_feeds.get(proper_feed_type).get( 's3_file_name').format(date_YYYY_MM_DD=context_date) assert response['s3_file_name'] == s3_file_name s3_file_full_path = '{}{}'.format(s3_path, s3_file_name) assert response['s3_file_full_path'] == s3_file_full_path def test_fetch_from_drop_location_ingested( monkeypatch, context_date, ftp_path, s3_path, ftp_file_name, s3_file_name): """Test fetch_from_drop_location task if flow was ingested already.""" overall_status = MagicMock(return_value=garcon_feed_status.STATUS_INGESTED) monkeypatch.setattr( garcon_feed_status, 'get_overall_status', value=overall_status) activity = MagicMock() response = tasks.fetch_from_drop_location( activity, context_date, feed_name, ftp_path, s3_path, ftp_file_name, s3_file_name) garcon_feed_status.get_overall_status.assert_called_with( feed_name, context_date) assert response['stop'] is True def test_fetch_from_drop_location( monkeypatch, context_date, feed_name, ftp_path, s3_path, ftp_file_name, s3_file_name): """Test normal execution of fetch_from_drop_location task.""" activity = MagicMock() monkeypatch.setattr(task_status, 'mark_completed_task', value=MagicMock()) copy_file_from_ftp_to_s3 = Mock() copy_file_from_ftp_to_s3.return_value = { 'file': s3_file_name, 'status': True} monkeypatch.setattr( garcon_ftp, 'copy_file_from_ftp_to_s3', copy_file_from_ftp_to_s3) monkeypatch.setattr(garcon_feed_status, 'set_status', value=MagicMock()) monkeypatch.setattr( garcon_feed_status, 'set_overall_status', value=MagicMock()) monkeypatch.setattr( garcon_feed_status, 'get_overall_status', value=MagicMock()) tasks.fetch_from_drop_location( activity, context_date, feed_name, ftp_path, s3_path, ftp_file_name, s3_file_name) garcon_feed_status.set_status.assert_called_with( feed_name, context_date, s3_file_name, status=garcon_feed_status.STATUS_DOWNLOADED) def test_fetch_from_drop_location_copy_fail( monkeypatch, context_date, feed_name, ftp_path, s3_path, ftp_file_name, s3_file_name): """Test flow when copy was unsuccessful.""" activity = MagicMock() copy_file_from_ftp_to_s3 = Mock(return_value={'status': False}) monkeypatch.setattr( garcon_ftp, 'copy_file_from_ftp_to_s3', copy_file_from_ftp_to_s3) monkeypatch.setattr( garcon_feed_status, 'set_overall_status', value=MagicMock()) monkeypatch.setattr(garcon_feed_status, 'set_status', value=MagicMock()) monkeypatch.setattr( garcon_feed_status, 'get_overall_status', value=MagicMock()) fetch_respone = tasks.fetch_from_drop_location( activity, context_date, feed_name, ftp_path, s3_path, ftp_file_name, s3_file_name) garcon_feed_status.set_overall_status.assert_called_with( feed_name, context_date, garcon_feed_status.STATUS_NOT_AVAILABLE) assert fetch_respone['stop'] is True @mock_aws def test_validate_csv_files( monkeypatch, context_date, file_to_validate, fieldnames, feed_name, feed_type_stock): """Test validation for files against schema.""" activity = MagicMock() monkeypatch.setattr( garcon_feed_status, 'set_overall_status', value=MagicMock()) monkeypatch.setattr(task_status, 'mark_completed_task', value=MagicMock()) validate_header_mock = MagicMock() validate_row_mock = MagicMock() monkeypatch.setattr(helpers, 'validate_header', validate_header_mock) monkeypatch.setattr(helpers, 'validate_row', validate_row_mock) s3_client = boto3.client('s3') s3_client.create_bucket(Bucket='bucket') bucket_name, bucket_path = garcon_s3.extract_bucket_path( file_to_validate) content = ','.join(fieldnames) + '\n' + ' ' s3_client.put_object( Body=content, Bucket=bucket_name, Key=bucket_path, ) tasks.validate_csv_file( activity, context_date, feed_name, feed_type_stock, file_to_validate, config.expected_bucket_owner) validate_header_mock.assert_called_with(fieldnames, fieldnames) assert validate_row_mock.call_count > 0 task_status.mark_completed_task.assert_called_with( feed_name, context_date, 'validate_csv_file') def test_validate_csv_files_without_schema(context_date, feed_name): """Test validation function with feeds without schema.""" activity = MagicMock() feed_without_schema = 'not existing schema' with pytest.raises(KeyError): tasks.validate_csv_file( activity, context_date, feed_name, feed_without_schema, file_to_validate, config.expected_bucket_owner) @mock_aws def test_validate_csv_files_failure( monkeypatch, context_date, file_to_validate, fieldnames, feed_name, feed_type_stock): """Test validation of a corrupted file.""" activity = MagicMock() monkeypatch.setattr( garcon_feed_status, 'set_overall_status', value=MagicMock()) monkeypatch.setattr(task_status, 'mark_completed_task', value=MagicMock()) validate_header_mock = MagicMock() validate_header_mock.side_effect = ValueError monkeypatch.setattr(helpers, 'validate_header', validate_header_mock) s3_client = boto3.client('s3') s3_client.create_bucket(Bucket='bucket') bucket_name, bucket_path = garcon_s3.extract_bucket_path( file_to_validate) content = ','.join(fieldnames) s3_client.put_object( Body=content, Bucket=bucket_name, Key=bucket_path, ) with pytest.raises(ValueError) as raised_exception: tasks.validate_csv_file( activity, context_date, feed_name, feed_type_stock, file_to_validate, config.expected_bucket_owner) assert str(raised_exception) == '' validate_header_mock.assert_called_with(fieldnames, fieldnames) def test_helpers_validate_header(): """Test validation of a proper header.""" helpers.validate_header(['AAA', 'BBB'], ['AAA', 'BBB']) def test_helpers_validate_header_failure(): """Test validation of a corrupted header.""" with pytest.raises(ValueError): helpers.validate_header(['AAA', 'AAA'], ['AAA', 'BBB']) def test_helpers_validate_data_row_stock(row_schema_stock, correct_row_stock): """Test validation of a correct data row of a Stock ESSN file.""" helpers.validate_row(row_schema_stock, correct_row_stock) def test_helpers_validate_data_row_stock_failure( row_schema_stock, incorrect_row_stock): """Test validation of a incorrect data row of a Stock ESSN file.""" with pytest.raises(jsonschema_exceptions.ValidationError): helpers.validate_row(row_schema_stock, incorrect_row_stock) # TODO(borisuvarov): Make the validation tests and fixtures more generic, # implement the JSON schema testing def test_helpers_validate_data_row_sales(row_schema_sales, correct_row_sales): """Test validation of a correct data row of a Sales ESSN file.""" helpers.validate_row(row_schema_sales, correct_row_sales) def test_helpers_validate_data_row_sales_failure( row_schema_sales, incorrect_row_sales): """Test validation of a incorrect data row of a Sales ESSN file.""" with pytest.raises(jsonschema_exceptions.ValidationError): helpers.validate_row(row_schema_sales, incorrect_row_sales) def test_ingest_stock_data_into_mysql_table( monkeypatch, file_to_validate, columns_names, feed_type_stock): """Test ingesting CSV file to MySQL table.""" bulk_insert_from_csv_file_on_s3 = MagicMock() mysql_db_config_mock = MagicMock() monkeypatch.setattr( garcon_mysql, 'bulk_insert_from_csv_file_on_s3', bulk_insert_from_csv_file_on_s3) monkeypatch.setattr(config, 'mysql_db_config', mysql_db_config_mock) mark_completed_task_mock = MagicMock() monkeypatch.setattr( task_status, 'mark_completed_task', value=mark_completed_task_mock) activity = MagicMock() feed_name = MagicMock() date = MagicMock() table_name = json_schemas.schemas['properties'][feed_type_stock][ 'sql_options']['table_name'] tasks.ingest_stock_data_into_mysql_table( activity, feed_name, date, feed_type_stock, file_to_validate) bulk_insert_from_csv_file_on_s3.assert_called_with( activity, file_to_validate, mysql_db_config_mock, table_name, columns_names, ignore_lines=1) mark_completed_task_mock.assert_called_with( feed_name, date, 'ingest_stock_data_into_mysql_table') def test_not_ingest_stock_data_into_mysql_table_if_no_feed_type(): """Test verifies if there is no schema for feed.""" activity = MagicMock() feed_name = MagicMock() file_to_validate = MagicMock() proper_feed_type = 'not_existing_feed' with pytest.raises(KeyError): tasks.ingest_stock_data_into_mysql_table( activity, feed_name, 'date', proper_feed_type, file_to_validate) def test_ingest_stock_data_into_mysql_table_mysql_failure( monkeypatch, feed_type_stock): """Test failed ingestion of CSV file to MySQL table.""" monkeypatch.setattr(task_status, 'mark_completed_task', value=MagicMock()) bulk_insert_from_csv_file_on_s3 = Mock( side_effect=pymysql.Error('Some MySQL error!')) mysql_db_config_mock = MagicMock() monkeypatch.setattr( garcon_mysql, 'bulk_insert_from_csv_file_on_s3', bulk_insert_from_csv_file_on_s3) monkeypatch.setattr(config, 'mysql_db_config', mysql_db_config_mock) activity = MagicMock() feed_name = MagicMock() date = MagicMock() file_to_validate = MagicMock() with pytest.raises(pymysql.Error): tasks.ingest_stock_data_into_mysql_table( activity, feed_name, date, feed_type_stock, file_to_validate) assert not task_status.mark_completed_task.called def test_add_release_id_to_proper_stock_essn_table(monkeypatch): """Test add_release_id_to_proper_stock_essn_table task.""" ctx_mock = MagicMock() cursor_mock = MagicMock(return_value=ctx_mock) db_connection = MagicMock() db_connection.cursor = cursor_mock monkeypatch.setattr( pymysql, 'connect', value=MagicMock(return_value=db_connection)) mark_completed_task_mock = MagicMock() monkeypatch.setattr( task_status, 'mark_completed_task', value=mark_completed_task_mock) activity = MagicMock() feed_name = 'proper_stock' date = '2016-10-26' tasks.add_release_id_to_proper_stock_essn_table( activity, feed_name, date) mark_completed_task_mock.assert_called_with( feed_name, date, 'add_release_id_to_proper_stock_essn_table') assert db_connection.cursor.call_count == 1 assert db_connection.close.call_count == 1 assert str(ctx_mock.mock_calls[1]) == ( "call.__enter__().execute('{sql}')".format( sql=config.update_proper_stock_essn_by_upc_sql)) assert str(ctx_mock.mock_calls[2]) == ( "call.__enter__().execute('{sql}')".format( sql=config.update_proper_stock_essn_by_display_upc_sql)) assert str(ctx_mock.mock_calls[3]) == ( "call.__enter__().execute('{sql}')".format( sql=config.update_proper_stock_essn_by_manufacturer_upc_sql)) assert str(ctx_mock.mock_calls[4]) == ( "call.__enter__().execute('{sql}')".format( sql=config.update_proper_stock_essn_by_catalog_number_sql)) def test_purge_old_data_from_snowflake( monkeypatch, snowflake_feeds_attrs): """Test purging old data from Snowflake table.""" fake_snowflake_config = { 'user': 'test', 'key': 'test', 'account': 'test', 'role': 'test', 'warehouse': 'test', 'db': 'test', 'schema': 'test' } monkeypatch.setattr( config, 'snowflake_db_config', fake_snowflake_config) monkeypatch.setattr(task_status, 'mark_completed_task', value=MagicMock()) util_snowflake_connect_mock = MagicMock() monkeypatch.setattr( garcon_snowflake, 'connect', util_snowflake_connect_mock) cursor_obj = MagicMock() enter_mock = MagicMock() ctx_mock = MagicMock() execute_mock = MagicMock() ctx_mock.execute = execute_mock enter_mock.return_value = ctx_mock cursor_obj.__enter__ = enter_mock util_snowflake_cursor_mock = MagicMock(return_value=cursor_obj) monkeypatch.setattr(garcon_snowflake, 'cursor', util_snowflake_cursor_mock) activity_obj = MagicMock() feed_type, feed_name, file_name = snowflake_feeds_attrs table_name = json_schemas.schemas['properties'][feed_type][ 'sql_options'].get('snowflake_table_name') tasks.purge_old_data_from_snowflake( activity_obj, feed_type, feed_name, 'date', file_name ) if feed_type in (config.FEED_TYPE_GOODSIN, config.FEED_TYPE_SHORTAGES): util_snowflake_connect_mock.assert_called_with( user='test', account='test', private_key='test', role='test') else: assert util_snowflake_connect_mock.call_count == 0 execute_mock.assert_called_with( "DELETE FROM PROD.PRODUCTION.{} WHERE " # noqa "file_name='{}';".format(table_name, file_name)) @mock_aws def test_transform_csv_file(monkeypatch, snowflake_feed_file): """Test transformation of CSV files and copying them to S3.""" monkeypatch.setattr(task_status, 'mark_completed_task', value=MagicMock()) activity_obj = MagicMock() expand_s3_csv_with_columns_mock = MagicMock() monkeypatch.setattr( s3_feed_ingestion_utils, 'expand_s3_csv_with_columns', expand_s3_csv_with_columns_mock) feed_type, file_to_transform, line_rstrip_char = snowflake_feed_file freezed_timestamp = datetime.now() with freeze_time(freezed_timestamp): tasks.transform_csv_file( activity_obj, feed_type, 'proper_daily', '2001-01-01', file_to_transform) transformed_file_name = file_to_transform.split('/')[-1] expand_s3_csv_with_columns_mock.assert_called_with( file_to_transform, ( 'file_name', 'file_date', 'ingestion_timestamp'), (transformed_file_name, '2016-09-29', datetime.strftime( datetime.now(), '%Y-%m-%dT%H:%M:%S')), '/ProperIncoming/snowflake/2001-01-01', line_rstrip_char=line_rstrip_char) assert task_status.mark_completed_task.called @mock_aws def test_transform_csv_file_not_called(monkeypatch): """Test transformation of CSV files and copying them to S3 not called. (If proper_feed_type not meant to ingest to Snowflake). """ monkeypatch.setattr(task_status, 'mark_completed_task', value=MagicMock()) activity_obj = activity.Activity(boto3.client('swf', 'us-east-1')) file_to_transform = ( 's3://bucket/ProperStock/archives/2016-09-29/' 'GoodsIn_ESSN_2016-09-29.csv') expand_s3_csv_with_columns_mock = MagicMock() monkeypatch.setattr( s3_feed_ingestion_utils, 'expand_s3_csv_with_columns', expand_s3_csv_with_columns_mock) tasks.transform_csv_file( activity_obj, config.FEED_TYPE_SALES, 'proper_daily', '2001-01-01', file_to_transform) assert expand_s3_csv_with_columns_mock.call_count == 0 assert task_status.mark_completed_task.call_count == 0 @mock_aws def test_load_data_into_snowflake(monkeypatch, snowflake_feeds_attrs): """Test load_data_into_snowflake task.""" fake_snowflake_config = { 'user': 'test', 'key': 'test', 'account': 'test', 'role': 'test', 'warehouse': 'test', 'db': 'test', 'schema': 'test' } monkeypatch.setattr( config, 'snowflake_db_config', fake_snowflake_config) monkeypatch.setattr(task_status, 'mark_completed_task', value=MagicMock()) util_snowflake_connect_mock = MagicMock() monkeypatch.setattr( garcon_snowflake, 'connect', util_snowflake_connect_mock) cursor_obj = MagicMock() enter_mock = MagicMock() ctx_mock = MagicMock() execute_mock = MagicMock() ctx_mock.execute = execute_mock enter_mock.return_value = ctx_mock cursor_obj.__enter__ = enter_mock util_snowflake_cursor_mock = MagicMock(return_value=cursor_obj) monkeypatch.setattr(garcon_snowflake, 'cursor', util_snowflake_cursor_mock) activity_obj = MagicMock() feed_type, feed_name, file_name = snowflake_feeds_attrs tasks.load_data_into_snowflake( activity_obj, feed_name, '2018-01-01', 's3://somepath', feed_type ) if feed_type in (config.FEED_TYPE_GOODSIN, config.FEED_TYPE_SHORTAGES): util_snowflake_connect_mock.assert_called_with( user='test', account='test', private_key='test', role='test') assert execute_mock.call_count == 4 else: assert util_snowflake_connect_mock.call_count == 0 assert execute_mock.call_count == 0 @patch('feed_ingestion.flows.proper_incoming.tasks.s3utils') def test_transform_csv_file_stock(mock_s3utils): """Test transform_csv_file_stock.""" filename = 'Stock_ESSN_2022-11-16.csv' s3_path_old = ( f's3://dev-cucumbers/ProperIncoming/archives/2022-11-16/{filename}') s3_path_new = ( f's3://dev-cucumbers/ProperIncoming/stock/2022-11-16/{filename}') df = pd.DataFrame([ {'CatNumber': 'CD1322', 'OnHand': '789', 'Allocated': '1', 'Faulty': '0', 'Consignment': '402', 'Available': '386', 'LabelCode': 'E283', 'EAN': '0600116132222'}, {'CatNumber': '911361', 'OnHand': '315', 'Allocated': '7', 'Faulty': '0', 'Consignment': '271', 'Available': '37', 'LabelCode': 'FOC', 'EAN': '4029759113614'}]) mock_s3utils.read_csv.return_value = df response = tasks.transform_csv_file_stock( MagicMock(), config.FEED_TYPE_STOCK, '2022-11-16', s3_path_old) mock_s3utils.read_csv.assert_called_with(s3_path_old) mock_s3utils.to_csv.assert_called_with(mock.ANY, s3_path_new) assert response == {'s3_file_full_path': s3_path_new} @patch('feed_ingestion.flows.proper_incoming.tasks.s3utils') def test_transform_csv_file_stock_failure(mock_s3utils): """Test transform_csv_file_stock with Label code less than 3 symbols.""" filename = 'Stock_ESSN_2022-11-16.csv' s3_path_old = ( f's3://dev-cucumbers/ProperIncoming/archives/2022-11-16/{filename}') df = pd.DataFrame([ {'CatNumber': 'CD1322', 'OnHand': '789', 'Allocated': '1', 'Faulty': '0', 'Consignment': '402', 'Available': '386', 'LabelCode': 'E283', 'EAN': '0600116132222'}, {'CatNumber': '911361', 'OnHand': '315', 'Allocated': '7', 'Faulty': '0', 'Consignment': '271', 'Available': '37', 'LabelCode': 'E11', 'EAN': '4029759113614'}]) mock_s3utils.read_csv.return_value = df with pytest.raises(ValueError): tasks.transform_csv_file_stock( MagicMock(), config.FEED_TYPE_STOCK, '2022-11-16', s3_path_old) mock_s3utils.read_csv.assert_called_with(s3_path_old) mock_s3utils.to_csv.assert_not_called()