"""Test for the document generation utilities.""" import csv from decimal import Decimal from unittest.mock import call from unittest.mock import MagicMock from unittest.mock import mock_open from unittest.mock import patch import pandas as pd import pytest from src.documents import utils from src.documents.utils import process_transactions from src.utils.constants import Encoding from src.utils.constants import NumberFormat from src.utils.error_handling import LambdaException @patch('src.documents.utils.csv') @patch('src.documents.utils.open') def test_write_file_csv(open_mock, csv_mock): """Test writing a CSV file.""" headers = ['one', 'two'] iterator = iter(['row1', 'row2', 'row3', 'row4']) max_rows = 3 destination_path = '/home/test.csv' utils._write_file(headers, iterator, max_rows, destination_path, file_type='csv') assert csv_mock.writer().writerow.call_count == 4 csv_mock.writer().writerow.assert_has_calls( [ call(headers), call('row1'), call('row2'), call('row3'), ] ) @patch('src.documents.utils.csv') @patch('src.documents.utils.open') def test_write_file_txt(open_mock, csv_mock): """Test writing a TXT (tab-delimited) file.""" headers = ['col1', 'col2'] iterator = iter([['v1', 'v2'], ['v3', 'v4'], ['v5', 'v6'], ['v7', 'v8']]) max_rows = 3 destination_path = '/home/test.txt' utils._write_file(headers, iterator, max_rows, destination_path, file_type='txt') assert csv_mock.writer().writerow.call_count == 4 csv_mock.writer().writerow.assert_has_calls( [ call(headers), call(['v1', 'v2']), call(['v3', 'v4']), call(['v5', 'v6']), ] ) @patch('src.documents.utils.csv') @patch('src.documents.utils.open') def test_write_file_xls(open_mock, csv_mock): """Test writing an XLS file.""" headers = ['colA', 'colB'] iterator = iter([['x1', 'x2'], ['x3', 'x4'], ['x5', 'x6']]) max_rows = 2 destination_path = '/home/test.xls' utils._write_file(headers, iterator, max_rows, destination_path, file_type='xls') assert csv_mock.writer().writerow.call_count == 3 csv_mock.writer().writerow.assert_has_calls( [ call(headers), call(['x1', 'x2']), call(['x3', 'x4']), ] ) @patch('src.documents.utils.csv') @patch('src.documents.utils.open') def test_write_file_with_row_processor(open_mock, csv_mock): """Test writing a file with row_processor.""" headers = ['one', 'two'] iterator = iter(['row1', 'row2', 'row3', 'row4']) max_rows = 3 destination_path = '/home/test.csv' row_processor = MagicMock() row_processor.return_value = ['processed1', 'processed2'] utils._write_file(headers, iterator, max_rows, destination_path, row_processor, file_type='csv') assert csv_mock.writer().writerow.call_count == 4 csv_mock.writer().writerow.assert_has_calls( [ call(headers), call(['processed1', 'processed2']), call(['processed1', 'processed2']), call(['processed1', 'processed2']), ] ) @pytest.mark.parametrize( 'value, number_format, expected', [ (Decimal('1234567.890123'), NumberFormat.EU, '1.234.567,890123'), (Decimal('0'), NumberFormat.EU, '0'), (Decimal('-1000'), NumberFormat.EU, '-1.000'), (Decimal('0.000035'), NumberFormat.EU, '0,000035'), (Decimal('1234567.890123'), NumberFormat.US, '1,234,567.890123'), (Decimal('0'), NumberFormat.US, '0'), (Decimal('-1000'), NumberFormat.US, '-1,000'), (Decimal('0.000035'), NumberFormat.US, '0.000035'), ], ) def test_make_float_formatter(value, number_format, expected): """Test making a float formatter function.""" formatter = utils._make_float_formatter(number_format) result = formatter(value) assert expected == result @patch('src.documents.utils._make_float_formatter') @patch('builtins.open', new_callable=mock_open) @pytest.mark.parametrize( 'file_type, number_format, separator, encoding, quoting', [ ('csv', NumberFormat.US, ',', 'utf-8-sig', csv.QUOTE_NONNUMERIC), ('txt', NumberFormat.EU, '\t', 'utf-8', csv.QUOTE_NONE), ('xls', NumberFormat.US, '\t', 'utf-16', csv.QUOTE_ALL), ], ) def test_write_file_pandas( m_open, mock_make_float_formatter, file_type, number_format, separator, encoding, quoting ): """Test writing a file (using Pandas).""" dataframe = MagicMock() dataframe.__len__.return_value = 100 dataframe.applymap.return_value = dataframe destination_path = '/tmp/test.out' mock_make_float_formatter.return_value = 'formatter' utils._write_file_pandas(dataframe, destination_path, file_type, number_format) m_open.assert_called_once_with(destination_path, 'w', encoding=encoding, newline='') dataframe.to_csv.assert_called_once() args, kwargs = dataframe.to_csv.call_args assert args and args[0] is not None assert kwargs['index'] is False assert kwargs['escapechar'] == '\\' assert kwargs['sep'] == separator assert kwargs['quoting'] == quoting assert kwargs['lineterminator'] == '\r\n' assert kwargs['float_format'] == mock_make_float_formatter.return_value assert kwargs['na_rep'] == '' @pytest.mark.parametrize( 'file_type, number_format, separator, encoding, quoting', [ ('csv', NumberFormat.US, ',', 'utf-8-sig', 'non_numeric'), ('txt', NumberFormat.EU, '\t', 'utf-8', 'never'), ('xls', NumberFormat.US, '\t', 'utf-16', 'always'), ], ) @patch('src.documents.utils._sanitize_for_tsv') @patch('src.documents.utils.io') @patch('src.documents.utils.pa') @patch('src.documents.utils.pl') def test_write_file_pandas_polars( mock_pl, mock_pa, mock_io, mock_sanitize, file_type, number_format, separator, encoding, quoting ): """Test writing file to the memory buffer (using Polars).""" dataframe = pd.DataFrame({'col1': [1, 2], 'col2': [3, 4]}) dataframe_polars = MagicMock(write_csv=MagicMock()) mock_arrow_table = MagicMock() mock_pa.Table.from_pandas.return_value = mock_arrow_table mock_pl.from_arrow.return_value = dataframe_polars mock_pl.from_pandas.return_value = dataframe_polars mock_internal_buffer = MagicMock( seek=MagicMock(), read=MagicMock( decode=MagicMock( encode=MagicMock(), ) ), ) mock_io.BytesIO.return_value = mock_internal_buffer mock_sanitize.return_value = dataframe utils._write_file_pandas_polars(dataframe, file_type, number_format, mock_internal_buffer) args, kwargs = dataframe_polars.write_csv.call_args assert args and args[0] is not None assert kwargs['separator'] == separator assert kwargs['include_header'] assert kwargs['line_terminator'] == '\r\n' assert kwargs['null_value'] == '' assert kwargs['decimal_comma'] == (number_format == NumberFormat.EU) assert kwargs['quote_style'] == quoting mock_pa.Table.from_pandas.assert_called_once_with(dataframe) mock_pl.from_arrow.assert_called_once_with(mock_arrow_table) mock_pl.from_pandas.assert_not_called() dataframe_polars.write_csv.assert_called() @pytest.mark.parametrize( 'file_type, number_format, separator, encoding, quoting', [ ('csv', NumberFormat.US, ',', 'utf-8-sig', 'non_numeric'), ('txt', NumberFormat.EU, '\t', 'utf-8', 'never'), ('xls', NumberFormat.US, '\t', 'utf-16', 'always'), ], ) @patch('src.documents.utils._sanitize_for_tsv') @patch('src.documents.utils.io') @patch('src.documents.utils.pa') @patch('src.documents.utils.pl') def test_write_file_pandas_polars_large_dataframe( mock_pl, mock_pa, mock_io, mock_sanitize, file_type, number_format, separator, encoding, quoting ): """Test writing file to the memory buffer (using Polars) with large dataframe.""" dataframe = pd.DataFrame({'col1': [1, 2], 'col2': [3, 4]}) dataframe_polars = MagicMock(write_csv=MagicMock()) mock_arrow_table = MagicMock() mock_pa.Table.from_pandas.return_value = mock_arrow_table mock_pl.from_arrow.return_value = dataframe_polars mock_pl.from_pandas.return_value = dataframe_polars mock_internal_buffer = MagicMock( seek=MagicMock(), read=MagicMock( decode=MagicMock( encode=MagicMock(), ) ), ) mock_io.BytesIO.return_value = mock_internal_buffer mock_sanitize.return_value = dataframe with patch('src.documents.utils.POLARS_ARROW_CONVERSION_THRESHOLD', 2): utils._write_file_pandas_polars(dataframe, file_type, number_format, mock_internal_buffer) args, kwargs = dataframe_polars.write_csv.call_args assert args and args[0] is not None assert kwargs['separator'] == separator assert kwargs['include_header'] assert kwargs['line_terminator'] == '\r\n' assert kwargs['null_value'] == '' assert kwargs['decimal_comma'] == (number_format == NumberFormat.EU) assert kwargs['quote_style'] == quoting mock_pa.Table.from_pandas.assert_not_called() mock_pl.from_arrow.assert_not_called() mock_pl.from_pandas.assert_called_once_with(dataframe) dataframe_polars.write_csv.assert_called() if file_type != 'txt': mock_internal_buffer.seek.assert_called() mock_internal_buffer.read.assert_called() mock_internal_buffer.read.return_value.decode.assert_called_once_with(Encoding.UTF8.value) # noqa: E501 mock_internal_buffer.read.return_value.decode.return_value.encode.assert_called_with( encoding, errors='replace' ) # noqa: E501 else: mock_sanitize.assert_called() @patch('builtins.open') def test_write_file_pandas_empty(mock_open): """Test writing a file with an empty DataFrame (using Pandas).""" dataframe = None destination_path = '/tmp/test.txt' utils._write_file_pandas(dataframe, destination_path, 'txt', NumberFormat.EU) mock_open.assert_called_once_with(destination_path, 'w') mock_open().close.assert_called_once_with() @patch('src.documents.utils.pl') @patch('src.documents.utils.pa') @patch('src.documents.utils.io') @patch('src.documents.utils._sanitize_for_tsv') def test_write_file_pandas_polars_empty_dataframe(mock_sanitize, mock_io, mock_pa, mock_pl): """Test writing empty DataFrame with headers using polars (after processing).""" processed_columns = ['Period', 'ISRC', 'Unit Price', 'Quantity'] dataframe = pd.DataFrame(columns=processed_columns) dataframe_polars = MagicMock() dataframe_polars.write_csv = MagicMock() mock_arrow_table = MagicMock() mock_pa.Table.from_pandas.return_value = mock_arrow_table mock_pl.from_arrow.return_value = dataframe_polars mock_pl.from_pandas.return_value = dataframe_polars mock_internal_buffer = MagicMock() mock_internal_buffer.seek = MagicMock() mock_internal_buffer.read = MagicMock(return_value=b'Period\tISRC\tUnit Price\tQuantity\r\n') mock_io.BytesIO.return_value = mock_internal_buffer file_type = 'xls' number_format = NumberFormat.US utils._write_file_pandas_polars(dataframe, file_type, number_format, mock_internal_buffer) args, kwargs = dataframe_polars.write_csv.call_args assert kwargs['include_header'] is True assert kwargs['separator'] == '\t' assert kwargs['line_terminator'] == '\r\n' assert kwargs['null_value'] == '' assert kwargs['quote_style'] == 'always' mock_pa.Table.from_pandas.assert_called_once_with(dataframe) mock_pl.from_arrow.assert_called_once_with(mock_arrow_table) mock_pl.from_pandas.assert_not_called() dataframe_polars.write_csv.assert_called_once() @patch('src.documents.utils._write_file') @patch('src.documents.utils.os') @patch('src.documents.utils.ZipFile') def test_create_report_file_single_file(ZipFile_mock, os_mock, _write_file_mock): """Test creating a report file with a single file as the result (now always zipped).""" headers = ['a', 'b'] iterator = iter([1, 2, 3]) total_rows = 3 destination_path = '/home/test.csv' row_processor = MagicMock() file_type = 'csv' mock_zip_file = MagicMock() ZipFile_mock().__enter__.return_value = mock_zip_file result = utils.create_report_file( headers, iterator, total_rows, destination_path, row_processor, file_type ) assert result == '/home/test.zip' _write_file_mock.assert_called_once_with( headers, iterator, utils.MAX_ROWS_PER_FILE, '/home/test_1.csv', row_processor, file_type ) mock_zip_file.write.assert_called_once_with('/home/test_1.csv', 'test_1.csv') os_mock.remove.assert_called_once_with('/home/test_1.csv') @patch('src.documents.utils._write_file') @patch('src.documents.utils.os') @patch('src.documents.utils.ZipFile') def test_create_report_file_single_file_tsv(ZipFile_mock, os_mock, _write_file_mock): """Test creating a TSV report file when total_rows is within limit (now always zipped).""" headers = ['col1', 'col2'] iterator = iter([['v1', 'v2'], ['v3', 'v4']]) total_rows = 2 destination_path = '/home/test.tsv' row_processor = MagicMock() file_type = 'txt' mock_zip_file = MagicMock() ZipFile_mock().__enter__.return_value = mock_zip_file result = utils.create_report_file( headers, iterator, total_rows, destination_path, row_processor, file_type ) assert result == '/home/test.zip' _write_file_mock.assert_called_once_with( headers, iterator, utils.MAX_ROWS_PER_FILE, '/home/test_1.tsv', row_processor, file_type ) mock_zip_file.write.assert_called_once_with('/home/test_1.tsv', 'test_1.tsv') os_mock.remove.assert_called_once_with('/home/test_1.tsv') @patch('src.documents.utils._write_file') @patch('src.documents.utils.os') @patch('src.documents.utils.ZipFile') def test_create_report_file_zip_file(ZipFile_mock, os_mock, _write_file_mock): """Test creating a report file with a zip file as the result.""" headers = ['a', 'b'] iterator = iter([1, 2, 3]) total_rows = utils.MAX_ROWS_PER_FILE * 3 destination_path = '/home/test.csv' row_processor = MagicMock() mock_zip_file = MagicMock() file_type = 'csv' ZipFile_mock().__enter__.return_value = mock_zip_file result = utils.create_report_file( headers, iterator, total_rows, destination_path, row_processor ) assert result == '/home/test.zip' assert _write_file_mock.call_count == 3 _write_file_mock.assert_has_calls( [ call( headers, iterator, utils.MAX_ROWS_PER_FILE, '/home/test_1.csv', row_processor, file_type, ), call( headers, iterator, utils.MAX_ROWS_PER_FILE, '/home/test_2.csv', row_processor, file_type, ), call( headers, iterator, utils.MAX_ROWS_PER_FILE, '/home/test_3.csv', row_processor, file_type, ), ] ) assert mock_zip_file.write.call_count == 3 mock_zip_file.write.assert_has_calls( [ call('/home/test_1.csv', 'test_1.csv'), call('/home/test_2.csv', 'test_2.csv'), call('/home/test_3.csv', 'test_3.csv'), ] ) assert os_mock.remove.call_count == 3 os_mock.remove.assert_has_calls( [ call('/home/test_1.csv'), call('/home/test_2.csv'), call('/home/test_3.csv'), ] ) @patch('src.documents.utils._write_file_pandas_polars') @patch('src.documents.utils.pandas.concat') @patch('src.documents.utils.ZipFile') def test_create_report_file_pandas_always_zip( ZipFile_mock, concat_mock, write_file_pandas_polars_mock ): """Test creating a pandas report file with large dataframe that exceeds MAX_ROWS_PER_FILE.""" large_df = MagicMock() large_df.__len__ = MagicMock(return_value=utils.MAX_ROWS_PER_FILE + 500) first_chunk = MagicMock() first_chunk.__len__ = MagicMock(return_value=utils.MAX_ROWS_PER_FILE) remaining_chunk = MagicMock() remaining_chunk.__len__ = MagicMock(return_value=500) def slice_side_effect(key): if isinstance(key, slice): if key.start == 0 and key.stop == utils.MAX_ROWS_PER_FILE: return first_chunk elif key.start == utils.MAX_ROWS_PER_FILE and key.stop is None: return remaining_chunk return MagicMock() large_df.__getitem__.side_effect = slice_side_effect dataframe_generator = iter([large_df]) concat_mock.return_value = large_df total_rows = utils.MAX_ROWS_PER_FILE + 500 process_dataframe = MagicMock() destination_path = '/home/test.csv' file_type = 'csv' number_format = NumberFormat.US mock_zip_file = MagicMock() ZipFile_mock().__enter__.return_value = mock_zip_file result = utils.create_report_file_pandas( dataframe_generator, total_rows, process_dataframe, destination_path, file_type, number_format, ) assert result == '/home/test.zip' assert process_dataframe.call_count == 1 process_dataframe.assert_called_once_with(large_df) assert mock_zip_file.writestr.call_count == 2 def test_extract_vat_info(): """Test extracting VAT info.""" ledger_vat_data = [ { 'base_amount_payee_currency': 50, 'vat_amount_payee_currency': 10, 'vat_amount_vat_currency': 1, 'vat_rate': 20, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, { 'base_amount_payee_currency': 50, 'vat_amount_payee_currency': 10, 'vat_amount_vat_currency': 1, 'vat_rate': 20, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, { 'base_amount_payee_currency': 50, 'vat_amount_payee_currency': 5, 'vat_amount_vat_currency': 0.5, 'vat_rate': 10, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, ] expected = { 'vat_items': [ { 'base_amount_payee_currency': 100, 'vat_rate': 20, 'vat_amount_payee_currency': 20, 'vat_amount_vat_currency': 2, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, { 'base_amount_payee_currency': 50, 'vat_rate': 10, 'vat_amount_payee_currency': 5, 'vat_amount_vat_currency': 0.5, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, ], 'base_amount_total': 150, 'payee_total_vat': 25, 'vat_total_vat': 2.5, 'has_vat': True, } result = utils.extract_vat_info(ledger_vat_data, False) assert result == expected def test_extract_wht_info(): """Test extracting WHT info.""" ledger_vat_data = [ { 'base_amount_payee_currency': 50, 'wht_amount_payee_currency': 10, 'wht_amount_vat_currency': 1, 'wht_rate': 20, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, { 'base_amount_payee_currency': 50, 'wht_amount_payee_currency': 10, 'wht_amount_vat_currency': 1, 'wht_rate': 20, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, { 'base_amount_payee_currency': 50, 'wht_amount_payee_currency': 5, 'wht_amount_vat_currency': 0.5, 'wht_rate': 10, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, ] expected = { 'wht_items': [ { 'base_amount_payee_currency': 100, 'wht_rate': 20, 'wht_amount_payee_currency': 20, 'wht_amount_vat_currency': 2, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, { 'base_amount_payee_currency': 50, 'wht_rate': 10, 'wht_amount_payee_currency': 5, 'wht_amount_vat_currency': 0.5, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, ], 'total_wht_amount_payee_currency': 25, 'total_wht_amount_vat_currency': 2.5, 'has_wht': True, } result = utils.extract_wht_info(ledger_vat_data, False) assert result == expected def test_extract_vat_info_zero(): """Test extracting VAT info when the sums are zero.""" ledger_vat_data = [ { 'payee_currency_code': 'EUR', 'vat_currency_code': 'GBP', 'base_amount_payee_currency': -100.0, 'vat_rate': 10.0, 'vat_amount_payee_currency': -20.0, 'vat_amount_vat_currency': -20.0, }, { 'payee_currency_code': 'EUR', 'vat_currency_code': 'GBP', 'base_amount_payee_currency': -1000.0, 'vat_rate': 20.0, 'vat_amount_payee_currency': -200.0, 'vat_amount_vat_currency': -165.65, }, { 'payee_currency_code': 'EUR', 'vat_currency_code': 'GBP', 'base_amount_payee_currency': 100.0, 'vat_rate': 10.0, 'vat_amount_payee_currency': 20.0, 'vat_amount_vat_currency': 20.0, }, { 'payee_currency_code': 'EUR', 'vat_currency_code': 'GBP', 'base_amount_payee_currency': 1000.0, 'vat_rate': 20.0, 'vat_amount_payee_currency': 200.0, 'vat_amount_vat_currency': 165.65, }, ] expected = { 'vat_items': [ { 'base_amount_payee_currency': 0, 'vat_rate': 10, 'vat_amount_payee_currency': 0, 'vat_amount_vat_currency': 0, 'vat_currency_code': 'GBP', 'payee_currency_code': 'EUR', }, { 'base_amount_payee_currency': 0, 'vat_rate': 20, 'vat_amount_payee_currency': 0, 'vat_amount_vat_currency': 0, 'vat_currency_code': 'GBP', 'payee_currency_code': 'EUR', }, ], 'base_amount_total': 0, 'payee_total_vat': 0, 'vat_total_vat': 0, 'has_vat': True, } result = utils.extract_vat_info(ledger_vat_data, False) assert result == expected def test_extract_vat_info_distro_invoice(): """Test extracting VAT info with distro invoice values.""" ledger_vat_data = [ { 'base_amount_payee_currency': -50, 'vat_amount_payee_currency': -10, 'vat_amount_vat_currency': -1, 'vat_rate': 20, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, { 'base_amount_payee_currency': -50, 'vat_amount_payee_currency': -5, 'vat_amount_vat_currency': -0.5, 'vat_rate': 10, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, ] expected = { 'vat_items': [ { 'base_amount_payee_currency': 50, 'vat_rate': 20, 'vat_amount_payee_currency': 10, 'vat_amount_vat_currency': 1, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, { 'base_amount_payee_currency': 50, 'vat_rate': 10, 'vat_amount_payee_currency': 5, 'vat_amount_vat_currency': 0.5, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, ], 'base_amount_total': 100, 'payee_total_vat': 15, 'vat_total_vat': 1.5, 'has_vat': True, } result = utils.extract_vat_info(ledger_vat_data, True) assert result == expected def test_extract_vat_info_null_values(): """Test extracting VAT info when VAT fields are NULL.""" ledger_vat_data = [ { 'base_amount_payee_currency': 50, 'vat_amount_payee_currency': None, 'vat_amount_vat_currency': None, 'vat_rate': None, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, { 'base_amount_payee_currency': 100, 'vat_amount_payee_currency': None, 'vat_amount_vat_currency': None, 'vat_rate': None, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, ] expected = { 'vat_items': [ { 'base_amount_payee_currency': 150, 'vat_rate': 0, 'vat_amount_payee_currency': 0, 'vat_amount_vat_currency': 0, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, ], 'base_amount_total': 150, 'payee_total_vat': 0, 'vat_total_vat': 0, 'has_vat': False, } result = utils.extract_vat_info(ledger_vat_data, False) assert result == expected def test_extract_wht_info_null_values(): """Test extracting WHT info when WHT fields are NULL.""" ledger_vat_data = [ { 'base_amount_payee_currency': 25, 'wht_amount_payee_currency': None, 'wht_amount_vat_currency': None, 'wht_rate': None, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, { 'base_amount_payee_currency': 25, 'wht_amount_payee_currency': None, 'wht_amount_vat_currency': None, 'wht_rate': None, 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', }, ] expected = { 'wht_items': [ { 'base_amount_payee_currency': Decimal(50), 'wht_rate': Decimal(0), 'wht_amount_payee_currency': Decimal(0), 'wht_amount_vat_currency': Decimal(0), 'vat_currency_code': 'GBP', 'payee_currency_code': 'NOK', } ], 'total_wht_amount_payee_currency': Decimal(0), 'total_wht_amount_vat_currency': Decimal(0), 'has_wht': False, } result = utils.extract_wht_info(ledger_vat_data, False) assert result == expected @pytest.mark.parametrize( 'input_data, expected', [ ( { 'country_code': 'BRA', 'address_1': '123 Street', 'address_2': ' ', 'city': 'Cityville', 'province': 'Fake State', 'zip': '24601', }, '123 Street Cityville 24601 Fake State BRA', ), ( { 'country_code': 'NOR', 'address_1': 'Schønings gt. 19A', 'address_2': 'Leil. 3A', 'city': 'Oslo', 'zip': '0362', }, 'Schønings gt. 19A Leil. 3A Oslo 0362 NOR', ), ({}, ''), ], ) def test_format_address(input_data, expected): """Test formatting address data.""" result = utils.format_address(input_data) assert result == expected @patch('src.documents.utils.pdfkit') @patch('src.documents.utils.render_template') def test_generate_pdf_file(render_template_mock, pdfkit_mock): """Test generating a PDF file.""" destination_path = 'test.pdf' template_file = 'page.html' data = {'name': 'Test'} utils.generate_pdf_file(destination_path, template_file, data) render_template_mock.assert_called_once_with(template_file, data) pdfkit_mock.from_string.assert_called_once() @pytest.mark.parametrize( 'sap_id,expected', [ # Altafonte ('2981', 'https://cdn.theorchard.io/assets/altafonte/icons/brand-text.png'), ('6020', 'https://cdn.theorchard.io/assets/altafonte/icons/brand-text.png'), # AWAL ('2820', 'https://cdn.theorchard.io/assets/awal/icons/brand-text.png'), ('2821', 'https://cdn.theorchard.io/assets/awal/icons/brand-text.png'), ('4914', 'https://cdn.theorchard.io/assets/awal/icons/brand-text.png'), ('4915', 'https://cdn.theorchard.io/assets/awal/icons/brand-text.png'), ('4916', 'https://cdn.theorchard.io/assets/awal/icons/brand-text.png'), # KNR ('4917', 'https://cdn.theorchard.io/assets/knr/icons/brand-text.png'), ('4918', 'https://cdn.theorchard.io/assets/knr/icons/brand-text.png'), ('4919', 'https://cdn.theorchard.io/assets/knr/icons/brand-text.png'), ('4920', 'https://cdn.theorchard.io/assets/knr/icons/brand-text.png'), # Orchard ('2963', 'https://cdn.theorchard.io/assets/orchard/icons/brand-text.png'), ], ) def test_get_logo(sap_id, expected): """Test getting a logo.""" result = utils.get_logo(sap_id) assert result == expected @pytest.mark.parametrize('sap_id', ['1', '1234', '999']) def test_get_logo_error(sap_id): """Test getting a logo that doesn't exist.""" with pytest.raises(LambdaException) as err: utils.get_logo(sap_id) assert str(err.value.message) == f'No logo for SAP ID: {sap_id}' @patch('src.documents.utils.jinja2') def test_render_template(jinja2_mock): """Test rendering a template file.""" template_file = 'page.html' data = {'name': 'Test'} expected = 'Template render yes!' jinja2_mock.Environment().get_template().render.return_value = expected result = utils.render_template(template_file, data) assert result == expected jinja2_mock.Environment().get_template.assert_called_with(template_file) jinja2_mock.Environment().get_template().render.assert_called_once_with(**data) @pytest.mark.parametrize( 'value, expected', [ ('This is a test', 'This_is_a_test'), ('"@,_i^-9b%X.', '____i_-9b_X_'), ('test123', 'test123'), ], ) def test_sanitize_filename(value, expected): """Test sanitizing filenames.""" result = utils.sanitize_filename(value) assert result == expected @pytest.mark.parametrize( 'value, expected', [ (123, 123), (Decimal('123.45'), Decimal('123.45')), ('Test', 'Test'), (Decimal('0.000000000'), 0), (Decimal('123.4567890000000000000'), Decimal('123.456789')), ], ) def test_sanitze_value(value, expected): """Test sanitizing values.""" result = utils.sanitize_value(value) assert str(result) == str(expected) @pytest.mark.parametrize( 'value, expected', [ (123, 123), (Decimal('123.45'), Decimal('123.45')), (Decimal('123.4567890123456789'), Decimal('123.456789')), (123.4567890123456789, 123.456789), ], ) def test_sanitze_value_with_rounding(value, expected): """Test sanitizing values with rounding.""" result = utils.sanitize_value(value, True) assert str(result) == str(expected) @pytest.mark.parametrize( 'row, number_format, expected', [ ( [Decimal('1234567.890123'), Decimal('0'), Decimal('1000'), Decimal('0.000035')], 'eu', ['1.234.567,890123', '0', '1.000', '0,000035'], ), ( [Decimal('1234567.890123'), Decimal('0'), Decimal('1000'), Decimal('0.000035')], 'us', ['1,234,567.890123', '0', '1,000', '0.000035'], ), ([Decimal('42'), 'not_a_number', Decimal('0.01')], 'eu', ['42', 'not_a_number', '0,01']), ( [Decimal('-0.00000001'), Decimal('0'), Decimal('-1.234')], 'eu', ['-0,00000001', '0', '-1,234'], ), ], ) def test_format_row_numbers(row, number_format, expected): """Test formating values according to number format.""" result = utils.format_row_numbers(row, number_format) assert result == expected @pytest.mark.parametrize( 'expected_transactions, flip_signs', [ ( [ { 'description': 'test ledger 1', 'quantity': 1, 'payee_currency_code': 'USD', 'amount': Decimal('106506.22'), 'amount_ex_tax': Decimal('106506.22'), 'vat_rate': Decimal('20.00'), 'wht_rate': Decimal('0'), } ], False, ), ( [ { 'description': 'test ledger 1', 'quantity': 1, 'payee_currency_code': 'USD', 'amount': Decimal('-106506.22'), 'amount_ex_tax': Decimal('-106506.22'), 'vat_rate': Decimal('20.00'), 'wht_rate': Decimal('0'), } ], True, ), ], ) def test_process_transactions( ledger_vat_summary_fixture, statement_period_fixture, expected_transactions, flip_signs ): """Test processing transactions.""" transactions = [] process_transactions( [ledger_vat_summary_fixture[0]], statement_period_fixture, transactions, flip_signs ) assert transactions == expected_transactions @patch('src.documents.utils._write_file_pandas_polars') @patch('src.documents.utils.ZipFile') def test_create_report_file_pandas_empty_data(ZipFile_mock, write_file_pandas_polars_mock): """Test creating a pandas report file with empty DataFrame that has headers.""" empty_df = pd.DataFrame(columns=['Period', 'ISRC', 'Unit Price', 'Quantity']) empty_generator = iter([empty_df]) process_dataframe = MagicMock() total_rows = 0 destination_path = '/home/test.xls' file_type = 'xls' number_format = NumberFormat.US mock_zip_file = MagicMock() ZipFile_mock().__enter__.return_value = mock_zip_file result = utils.create_report_file_pandas( empty_generator, total_rows, process_dataframe, destination_path, file_type, number_format ) assert result == '/home/test.zip' process_dataframe.assert_called_once() write_file_pandas_polars_mock.assert_called_once() mock_zip_file.writestr.assert_called_once()