jacob-mathison/doczyai-pipelines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Merged in feature/split-and-filter-reimbursements (pull request #587)

Browse Source 
Feature/filter reimbursements

* First-pass implementation for reimbursement filtering

* Enhance reimbursement level processing by filtering out services without reimbursement terms and logging when no valid pairs are found.

* Update mock return values in reimbursement level tests to reflect new reimbursement terms

* Merge remote-tracking branch 'origin/main' into feature/split-and-filter-reimbursements

* Refine reimbursement filtering by adding 'billed' to indicators and ensuring REIMB_TERM is a string before processing.

* Refactor is_empty function signature to support multiple input types and clarify return values

* adding LLM review for ambiguous reimbursement methodology cases

* more stringent keywords for stage 1 of reimbursement filtering

* refining rate and cost patterns for reimbursement filtering

* Enhance LLM response handling in reimbursement methodology check to improve accuracy and logging for ambiguous cases.

* Merge remote-tracking branch 'origin/main' into feature/split-and-filter-reimbursements

* Update reimbursement test cases to reflect accurate reimbursement terms and improve deduplication logic

* move VALIDATE_REIMBURSEMENTS_PROMPT to investment_prompts.py

* black, isort formatting

* Remove IDENTIFY_REIMBURSEMENT_EXHIBITS_PROMPT function (unused)

* Add MODEL_ID_CLAUDE4_SONNET for new model integration (doesn't work right now)

* pipe delimited-output and parsing

* Merge remote-tracking branch 'origin/main' into feature/split-and-filter-reimbursements

* Refactor reimbursement filtering by moving clear indicators to constants


Approved-by: Katon Minhas
This commit is contained in:
Alex Galarce
2025-06-26 21:26:27 +00:00
parent 1531ad6bac
commit 724e31b15d
7 changed files with 267 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fieldExtraction/src/config.py
+1
View File
@@ -236,6 +236,7 @@ MODEL_ID_CLAUDE35_SONNET = "anthropic.claude-3-5-sonnet-20240620-v1:0"
MODEL_ID_CLAUDE2 = "anthropic.claude-instant-v1"
MODEL_ID_CLAUDE35_SONNET_V2 = "anthropic.claude-3-5-sonnet-20241022-v2:0"
MODEL_ID_CLAUDE37_SONNET = "anthropic.claude-3-7-sonnet-20250219-v1:0"
MODEL_ID_CLAUDE4_SONNET = "anthropic.claude-sonnet-4-20250514-v1:0"
######################################## CLAUDE AND PROMPT TRACKING ########################################
ENABLE_ANSWER_TRACKING = False
fieldExtraction/src/constants/investment_values.py
+50
View File
@@ -149,3 +149,53 @@ VALID_SPECIAL = {
proc_levels = ["cpt", "hcpcs", "cpt_level3", "cpt_level2", "hcpcs_level2", "cpt_level1", "hcpcs_level1"]
diag_levels = ["diag"]
### Reimbursement Filtering Keywords
clear_reimbursement_indicators = [
"paid at",
"reimbursed at",
"will be paid",
"will be reimbursed at",
"%",
"$",
"conversion factor",
"lesser of",
"greater of",
"not to exceed",
"billed",
"allowable",
# Rate patterns
"reimbursement rate",
"payment rate",
"rate of $",
"hourly rate",
"daily rate",
"per diem rate",
"capitation rate",
"flat rate",
# Cost patterns
"cost sharing",
"cost plus",
"cost based",
"at cost",
"cost per",
# Fee schedule patterns
"% of fee schedule",
"fee schedule rate",
"according to fee schedule",
"based on fee schedule",
"per fee schedule",
# Medicare patterns
"% of medicare",
"medicare rate",
"medicare allowable",
"according to medicare",
"based on medicare",
# Contract patterns
"contract rate",
"contracted rate",
"contract amount",
"per contract",
"according to contract",
]
fieldExtraction/src/investment/one_to_n_funcs.py
+156 -34
View File
@@ -1,7 +1,7 @@
import hashlib
import logging
import re
from typing import Callable
import logging
import src.constants.investment_values as investment_values
import src.investment.code_funcs as code_funcs
@@ -9,12 +9,10 @@ import src.prompts.investment_prompts as investment_prompts
import src.utils.llm_utils as llm_utils
import src.utils.string_utils as string_utils
from src import config
from src.config import FIELD_JSON_PATH
from src.constants.special_case_configs import SPECIAL_CASE_CONFIGS
from src.enums.delimiters import Delimiter
from src.prompts.investment_prompts import Field, FieldSet, EXHIBIT_LEVEL
from src.config import FIELD_JSON_PATH
import yaml # This is for experimental YAML parsing
from src.prompts.investment_prompts import EXHIBIT_LEVEL, Field, FieldSet
METHODOLOGY_BREAKOUT_QUESTIONS = FieldSet(
file_path=config.FIELD_JSON_PATH, field_type="methodology_breakout"
@@ -25,7 +23,9 @@ FS_BREAKOUT_QUESTIONS = FieldSet(
def get_exhibit_level_answers(exhibit_chunk, filename):
exhibit_level_fields = FieldSet(relationship="one_to_n", field_type="exhibit_level", file_path=FIELD_JSON_PATH)
exhibit_level_fields = FieldSet(
relationship="one_to_n", field_type="exhibit_level", file_path=FIELD_JSON_PATH
)
if not exhibit_level_fields.contains_fields():
return {}
@@ -41,18 +41,22 @@ def get_exhibit_level_answers(exhibit_chunk, filename):
def get_reimbursement_primary(reimbursement_level_fields, exhibit_text, filename):
field_prompts = reimbursement_level_fields.print_prompt_dict()
prompt = investment_prompts.REIMBURSEMENT_LEVEL_PRIMARY(exhibit_text, field_prompts)
logging.debug(f"""Running reimbursement primary prompt for {filename} with prompt:
{prompt}""")
logging.debug(
f"""Running reimbursement primary prompt for {filename} with prompt:
{prompt}"""
)
llm_answer_raw = llm_utils.invoke_claude(
prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, max_tokens=25000
)
logging.debug(f"""LLM raw output for {filename}:
{llm_answer_raw}""")
logging.debug(
f"""LLM raw output for {filename}:
{llm_answer_raw}"""
)
# Check for the special "no results" case
if "NO_REIMBURSEMENT_TERMS_FOUND" in llm_answer_raw:
return [] # Return empty list if no reimbursement terms found
try:
llm_answer_final = string_utils.universal_json_load(llm_answer_raw)
except ValueError as e:
@@ -147,7 +151,7 @@ def process_carveout_or_special_case(
methodology_prompt_template (Callable): Template for methodology prompt.
fee_schedule_prompt_template (Callable): Template for fee schedule prompt.
term_key (str): Indicates if the term key should be included.
If None, it will not be included.
If None, it will not be included.
If "methodology", the methodology only will be included.
If "service and methodology", both will be included.
@@ -251,8 +255,8 @@ def methodology_breakout_single_row(
# Methodology Breakout: AARETE_DERIVED_REIMB_METHOD, REIMB_FEE_RATE, REIMB_PCT_RATE, LESSER_OF_IND, GREATER_OF_IND, AARETE_DERIVED_PROV_TYPE
if contract_reimbursement_method:
prompt = methodology_prompt_template(
service, contract_reimbursement_method, methodology_breakout_questions
)
service, contract_reimbursement_method, methodology_breakout_questions
)
llm_response = llm_utils.invoke_claude(
prompt,
config.MODEL_ID_CLAUDE35_SONNET,
@@ -335,13 +339,20 @@ def get_methodology_breakout(
if isinstance(answer_dict, list): # if a nested list is found, flatten it
# This shouldn't happen but is in for defensive programming purposes
for nested_dict in answer_dict:
_process_carveout_or_breakout(nested_dict, methodology_breakout_answers, filename)
_process_carveout_or_breakout(
nested_dict, methodology_breakout_answers, filename
)
else: # normal processing for dict items
_process_carveout_or_breakout(answer_dict, methodology_breakout_answers, filename)
_process_carveout_or_breakout(
answer_dict, methodology_breakout_answers, filename
)
return methodology_breakout_answers
def _process_carveout_or_breakout(answer_dict: dict, methodology_breakout_answers: list, filename: str):
def _process_carveout_or_breakout(
answer_dict: dict, methodology_breakout_answers: list, filename: str
):
"""Processes a single answer dictionary for methodology breakout.
Args:
@@ -365,6 +376,7 @@ def _process_carveout_or_breakout(answer_dict: dict, methodology_breakout_answer
)
return True
def get_service_breakout(answer_dicts, filename):
"""
Runs service breakout prompt.
@@ -437,7 +449,12 @@ def combine_one_to_n_answers(
def reimbursement_level(
exhibit_text, filename, reimbursement_level_fields, all_dataset, exhibit_page: str, seen_pairs: set
exhibit_text,
filename,
reimbursement_level_fields,
all_dataset,
exhibit_page: str,
seen_pairs: set,
):
"""
Processes reimbursement-level fields.
@@ -451,7 +468,7 @@ def reimbursement_level(
exhibit_page (str): The page number of the exhibit being processed (for a unique identifier)
seen_pairs (set): A set accumulator to track pairs of SERVICE_TERM and REIMB_TERM to avoid duplicate processing.
This is currently across all exhibits, but could be modified to be per-superexhibit if needed.
(i.e. if a exhibit 23.0 [superexhibit 23.0] has subexhibits 23.0, 23.1, 23.2, etc.,
(i.e. if a exhibit 23.0 [superexhibit 23.0] has subexhibits 23.0, 23.1, 23.2, etc.,
then the pairs could be tracked per superexhibit.) This is NOT currently implemented.
Returns:
@@ -468,21 +485,32 @@ def reimbursement_level(
reimbursement_primary_answers, seen_pairs, exhibit_page, filename
)
if deduplicated_answers: # Only process if we have unique pairs
carveout_answers = get_special_cases(deduplicated_answers, filename)
code_breakout_answers = code_funcs.get_code_breakout(
carveout_answers, filename, all_dataset
if deduplicated_answers: # Only process if we have unique pairs
# Filter out services without actual reimbursement terms
filtered_answers = filter_services_without_reimbursements(
deduplicated_answers, filename
)
return code_breakout_answers
if filtered_answers: # Only continue if we have valid reimbursement pairs
carveout_answers = get_special_cases(filtered_answers, filename)
code_breakout_answers = code_funcs.get_code_breakout(
carveout_answers, filename, all_dataset
)
return code_breakout_answers
else:
logging.info(
f"No valid reimbursement pairs found in exhibit {exhibit_page} of {filename}."
)
return [] # Return empty list if no valid reimbursement pairs
else:
return [] # All pairs were duplicates, return empty list
return [] # All pairs were duplicates, return empty list
else:
return []
def deduplicate_with_accumulator(
answers: list[dict], seen_pairs: set, exhibit_page: str, filename: str
) -> list[dict]:
answers: list[dict], seen_pairs: set, exhibit_page: str, filename: str
) -> list[dict]:
"""Remove service-reimbursement pairs that have been seen in previous exhibits.
Args:
@@ -505,13 +533,107 @@ def deduplicate_with_accumulator(
pair_key = (service_term, reimb_term)
if pair_key not in seen_pairs:
seen_pairs.add(pair_key) # Update accumulator
seen_pairs.add(pair_key) # Update accumulator
deduplicated.append(answer_dict)
else:
duplicates_found += 1
logging.debug(f"Cross-exhibit duplicate found in {filename}, exhibit {exhibit_page}: {service_term} - {reimb_term}")
if duplicates_found > 0:
logging.info(f"Removed {duplicates_found} duplicate pairs from exhibit {exhibit_page} in {filename}")
logging.debug(
f"Cross-exhibit duplicate found in {filename}, exhibit {exhibit_page}: {service_term} - {reimb_term}"
)
return deduplicated
if duplicates_found > 0:
logging.info(
f"Removed {duplicates_found} duplicate pairs from exhibit {exhibit_page} in {filename}"
)
return deduplicated
def filter_services_without_reimbursements(
answers: list[dict], filename: str
) -> list[dict]:
"""
Filters out services that lack actual reimbursement methodologies.
Args:
answers (list[dict]): List of dictionaries containing reimbursement details.
filename (str): The name of the file being processed.
Returns:
list[dict]: Filtered list of dictionaries containing only services with reimbursement methodologies.
Example:
>>> answers = [
{"SERVICE_TERM": "Routine Vision", "REIMB_TERM": "Reimbursement will be at 100% of the first published DMAP fee
schedule for year prior. Repairs for optical hardware are not covered."},
{"SERVICE_TERM": "Eyeglass Lenses", "REIMB_TERM": "Standard eyeglass lenses as defined by DMAP are CR39
lenses including scratch resistant coating, +/- 7D sphere up to -4D cylinder."}, # Note that there is no actual reimbursement information here
]
>>> filtered_answers = filter_services_without_reimbursements(answers, "example_file.txt")
>>> print(filtered_answers)
[{"SERVICE_TERM": "Routine Vision", "REIMB_TERM": "Reimbursement will be at 100% of the first published DMAP fee
schedule for year prior. Repairs for optical hardware are not covered."}]
"""
# Stage 1: Allow obvious reimbursement cases
clear_reimbursement = []
needs_llm_review = []
for answer_dict in answers:
reimb_term = str(answer_dict.get("REIMB_TERM", "")).strip().lower()
if any(indicator in reimb_term for indicator in investment_values.clear_reimbursement_indicators):
clear_reimbursement.append(answer_dict)
else:
needs_llm_review.append(answer_dict)
# Stage 2: LLM review for ambiguous cases
if needs_llm_review:
logging.debug(
f"Sending {len(needs_llm_review)} reimbursement terms for LLM validation in {filename}."
)
for answer_dict in needs_llm_review:
if validate_reimbursements_for_llm(answer_dict, filename):
clear_reimbursement.append(answer_dict)
else:
logging.debug(f"LLM filtered out: {answer_dict}")
filtered_count = len(answers) - len(clear_reimbursement)
if filtered_count > 0:
logging.info(
f"Filtered {filtered_count} services without clear reimbursement methodologies in {filename}."
)
return clear_reimbursement
def validate_reimbursements_for_llm(answer_dict: dict, filename: str) -> bool:
"""Use LLM to determine if a service has actual reimbursement methodology.
Args:
answer_dict (dict): The service-reimbursement answer dictionary.
filename (str): The name of the file being processed.
Returns:
bool: True if the service has a clear reimbursement methodology, False otherwise.
"""
service_term = str(answer_dict.get("SERVICE_TERM", "")).strip()
reimb_term = str(answer_dict.get("REIMB_TERM", "")).strip()
if (string_utils.is_empty(service_term)
or string_utils.is_empty(reimb_term)
):
logging.warning(
f"Missing SERVICE_TERM or REIMB_TERM in {filename}: {answer_dict}"
)
return False
prompt = investment_prompts.VALIDATE_REIMBURSEMENTS_PROMPT(service_term, reimb_term)
llm_response = llm_utils.invoke_claude(
prompt, config.MODEL_ID_CLAUDE3_HAIKU, filename # try haiku for speed/cost
)
final_answer = string_utils.extract_text_from_delimiters(
llm_response, Delimiter.PIPE
).strip().upper()
return final_answer == "YES"
fieldExtraction/src/prompts/investment_prompts.py
+30 -19
View File
@@ -312,31 +312,42 @@ Header 2:
Briefly explain your answer, then enclose your final answer in |pipes|.
"""
def VALIDATE_REIMBURSEMENTS_PROMPT(service_term: str, reimb_term: str) -> str:
""" Validates whether a service-reimbursement pair represents a complete reimbursement methodology.
Args:
service_term (str): The service term to validate.
reimb_term (str): The reimbursement term to validate.
Returns:
str: A prompt for the AI to determine if the pair is a complete reimbursement methodology.
"""
return f"""Analyze this service-reimbursement pair from a healthcare contract:
def IDENTIFY_REIMBURSEMENT_EXHIBITS_PROMPT(exhibit_headers: dict, yaml_output: bool = False):
formatted_headers = [f"Page {page}: {header}" for page, header in exhibit_headers.items()]
return f"""Review these exhibit headers and identify which ones are likely to contain reimbursement methodologies, payment terms, or compensation schedules.
SERVICE: {service_term}
REIMBURSEMENT TERM: {reimb_term}
Consider both explicit and implicit indicators of reimbursement information:
Determine if this represents a COMPLETE reimbursement methodology that specifies how payment will be calculated.
EXPLICIT INDICATORS:
- Exhibits that have "Compensation Schedule", "Reimbursement Schedule", "Rates", or similar in the title. The title must somehow explicitly indicate that the Exhibit contains payment terms.
VALID reimbursement methodologies include:
- Specific payment rates (percentages, dollar amounts)
- Fee schedule references with rates
- Calculation formulas
- Payment limitations with amounts
- "Lesser of" or "greater of" statements with rates
IMPLICIT INDICATORS:
- Sections titled "COVERED SERVICES" for specific insurance products typically include payment terms because they define what services are reimbursable and how
- Attachments specifying plan types (Medicare, Medicaid, Commercial, Marketplace) usually contain reimbursement details
- References to specific programs (Medicare Advantage, CHIP) that have distinct payment methodologies
- Attachments that appear to define specific services or service categories
INVALID entries (return NO):
- Service definitions without payment terms
- Authorization requirements only
- Coverage limitations without payment rates
- General benefit descriptions
- Exclusions or non-covered statements
For each header, return 'EXPLICIT' if there is an EXPLICIT INDICATOR, 'IMPLICIT' if there is an IMPLICIT INDICATOR, or 'NO' if the header indicates that the exhibit is unlikely to contain reimbursement information.
Examples:
✅ "reimbursed at 100% of DMAP fee schedule" → YES (clear payment method)
✅ "flat rate of $20.00 per frame" → YES (specific rate)
❌ "Standard eyeglass lenses as defined by DMAP are CR39 lenses..." → NO (definition only)
❌ "Contact lenses must be prior authorized" → NO (authorization only)
Headers to review:
{"\n".join(formatted_headers)}
Return your answer as a {"JSON object" if not yaml_output else "YAML list"} where keys are PAGE NUMBERS and values are 'EXPLICIT', 'IMPLICIT', or 'NO'.
Feel free to explain your reasoning, but place your final {"JSON" if not yaml_output else "YAML"} output between triple backticks.
If you're uncertain about a header, err on the side of inclusion ('IMPLICIT').
"""
Briefly explain your reasoning, then return YES or NO in |pipes|."""
def EXHIBIT_LEVEL(context, fields):
fieldExtraction/src/utils/string_utils.py
+7 -2
View File
@@ -341,7 +341,7 @@ def count_reimbursements_in_exhibit(exhibit_text: str) -> int: #JUST by regex
"""
return len(re.findall(reimb_regex, exhibit_text))
def is_empty(value, pd_mask=True):
def is_empty(value: str | list | pd.Series, pd_mask: bool=True) -> bool | pd.Series:
"""
Checks if a value is considered empty or invalid.
@@ -353,7 +353,12 @@ def is_empty(value, pd_mask=True):
Otherwise, the function returns True iff the entire Series is empty.
Returns:
bool: True if the value is empty or invalid, False otherwise.
bool | pd.Series:
- When the input is a string, returns True if the value is empty or invalid, False otherwise.
- When the input is a list, returns True if the list is empty or all elements are empty/invalid.
- When the input is a pandas Series:
- If `pd_mask` is True, returns a boolean mask indicating which elements are empty or invalid.
- If `pd_mask` is False, returns True iff all elements are empty/invalid, or if the entire Series is empty.
"""
empty_values = [None, "", "N/A", "NA", "null", "none", "NaN", np.nan, "nan", "None"]
fieldExtraction/tests/dynamic_test.py
+3 -3
View File
@@ -135,9 +135,9 @@ class TestOneToN(unittest.TestCase):
def test_reimbursement_level(self, mock_code_breakout,
mock_special_cases, mock_reimbursement_primary):
# Setup
mock_reimbursement_primary.return_value = [{"REIMB_TERM": "term1"}]
mock_special_cases.return_value = [{"REIMB_TERM": "term1", "CARVEOUT_IND": "N"}]
mock_code_breakout.return_value = [{"REIMB_TERM": "term1", "CODE": "code1"}]
mock_reimbursement_primary.return_value = [{"REIMB_TERM": "term1$"}]
mock_special_cases.return_value = [{"REIMB_TERM": "term1$", "CARVEOUT_IND": "N"}]
mock_code_breakout.return_value = [{"REIMB_TERM": "term1$", "CODE": "code1"}]
mock_fields = MagicMock()
mock_dataset = {}
fieldExtraction/tests/one_to_n_test.py
+20 -20
View File
@@ -97,10 +97,10 @@ class TestOneToN(unittest.TestCase):
def test_reimbursement_level(self, mock_code_breakout,
mock_special_cases, mock_reimbursement_primary):
# Setup
mock_reimbursement_primary.return_value = [{"REIMB_TERM": "term1"}]
mock_special_cases.return_value = [{"REIMB_TERM": "term1", "CARVEOUT_IND": "N"}]
mock_code_breakout.return_value = [{"REIMB_TERM": "term1", "CODE": "code1"}]
mock_reimbursement_primary.return_value = [{"REIMB_TERM": "term1 paid at rate"}]
mock_special_cases.return_value = [{"REIMB_TERM": "term1 paid at rate", "CARVEOUT_IND": "N"}]
mock_code_breakout.return_value = [{"REIMB_TERM": "term1 paid at rate", "CODE": "code1"}]
mock_fields = MagicMock()
mock_dataset = {}
exhibit_page = "1"
@@ -123,9 +123,9 @@ class TestOneToN(unittest.TestCase):
"""Test that duplicate service-reimbursement pairs are properly filtered"""
# Setup - simulate getting duplicate pairs
mock_reimbursement_primary.return_value = [
{"SERVICE_TERM": "Service A", "REIMB_TERM": "Rate A"},
{"SERVICE_TERM": "Service B", "REIMB_TERM": "Rate B"},
{"SERVICE_TERM": "Service A", "REIMB_TERM": "Rate A"} # Duplicate
{"SERVICE_TERM": "Service A", "REIMB_TERM": "paid at 80% of fee schedule"},
{"SERVICE_TERM": "Service B", "REIMB_TERM": "$50 per visit"},
{"SERVICE_TERM": "Service A", "REIMB_TERM": "paid at 80% of fee schedule"} # Duplicate
]
mock_fields = MagicMock()
@@ -150,19 +150,19 @@ class TestOneToN(unittest.TestCase):
self.assertEqual(len(result), 2)
# Verify seen_pairs was updated
self.assertEqual(len(seen_pairs), 2)
self.assertIn(("Service A", "Rate A"), seen_pairs)
self.assertIn(("Service B", "Rate B"), seen_pairs)
self.assertIn(("Service A", "paid at 80% of fee schedule"), seen_pairs)
self.assertIn(("Service B", "$50 per visit"), seen_pairs)
@patch('src.investment.one_to_n_funcs.get_reimbursement_primary')
def test_reimbursement_level_cross_exhibit_deduplication(self, mock_reimbursement_primary):
"""Test that pairs seen in previous exhibits are filtered out"""
# Setup - pre-populate seen_pairs as if previous exhibit processed
seen_pairs = {("Service A", "Rate A")}
seen_pairs = {("Service A", "paid at 80% of fee schedule"),}
# Current exhibit has same pair plus new one
mock_reimbursement_primary.return_value = [
{"SERVICE_TERM": "Service A", "REIMB_TERM": "Rate A"}, # Should be filtered
{"SERVICE_TERM": "Service B", "REIMB_TERM": "Rate B"} # Should be kept
{"SERVICE_TERM": "Service A", "REIMB_TERM": "paid at 80% of fee schedule"}, # Should be filtered
{"SERVICE_TERM": "Service B", "REIMB_TERM": "$50 per visit"} # Should be kept
]
mock_fields = MagicMock()
@@ -189,12 +189,12 @@ class TestOneToN(unittest.TestCase):
def test_reimbursement_level_all_duplicates(self):
"""Test behavior when all pairs are duplicates"""
seen_pairs = {("Service A", "Rate A"), ("Service B", "Rate B")}
seen_pairs = {("Service A", "paid at 80% of fee schedule"), ("Service B", "$50 per visit")}
with patch('src.investment.one_to_n_funcs.get_reimbursement_primary') as mock_primary:
mock_primary.return_value = [
{"SERVICE_TERM": "Service A", "REIMB_TERM": "Rate A"},
{"SERVICE_TERM": "Service B", "REIMB_TERM": "Rate B"}
{"SERVICE_TERM": "Service A", "REIMB_TERM": "paid at 80% of fee schedule"},
{"SERVICE_TERM": "Service B", "REIMB_TERM": "$50 per visit"}
]
mock_fields = MagicMock()
@@ -213,10 +213,10 @@ class TestOneToN(unittest.TestCase):
def test_deduplicate_with_accumulator(self):
"""Test the deduplication function directly"""
answers = [
{"SERVICE_TERM": "Service A", "REIMB_TERM": "Rate A"},
{"SERVICE_TERM": "Service B", "REIMB_TERM": "Rate B"},
{"SERVICE_TERM": "Service A", "REIMB_TERM": "Rate A"}, # Duplicate
{"SERVICE_TERM": " Service C ", "REIMB_TERM": " Rate C "} # Test stripping
{"SERVICE_TERM": "Service A", "REIMB_TERM": "paid at 80% of fee schedule"},
{"SERVICE_TERM": "Service B", "REIMB_TERM": "$50 per visit"},
{"SERVICE_TERM": "Service A", "REIMB_TERM": "paid at 80% of fee schedule"}, # Duplicate
{"SERVICE_TERM": " Service C ", "REIMB_TERM": " $150 "} # Test stripping
]
seen_pairs = set()
@@ -230,7 +230,7 @@ class TestOneToN(unittest.TestCase):
self.assertEqual(len(seen_pairs), 3)
# Verify stripping works
self.assertIn(("Service C", "Rate C"), seen_pairs)
self.assertIn(("Service C", "$150"), seen_pairs)
if __name__ == '__main__':
unittest.main()