"""Tests for the legacy revenue report generation (using Pandas).""" from datetime import datetime from decimal import Decimal from os import path from unittest.mock import MagicMock from unittest.mock import patch import pandas import pytest from config import FILE_OUTPUT_PATH from src.documents import legacy_revenue_report_pandas def test_make_process_dataframe(): """Test the process dataframe function.""" dataframe = pandas.DataFrame( [ { 'CUSTOMER_NAME': 'One', 'FX_GROSS': Decimal('123.4'), 'FX_NET_RECEIPT': Decimal('100.00'), }, { 'CUSTOMER_NAME': 'Two', 'FX_GROSS': Decimal('432.1'), 'FX_NET_RECEIPT': Decimal('200.00'), }, { 'CUSTOMER_NAME': 'Three', 'FX_GROSS': Decimal('0.1234567'), 'FX_NET_RECEIPT': Decimal('300'), }, ] ) expected = pandas.DataFrame( [ {'Retailer': 'One', 'Total': '123.400000', 'Label Share Net Receipts': '100.000000'}, {'Retailer': 'Two', 'Total': '432.100000', 'Label Share Net Receipts': '200.000000'}, {'Retailer': 'Three', 'Total': '0.123457', 'Label Share Net Receipts': '300.000000'}, ] ) result = legacy_revenue_report_pandas._make_process_dataframe(None, True) assert callable(result) result(dataframe) pandas.testing.assert_frame_equal(dataframe, expected) def test_make_process_dataframe_subaccount(): """Test the process dataframe function with a subaccount.""" dataframe = pandas.DataFrame( [ { 'CUSTOMER_NAME': 'One', 'FX_GROSS': Decimal('100.0'), 'FX_NET_RECEIPT': Decimal('100.0'), }, { 'CUSTOMER_NAME': 'Two', 'FX_GROSS': Decimal('200.0'), 'FX_NET_RECEIPT': Decimal('200.0'), }, ] ) subaccount = { 'SUBACCOUNTNAME': 'Subbb', 'SUBACCOUNT_SPLIT_TYPE': 'Gross', 'COMMISSIONOVERRIDE': Decimal('0.80'), } expected = pandas.DataFrame( [ {'Retailer': 'One', 'Label Share Net Receipts': '80.000000'}, {'Retailer': 'Two', 'Label Share Net Receipts': '160.000000'}, ] ) result = legacy_revenue_report_pandas._make_process_dataframe(subaccount, True) assert callable(result) result(dataframe) pandas.testing.assert_frame_equal(dataframe, expected) @patch('src.documents.legacy_revenue_report_pandas.create_report_file_pandas') @patch('src.documents.legacy_revenue_report_pandas.snowflake') @patch('src.documents.legacy_revenue_report_pandas.OwsAbacusAccount.get_account') @patch('src.documents.legacy_revenue_report_pandas.OwsRoyalties.get_statement_period') @patch('src.documents.legacy_revenue_report_pandas._make_process_dataframe') def test_build_full_document( _make_process_dataframe_mock, get_statement_period_mock, get_account_mock, snowflake_mock, create_report_file_pandas_mock, account_fixture, statement_period_fixture, ): """Test building a legacy revenue full report.""" account_id = 24601 statement_period_ids = '123' file_type = 'xls' number_format = 'us' mock_generator = MagicMock() data = mock_generator total_rows = 3 generation_date = datetime.today().strftime('%Y%m%d') expected_path = path.join( FILE_OUTPUT_PATH, f'{generation_date}_July_2022_fullreport_Jeans_Nest.xls' ) get_account_mock.return_value = account_fixture get_statement_period_mock.return_value = statement_period_fixture snowflake_mock.get_workstation_fact_sales_pandas.return_value = (data, total_rows) snowflake_mock.is_distributor.return_value = {'IS_DISTRIBUTOR': 'Y'} create_report_file_pandas_mock.return_value = expected_path _make_process_dataframe_mock.return_value = 'FUNC' result = legacy_revenue_report_pandas.build_full_document( account_id, statement_period_ids, None, file_type, number_format ) call_args = create_report_file_pandas_mock.call_args[0] actual_generator = call_args[0] assert result == expected_path create_report_file_pandas_mock.assert_called_once() _make_process_dataframe_mock.assert_called_once_with(None, True, number_format) assert hasattr(actual_generator, '__iter__') and hasattr(actual_generator, '__next__') assert call_args[1] == total_rows assert call_args[2] == 'FUNC' assert call_args[3] == expected_path assert call_args[4] == file_type assert call_args[5] == number_format snowflake_mock.get_workstation_fact_sales_pandas.assert_called_once_with( (123,), account_id, None, None ) @patch('src.documents.legacy_revenue_report_pandas.create_report_file_pandas') @patch('src.documents.legacy_revenue_report_pandas.snowflake') @patch('src.documents.legacy_revenue_report_pandas.OwsRoyalties.get_statement_period') @patch('src.documents.legacy_revenue_report_pandas._make_process_dataframe') def test_build_full_document_subaccount( _make_process_dataframe_mock, get_statement_period_mock, snowflake_mock, create_report_file_pandas_mock, statement_period_fixture, ): """Test building a legacy revenue full report for a subaccount (which is also a non-d3).""" account_id = 24601 statement_period_ids = '123' subaccount_id = 54321 subaccount = { 'SUBACCOUNTNAME': 'Subbb', 'SUBACCOUNT_SPLIT_TYPE': 'Gross', 'COMMISSIONOVERRIDE': Decimal('0.80'), } file_type = 'xls' number_format = 'eu' fact_sales = [ {'FX_GROSS': Decimal('10.0')}, {'FX_GROSS': Decimal('20.0')}, {'FX_GROSS': Decimal('30.0')}, ] total_rows = 3 generation_date = datetime.today().strftime('%Y%m%d') expected_path = path.join(FILE_OUTPUT_PATH, f'{generation_date}_July_2022_fullreport_Subbb.xls') get_statement_period_mock.return_value = statement_period_fixture snowflake_mock.get_workstation_fact_sales_pandas.return_value = (fact_sales, total_rows) snowflake_mock.is_distributor.return_value = {'IS_DISTRIBUTOR': 'N'} snowflake_mock.get_subaccount.return_value = subaccount create_report_file_pandas_mock.return_value = expected_path _make_process_dataframe_mock.return_value = 'FUNC' result = legacy_revenue_report_pandas.build_full_document( account_id, statement_period_ids, subaccount_id, file_type, number_format ) call_args = create_report_file_pandas_mock.call_args[0] actual_generator = call_args[0] assert result == expected_path _make_process_dataframe_mock.assert_called_once_with(subaccount, False, number_format) create_report_file_pandas_mock.assert_called_once() assert hasattr(actual_generator, '__iter__') and hasattr(actual_generator, '__next__') assert call_args[1] == total_rows assert call_args[2] == 'FUNC' assert call_args[3] == expected_path assert call_args[4] == file_type assert call_args[5] == number_format @patch('src.documents.legacy_revenue_report_pandas.create_report_file_pandas') @patch('src.documents.legacy_revenue_report_pandas.snowflake') @patch('src.documents.legacy_revenue_report_pandas.OwsAbacusAccount.get_account') @patch('src.documents.legacy_revenue_report_pandas.OwsRoyalties.get_statement_period') @patch('src.documents.legacy_revenue_report_pandas._make_process_dataframe') def test_build_physical_document( _make_process_dataframe_mock, get_statement_period_mock, get_account_mock, snowflake_mock, create_report_file_pandas_mock, account_fixture, statement_period_fixture, ): """Test building a legacy revenue physical report.""" account_id = 24601 statement_period_ids = '123' file_type = 'xls' number_format = 'us' data = [1, 2, 3] total_rows = 3 generation_date = datetime.today().strftime('%Y%m%d') expected_path = path.join( FILE_OUTPUT_PATH, f'{generation_date}_July_2022_physicalreport_Jeans_Nest.xls' ) get_account_mock.return_value = account_fixture get_statement_period_mock.return_value = statement_period_fixture snowflake_mock.get_workstation_physical_sales_pandas.return_value = (data, total_rows) snowflake_mock.is_distributor.return_value = {'IS_DISTRIBUTOR': 'Y'} create_report_file_pandas_mock.return_value = expected_path _make_process_dataframe_mock.return_value = 'FUNC' result = legacy_revenue_report_pandas.build_physical_document( account_id, statement_period_ids, None, file_type, number_format ) call_args = create_report_file_pandas_mock.call_args[0] actual_generator = call_args[0] assert result == expected_path _make_process_dataframe_mock.assert_called_once_with(None, True, number_format) create_report_file_pandas_mock.assert_called_once() assert hasattr(actual_generator, '__iter__') and hasattr(actual_generator, '__next__') assert call_args[1] == total_rows assert call_args[2] == 'FUNC' assert call_args[3] == expected_path assert call_args[4] == file_type assert call_args[5] == number_format @patch('src.documents.legacy_revenue_report_pandas.create_report_file_pandas') @patch('src.documents.legacy_revenue_report_pandas.snowflake') @patch('src.documents.legacy_revenue_report_pandas.OwsAbacusAccount.get_account') @patch('src.documents.legacy_revenue_report_pandas.OwsRoyalties.get_statement_period') @patch('src.documents.legacy_revenue_report_pandas._make_process_dataframe') def test_build_full_document_empty_report( _make_process_dataframe_mock, get_statement_period_mock, get_account_mock, snowflake_mock, create_report_file_pandas_mock, account_fixture, statement_period_fixture, ): """Test building a legacy revenue full report with 0 rows (empty report).""" account_id = 24601 statement_period_ids = '123' file_type = 'xls' number_format = 'us' mock_generator = MagicMock() data = mock_generator total_rows = 0 # Empty report generation_date = datetime.today().strftime('%Y%m%d') expected_path = path.join( FILE_OUTPUT_PATH, f'{generation_date}_July_2022_fullreport_Jeans_Nest.xls' ) get_account_mock.return_value = account_fixture get_statement_period_mock.return_value = statement_period_fixture snowflake_mock.get_workstation_fact_sales_pandas.return_value = (data, total_rows) snowflake_mock.is_distributor.return_value = {'IS_DISTRIBUTOR': 'Y'} create_report_file_pandas_mock.return_value = expected_path _make_process_dataframe_mock.return_value = 'FUNC' result = legacy_revenue_report_pandas.build_full_document( account_id, statement_period_ids, None, file_type, number_format ) call_args = create_report_file_pandas_mock.call_args[0] actual_generator = call_args[0] assert result == expected_path create_report_file_pandas_mock.assert_called_once() _make_process_dataframe_mock.assert_called_once_with(None, True, number_format) assert hasattr(actual_generator, '__iter__') and hasattr(actual_generator, '__next__') first_df = next(actual_generator) assert isinstance(first_df, pandas.DataFrame) assert len(first_df) == 0 assert list(first_df.columns) == list(legacy_revenue_report_pandas._COLUMN_MAPPING.keys()) assert call_args[1] == total_rows assert call_args[2] == 'FUNC' assert call_args[3] == expected_path assert call_args[4] == file_type assert call_args[5] == number_format @pytest.mark.parametrize( 'generation_date,statement_period_ids,filters,account_name,extension,period_name,transaction_type_names,expected_filename,expected_type_ids_call', [ # noqa:E501 ( '20240115', (123,), {'transaction_type_ids': [1, 2, 3]}, 'Test_Account', 'xls', 'December 2018', ['Download', 'Stream'], '20240115_December_2018_RevDetLegacy_Download-Stream_Test_Account.xls', [1, 2, 3], ), ( '20240115', (123,), {'transaction_type_ids': [1, 2, 3, 4, 5]}, 'Test_Account', 'txt', 'January 2019', ['Download', 'Stream', 'Ad-Supported', 'Physical', 'Ringtone'], '20240115_January_2019_RevDetLegacy_Download-Stream-Ad-Supported-andmore_Test_Account.txt', [1, 2, 3, 4, 5], ), ( '20240115', (123,), {'exclude_transaction_type_ids': [5, 6]}, 'Test_Account', 'xls', 'March 2020', ['Physical', 'Ringtone'], '20240115_March_2020_RevDetLegacy_Excl-Physical-Ringtone_Test_Account.xls', [5, 6], ), ( '20240115', (123,), {'variant': 'physical'}, 'Test_Account', 'xls', 'June 2021', None, '20240115_June_2021_RevDetLegacy_Physical_Test_Account.xls', None, ), ( '20240115', (123,), {'variant': 'digital'}, 'Test_Account', 'xls', 'June 2021', None, '20240115_June_2021_RevDetLegacy_Digital_Test_Account.xls', None, ), ( '20240115', (123,), {'variant': 'all'}, 'Test_Account', 'xls', 'June 2021', None, '20240115_June_2021_RevDetLegacy_Test_Account.xls', None, ), ( '20240115', (123,), {}, 'Test_Account', 'txt', 'August 2022', None, '20240115_August_2022_RevDetLegacy_Test_Account.txt', None, ), ( '20240115', (123,), {'transaction_type_ids': [1, 2], 'variant': 'physical'}, 'Test_Account', 'xls', 'October 2023', ['Download', 'Stream'], '20240115_October_2023_RevDetLegacy_Download-Stream_Test_Account.xls', [1, 2], ), ], ) @patch('src.documents.legacy_revenue_report_pandas.snowflake') def test_generate_custom_filename( snowflake_mock, generation_date, statement_period_ids, filters, account_name, extension, period_name, transaction_type_names, expected_filename, expected_type_ids_call, ): """Test generating custom filename with various filter configurations.""" snowflake_mock.get_statement_periods_name.return_value = {'STATEMENT_PERIOD_NAME': period_name} if transaction_type_names is not None: snowflake_mock.get_transaction_types_names.return_value = transaction_type_names result = legacy_revenue_report_pandas._generate_custom_filename( generation_date, statement_period_ids, filters, account_name, extension ) assert result == expected_filename snowflake_mock.get_statement_periods_name.assert_called_once_with(statement_period_ids[0]) if expected_type_ids_call is not None: snowflake_mock.get_transaction_types_names.assert_called_once_with(expected_type_ids_call) else: snowflake_mock.get_transaction_types_names.assert_not_called() @patch('src.documents.legacy_revenue_report_pandas.create_report_file_pandas') @patch('src.documents.legacy_revenue_report_pandas.snowflake') @patch('src.documents.legacy_revenue_report_pandas.OwsAbacusAccount.get_account') @patch('src.documents.legacy_revenue_report_pandas.OwsRoyalties.get_statement_period') @patch('src.documents.legacy_revenue_report_pandas._make_process_dataframe') def test_build_full_document_with_filters( _make_process_dataframe_mock, get_statement_period_mock, get_account_mock, snowflake_mock, create_report_file_pandas_mock, account_fixture, statement_period_fixture, ): """Test building a legacy revenue full report with filters.""" account_id = 24601 statement_period_ids = '123' file_type = 'xls' number_format = 'us' filters = {'transaction_type_ids': [1, 2], 'variant': 'digital'} mock_generator = MagicMock() data = mock_generator total_rows = 5 generation_date = datetime.today().strftime('%Y%m%d') expected_filename = ( f'{generation_date}_December_2018_RevDetLegacy_Download-Stream_Jeans_Nest.xls' # noqa:E501 ) expected_path = path.join(FILE_OUTPUT_PATH, expected_filename) get_account_mock.return_value = account_fixture get_statement_period_mock.return_value = statement_period_fixture snowflake_mock.get_workstation_fact_sales_pandas.return_value = (data, total_rows) snowflake_mock.is_distributor.return_value = {'IS_DISTRIBUTOR': 'Y'} snowflake_mock.get_statement_periods_name.return_value = { 'STATEMENT_PERIOD_NAME': 'December 2018' } snowflake_mock.get_transaction_types_names.return_value = ['Download', 'Stream'] create_report_file_pandas_mock.return_value = expected_path _make_process_dataframe_mock.return_value = 'FUNC' result = legacy_revenue_report_pandas.build_full_document( account_id, statement_period_ids, None, file_type, number_format, filters ) assert result == expected_path snowflake_mock.get_workstation_fact_sales_pandas.assert_called_once_with( (123,), account_id, None, filters ) snowflake_mock.get_statement_periods_name.assert_called_once_with(123) snowflake_mock.get_transaction_types_names.assert_called_once_with([1, 2]) @patch('src.documents.legacy_revenue_report_pandas.snowflake') def test_generate_custom_filename_max_length(snowflake_mock): """Test that generated filename doesn't exceed 255 character limit by truncating filter.""" generation_date = '20240115' statement_period_ids = (123,) account_name = 'Test_Account' extension = 'txt' long_filter_names = ['VeryLongTransactionTypeName' * 3 + str(i) for i in range(10)] filters = {'transaction_type_ids': list(range(1, 11))} snowflake_mock.get_statement_periods_name.return_value = { 'STATEMENT_PERIOD_NAME': 'December 2023' } snowflake_mock.get_transaction_types_names.return_value = long_filter_names result = legacy_revenue_report_pandas._generate_custom_filename( generation_date, statement_period_ids, filters, account_name, extension ) assert len(result) <= 255 assert result.endswith('.txt') assert 'andmore' in result assert account_name in result snowflake_mock.get_statement_periods_name.assert_called_once_with(statement_period_ids[0]) @patch('src.documents.legacy_revenue_report_pandas.create_report_file_pandas') @patch('src.documents.legacy_revenue_report_pandas.snowflake') @patch('src.documents.legacy_revenue_report_pandas.OwsAbacusAccount.get_account') @patch('src.documents.legacy_revenue_report_pandas.OwsRoyalties.get_statement_period') @patch('src.documents.legacy_revenue_report_pandas._make_process_dataframe') def test_build_physical_document_with_filters( _make_process_dataframe_mock, get_statement_period_mock, get_account_mock, snowflake_mock, create_report_file_pandas_mock, account_fixture, statement_period_fixture, ): """Test building a legacy physical report with filters.""" account_id = 24601 statement_period_ids = '123' file_type = 'txt' number_format = 'eu' filters = {'variant': 'physical'} data = [1, 2, 3] total_rows = 3 generation_date = datetime.today().strftime('%Y%m%d') expected_filename = f'{generation_date}_September_2020_RevDetLegacy_Physical_Jeans_Nest.txt' # noqa:E501 expected_path = path.join(FILE_OUTPUT_PATH, expected_filename) get_account_mock.return_value = account_fixture get_statement_period_mock.return_value = statement_period_fixture snowflake_mock.get_workstation_physical_sales_pandas.return_value = (data, total_rows) snowflake_mock.is_distributor.return_value = {'IS_DISTRIBUTOR': 'Y'} snowflake_mock.get_statement_periods_name.return_value = { 'STATEMENT_PERIOD_NAME': 'September 2020' } create_report_file_pandas_mock.return_value = expected_path _make_process_dataframe_mock.return_value = 'FUNC' result = legacy_revenue_report_pandas.build_physical_document( account_id, statement_period_ids, None, file_type, number_format, filters ) assert result == expected_path snowflake_mock.get_workstation_physical_sales_pandas.assert_called_once_with( (123,), account_id, None, filters ) snowflake_mock.get_statement_periods_name.assert_called_once_with(123)