jacob-mathison/doczyai-pipelines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity
Files
3608a5650915b6ba135dfeabe2d0bce7f350dfbe
doczyai-pipelines/fieldExtraction/tests/postprocessing_funcs_test.py
T
Siddhant Medar 04d268a2f8 Merged in task/unit-tests-investment-and-client (pull request #354) 
Task/unit tests investment and client

* move io_utils.py

* add __init__.py

* refactor: move string_funcs and update import paths to use utils namespace

* rename string_funcs to string_utils

* move and rename table_utils.py

* refactor: rename claude_funcs to llm_utils in multiple files

* update unit tests

* added unit tests for src/string_utils.py

* added unit tests for src/table_utils.py

* added unit tests for src/llm_utils.py

* added unit tests for src/io_utils.py

* Merge branch 'main' into task/unit-tests-investment-and-client; TODO: fix unit tests

* Unit tests to fix

* fix unit tests

* fix invalid references

* add unit tests for preprocessing_funcs.py

* move pytest-mock to test dependencies

* Merge remote-tracking branch 'origin/main' into task/unit-tests-investment-and-client

* add tests/string_utils_test.py

* removed filename parameters for a few functions

* save postprocessing_funcs_test midway

* save postprocessing_funcs_test

* postprocessing_funcs unit tests complete

* Merge branch 'main' into task/unit-tests-investment-and-client


Approved-by: Katon Minhas
2025-01-17 17:15:32 +00:00

626 lines
26 KiB
Python
Raw Blame History

import pytest
import pandas as pd
import numpy as np
from src.postprocessing_funcs import check_add_on
from src.postprocessing_funcs import (
InvalidDateException,
yn_fixes,
convert_to_us_date_format,
filter_service_column,
filter_methodology_column,
clean_td,
get_parent_agreement_code,
consolidate_subheader,
clean_msr_lesser,
clean_default_term,
clean_lesser_rate,
clean_prov_2,
clean_term_clause,
clean_auto_renewal_ind,
clean_npi,
clean_network_access_fees,
clean_health_plan_state,
get_health_plan_state,
state_check,
clean_health_plan_state_from_hotfix,
clean_notice_provider_name_and_address,
get_tin_from_filename,
clean_policies_and_procedures,
clean_tin,
clean_tin_npi_other,
filter_add_ons,
add_scmr,
add_hyphen_if_needed,
remove_unnamed_columns,
format_tin,
count_dollar_values
)
class TestPostprocessingFuncs:
@pytest.fixture
def valid_dates(self):
return [
("2023-12-01", "12/01/2023"),
("2023-01-12", "01/12/2023"),
("2023-13-01", "01/13/2023"),
]
@pytest.fixture
def invalid_dates(self):
return ["2023-13-32", "2023-00-01", "2023-12-00", "2023-12-32", "2023-13-13"]
@pytest.fixture
def empty_dates(self):
return ["", "N/A", "null", "none"]
@pytest.fixture
def invalid_format_dates(self):
return ["2023/12/01", "12-01-2023", "2023.12.01"]
@pytest.fixture
def invalid_type_dates(self):
return [None]
def test_convert_to_us_date_format(
self,
valid_dates,
invalid_dates,
empty_dates,
invalid_format_dates,
invalid_type_dates,
):
for input_date, expected_output in valid_dates:
assert convert_to_us_date_format(input_date) == expected_output
for input_date in invalid_dates:
with pytest.raises(InvalidDateException):
convert_to_us_date_format(input_date)
for input_date in empty_dates:
with pytest.raises(InvalidDateException):
convert_to_us_date_format(input_date)
for input_date in invalid_format_dates:
with pytest.raises(InvalidDateException):
convert_to_us_date_format(input_date)
for input_date in invalid_type_dates:
with pytest.raises(TypeError):
convert_to_us_date_format(input_date)
@pytest.fixture
def sample_answer_dicts(self):
return [
{"FULL_SERVICE": "Service Risk", "FULL_METHODOLOGY": "Methodology interest"},
{"FULL_SERVICE": "Service B", "FULL_METHODOLOGY": "PMPM"},
{"FULL_SERVICE": "damages", "FULL_METHODOLOGY": "Invalid Methodology"},
]
def test_filter_service_column(self, sample_answer_dicts):
filtered = filter_service_column(sample_answer_dicts)
assert len(filtered) == 1
assert all(d["FULL_SERVICE"] == "Service B" for d in filtered)
def test_filter_methodology_column(self, sample_answer_dicts):
filtered = filter_methodology_column(sample_answer_dicts)
assert len(filtered) == 1
assert all(d["FULL_METHODOLOGY"] == "Invalid Methodology" for d in filtered)
@pytest.mark.parametrize("input_td, expected_output", [
(
{"DATE": "01/01/2023", "Health Plan State": "Chicago", "page_num": 1, "Filename": "file1.txt"},
{"DATE": ["01/01/2023"], "Health Plan State": ["Chicago"], "page_num": 1, "Filename": "file1.txt"}
),
(
{"DATE": "01/01/2023", "Health Plan State": ["New York", "Michigan"], "page_num": 1, "Filename": "file1.txt"},
{"DATE": ["01/01/2023"], "Health Plan State": ["New York", "Michigan"], "page_num": 1, "Filename": "file1.txt"}
),
(
{"DATE": "N/A", "Health Plan State": "N/A", "page_num": 2, "Filename": "file2.txt"},
{"DATE": ["N/A"], "Health Plan State": [], "page_num": 2, "Filename": "file2.txt"}
),
(
{"DATE": "02/01/2023", "Health Plan State": "California, Texas", "page_num": 3, "Filename": "file3.txt"},
{"DATE": ["02/01/2023"], "Health Plan State": ["California", "Texas"], "page_num": 3, "Filename": "file3.txt"}
),
])
def test_clean_td(self, input_td, expected_output):
cleaned = clean_td([input_td])
assert cleaned[0] == expected_output
@pytest.mark.parametrize("filename, expected_code", [
("contract_1234(1).txt", "1234"),
("contract_5678.txt", "5678"),
("invalid_filename.txt", "N/A"),
])
def test_get_parent_agreement_code(self, filename, expected_code):
code = get_parent_agreement_code(filename)
assert code == expected_code
@pytest.mark.parametrize("input_dict, expected_output", [
({"FULL_SERVICE": "Service A", "SUBHEADER": "Header A"}, {"FULL_SERVICE": "Header A - Service A"}),
({"FULL_SERVICE": "Service B", "SUBHEADER": ""}, {"FULL_SERVICE": " - Service B"}),
({"FULL_SERVICE": "Service C", "SUBHEADER": "N/A"}, {"FULL_SERVICE": "Service C"}),
])
def test_consolidate_subheader(self, input_dict, expected_output):
consolidated = consolidate_subheader([input_dict])
assert consolidated[0] == expected_output
@pytest.mark.parametrize("input_df, expected_lesser, expected_lesser_rate", [
(
pd.DataFrame({
"FULL_SERVICE": ["Multiple Outpatient", "Multiple Outpatient", "Not multiple"],
"EXHIBIT": ["Type A", "Type A", "Type B"],
"LESSER": ["", "Y", "N"],
"LESSER_RATE": [100, 200, 300]
}),
["Y", "Y", "N"],
[200, 200, 300]
),
(
pd.DataFrame({
"FULL_SERVICE": ["Multiple Outpatient", "Multiple Outpatient", "Not multiple"],
"EXHIBIT": ["Type A", "Type A", "Type B"],
"LESSER": ["Y", "Y", "N"],
"LESSER_RATE": [100, 200, 300]
}),
["Y", "Y", "N"],
[100, 100, 300]
),
])
def test_clean_msr_lesser(self, input_df, expected_lesser, expected_lesser_rate):
cleaned_df = clean_msr_lesser(input_df)
assert cleaned_df["LESSER"].tolist() == expected_lesser
assert cleaned_df["LESSER_RATE"].tolist() == expected_lesser_rate
@pytest.mark.parametrize("input_df, expected_df", [
(
pd.DataFrame({
"DEFAULT_TERM": ["Except as otherwise provided"],
"DEFAULT_RATE": ["Invalid"],
"IP_OP": ["IP"]
}),
pd.DataFrame({
"DEFAULT_TERM": ["N/A"],
"DEFAULT_RATE": ["N/A"],
"IP_OP": ["IP"]
})
),
(
pd.DataFrame({
"DEFAULT_TERM": ["Medicare Outpatient", "Medicaid Inpatient", "Invalid"],
"DEFAULT_RATE": ["AC", "BC", "Amount Payable by Medicaid:NA"],
"IP_OP": ["IP", "OP", "IP"]
}),
pd.DataFrame({
"DEFAULT_TERM": ["N/A", "N/A", "Invalid"],
"DEFAULT_RATE": ["N/A", "N/A", "Amount Payable by MCD:NA"],
"IP_OP": ["IP", "OP", "IP"]
})
),
])
def test_clean_default_term(self, input_df, expected_df):
cleaned_df = clean_default_term(input_df)
pd.testing.assert_frame_equal(cleaned_df, expected_df)
@pytest.mark.parametrize("input_df, expected_output", [
(
pd.DataFrame({
"LESSER_RATE": [np.nan, "200", "300"],
"RATE_STANDARD": ["150", "250", "350"],
"LESSER": ["Y", "Y", "N"]
}),
pd.DataFrame({
"LESSER_RATE": ["150", "200", "300"],
"RATE_STANDARD": ["N/A", "250", "350"],
"LESSER": ["Y", "Y", "N"]
})
)
])
def test_clean_lesser_rate(self, input_df, expected_output):
cleaned_df = clean_lesser_rate(input_df)
pd.testing.assert_frame_equal(cleaned_df, expected_output)
@pytest.mark.parametrize("input_df, expected_output", [
(
pd.DataFrame({
"PROV_TYPE_LEVEL_2": ["", "Type A", "Type B"],
"FULL_SERVICE": ["Specialist", "ASC", "Invalid Service - Skilled Nursing Facility"],
"EXHIBIT": ["Exhibit A", "Exhibit B", "Exhibit C"],
"PROV_TYPE": ["Professional", "Ancilliary","Invalid"]
}),
["Specialist", "ASC", "Skilled Nursing Facility"],
),
(
pd.DataFrame({
"PROV_TYPE_LEVEL_2": [""],
"FULL_SERVICE": [""],
"EXHIBIT": ["This has keyword, Primary Care Provider"],
"PROV_TYPE": ["Invalid"]
}),
["Primary Care Provider"],
),
(
pd.DataFrame({
"PROV_TYPE_LEVEL_2": ["", "Lab","",""],
"FULL_SERVICE": ["", "ASC","","Hospital"],
"EXHIBIT": ["Invalid", "Invalid","Invalid2","Invalid2"],
"PROV_TYPE": ["", "Ancilliary","","Facility"]
}),
["Lab", "ASC", "Hospital", "Hospital"],
)
])
def test_clean_prov_2(self, input_df, expected_output):
cleaned_df = clean_prov_2(input_df)
cleaned_df['PROV_TYPE_LEVEL_2'].tolist() == expected_output
@pytest.mark.parametrize("input_df, expected_output", [
(
pd.DataFrame({
"TERM_CLAUSE": ["IL-4 Termination", "bonus payment shall be effective", "No term or termination", "does not contain", "valid term clause"]
}),
pd.DataFrame({
"TERM_CLAUSE": ["N/A", "N/A", "N/A", "N/A", "valid term clause"]
})
)
])
def test_clean_term_clause(self, input_df, expected_output):
cleaned_df = clean_term_clause(input_df)
pd.testing.assert_frame_equal(cleaned_df, expected_output)
@pytest.mark.parametrize("input_df, column, expected_output", [
(
pd.DataFrame({
"TERM_CLAUSE": ["Contract automatically renews", "coterminous with the agreement", "N/A", " "],
"CONTRACT_AUTO_RENEWAL_IND": ["N", "N", "N", "N"]}),
"CONTRACT_AUTO_RENEWAL_IND",
["Y", "Y", "N", "N"]
),
(
pd.DataFrame({
"TERM_CLAUSE": ["N/A", ""],
"CONTRACT_TERMINATION_DT": ["Y", "Y"]}),
"CONTRACT_TERMINATION_DT",
[np.nan, np.nan]
),
(
pd.DataFrame({
"CONTRACT_AUTO_RENEWAL_IND": ["Yes", "Y", "N"],
"CONTRACT_TERMINATION_DT": ["", "N/A", "12/31/2022"],
}),
"CONTRACT_TERMINATION_DT",
[np.nan, np.nan, "12/31/2022"]
),
(
pd.DataFrame({
"CONTRACT_AUTO_RENEWAL_IND": ["Y",""],
}),
"CONTRACT_AUTO_RENEWAL_IND",
["Y", "N"]
),
(
pd.DataFrame({
"Contract Name": ["1234_contract.txt"]}),
"CONTRACT_AUTO_RENEWAL_IND",
["N"]
)
])
def test_clean_auto_renewal_ind(self, input_df, column, expected_output):
cleaned_df = clean_auto_renewal_ind(input_df)
assert cleaned_df[column].tolist() == expected_output
@pytest.mark.parametrize("input_df, expected_npi", [
(
pd.DataFrame({"PROV_GROUP_NPI": ["1234567890", "123", "N/A"]}),
["1234567890", "N/A - (model detected: 123)", "N/A"]
)
])
def test_clean_npi(self, input_df, expected_npi):
cleaned_df = clean_npi(input_df)
assert cleaned_df["PROV_GROUP_NPI"].tolist() == expected_npi
@pytest.mark.parametrize("input_df, expected_output", [
(
pd.DataFrame({
"NETWORK_ACCESS_FEES_IND": ["Yes", "N/A", np.nan, None, "nan","Test", "No"]
}),
["Yes", "N/A", np.nan, None, "nan", "Yes", "Yes"]
)
])
def test_clean_network_access_fees(self, input_df, expected_output):
cleaned_df = clean_network_access_fees(input_df)
assert cleaned_df["NETWORK_ACCESS_FEES_IND"].tolist() == expected_output
@pytest.mark.parametrize("input_df, expected_state", [
(pd.DataFrame({"HEALTH_PLAN_STATE": ["IL", "TX", "CA"],
"PAYER_NAME":["Illinios Payer","",""]}), ["Illinois","Texas","California"]),
(pd.DataFrame({"HEALTH_PLAN_STATE": ["TX", "TX", "CA"]}), ["Texas","Texas","California"]),
(pd.DataFrame({"HEALTH_PLAN_STATE": ["CA", "TX", "CA"]}), ["California","Texas","California"]),
])
def test_clean_health_plan_state(self, input_df, expected_state):
cleaned_df = clean_health_plan_state(input_df)
assert cleaned_df["HEALTH_PLAN_STATE"].tolist() == expected_state
@pytest.mark.parametrize("text_dict, expected_state", [
({"page1": "This contract is for Illinois and Texas."}, "Multiple"),
({"page1": "This contract is for IL."}, "Illinois"),
({"page1": "This contract is for TX and Texas."}, "Texas"),
({"page1": "This contract is for California."}, "California"),
({"page1": "This contract is for New York."}, "New York"),
({"page1": "This contract is for Illinois, Texas, and California."}, "Multiple"),
({"page1": "This contract is for Illinois and California."}, "Multiple"),
({"page1": "This contract is for Texas and New York."}, "Multiple"),
({"page1": "This contract is for."}, "N/A"),
({"page1": ""}, "N/A"),
])
def test_get_health_plan_state(self, text_dict, expected_state):
state = get_health_plan_state(text_dict)
assert state == expected_state
@pytest.mark.parametrize("state_abbr, text_dict, expected_state", [
("IL", {"page1": "This contract is for Illinois and Texas."}, "Illinois"),
("TX", {"page1": "This contract is for Illinois and Texas."}, "Texas"),
("CA", {"page1": "This contract is for California."}, "California"),
("NY", {"page1": "This contract is for New York."}, "New York"),
("FL", {"page1": "This contract is for Hlorida."}, "Florida"),
("", {"page1": "This contract is for Illinois and Texas."}, "Multiple"),
])
def test_state_check(self, state_abbr, text_dict, expected_state):
state = state_check(state_abbr, text_dict)
assert state == expected_state
@pytest.mark.parametrize("input_df, contract_name, text_dict, expected_state", [
(
pd.DataFrame({"HEALTH_PLAN_STATE": [["IL", "TX", "CA"]]}),
"contract_1234_v1.txt",
{"page1": "This is a sample contract contract"},
"Multiple"
),
(
pd.DataFrame({"HEALTH_PLAN_STATE": ["IL,"]}),
"contract_1234_v1.txt",
{"page1": "This is a sample contract"},
"Illinois"
),
(
pd.DataFrame({"HEALTH_PLAN_STATE": [""]}),
"contract_1234_v1.txt",
{"page1": "This contract is for Illinois and Texas."},
"Multiple"
)
])
def test_clean_health_plan_state_from_hotfix(self, input_df, contract_name, text_dict, expected_state):
input_df["HEALTH_PLAN_STATE"] = input_df["HEALTH_PLAN_STATE"].apply(
lambda x: f"[{','.join(x)}]" if isinstance(x, list) else x
)
cleaned_df = clean_health_plan_state_from_hotfix(input_df, contract_name, text_dict)
assert cleaned_df.loc[0, "HEALTH_PLAN_STATE"] == expected_state
@pytest.mark.parametrize("input_df, expected_name, expected_address", [
(
pd.DataFrame({
"NOTICE_PROVIDER_NAME": ["Superior HealthPlan", "Provider A", "Provider B"],
"NOTICE_PROVIDER_ADDRESS": ["Address A", "Address B", "Address C"]
}),
[np.nan, "Provider A", "Provider B"],
[np.nan, "Address B", "Address C"]
),
])
def test_clean_notice_provider_name_and_address(self, input_df, expected_name, expected_address):
cleaned_df = clean_notice_provider_name_and_address(input_df)
assert cleaned_df["NOTICE_PROVIDER_NAME"].tolist() == expected_name
assert cleaned_df["NOTICE_PROVIDER_ADDRESS"].tolist() == expected_address
@pytest.mark.parametrize("input_df, expected_tin", [
(pd.DataFrame({"Contract Name":["12-3456789_contract(1)", "12-3456789_contract(2)"],
"PROV_GROUP_TIN": ["", "123456789"]}
), ["123456789", "123456789"]
),
])
def test_get_tin_from_filename(self, input_df, expected_tin):
cleaned_df = get_tin_from_filename(input_df)
assert cleaned_df["PROV_GROUP_TIN"].tolist() == expected_tin
@pytest.mark.parametrize("input_df, expected_output", [
(
pd.DataFrame({
"POLICIES_AND_PROCEDURES": ["Policy A", "Procedure B", "Invalid", ""]
}),
pd.DataFrame({
"POLICIES_AND_PROCEDURES": ["Policy A", "Procedure B", "N/A", "N/A"]
})
),
])
def test_clean_policies_and_procedures(self, input_df, expected_output):
cleaned_df = clean_policies_and_procedures(input_df)
pd.testing.assert_frame_equal(cleaned_df, expected_output)
@pytest.mark.parametrize("input_df, expected_tin", [
(pd.DataFrame({"PROV_GROUP_TIN": ["123-45-6789"]}), "N/A"),
(pd.DataFrame({"PROV_GROUP_TIN": ["12-34-56"]}), "N/A - (model detected: 12-34-56)"),
(pd.DataFrame({"PROV_GROUP_TIN": ["12-3456789"]}), "12-3456789"),
(pd.DataFrame({"PROV_GROUP_TIN": ["N/A"]}), "N/A"),
])
def test_clean_tin(self, input_df, expected_tin):
cleaned_df = clean_tin(input_df)
assert cleaned_df.loc[0, "PROV_GROUP_TIN"] == expected_tin
@pytest.mark.parametrize("input_df, column, expected_tin", [
(pd.DataFrame({"PROV_TIN_OTHER": [["123456789", "123456789"], "N/A"]}), "PROV_TIN_OTHER", ["123456789,123456789","N/A"]),
(pd.DataFrame({"PROV_NPI_OTHER": ["12-3456789"]}), "PROV_NPI_OTHER", ["12-3456789"]),
(pd.DataFrame({"PROV_TIN_OTHER": [""]}), "PROV_TIN_OTHER", [""]),
(pd.DataFrame({"PROV_NPI_OTHER": [""]}), "PROV_NPI_OTHER", [""]),
])
def test_clean_tin_npi_other(self, input_df, column, expected_tin):
input_df[column] = input_df[column].apply(lambda x: f"[{','.join(x)}]" if isinstance(x, list) else x)
cleaned_df = clean_tin_npi_other(input_df)
assert cleaned_df[column].tolist() == expected_tin
@pytest.mark.parametrize("input_df, expected_output", [
(
pd.DataFrame({
"ADD_ON_REIMBURSEMENT_LANGUAGE": ["in no event", "forward", "Valid"],
"ADD_ON_REIMBURSEMENT_IND": ["Y", "N", "Y"],
"FULL_SERVICE": ["Add-on service", "No add-on", "Valid service"],
"Contract Name": ["Contract A", "Contract A", "Contract B"],
"EXHIBIT": ["Exhibit 1", "Exhibit 1", "Exhibit 2"]
}),
pd.DataFrame({
"ADD_ON_REIMBURSEMENT_LANGUAGE": ["N/A", "N/A", "Valid"],
"ADD_ON_REIMBURSEMENT_IND": ["N", "N", "Y"],
"FULL_SERVICE": ["Add-on service", "No add-on", "Valid service"],
"Contract Name": ["Contract A", "Contract A", "Contract B"],
"EXHIBIT": ["Exhibit 1", "Exhibit 1", "Exhibit 2"]
})
),
(
pd.DataFrame({
"ADD_ON_REIMBURSEMENT_LANGUAGE": ["", "additional fees", "N/A"],
"ADD_ON_REIMBURSEMENT_IND": ["Y", "Y", "N"],
"FULL_SERVICE": ["No add-on", "Add On service", "N/A"],
"Filename": ["File1", "File1", "File2"],
"EXHIBIT": ["Exhibit 1", "Exhibit 1", "Exhibit 2"]
}),
pd.DataFrame({
"ADD_ON_REIMBURSEMENT_LANGUAGE": ["N/A", "N/A", "N/A"],
"ADD_ON_REIMBURSEMENT_IND": ["N", "N", "N"],
"FULL_SERVICE": ["No add-on", "Add On service", "N/A"],
"Filename": ["File1", "File1", "File2"],
"EXHIBIT": ["Exhibit 1", "Exhibit 1", "Exhibit 2"]
})
)
])
@pytest.mark.filterwarnings("ignore::UserWarning")
@pytest.mark.filterwarnings("ignore::DeprecationWarning")
def test_filter_add_ons(self, input_df, expected_output):
# Call the function and reset the index
cleaned_df = filter_add_ons(input_df).reset_index(drop=True)
# Assert that the cleaned DataFrame matches the expected output
pd.testing.assert_frame_equal(cleaned_df, expected_output)
@pytest.mark.parametrize("input_df, expected_df", [
(pd.DataFrame({"FLAT_FEE_STANDARD": ["$100", "$200 30%", "$400"]}),
pd.DataFrame({
"Single Code Multiple Rates (Language)": ["N/A", "$200 30%", "N/A"],
"Single Code Multiple Rates (Y/N)": ["N", "Y", "N"]
}))
])
def test_add_scmr(self, input_df, expected_df):
cleaned_df = add_scmr(input_df)
pd.testing.assert_series_equal(
cleaned_df["Single Code Multiple Rates (Language)"],
expected_df["Single Code Multiple Rates (Language)"],
)
pd.testing.assert_series_equal(
cleaned_df["Single Code Multiple Rates (Y/N)"],
expected_df["Single Code Multiple Rates (Y/N)"],
)
@pytest.mark.parametrize("input_df, expected_df", [
(
pd.DataFrame({
"DUMMY_COL(Y/N)": ["Yes", "No", ""],
"SINGLE_CODE_MULTIPLE_RATES_IND": ["Y", "N", "Y"],
"SINGLE_CODE_MULTIPLE_RATES_LANGUAGE": ["Language 1", "Language 2", "N/A"],
"RATE_ESCALATOR_IND": ["Y", "N", "N"],
"RATE_ESCALATOR_DT": ["12/1/2022", "12/11/2022", "01/05/2018"],
}),
pd.DataFrame({
"DUMMY_COL(Y/N)": ["Y", "N", "N"],
"SINGLE_CODE_MULTIPLE_RATES_IND": ["Y", "N", "Y"],
"SINGLE_CODE_MULTIPLE_RATES_LANGUAGE": ["Language 1", "", "N/A"],
"RATE_ESCALATOR_IND": ["Y", "N", "N"],
"RATE_ESCALATOR_DT": ["12/1/2022", "", ""],
})
),
])
def test_yn_fixes(self, input_df, expected_df):
cleaned_df = yn_fixes(input_df)
pd.testing.assert_frame_equal(cleaned_df, expected_df)
@pytest.mark.parametrize("input_tin, expected_output", [
("12345678", "01-2345678"),
("123456789", "12-3456789"),
("1-2345678", "01-2345678"),
("123-45", None),
("", None),
])
def test_add_hyphen_if_needed(self, input_tin, expected_output):
assert add_hyphen_if_needed(input_tin) == expected_output
@pytest.mark.parametrize("input_date, expected_output", [
("2023-12-31", "12/31/2023"),
("2023-01-12", "01/12/2023"),
])
def test_convert_to_us_date_format_valid(self, input_date, expected_output):
assert convert_to_us_date_format(input_date) == expected_output
@pytest.mark.parametrize("input_date", [
"N/A",
"2023-13-32",
"2023-00-01",
"2023-12-00",
"2023-12-32",
"2023-13-13",
"",
"null",
"none",
"2023/12/01",
"12-01-2023",
"2023.12.01",
None,
])
def test_convert_to_us_date_format_invalid(self, input_date):
with pytest.raises((InvalidDateException, TypeError)):
convert_to_us_date_format(input_date)
@pytest.mark.parametrize("input_df, expected_columns", [
(pd.DataFrame({"Unnamed: 0": [1, 2], "A": [3, 4]}), ["A"]),
(pd.DataFrame({"Unnamed: 0": [1, 2], "Unnamed: 1": [3, 4], "B": [5, 6]}), ["B"]),
(pd.DataFrame({"A": [1, 2], "B": [3, 4]}), ["A", "B"]),
])
def test_remove_unnamed_columns(self, input_df, expected_columns):
cleaned_df = remove_unnamed_columns(input_df)
assert list(cleaned_df.columns) == expected_columns
@pytest.mark.parametrize("input_tin, expected_output", [
("123456789", "12-3456789"),
("12-3456789", "12-3456789"),
("123-45-6789", "12-3456789"),
("N/A", "N/A"),
("invalidTIN", "N/A - (model detected: invalidTIN)"),
("", "N/A - (model detected: )"),
])
def test_format_tin(self, input_tin, expected_output):
assert format_tin(input_tin) == expected_output
@pytest.mark.parametrize("input_str, expected_count", [
("$100", 1),
("100%", 1),
("$100 and 50%", 2),
("No dollar or percentage signs", 0),
("$$$ and %%%", 6),
("$100 $200 $300", 3),
("100% 200% 300%", 3),
("", 0),
("$%$%", 4),
])
def test_count_dollar_values(self, input_str, expected_count):
assert count_dollar_values(input_str) == expected_count
@pytest.mark.parametrize("input_df, expected_df", [
(
pd.DataFrame({
"FULL_SERVICE": ["Add-on Service", "Add-on Service"],
"ADD_ON_REIMBURSEMENT_LANGUAGE": ["Language A", "Language B"],
"ADD_ON_REIMBURSEMENT_IND": ["Y", "Y"]
}),
pd.DataFrame({
"FULL_SERVICE": ["Add-on Service", "Add-on Service"],
"ADD_ON_REIMBURSEMENT_LANGUAGE": ["N/A", "N/A"],
"ADD_ON_REIMBURSEMENT_IND": ["N", "N"]
})
),
(
pd.DataFrame({
"FULL_SERVICE": ["Add On Service", "Add On Service"],
"ADD_ON_REIMBURSEMENT_LANGUAGE": ["Language A", "Language B"],
"ADD_ON_REIMBURSEMENT_IND": ["Y", "Y"]
}),
pd.DataFrame({
"FULL_SERVICE": ["Add On Service", "Add On Service"],
"ADD_ON_REIMBURSEMENT_LANGUAGE": ["N/A", "N/A"],
"ADD_ON_REIMBURSEMENT_IND": ["N", "N"]
})
),
(
pd.DataFrame({
"FULL_SERVICE": ["Regular Service", "Regular Service"],
"ADD_ON_REIMBURSEMENT_LANGUAGE": ["Language A", "Language B"],
"ADD_ON_REIMBURSEMENT_IND": ["Y", "Y"]
}),
pd.DataFrame({
"FULL_SERVICE": ["Regular Service", "Regular Service"],
"ADD_ON_REIMBURSEMENT_LANGUAGE": ["Language A", "Language B"],
"ADD_ON_REIMBURSEMENT_IND": ["Y", "Y"]
})
)
])
def test_check_add_on(self, input_df, expected_df):
cleaned_df = check_add_on(input_df)
pd.testing.assert_frame_equal(cleaned_df, expected_df)
Reference in New Issue View Git Blame Copy Permalink