-
Notifications
You must be signed in to change notification settings - Fork 10
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
8cce6fe
commit 2ecbf6a
Showing
2 changed files
with
230 additions
and
321 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,93 +1,156 @@ | ||
| import datetime | ||
| import unittest | ||
| from pain001.core.core import process_files | ||
| from pain001.csv.load_csv_data import load_csv_data | ||
| from pain001.csv.validate_csv_data import validate_csv_data | ||
|
|
||
|
|
||
| def validate_csv_data(data): | ||
| """Validate the CSV data before processing it. | ||
| class TestLoadCsvData(unittest.TestCase): | ||
| def test_load_valid_csv(self): | ||
| file_path = "tests/data/valid_data.csv" | ||
| data = load_csv_data(file_path) | ||
| self.assertEqual(len(data), 2) | ||
| self.assertEqual(data[0]["id"], "1") | ||
|
|
||
| Args: | ||
| data (list): A list of dictionaries containing the CSV data. | ||
| def test_load_empty_csv(self): | ||
| file_path = "tests/data/empty.csv" | ||
| with self.assertRaises(ValueError): | ||
| load_csv_data(file_path) | ||
|
|
||
| Returns: | ||
| bool: True if the data is valid, False otherwise. | ||
| """ | ||
| required_columns = { | ||
| "id": int, | ||
| "date": datetime.datetime, | ||
| "nb_of_txs": int, | ||
| "ctrl_sum": float, | ||
| "initiator_name": str, | ||
| "payment_information_id": str, | ||
| "payment_method": str, | ||
| "batch_booking": bool, | ||
| "service_level_code": str, | ||
| "requested_execution_date": datetime.datetime, | ||
| "debtor_name": str, | ||
| "debtor_account_IBAN": str, | ||
| "debtor_agent_BIC": str, | ||
| "forwarding_agent_BIC": str, | ||
| "charge_bearer": str, | ||
| "payment_id": str, | ||
| "payment_amount": float, | ||
| "currency": str, | ||
| "creditor_agent_BIC": str, | ||
| "creditor_name": str, | ||
| "creditor_account_IBAN": str, | ||
| "remittance_information": str, | ||
| } | ||
| def test_load_csv_with_invalid_data(self): | ||
| file_path = "tests/data/invalid_data.csv" | ||
| data = load_csv_data(file_path) | ||
| self.assertFalse(validate_csv_data(data)) | ||
|
|
||
| if not data: | ||
| print("Error: The CSV data is empty.") | ||
| return False | ||
| def test_load_single_row_csv(self): | ||
| file_path = "tests/data/single_row.csv" | ||
| data = load_csv_data(file_path) | ||
| self.assertEqual(len(data), 1) | ||
| self.assertEqual(data[0]["id"], "1") | ||
|
|
||
| is_valid = True | ||
| def test_load_single_column_csv(self): | ||
| file_path = "tests/data/single_column.csv" | ||
| data = load_csv_data(file_path) | ||
| self.assertEqual(len(data), 3) | ||
| self.assertEqual(data[0]["id"], "1") | ||
|
|
||
| for row in data: | ||
| missing_columns = [] | ||
| invalid_columns = [] | ||
| for column, data_type in required_columns.items(): | ||
| value = row.get(column) | ||
| if value is None or value.strip() == "": | ||
| missing_columns.append(column) | ||
| is_valid = False | ||
| else: | ||
| try: | ||
| if data_type == int: | ||
| int(value) | ||
| elif data_type == float: | ||
| float(value) | ||
| elif data_type == bool: | ||
| if value.strip().lower() not in [ | ||
| "true", | ||
| "false", | ||
| ]: | ||
| raise ValueError | ||
| elif data_type == datetime.datetime: | ||
| try: | ||
| if value.endswith("Z"): | ||
| value = value[:-1] + "+00:00" | ||
| datetime.datetime.fromisoformat(value) | ||
| except ValueError: | ||
| datetime.datetime.strptime( | ||
| value, "%Y-%m-%d" | ||
| ) | ||
| else: | ||
| str(value) | ||
| except ValueError: | ||
| invalid_columns.append(column) | ||
| is_valid = False | ||
| if missing_columns: | ||
| print( | ||
| f"Error: Missing value(s) for column(s) {missing_columns} " | ||
| f"in row: {row}" | ||
| ) | ||
| if invalid_columns: | ||
| expected_types = [ | ||
| required_columns[col].__name__ | ||
| for col in invalid_columns | ||
| ] | ||
| print( | ||
| f"Error: Invalid data type for column(s) {invalid_columns}, " | ||
| f"expected {expected_types} in row: {row}" | ||
| ) | ||
|
|
||
| return is_valid | ||
| class TestProcessFiles(unittest.TestCase): | ||
| def test_successful_execution(self): | ||
| process_files( | ||
| "pain.001.001.03", | ||
| "tests/data/template.xml", | ||
| "tests/data/template.xsd", | ||
| "tests/data/template.csv", | ||
| ) | ||
|
|
||
| def test_valid_xml_message_type(self): | ||
| process_files( | ||
| "pain.001.001.03", | ||
| "tests/data/template.xml", | ||
| "tests/data/template.xsd", | ||
| "tests/data/template.csv", | ||
| ) | ||
|
|
||
|
|
||
| class TestValidateCsvData(unittest.TestCase): | ||
| def test_validate_csv_with_valid_data(self): | ||
| data = [ | ||
| { | ||
| "id": "1", | ||
| "date": "2023-03-10T15:30:47.000Z", | ||
| "nb_of_txs": "2", | ||
| "initiator_name": "John Doe", | ||
| "initiator_street_name": "John's Street", | ||
| "initiator_building_number": "1", | ||
| "initiator_postal_code": "12345", | ||
| "initiator_town_name": "John's Town", | ||
| "initiator_country_code": "DE", | ||
| "payment_information_id": "Payment-Info-12345", | ||
| "payment_method": "TRF", | ||
| "batch_booking": "true", | ||
| "requested_execution_date": "2023-03-12", | ||
| "debtor_name": "Acme Corp", | ||
| "debtor_street_name": "Acme Street", | ||
| "debtor_building_number": "2", | ||
| "debtor_postal_code": "67890", | ||
| "debtor_town_name": "Acme Town", | ||
| "debtor_country_code": "DE", | ||
| "debtor_account_IBAN": "DE75512108001245126162", | ||
| "debtor_agent_BIC": "BANKDEFFXXX", | ||
| "charge_bearer": "SLEV", | ||
| "payment_id": "PaymentID6789", | ||
| "payment_amount": "150", | ||
| "currency": "EUR", | ||
| "payment_currency": "EUR", | ||
| "ctrl_sum": "15000", | ||
| "creditor_agent_BIC": "SPUEDE2UXXX", | ||
| "creditor_name": "Global Tech", | ||
| "creditor_street_name": "Global Street", | ||
| "creditor_building_number": "3", | ||
| "creditor_postal_code": "11223", | ||
| "creditor_town_name": "Global Town", | ||
| "creditor_country_code": "DE", | ||
| "creditor_account_IBAN": "DE68210501700024690959", | ||
| "purpose_code": "OTHR", | ||
| "reference_number": "Invoice-98765", | ||
| "reference_date": "2023-03-09", | ||
| "service_level_code": "SEPA", | ||
| "forwarding_agent_BIC": "SPUEDE2UXXX", | ||
| "remittance_information": "Invoice-12345", | ||
| "charge_account_IBAN": "CHARGE-IBAN-12345", | ||
| } | ||
| ] | ||
| self.assertTrue(validate_csv_data(data)) | ||
|
|
||
| def test_validate_csv_with_invalid_data(self): | ||
| data = [ | ||
| { | ||
| "id": "1", | ||
| "date": "not-a-date", | ||
| "nb_of_txs": "2", | ||
| "initiator_name": "John Doe", | ||
| "initiator_street_name": "John's Street", | ||
| "initiator_building_number": "1", | ||
| "initiator_postal_code": "12345", | ||
| "initiator_town_name": "John's Town", | ||
| "initiator_country_code": "DE", | ||
| "payment_information_id": "Payment-Info-12345", | ||
| "payment_method": "TRF", | ||
| "batch_booking": "true", | ||
| "requested_execution_date": "2023-03-12", | ||
| "debtor_name": "Acme Corp", | ||
| "debtor_street_name": "Acme Street", | ||
| "debtor_building_number": "2", | ||
| "debtor_postal_code": "67890", | ||
| "debtor_town_name": "Acme Town", | ||
| "debtor_country_code": "DE", | ||
| "debtor_account_IBAN": "DE75512108001245126162", | ||
| "debtor_agent_BIC": "BANKDEFFXXX", | ||
| "charge_bearer": "SLEV", | ||
| "payment_id": "PaymentID6789", | ||
| "payment_amount": "150", | ||
| "currency": "EUR", | ||
| "payment_currency": "EUR", | ||
| "ctrl_sum": "15000", | ||
| "creditor_agent_BIC": "SPUEDE2UXXX", | ||
| "creditor_name": "Global Tech", | ||
| "creditor_street_name": "Global Street", | ||
| "creditor_building_number": "3", | ||
| "creditor_postal_code": "11223", | ||
| "creditor_town_name": "Global Town", | ||
| "creditor_country_code": "DE", | ||
| "creditor_account_IBAN": "DE68210501700024690959", | ||
| "purpose_code": "OTHR", | ||
| "reference_number": "Invoice-98765", | ||
| "reference_date": "2023-03-09", | ||
| "service_level_code": "SEPA", | ||
| "forwarding_agent_BIC": "SPUEDE2UXXX", | ||
| "remittance_information": "Invoice-12345", | ||
| "charge_account_IBAN": "CHARGE-IBAN-12345", | ||
| } | ||
| ] | ||
| self.assertFalse(validate_csv_data(data)) | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| unittest.main() |
Oops, something went wrong.