Merged in QCQA_Additions (pull request #331)
QCQA Additions * tested on batch 10, small fixes + change to s3 output * Merged main into QCQA_Additions * unit tests for qc qa helpers * static error fix * another static fix * static fix again * fixed None check * static fix none * Merged main into QCQA_Additions * Small changes based on Katon's review Approved-by: Katon Minhas
This commit is contained in:
Chris Bull
@@ -18,7 +18,7 @@ from qa_qc_helpers import (
|
||||
from consolidate_output import consolidate_adhoc
|
||||
|
||||
|
||||
file_name_column = config.CONTRACT_FILE_NAME
|
||||
file_name_column = config.FILE_NAME_COLUMN
|
||||
|
||||
"""python scripts/qa_qc.py input_dir=______ output_dir=_______ ac_df=________ b_df=______ read_mode=____ df_read_mode=________
|
||||
You should have both input and output (ouput is just an empty folder) and then one of ac_df or b_df or both
|
||||
|
||||
@@ -225,6 +225,7 @@ ANSWER_TRACKING_DICT = {}
|
||||
######################################## POSTPROCESSING SETTINGS ########################################
|
||||
FUZZY_MATCH_THRESHOLD = 0.8
|
||||
FILE_NAME_COLUMN = 'Contract Name'
|
||||
BLANKS_THRESHOLD = 0.5
|
||||
|
||||
######################################## COMPLEX FLAG SETTINGS ########################################
|
||||
COMPLEX_MAX_WORKERS = 50
|
||||
|
||||
@@ -137,8 +137,10 @@ def consolidate_adhoc(ac_df, b_df):
|
||||
|
||||
# If ABC
|
||||
|
||||
b_df.drop([col for col in ["Pages", "Parent Agreement Code"] if col in b_df.columns], axis=1, inplace=True)
|
||||
|
||||
abc_final_df = pd.merge(
|
||||
b_df, ac_df, on=[config.FILE_NAME_COLUMN, "Pages", "Parent Agreement Code"], how="right"
|
||||
b_df, ac_df, on=[config.FILE_NAME_COLUMN], how="right"
|
||||
)
|
||||
abc_final_df = abc_final_df[
|
||||
[col for col in valid.ABC_COLUMNS if col in abc_final_df]
|
||||
|
||||
@@ -10,6 +10,7 @@ import preprocessing_funcs
|
||||
import config
|
||||
from openpyxl import Workbook
|
||||
from openpyxl.utils.dataframe import dataframe_to_rows
|
||||
from hotfix_helper_funcs import keywords_to_detect_invalid_state_name
|
||||
|
||||
def detect_date_anomalies(series, field_name, is_merged=False):
|
||||
total_count = len(series)
|
||||
@@ -48,17 +49,20 @@ def detect_date_anomalies(series, field_name, is_merged=False):
|
||||
return result
|
||||
|
||||
def tin_check(tin: str) -> tuple[str, bool]:
|
||||
pattern = r'\b\d{2}-\d{7}\b'
|
||||
pattern = r'^\b\d{2}-\d{7}\b$'
|
||||
|
||||
if not isinstance(tin, str):
|
||||
try:
|
||||
tin = str(tin)
|
||||
except TypeError:
|
||||
except Exception:
|
||||
check = False
|
||||
|
||||
return tin, check
|
||||
|
||||
check=bool(re.match(pattern, tin))
|
||||
match=re.match(pattern, tin)
|
||||
check = bool(match)
|
||||
|
||||
tin = match.group(0) if match else tin
|
||||
|
||||
return tin, check
|
||||
|
||||
@@ -68,12 +72,15 @@ def npi_check(npi: str) -> tuple[str, bool]:
|
||||
if not isinstance(npi, str):
|
||||
try:
|
||||
npi = str(npi)
|
||||
except TypeError:
|
||||
except Exception:
|
||||
check = False
|
||||
|
||||
return npi, check
|
||||
|
||||
check = bool(re.match(pattern, npi))
|
||||
match = re.match(pattern, npi)
|
||||
check = bool(match)
|
||||
|
||||
npi = match.group(0) if match else npi
|
||||
|
||||
return npi, check
|
||||
|
||||
@@ -81,9 +88,15 @@ def date_format_check(date: str) -> tuple:
|
||||
|
||||
format = '%m/%d/%Y'
|
||||
|
||||
if not isinstance(date, str):
|
||||
try:
|
||||
date = str(date)
|
||||
except TypeError:
|
||||
check = False
|
||||
date = date
|
||||
try:
|
||||
dt = datetime.strptime(date, format)
|
||||
date = dt.strftime('%m/%d/%Y')
|
||||
date = dt.strftime('%-m/%-d/%Y')
|
||||
check = True
|
||||
except ValueError:
|
||||
date = date
|
||||
@@ -112,35 +125,66 @@ def state_check(state: str) -> tuple[str, bool]:
|
||||
check = False
|
||||
return state, check
|
||||
|
||||
if state in STATE_MAP.keys():
|
||||
state = STATE_MAP[state]
|
||||
if state.upper() in STATE_MAP.keys():
|
||||
state = STATE_MAP[state.upper()]
|
||||
check = True
|
||||
elif state.strip().title() in STATE_MAP.values():
|
||||
check = True
|
||||
state = state.strip().title()
|
||||
else:
|
||||
check = False
|
||||
|
||||
return state, check
|
||||
|
||||
def pages_pagenum_check(pages: str, pagenum: str) -> bool:
|
||||
def pages_pagenum_check(pages: int, pagenum: int) -> bool:
|
||||
|
||||
if not isinstance(pages, str):
|
||||
if not isinstance(pages, int):
|
||||
try:
|
||||
pages = str(pages)
|
||||
except TypeError:
|
||||
pages = int(pages)
|
||||
except (TypeError, ValueError) as e:
|
||||
check = False
|
||||
return check
|
||||
elif not isinstance(pagenum, str):
|
||||
try:
|
||||
pagenum = str(pagenum)
|
||||
except TypeError:
|
||||
check = False
|
||||
return check
|
||||
else:
|
||||
check = (pagenum < pages)
|
||||
|
||||
return check
|
||||
return check
|
||||
|
||||
if not isinstance(pagenum, int):
|
||||
try:
|
||||
pagenum = int(pagenum)
|
||||
except (TypeError, ValueError) as e:
|
||||
check = False
|
||||
|
||||
return check
|
||||
|
||||
check = (pages >= pagenum)
|
||||
|
||||
return check
|
||||
|
||||
def blanks_check(df: pd.DataFrame, threshold=config.BLANKS_THRESHOLD) -> pd.DataFrame:
|
||||
|
||||
highly_blank_cols = []
|
||||
for col in df.columns:
|
||||
if df[col].apply(lambda x: utils.is_empty(x)).sum() / len(df[col]) > threshold:
|
||||
highly_blank_cols.append(col)
|
||||
|
||||
return '\n'.join(highly_blank_cols)
|
||||
|
||||
def self_talk_check(df: pd.DataFrame):
|
||||
|
||||
lang_cols = [col for col in df.columns if ('Language' in col or 'Clause' in col) and ('(Y/N)' not in col)]
|
||||
|
||||
selftalkers = []
|
||||
|
||||
for col in lang_cols:
|
||||
|
||||
df['selftalk'] = df[col].apply(lambda x: any([(kw in x) for kw in keywords_to_detect_invalid_state_name]) if not utils.is_empty(x) else False)
|
||||
|
||||
if df['selftalk'].any():
|
||||
|
||||
pct = df['selftalk'].sum() / len(df['selftalk'])
|
||||
|
||||
selftalkers.append(f'{col} - {pct:.1%}')
|
||||
|
||||
|
||||
return '\n'.join(selftalkers)
|
||||
|
||||
def perform_qc_qa_tests_both(ac_df, b_df, merged_df, input_dict, processed_files):
|
||||
|
||||
@@ -192,6 +236,8 @@ def perform_qc_qa_tests_both(ac_df, b_df, merged_df, input_dict, processed_files
|
||||
"Number of duplicate rows in merged DataFrame": merged_df.duplicated().sum(),
|
||||
"# of fields for AC run" : len(ac_df.columns),
|
||||
"# of fields for B run" : len(b_df.columns),
|
||||
"Highly Blank Fields in merged DataFrame" : blanks_check(merged_df),
|
||||
"Potential 'Self-Talk' columns in merged DataFrame" : self_talk_check(merged_df),
|
||||
"# of incorrectly formatted Provider Group TINs" : ac_df["IRS #"].apply(lambda x: not tin_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||||
"# of incorrectly formatted Provider Group NPIs" : ac_df["NPI (10-digits)"].apply(lambda x: not npi_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||||
"# of incorrectly formatted Provider Group Signatory TINs": ac_df["PROV_GROUP_TIN_SIGNATORY"].apply(lambda x: not tin_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||||
@@ -199,7 +245,7 @@ def perform_qc_qa_tests_both(ac_df, b_df, merged_df, input_dict, processed_files
|
||||
"# of incorrectly formatted Provider NPIs (other)": ac_df["PROV_NPI_OTHER"].apply(lambda x: not npi_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||||
"# of incorrect Health Plan States" : ac_df['Health Plan State'].apply(lambda x: not state_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||||
"# of rows with Pages greater than Page Num" : merged_df.apply(lambda row: pages_pagenum_check(row['Pages'], row['Page_Num']) if (not utils.is_empty(row['Pages']) and utils.is_empty(row['Page_Num'])) else False, axis=1).sum(),
|
||||
"# of Payer Names found in Notice to Provider Name" : ac_df.apply(lambda row: row['PAYER NAME'] in row['Notice to Provider Name'] if not utils.is_empty(row['Notice to Provider Name']) else False, axis = 1).sum(),
|
||||
"# of Payer Names found in Notice to Provider Name" : ac_df.apply(lambda row: row['PAYER NAME'] in row['Notice to Provider Name'] if not utils.is_empty(row['Notice to Provider Name']) and not utils.is_empty(row['PAYER NAME']) else False, axis = 1).sum(),
|
||||
"# of invalid Y/N values for B run" : b_df[yn_cols_b].apply(lambda x: yn_check(x)[1], axis = 1).sum().sum(),
|
||||
"# of invalid Y/N values for AC run" : ac_df[yn_cols_ac].apply(lambda x: yn_check(x)[1], axis = 1).sum().sum(),
|
||||
"# of invalid Termination Dates" : ac_df['Termination Date'].apply(lambda x: not date_format_check(x) if not utils.is_empty(x) else x).sum(),
|
||||
@@ -269,6 +315,8 @@ def perform_qc_qa_tests_ac(ac_df, input_dict, processed_files):
|
||||
.sum(),
|
||||
"# of duplicate rows for AC run": ac_df.duplicated().sum(),
|
||||
"# of fields for AC run" : len(ac_df.columns),
|
||||
"Highly Blank Fields in AC DataFrame" : blanks_check(ac_df),
|
||||
"Potential 'Self-Talk' columns in merged DataFrame" : self_talk_check(ac_df),
|
||||
"# of incorrectly formatted Provider Group TINs" : ac_df["IRS #"].apply(lambda x: not tin_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||||
"# of incorrectly formatted Provider Group NPIs" : ac_df["NPI (10-digits)"].apply(lambda x: not npi_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||||
"# of incorrectly formatted Provider Group Signatory TINs": ac_df["PROV_GROUP_TIN_SIGNATORY"].apply(lambda x: not tin_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||||
@@ -321,6 +369,8 @@ def perform_qc_qa_tests_b(b_df, input_dict, processed_files):
|
||||
.sum(),
|
||||
"# of duplicate rows for B run": b_df.duplicated().sum(),
|
||||
"# of fields for B run" : len(b_df.columns),
|
||||
"Highly Blank Fields in B DataFrame" : blanks_check(b_df),
|
||||
"Potential 'Self-Talk' columns in merged DataFrame" : self_talk_check(b_df),
|
||||
"# of rows with Pages greater than Page Num" : b_df.apply(lambda row: pages_pagenum_check(row['Pages'], row['Page_Num']), axis=1).sum(),
|
||||
"# of invalid Y/N values for B run" : b_df[yn_cols_b].apply(lambda x: yn_check(x)[1], axis = 1).sum().sum()}
|
||||
|
||||
@@ -500,7 +550,7 @@ def save_qcqa_wb(wb, run_timestamp):
|
||||
# Upload to S3
|
||||
if config.WRITE_TO_S3:
|
||||
BUCKET_NAME = config.S3_OUTPUT_BUCKET
|
||||
print(f"Uploading final outputs to s3://{BUCKET_NAME}/{run_timestamp}")
|
||||
print(f"Uploading final outputs to s3://{BUCKET_NAME}/qcqa/{run_timestamp}")
|
||||
|
||||
|
||||
# Upload consolidated outputs
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
import pytest
|
||||
import pandas as pd
|
||||
import sys
|
||||
sys.path.append('src')
|
||||
from qa_qc_helpers import blanks_check
|
||||
|
||||
|
||||
def test_blanks_check_no_highly_blank_columns():
|
||||
# Create a sample DataFrame with no highly blank columns
|
||||
df = pd.DataFrame({
|
||||
"A": [1, 2, 3, 4, 5],
|
||||
"B": ["a", "b", "c", "d", "e"],
|
||||
"C": [10.5, 11.2, 12.3, 13.4, 14.5]
|
||||
})
|
||||
|
||||
result = blanks_check(df)
|
||||
assert result == ""
|
||||
|
||||
def test_blanks_check_with_highly_blank_columns():
|
||||
# Create a sample DataFrame with highly blank columns
|
||||
df = pd.DataFrame({
|
||||
"A": [1, 2, 3, None, 5, None, 6],
|
||||
"B": ["a", "b", "", "d", "", "", ""],
|
||||
"C": [10.5, None, 12.3, None, 14.5, None, None]
|
||||
})
|
||||
|
||||
result = blanks_check(df)
|
||||
expected_output = "B\nC"
|
||||
assert result == expected_output
|
||||
|
||||
def test_blanks_check_all_columns_highly_blank():
|
||||
# Create a sample DataFrame where all columns are highly blank
|
||||
df = pd.DataFrame({
|
||||
"A": [None, None, None, None, None],
|
||||
"B": ["", "", "", "", ""],
|
||||
"C": [None, None, None, None, None]
|
||||
})
|
||||
|
||||
result = blanks_check(df)
|
||||
expected_output = "A\nB\nC"
|
||||
assert result == expected_output
|
||||
@@ -0,0 +1,84 @@
|
||||
import pytest
|
||||
from datetime import datetime
|
||||
import sys
|
||||
sys.path.append('src')
|
||||
from qa_qc_helpers import date_format_check
|
||||
|
||||
def test_valid_date_formats():
|
||||
"""Test valid date formats are correctly validated"""
|
||||
valid_dates = {
|
||||
"01/01/2024" : '1/1/2024',
|
||||
"12/31/2023" : '12/31/2023',
|
||||
"02/29/2024" : '2/29/2024', # Leap year
|
||||
"07/04/1776" : '7/4/1776',
|
||||
"09/08/2024" : '9/8/2024'
|
||||
}
|
||||
|
||||
for date in valid_dates.keys():
|
||||
result, is_valid = date_format_check(date)
|
||||
assert result == valid_dates[date]
|
||||
assert is_valid is True
|
||||
|
||||
def test_invalid_date_formats():
|
||||
"""Test various invalid date formats"""
|
||||
invalid_dates = [
|
||||
"13/01/2024", # Invalid month
|
||||
"01/32/2024", # Invalid day
|
||||
"02/29/2023", # Not a leap year
|
||||
"00/01/2024", # Zero month
|
||||
"01/00/2024", # Zero day
|
||||
"01-01-2024", # Wrong separator
|
||||
"2024/01/01", # Wrong order
|
||||
"01/01/24", # Two-digit year
|
||||
"Jan/01/2024", # Text month
|
||||
]
|
||||
|
||||
for date in invalid_dates:
|
||||
result, is_valid = date_format_check(date)
|
||||
assert result == date
|
||||
assert is_valid is False
|
||||
|
||||
def test_non_string_inputs():
|
||||
"""Test handling of non-string inputs"""
|
||||
test_cases = [
|
||||
(20240101, "20240101"),
|
||||
(None, None),
|
||||
(1.234, "1.234"),
|
||||
(["01/01/2024"], "['01/01/2024']"),
|
||||
({"date": "01/01/2024"}, "{'date': '01/01/2024'}"),
|
||||
]
|
||||
|
||||
for input_val, expected_str in test_cases:
|
||||
result, is_valid = date_format_check(input_val)
|
||||
assert result == str(input_val)
|
||||
assert is_valid is False
|
||||
|
||||
def test_edge_cases():
|
||||
"""Test edge cases with valid-looking but incorrect formats"""
|
||||
test_cases = [
|
||||
"01/01/2024 ", # Extra space at end
|
||||
" 01/01/2024", # Extra space at start
|
||||
"01/01/2024\n", # Newline at end
|
||||
"01 /01/2024", # Space in middle
|
||||
"01/01/ 2024", # Space in middle
|
||||
]
|
||||
|
||||
for date in test_cases:
|
||||
result, is_valid = date_format_check(date)
|
||||
assert result == date
|
||||
assert is_valid is False
|
||||
|
||||
def test_special_dates():
|
||||
"""Test boundary cases for dates"""
|
||||
test_cases = [
|
||||
("12/31/9999", True), # Far future
|
||||
("2/28/2024", True), # Day before leap day
|
||||
("3/1/2024", True), # Day after leap day
|
||||
("12/31/2023", True), # Last day of year
|
||||
("1/1/2024", True), # First day of year
|
||||
]
|
||||
|
||||
for date, expected_valid in test_cases:
|
||||
result, is_valid = date_format_check(date)
|
||||
assert result == date
|
||||
assert is_valid == expected_valid
|
||||
@@ -0,0 +1,67 @@
|
||||
import pytest
|
||||
import sys
|
||||
sys.path.append('src')
|
||||
from qa_qc_helpers import npi_check
|
||||
import re
|
||||
|
||||
def test_valid_npi_format():
|
||||
"""Test that a correctly formatted NPI returns True"""
|
||||
npi = "1234567890"
|
||||
result, is_valid = npi_check(npi)
|
||||
assert result == npi
|
||||
assert is_valid is True
|
||||
|
||||
def test_invalid_npi_format():
|
||||
"""Test various invalid NPI formats"""
|
||||
invalid_npis = [
|
||||
"123456789", # Too few digits
|
||||
"12345678900", # Too many digits
|
||||
"12-4567890", # TIN format
|
||||
"1ab3456789", # Letters instead of numbers
|
||||
"3234567890", # Doesn't start with 1 or 2
|
||||
]
|
||||
|
||||
for npi in invalid_npis:
|
||||
result, is_valid = npi_check(npi)
|
||||
assert result == npi
|
||||
assert is_valid is False
|
||||
|
||||
def test_non_string_inputs():
|
||||
"""Test handling of non-string inputs"""
|
||||
test_cases = [
|
||||
(1234567890, "1234567890", True),
|
||||
(12.3456789, "12.3456789", False),
|
||||
(None, 'None', False),
|
||||
([12, 3456789], "[12, 3456789]", False),
|
||||
({"npi" : "1234567890"}, "{'npi': '1234567890'}", False),
|
||||
]
|
||||
|
||||
for input_val, expected_str, expected_valid in test_cases:
|
||||
result, is_valid = npi_check(input_val)
|
||||
assert result == expected_str
|
||||
assert is_valid is expected_valid
|
||||
|
||||
def test_edge_cases_npi():
|
||||
"""Test edge cases with valid-looking but incorrect formats"""
|
||||
test_cases = [
|
||||
"12345 67890" # Space in middle
|
||||
]
|
||||
|
||||
for npi in test_cases:
|
||||
result, is_valid = npi_check(npi)
|
||||
assert result == npi
|
||||
assert is_valid is False
|
||||
|
||||
def test_multiple_tins_in_string():
|
||||
"""Test strings containing multiple TIN-like patterns"""
|
||||
test_cases = [
|
||||
"1234567890 1256789001", # Two TINs with space
|
||||
"12345678901234567890", # Two TINs without space
|
||||
"Some text 1234567890", # TIN with preceding text
|
||||
"1234567890 some text", # TIN with following text
|
||||
]
|
||||
|
||||
for npi in test_cases:
|
||||
result, is_valid = npi_check(npi)
|
||||
assert result == npi
|
||||
assert is_valid is False
|
||||
@@ -0,0 +1,41 @@
|
||||
import pytest
|
||||
import sys
|
||||
sys.path.append('src')
|
||||
from qa_qc_helpers import pages_pagenum_check
|
||||
|
||||
def test_pages_pagenum_check_valid_inputs():
|
||||
# Test with valid string inputs
|
||||
assert pages_pagenum_check("100", "50") is True
|
||||
assert pages_pagenum_check(50, 10) is True
|
||||
|
||||
def test_pages_pagenum_check_invalid_inputs():
|
||||
# Test with invalid string inputs
|
||||
assert pages_pagenum_check("abc", "50") is False
|
||||
assert pages_pagenum_check("50", "xyz") is False
|
||||
|
||||
def test_pages_pagenum_check_non_string_inputs():
|
||||
# Test with non-string inputs for pages
|
||||
assert pages_pagenum_check(100, "50") is True
|
||||
assert pages_pagenum_check(50, 10) is True
|
||||
|
||||
# Test with non-string inputs for pagenum
|
||||
assert pages_pagenum_check("100", 50) is True
|
||||
assert pages_pagenum_check("50", 10) is True
|
||||
|
||||
def test_pages_pagenum_check_non_convertible_inputs():
|
||||
# Test with inputs that cannot be converted to string for pages
|
||||
def raise_type_error():
|
||||
raise TypeError("Cannot convert to string")
|
||||
|
||||
assert pages_pagenum_check(raise_type_error, "50") is False
|
||||
|
||||
# Test with inputs that cannot be converted to string for pagenum
|
||||
assert pages_pagenum_check("100", raise_type_error) is False
|
||||
|
||||
def test_pages_pagenum_check_equality():
|
||||
# Test when pages and pagenum are equal
|
||||
assert pages_pagenum_check("100", "100") is True
|
||||
|
||||
def test_pages_pagenum_check_pagenum_greater():
|
||||
# Test when pagenum is greater than pages
|
||||
assert pages_pagenum_check("50", "100") is False
|
||||
@@ -0,0 +1,42 @@
|
||||
import pytest
|
||||
import pandas as pd
|
||||
import sys
|
||||
sys.path.append('src')
|
||||
from qa_qc_helpers import self_talk_check
|
||||
from hotfix_helper_funcs import keywords_to_detect_invalid_state_name
|
||||
|
||||
|
||||
def test_self_talk_check_no_self_talk():
|
||||
# Create a sample DataFrame with no self-talk columns
|
||||
df = pd.DataFrame({
|
||||
"Name": ["John", "Jane", "Bob", "Alice", "Tom"],
|
||||
"Age": [25, 30, 35, 40, 45],
|
||||
"City": ["New York", "Los Angeles", "Chicago", "Houston", "Miami"]
|
||||
})
|
||||
|
||||
result = self_talk_check(df)
|
||||
assert result == ""
|
||||
|
||||
def test_self_talk_check_with_self_talk():
|
||||
# Create a sample DataFrame with self-talk columns
|
||||
df = pd.DataFrame({
|
||||
"Name": ["John", "Jane", "Bob", "Alice", "Tom"],
|
||||
"Language": ["English", "Spanish", "The Contract Text suggests Armenian", "French", "German"],
|
||||
"Clause": ["", "this suggests", "Mike refers to xyz", "The Contract Text says", "My Answer is this: "]
|
||||
})
|
||||
|
||||
result = self_talk_check(df)
|
||||
expected_output = "Language - 20.0%\nClause - 80.0%"
|
||||
assert result == expected_output
|
||||
|
||||
def test_self_talk_check_all_self_talk():
|
||||
# Create a sample DataFrame where all language columns have self-talk
|
||||
df = pd.DataFrame({
|
||||
"Name": ["John", "Jane", "Bob", "Alice", "Tom"],
|
||||
"Language": ["Statewide", "Multiple sources imply", "The contract text says", "For this text, it would seem", "This suggests that"],
|
||||
"Clause": ["The Contract Text contains this", "My Answer Is This: ", "It would seem that multiple clauses say", "I conclude that the exhibit suggests", "For each clause, this is the case"]
|
||||
})
|
||||
|
||||
result = self_talk_check(df)
|
||||
expected_output = "Language - 100.0%\nClause - 100.0%"
|
||||
assert result == expected_output
|
||||
@@ -0,0 +1,32 @@
|
||||
import pytest
|
||||
import sys
|
||||
sys.path.append('src')
|
||||
from qa_qc_helpers import state_check
|
||||
|
||||
|
||||
def test_state_check_valid_abbreviations():
|
||||
# Test with valid state abbreviations
|
||||
assert state_check("CA") == ("California", True)
|
||||
assert state_check("NY") == ("New York", True)
|
||||
|
||||
def test_state_check_valid_full_names():
|
||||
# Test with valid state full names
|
||||
assert state_check("California") == ("California", True)
|
||||
assert state_check("new york") == ("New York", True)
|
||||
|
||||
def test_state_check_invalid_input():
|
||||
# Test with invalid input
|
||||
assert state_check("ABC") == ("ABC", False)
|
||||
assert state_check("Invalid State") == ("Invalid State", False)
|
||||
|
||||
def test_state_check_non_string_input():
|
||||
# Test with non-string input
|
||||
assert state_check(123) == ("123", False)
|
||||
assert state_check(True) == ("True", False)
|
||||
|
||||
def test_state_check_non_convertible_input():
|
||||
# Test with input that cannot be converted to string
|
||||
def raise_type_error():
|
||||
raise TypeError("Cannot convert to string")
|
||||
|
||||
assert state_check(raise_type_error)[1] == False
|
||||
@@ -0,0 +1,74 @@
|
||||
import pytest
|
||||
import sys
|
||||
sys.path.append('src')
|
||||
from qa_qc_helpers import tin_check
|
||||
import re
|
||||
|
||||
def test_valid_tin_format():
|
||||
"""Test that a correctly formatted TIN returns True"""
|
||||
tin = "12-3456789"
|
||||
result, is_valid = tin_check(tin)
|
||||
assert result == tin
|
||||
assert is_valid is True
|
||||
|
||||
def test_invalid_tin_format():
|
||||
"""Test various invalid TIN formats"""
|
||||
invalid_tins = [
|
||||
"123-456789", # Too many digits before hyphen
|
||||
"12-456789", # Too few digits after hyphen
|
||||
"12-45678901", # Too many digits after hyphen
|
||||
"12_3456789", # Wrong separator
|
||||
"ab-3456789", # Letters instead of numbers
|
||||
"12-abcdefg", # Letters instead of numbers
|
||||
"123456789", # No hyphen
|
||||
"-12345678", # Hyphen at start
|
||||
"12-34567-", # Hyphen at end
|
||||
]
|
||||
|
||||
for tin in invalid_tins:
|
||||
result, is_valid = tin_check(tin)
|
||||
assert result == tin
|
||||
assert is_valid is False
|
||||
|
||||
def test_non_string_inputs():
|
||||
"""Test handling of non-string inputs"""
|
||||
test_cases = [
|
||||
(123456789, "123456789", False),
|
||||
(12.3456789, "12.3456789", False),
|
||||
(None, 'None', False),
|
||||
([12, 3456789], "[12, 3456789]", False),
|
||||
({"tin":"12-3456789"}, "{'tin': '12-3456789'}", False),
|
||||
]
|
||||
|
||||
for input_val, expected_str, expected_valid in test_cases:
|
||||
result, is_valid = tin_check(input_val)
|
||||
assert result == expected_str
|
||||
assert is_valid is expected_valid
|
||||
|
||||
def test_edge_cases():
|
||||
"""Test edge cases with valid-looking but incorrect formats"""
|
||||
test_cases = [
|
||||
"12 3456789", #Space instead of dash
|
||||
"1 2-3456789", # Space at beginning
|
||||
"12-345 6789", # Space in middle
|
||||
]
|
||||
|
||||
for tin in test_cases:
|
||||
result, is_valid = tin_check(tin)
|
||||
assert result == tin
|
||||
assert is_valid is False
|
||||
|
||||
def test_multiple_tins_in_string():
|
||||
"""Test strings containing multiple TIN-like patterns"""
|
||||
test_cases = [
|
||||
"12-3456789 34-5678901", # Two TINs with space
|
||||
"12-345678912-3456789", # Two TINs without space
|
||||
"Some text 12-3456789", # TIN with preceding text
|
||||
"12-3456789 some text", # TIN with following text
|
||||
]
|
||||
|
||||
for tin in test_cases:
|
||||
result, is_valid = tin_check(tin)
|
||||
assert result == tin
|
||||
assert is_valid is False
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
import pytest
|
||||
import sys
|
||||
sys.path.append('src')
|
||||
from qa_qc_helpers import yn_check
|
||||
|
||||
def test_yn_check_with_valid_string_input():
|
||||
# Test with valid string inputs
|
||||
assert yn_check("Y") == ("Y", True)
|
||||
assert yn_check("N") == ("N", True)
|
||||
|
||||
def test_yn_check_with_invalid_string_input():
|
||||
# Test with invalid string inputs
|
||||
assert yn_check("a") == ("a", False)
|
||||
assert yn_check("yes") == ("yes", False)
|
||||
|
||||
def test_yn_check_with_non_string_input():
|
||||
# Test with non-string inputs
|
||||
assert yn_check(1) == ("1", False)
|
||||
assert yn_check(True) == ("True", False)
|
||||
|
||||
def test_yn_check_with_non_convertible_input():
|
||||
# Test with inputs that cannot be converted to string
|
||||
def raise_type_error():
|
||||
raise TypeError("Cannot convert to string")
|
||||
|
||||
assert yn_check(raise_type_error)[1] == False
|
||||
Reference in New Issue
Block a user