diff --git a/flux_sdk/pension/utils/ascensus.py b/flux_sdk/pension/utils/ascensus_report_payroll_contributions.py similarity index 75% rename from flux_sdk/pension/utils/ascensus.py rename to flux_sdk/pension/utils/ascensus_report_payroll_contributions.py index 82ab0b97..75ee5f7d 100644 --- a/flux_sdk/pension/utils/ascensus.py +++ b/flux_sdk/pension/utils/ascensus_report_payroll_contributions.py @@ -4,12 +4,10 @@ from datetime import datetime from decimal import ROUND_HALF_UP, Decimal from enum import Enum -from io import IOBase, StringIO -from typing import Any, Optional, Union +from io import StringIO from flux_sdk.flux_core.data_models import ( ContributionType, - DeductionType, Employee, File, ) @@ -18,9 +16,6 @@ PayrollRunContribution, PayrollUploadSettings, ) -from flux_sdk.pension.capabilities.update_deduction_elections.data_models import ( - EmployeeDeductionSetting, -) logger = logging.getLogger(__name__) @@ -62,23 +57,6 @@ "EMPLOYEE WORK EMAIL", ] -COLUMNS_360 = [ - "RecordType", - "PlanId", - "EmployeeLastName", - "EmployeeFirstName", - "EmployeeMiddleInitial", - "EmployeeSSN", - "EffectiveDate", - "ContributionCode", - "DeferralPercent", - "DeferralAmount", - "EmployeeEligibilityDate", - "LoanNumber", - "LoanPaymentAmount", - "TotalLoanAmount", -] - STANDARD_DATE_FORMAT = "%m/%d/%Y" TWO_PLACES = Decimal(".01") @@ -403,141 +381,3 @@ def format_contributions_for_ascensus_vendor( file.content = ReportPayrollContributionsAscensusUtil.to_bytes(header + output.getvalue()) return file - - -class UpdateDeductionElectionsAscensusUtil: - """ - This class represents the "update deduction elections" capability for vendors utilizing - the Ascensus. The developer is supposed to implement - parse_deductions_for_ascensus method in their implementation. For further details regarding their - implementation details, check their documentation. - """ - - @staticmethod - def _create_eds_for_value( - deduction_type: DeductionType, - value: Union[str, Decimal], - percentage: bool, - ssn: str, - effective_date: datetime, - ) -> EmployeeDeductionSetting: - eds = EmployeeDeductionSetting() - eds.ssn = ssn - eds.effective_date = effective_date - eds.deduction_type = deduction_type - eds.value = Decimal(value) # type: ignore - eds.is_percentage = percentage - return eds - - @staticmethod - def _is_valid_amount(value) -> bool: - try: - Decimal(value) - return True - except Exception: - return False - - @staticmethod - def get_deduction_type(given_ded_type) -> Optional[DeductionType]: - ded_match_map = { - "4ROTH": DeductionType.ROTH_401K, - "4ROTC": DeductionType.ROTH_401K, - "401K": DeductionType._401K, - "401KC": DeductionType._401K, - "401L": DeductionType._401K_LOAN_PAYMENT, - "403B": DeductionType._403B, - "401A": DeductionType.AFTER_TAX_401K, - "401O": DeductionType._401K, - } - return ded_match_map.get(given_ded_type, None) - - @staticmethod - def _parse_deduction_rows( - row: dict[str, Any], result: list[EmployeeDeductionSetting] - ) -> list[EmployeeDeductionSetting]: - ssn = row["EmployeeSSN"] - deduction_type = UpdateDeductionElectionsAscensusUtil.get_deduction_type(row["ContributionCode"]) - eligibility_date = ( - datetime.strptime(row["EmployeeEligibilityDate"], "%m%d%Y") - if row["EmployeeEligibilityDate"] - else datetime.now() - ) - - if ( - UpdateDeductionElectionsAscensusUtil._is_valid_amount(row["DeferralAmount"]) - and UpdateDeductionElectionsAscensusUtil._is_valid_amount(row["DeferralPercent"]) - and deduction_type - ): - result.append( - UpdateDeductionElectionsAscensusUtil._create_eds_for_value( - deduction_type=deduction_type, - value=row["DeferralAmount"] - if row["DeferralAmount"] > row["DeferralPercent"] - else row["DeferralPercent"], - percentage=row["DeferralPercent"] > row["DeferralAmount"], - ssn=ssn, - effective_date=eligibility_date, - ) - ) - - return result - - @staticmethod - def _parse_loan_rows(row: dict[str, Any], ssn_to_loan_sum_map: dict[str, Decimal]) -> dict[str, Decimal]: - ssn = row["EmployeeSSN"] - if UpdateDeductionElectionsAscensusUtil._is_valid_amount(row["LoanPaymentAmount"]): - loan_value = Decimal(row["LoanPaymentAmount"]) - if ssn in ssn_to_loan_sum_map: - ssn_to_loan_sum_map[ssn] += loan_value - else: - ssn_to_loan_sum_map[ssn] = loan_value - - return ssn_to_loan_sum_map - - @staticmethod - def parse_deductions_for_ascensus(uri: str, stream: IOBase) -> list[EmployeeDeductionSetting]: - """ - This method receives a stream from which the developer is expected to return a list of EmployeeDeductionSetting - for each employee identifier (SSN). - :param uri: Contains the path of file - :param stream: Contains the stream - :return: list[EmployeeDeductionSetting] - """ - result: list[EmployeeDeductionSetting] = [] - - try: - reader = csv.DictReader(stream) # type: ignore - except Exception as e: - logger.error(f"[UpdateDeductionElectionsImpl.parse_deductions] Parse deductions failed due to message {e}") - return result - - ssn_to_loan_sum_map: dict[str, Decimal] = {} - - for row in reader: - try: - ssn = row["EmployeeSSN"] - record_type = row["RecordType"] - - if record_type == "D": - UpdateDeductionElectionsAscensusUtil._parse_deduction_rows(row, result) - elif record_type == "L": - UpdateDeductionElectionsAscensusUtil._parse_loan_rows(row, ssn_to_loan_sum_map) - else: - logger.error(f"Unknown transaction type in row: {row}") - - except Exception as e: - logger.error(f"[UpdateDeductionElectionsImpl.parse_deductions] Parse row failed due to error {e}") - - for ssn in ssn_to_loan_sum_map: - loan_sum = ssn_to_loan_sum_map[ssn] - result.append( - UpdateDeductionElectionsAscensusUtil._create_eds_for_value( - deduction_type=DeductionType._401K_LOAN_PAYMENT, - value=Decimal(loan_sum), - percentage=False, - ssn=ssn, - effective_date=datetime.now(), - ) - ) - - return result diff --git a/flux_sdk/pension/utils/ascensus_update_deduction_elections.py b/flux_sdk/pension/utils/ascensus_update_deduction_elections.py new file mode 100644 index 00000000..95c2cf3f --- /dev/null +++ b/flux_sdk/pension/utils/ascensus_update_deduction_elections.py @@ -0,0 +1,154 @@ +import csv +import logging +from datetime import datetime +from decimal import Decimal +from io import IOBase +from typing import Any, Union + +from flux_sdk.flux_core.data_models import ( + DeductionType, +) +from flux_sdk.pension.capabilities.update_deduction_elections.data_models import ( + EmployeeDeductionSetting, +) +from flux_sdk.pension.utils.common import ( + RecordTypeKeys, + get_deduction_type, +) + +logger = logging.getLogger(__name__) + +COLUMNS_360 = [ + "RecordType", ## 'D' represents Contribution Change, 'L' represents Loan + "PlanId", ## Plan ID or Contract number + "EmployeeLastName", + "EmployeeFirstName", + "EmployeeMiddleInitial", + "EmployeeSSN", + "EffectiveDate", ## The date that the change is effective + "ContributionCode", + "DeferralPercent", + "DeferralAmount", + "EmployeeEligibilityDate", ## The date the employee became eligible + "LoanNumber", + "LoanPaymentAmount", + "TotalLoanAmount", +] + + +class UpdateDeductionElectionsAscensusUtil: + """ + This class represents the "update deduction elections" capability for vendors utilizing + the Ascensus. The developer is supposed to implement + parse_deductions_for_ascensus method in their implementation. For further details regarding their + implementation details, check their documentation. + """ + + @staticmethod + def _create_eds_for_value( + deduction_type: DeductionType, + value: Union[str, Decimal], + is_percentage: bool, + ssn: str, + effective_date: datetime, + ) -> EmployeeDeductionSetting: + eds = EmployeeDeductionSetting() + eds.ssn = ssn + eds.effective_date = effective_date + eds.deduction_type = deduction_type + eds.value = Decimal(value) # type: ignore + eds.is_percentage = is_percentage + return eds + + @staticmethod + def _is_valid_amount(value) -> bool: + try: + Decimal(value) + return True + except Exception: + return False + + @staticmethod + def _parse_deduction_rows(row: dict[str, Any], result: list[EmployeeDeductionSetting]) -> None: + ssn = row["EmployeeSSN"] + deduction_type = get_deduction_type(row["ContributionCode"]) + eligibility_date = ( + datetime.strptime(row["EmployeeEligibilityDate"], "%m%d%Y") + if row["EmployeeEligibilityDate"] + else datetime.now() + ) + + if ( + UpdateDeductionElectionsAscensusUtil._is_valid_amount(row["DeferralAmount"]) + and UpdateDeductionElectionsAscensusUtil._is_valid_amount(row["DeferralPercent"]) + and deduction_type + ): + result.append( + UpdateDeductionElectionsAscensusUtil._create_eds_for_value( + deduction_type=deduction_type, + value=row["DeferralAmount"] + if row["DeferralAmount"] > row["DeferralPercent"] + else row["DeferralPercent"], + is_percentage=row["DeferralPercent"] > row["DeferralAmount"], + ssn=ssn, + effective_date=eligibility_date, + ) + ) + + @staticmethod + def _parse_loan_rows(row: dict[str, Any], ssn_to_loan_sum_map: dict[str, Decimal]) -> None: + ssn = row["EmployeeSSN"] + if UpdateDeductionElectionsAscensusUtil._is_valid_amount(row["LoanPaymentAmount"]): + loan_value = Decimal(row["LoanPaymentAmount"]) + if ssn in ssn_to_loan_sum_map: + ssn_to_loan_sum_map[ssn] += loan_value + else: + ssn_to_loan_sum_map[ssn] = loan_value + + @staticmethod + def parse_deductions_for_ascensus(uri: str, stream: IOBase) -> list[EmployeeDeductionSetting]: + """ + This method receives a stream from which the developer is expected to return a list of EmployeeDeductionSetting + for each employee identifier (SSN). + :param uri: Contains the path of file + :param stream: Contains the stream + :return: list[EmployeeDeductionSetting] + """ + result: list[EmployeeDeductionSetting] = [] + + try: + reader = csv.DictReader(stream) # type: ignore + except Exception as e: + logger.error(f"[UpdateDeductionElectionsImpl.parse_deductions] Parse deductions failed due to message {e}") + return result + + ssn_to_loan_sum_map: dict[str, Decimal] = {} + + for row in reader: + try: + ssn = row["EmployeeSSN"] + record_type = row["RecordType"] + + if record_type == RecordTypeKeys.DeductionType.value: + UpdateDeductionElectionsAscensusUtil._parse_deduction_rows(row, result) + elif record_type == RecordTypeKeys.LoanType.value: + UpdateDeductionElectionsAscensusUtil._parse_loan_rows(row, ssn_to_loan_sum_map) + else: + logger.error(f"Unknown transaction type in row: {row}") + + except Exception as e: + logger.error(f"[UpdateDeductionElectionsImpl.parse_deductions] Parse row failed due to error {e}") + + for ssn in ssn_to_loan_sum_map: + loan_sum = ssn_to_loan_sum_map[ssn] + result.append( + UpdateDeductionElectionsAscensusUtil._create_eds_for_value( + deduction_type=DeductionType._401K_LOAN_PAYMENT, + value=Decimal(loan_sum), + is_percentage=False, + ssn=ssn, + effective_date=datetime.now(), + ) + ) + + return result diff --git a/flux_sdk/pension/utils/common.py b/flux_sdk/pension/utils/common.py new file mode 100644 index 00000000..29d8afbc --- /dev/null +++ b/flux_sdk/pension/utils/common.py @@ -0,0 +1,23 @@ +from enum import Enum +from typing import Optional + +from flux_sdk.flux_core.data_models import DeductionType + + +class RecordTypeKeys(Enum): + DeductionType = "D" + LoanType = "L" + + +def get_deduction_type(given_ded_type: str) -> Optional[DeductionType]: + ded_match_map = { + "4ROTH": DeductionType.ROTH_401K, + "4ROTC": DeductionType.ROTH_401K, + "401K": DeductionType._401K, + "401KC": DeductionType._401K, + "401L": DeductionType._401K_LOAN_PAYMENT, + "403B": DeductionType._403B, + "401A": DeductionType.AFTER_TAX_401K, + "401O": DeductionType._401K, + } + return ded_match_map.get(given_ded_type, None) diff --git a/flux_sdk/pension/utils/pay_konnect.py b/flux_sdk/pension/utils/pay_konnect_report_payroll_contributions.py similarity index 99% rename from flux_sdk/pension/utils/pay_konnect.py rename to flux_sdk/pension/utils/pay_konnect_report_payroll_contributions.py index bc2657fe..d317c8e9 100644 --- a/flux_sdk/pension/utils/pay_konnect.py +++ b/flux_sdk/pension/utils/pay_konnect_report_payroll_contributions.py @@ -355,7 +355,7 @@ def format_contributions_for_pay_konnect_vendor( pay_type = ReportPayrollContributionsPayKonnectUtil._get_employee_pay_type(employee) termination_date = getattr(employee, "termination_date", None) termination_date = termination_date.strftime(STANDARD_DATE_FORMAT) if termination_date else "" - birth_day = employee.dob.strftime(STANDARD_DATE_FORMAT) + birth_day = employee.dob.strftime(STANDARD_DATE_FORMAT) if employee.dob else "" phone_number = employee.phone_number if employee.phone_number else "" rehire_date = employee.start_date.strftime(STANDARD_DATE_FORMAT) hire_date = employee.original_hire_date.strftime(STANDARD_DATE_FORMAT) diff --git a/flux_sdk/pension/utils/pay_konnect_update_deduction_elections.py b/flux_sdk/pension/utils/pay_konnect_update_deduction_elections.py new file mode 100644 index 00000000..0099a041 --- /dev/null +++ b/flux_sdk/pension/utils/pay_konnect_update_deduction_elections.py @@ -0,0 +1,144 @@ +import csv +import datetime +import logging +from decimal import Decimal +from io import IOBase +from typing import Any, Union + +from flux_sdk.flux_core.data_models import DeductionType +from flux_sdk.pension.capabilities.update_deduction_elections.data_models import ( + EmployeeDeductionSetting, +) +from flux_sdk.pension.utils.common import ( + RecordTypeKeys, + get_deduction_type, +) + +logger = logging.getLogger(__name__) + +columns_360 = [ + "Record Type", + "Plan Number", + "SSN", + "Effective Date", + "Eligibility Date", + "Transaction Date", + "Transaction Type", + "Code", + "Value Type", + "Value", + "Loan Reference Number", + "Loan Goal", +] + + +class UpdateDeductionElectionsPayKonnectUtil: + """ + This class represents the "update deduction elections" capability for vendors utilizing + the PayKonnect. The app developer is supposed to implement + parse_deductions_for_pay_konnect method in their implementation. For further details regarding their + implementation details, check their documentation. + """ + + @staticmethod + def _parse_loan_rows(row: dict[str, Any], ssn_to_loan_sum_map: dict[str, Decimal]) -> None: + ssn = row["SSN"] + if UpdateDeductionElectionsPayKonnectUtil._is_valid_amount(row["Value"]): + loan_value = Decimal(row["Value"]) + if ssn in ssn_to_loan_sum_map: + ssn_to_loan_sum_map[ssn] += loan_value + else: + ssn_to_loan_sum_map[ssn] = loan_value + + @staticmethod + def _create_eds_for_value( + deduction_type: DeductionType, + value: Union[str, Decimal], + percentage: bool, + ssn: str, + effective_date: datetime.datetime, + ) -> EmployeeDeductionSetting: + eds = EmployeeDeductionSetting() + eds.ssn = ssn + eds.effective_date = effective_date + eds.deduction_type = deduction_type + eds.value = Decimal(value) # type: ignore + eds.is_percentage = percentage + return eds + + @staticmethod + def _is_valid_amount(value) -> bool: + try: + Decimal(value) + return True + except Exception: + return False + + @staticmethod + def _parse_deduction_rows(row: dict[str, Any], result: list[EmployeeDeductionSetting]) -> None: + ssn = row["SSN"] + deduction_type = get_deduction_type(row["Code"]) + eligibility_date = ( + datetime.datetime.strptime(row["Eligibility Date"], "%m%d%Y") + if row["Eligibility Date"] + else datetime.datetime.now() + ) + + if UpdateDeductionElectionsPayKonnectUtil._is_valid_amount(row["Value"]) and deduction_type: + result.append( + UpdateDeductionElectionsPayKonnectUtil._create_eds_for_value( + deduction_type=deduction_type, + value=row["Value"], + percentage=True if row["Value Type"] == "Percent" else False, + ssn=ssn, + effective_date=eligibility_date, + ) + ) + + @staticmethod + def parse_deductions_for_pay_konnect(uri: str, stream: IOBase) -> list[EmployeeDeductionSetting]: + """ + This method receives a stream from which the developer is expected to return a list of EmployeeDeductionSetting + for each employee identifier (SSN). + :param uri: Contains the path of file + :param stream: Contains the stream + :return: list[EmployeeDeductionSetting] + """ + result: list[EmployeeDeductionSetting] = [] + + try: + reader = csv.DictReader(stream) # type: ignore + except Exception as e: + logger.error(f"[UpdateDeductionElectionsImpl.parse_deductions] Parse deductions failed due to message {e}") + return result + + ssn_to_loan_sum_map: dict[str, Decimal] = {} + + for row in reader: + try: + ssn = row["SSN"] + record_type = row["Record Type"] + + if record_type == RecordTypeKeys.DeductionType.value: + UpdateDeductionElectionsPayKonnectUtil._parse_deduction_rows(row, result) + elif record_type == RecordTypeKeys.LoanType.value: + UpdateDeductionElectionsPayKonnectUtil._parse_loan_rows(row, ssn_to_loan_sum_map) + else: + logger.error(f"Unknown transaction type in row: {row}") + + except Exception as e: + logger.error(f"[UpdateDeductionElectionsImpl.parse_deductions] Parse row failed due to error {e}") + + for ssn in ssn_to_loan_sum_map: + loan_sum = ssn_to_loan_sum_map[ssn] + result.append( + UpdateDeductionElectionsPayKonnectUtil._create_eds_for_value( + deduction_type=DeductionType._401K_LOAN_PAYMENT, + value=Decimal(loan_sum), + percentage=False, + ssn=ssn, + effective_date=datetime.datetime.now(), + ) + ) + + return result diff --git a/flux_sdk/pension/utils/tests/test_ascensus/contributions_180.csv b/flux_sdk/pension/utils/tests/test_ascensus/contributions_report_payroll_contributions.csv similarity index 100% rename from flux_sdk/pension/utils/tests/test_ascensus/contributions_180.csv rename to flux_sdk/pension/utils/tests/test_ascensus/contributions_report_payroll_contributions.csv diff --git a/flux_sdk/pension/utils/tests/test_ascensus/test_ascensus_180.py b/flux_sdk/pension/utils/tests/test_ascensus/test_ascensus_report_payroll_contributions.py similarity index 98% rename from flux_sdk/pension/utils/tests/test_ascensus/test_ascensus_180.py rename to flux_sdk/pension/utils/tests/test_ascensus/test_ascensus_report_payroll_contributions.py index 88ad402a..f65d05bc 100644 --- a/flux_sdk/pension/utils/tests/test_ascensus/test_ascensus_180.py +++ b/flux_sdk/pension/utils/tests/test_ascensus/test_ascensus_report_payroll_contributions.py @@ -17,7 +17,7 @@ PayrollUploadSettings, PayrunInfo, ) -from flux_sdk.pension.utils.ascensus import ReportPayrollContributionsAscensusUtil +from flux_sdk.pension.utils.ascensus_report_payroll_contributions import ReportPayrollContributionsAscensusUtil class TestReportPayrollContributionsAscensusUtil(unittest.TestCase): @@ -171,7 +171,7 @@ def test_format_contributions_for_ascensus_vendor(self) -> None: ) file_content = contributions_file.content.decode() with open( - os.path.join(os.path.dirname(__file__), "contributions_180.csv") + os.path.join(os.path.dirname(__file__), "contributions_report_payroll_contributions.csv") ) as contribution_file: contribution_file_contents = contribution_file.read() self.assertEqual( diff --git a/flux_sdk/pension/utils/tests/test_ascensus/test_ascensus_360.py b/flux_sdk/pension/utils/tests/test_ascensus/test_ascensus_update_deduction_elections.py similarity index 96% rename from flux_sdk/pension/utils/tests/test_ascensus/test_ascensus_360.py rename to flux_sdk/pension/utils/tests/test_ascensus/test_ascensus_update_deduction_elections.py index 44fde487..ab767440 100644 --- a/flux_sdk/pension/utils/tests/test_ascensus/test_ascensus_360.py +++ b/flux_sdk/pension/utils/tests/test_ascensus/test_ascensus_update_deduction_elections.py @@ -4,7 +4,7 @@ from typing import List from flux_sdk.pension.capabilities.update_deduction_elections.data_models import EmployeeDeductionSetting -from flux_sdk.pension.utils.ascensus import UpdateDeductionElectionsAscensusUtil +from flux_sdk.pension.utils.ascensus_update_deduction_elections import UpdateDeductionElectionsAscensusUtil class TestUpdateDeductionElections(unittest.TestCase): diff --git a/flux_sdk/pension/utils/tests/test_pay_konnect/test_pay_konnect.py b/flux_sdk/pension/utils/tests/test_pay_konnect/test_pay_konnect_report_payroll_contributions.py similarity index 98% rename from flux_sdk/pension/utils/tests/test_pay_konnect/test_pay_konnect.py rename to flux_sdk/pension/utils/tests/test_pay_konnect/test_pay_konnect_report_payroll_contributions.py index 02035587..66b2419a 100644 --- a/flux_sdk/pension/utils/tests/test_pay_konnect/test_pay_konnect.py +++ b/flux_sdk/pension/utils/tests/test_pay_konnect/test_pay_konnect_report_payroll_contributions.py @@ -20,7 +20,7 @@ PayrollUploadSettings, PayrunInfo, ) -from flux_sdk.pension.utils.pay_konnect import ReportPayrollContributionsPayKonnectUtil +from flux_sdk.pension.utils.pay_konnect_report_payroll_contributions import ReportPayrollContributionsPayKonnectUtil class TestReportPayrollContributionsPayKonnectUtil(unittest.TestCase): diff --git a/flux_sdk/pension/utils/tests/test_pay_konnect/test_pay_konnect_update_deduction_elections.py b/flux_sdk/pension/utils/tests/test_pay_konnect/test_pay_konnect_update_deduction_elections.py new file mode 100644 index 00000000..1c447582 --- /dev/null +++ b/flux_sdk/pension/utils/tests/test_pay_konnect/test_pay_konnect_update_deduction_elections.py @@ -0,0 +1,67 @@ +import io +import unittest +from decimal import Decimal +from typing import List + +from flux_sdk.pension.capabilities.update_deduction_elections.data_models import EmployeeDeductionSetting +from flux_sdk.pension.utils.pay_konnect_update_deduction_elections import UpdateDeductionElectionsPayKonnectUtil + + +class TestUpdateDeductionElections(unittest.TestCase): + """ + Tests for functions for the UpdateDeductionElections capability. + """ + + def get_file_data_for_test_update_deduction(self, date, ssn1, ssn2): + + sample_deferral_file = ( + "Record Type,Plan Number,SSN,Effective Date,Eligibility Date," + "Transaction Date,Transaction Type,Code,Value Type,Value,Loan Reference Number,Loan Goal\n" + "D,222222-00000,{},,03312019,,,401K,Amount,1.0,123,\n" + "D,222222-00000,{},,03312019,,,4ROTH,Percent,12.0,123,\n" + "D,222222-00000,{},,03312019,,,401K,Amount,15.00,123,\n" + "D,222222-00000,{},,03312019,,,4ROTH,Amount,25.00,123,\n" + "L,222222-00000,{},,04122018,,,401L,Amount,20.00,123,\n" + "L,222222-00000,{},,04122018,,,401L,Amount,30.00,123,\n" + "L,222222-00000,{},,04122018,,,401L,Amount,40.00,123,\n" + "L,222222-00000,{},,04122018,,,401L,Amount,50.00,123,\n" + ).format(ssn1, ssn1, ssn2, ssn2, ssn1, ssn2, ssn1, ssn2) + + return sample_deferral_file + + def test_parse_deductions(self): + ssn1 = "523546780" + ssn2 = "523546781" + sample_deferral_file = self.get_file_data_for_test_update_deduction("3/6/2023", ssn1, ssn2) + + result: List[ + EmployeeDeductionSetting + ] = UpdateDeductionElectionsPayKonnectUtil.parse_deductions_for_pay_konnect( + "", io.StringIO(sample_deferral_file) + ) + + count = 0 + for ed in result: + count = count + 1 + if ed.ssn == ssn1: + if ed.deduction_type == "_401K": + self.assertEqual(ed.value, Decimal("1.00")) + self.assertEqual(ed.is_percentage, True) + if ed.deduction_type == "ROTH_401K": + self.assertEqual(ed.value, Decimal("12.00")) + self.assertEqual(ed.is_percentage, True) + if ed.deduction_type == "_401K_LOAN_PAYMENT": + self.assertEqual(ed.value, Decimal("60")) + self.assertEqual(ed.is_percentage, False) + else: + if ed.deduction_type == "_401K": + self.assertEqual(ed.value, Decimal("15.00")) + self.assertEqual(ed.is_percentage, False) + if ed.deduction_type == "ROTH_401K": + self.assertEqual(ed.value, Decimal("25.00")) + self.assertEqual(ed.is_percentage, False) + if ed.deduction_type == "_401K_LOAN_PAYMENT": + self.assertEqual(ed.value, Decimal("80")) + self.assertEqual(ed.is_percentage, False) + + self.assertEqual(count, 6) diff --git a/pyproject.toml b/pyproject.toml index 13a264d7..2f307f1c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "rippling-flux-sdk" -version = "0.20" +version = "0.21" description = "Defines the interfaces and data-models used by Rippling Flux Apps." authors = ["Rippling Apps "] readme = "README.md"