from pathlib import Path from unittest.mock import patch, MagicMock, call from flows.sme_latam import api THIS_DIR = Path(__file__).parent FILENAME_LATAM_UNIFIED_WEEKLY_CHART = 'BMAT - Sony - Sony Latam Unified Weekly Chart - Week 39.xlsx' def test_load_latam_unified_weekly_chart(): xlsx_file = THIS_DIR / FILENAME_LATAM_UNIFIED_WEEKLY_CHART df, year, week = api.load_latam_unified_weekly_chart( xlsx_file=xlsx_file, ) assert len(df) == 10 columns = list(df.columns) assert columns == list(api.LATAM_UNIFIED_WEEKLY_CHART_COLUMNS.values()) assert year == 2022 assert week == 39 assert '-' not in df['POS_2_BEFORE'].value_counts() assert '-' not in df['POS_BEFORE'].value_counts() @patch.object(api, 'pandas_tools') def test_save_dataframe_to_table(pandas_tools_mock): pandas_tools_mock.write_pandas.return_value = True, 1, 10, 'OK' connection_mock = MagicMock() df_mock = MagicMock() success, n_rows = api.save_dataframe_to_table( connection=connection_mock, df=df_mock, table_name='temp_sme_latam' ) assert success assert n_rows == 10 assert pandas_tools_mock.write_pandas.call_args_list == [ call(conn=connection_mock, df=df_mock, table_name='TEMP_SME_LATAM', table_type='transient', auto_create_table=True, overwrite=True) ]