Merged in bugfix/ut-errors (pull request #756)
Bugfix/ut errors * Update universal_json_load and unit tests * Update tests * Fix unit tests * remove test * remove imports * Update function * update pd Series is_empty * Update reimbursement check * added some test cases * Merged main into bugfix/ut-errors * Merged main into bugfix/ut-errors * remove import * Merged main into bugfix/ut-errors Approved-by: Karan Desai
This commit is contained in:
Katon Minhas
@@ -182,76 +182,47 @@ def json_parsing_search(response_text: str, field_list: list[str]) -> dict[str,
|
||||
return dict(zip(field_l, answer_l))
|
||||
|
||||
|
||||
def universal_json_load(string_dict: str):
|
||||
"""Matches json dicts and list of dicts - NOT simple lists
|
||||
Args:
|
||||
string_dict (str): The string containing the JSON-like content.
|
||||
Returns:
|
||||
list[dict] | dict: The dictionary or list of dictionaries obtained from parsing the cleaned and corrected string.
|
||||
Raises:
|
||||
ValueError: If no valid JSON object is found in the input string. In addition,
|
||||
the traceback as well as the offending string are printed.
|
||||
|
||||
Note: if more than one json-like (or list-like) is found in the input string, this function will
|
||||
take the last one positionally.
|
||||
def universal_json_load(s: str):
|
||||
"""
|
||||
# Try to match dictionary or list of dictionaries first
|
||||
# It will find any substring starting with { or [ and ending with } or ].
|
||||
# dict_matches = re.findall(r"\{.*?\}|\[\s*?\{.*?\}\s*?\]", string_dict, re.DOTALL)
|
||||
dict_matches = re.findall(r'(\[.*\]|\{.*?\})', string_dict, re.DOTALL)
|
||||
if dict_matches:
|
||||
matched_str = dict_matches[-1] # Take the last match in the string
|
||||
try:
|
||||
return json.loads(matched_str)
|
||||
except json.JSONDecodeError:
|
||||
# If it fails, it might be a malformed list (e.g., "[{...} text {...}]").
|
||||
# As a fallback, find the LAST dictionary within that malformed string.
|
||||
fallback_matches = re.findall(r'\{.*?\}', matched_str, re.DOTALL)
|
||||
if fallback_matches:
|
||||
last_dict_in_string = fallback_matches[-1]
|
||||
try:
|
||||
return json.loads(last_dict_in_string)
|
||||
except json.JSONDecodeError as e:
|
||||
logging.error(f"Failed to parse JSON in fallback: {e}")
|
||||
logging.error(f"Original string: {string_dict}")
|
||||
logging.error(f"Traceback: {traceback.format_exc()}")
|
||||
# Use placeholders to preserve JSON structure characters for CSV compatibility
|
||||
csv_safe_string = (
|
||||
string_dict.replace(",", "[COMMA]")
|
||||
.replace("\n", "[NEWLINE]")
|
||||
.replace('"', "[QUOTE]")
|
||||
)
|
||||
raise ValueError(
|
||||
f"JSON parsing failed in fallback: {e.msg} at position {e.pos}; Original string:\n{csv_safe_string}"
|
||||
) from e
|
||||
|
||||
# If no dict pattern found, try to match list of strings
|
||||
list_matches = re.findall(r"\[(.*?)\]", string_dict, re.DOTALL)
|
||||
if list_matches:
|
||||
matched_str = f"[{list_matches[-1]}]" # Take the last match in the string
|
||||
try:
|
||||
# Clean up the string list before parsing
|
||||
cleaned_str = re.sub(r'(?<!\\)"', '"', matched_str) # Handle escaped quotes
|
||||
cleaned_str = re.sub(
|
||||
r"'", '"', cleaned_str
|
||||
) # Replace single quotes with double quotes
|
||||
|
||||
return json.loads(cleaned_str)
|
||||
except json.JSONDecodeError as e:
|
||||
logging.error(f"Failed to parse list: {e}")
|
||||
logging.error(f"Original string: {string_dict}")
|
||||
logging.error(f"Traceback: {traceback.format_exc()}")
|
||||
# Use placeholders to preserve JSON structure characters for CSV compatibility
|
||||
csv_safe_string = (
|
||||
string_dict.replace(",", "[COMMA]")
|
||||
.replace("\n", "[NEWLINE]")
|
||||
.replace('"', "[QUOTE]")
|
||||
)
|
||||
raise ValueError(
|
||||
f"JSON parsing failed: {e.msg} at position {e.pos}; Original string:\n{csv_safe_string}"
|
||||
) from e
|
||||
|
||||
return {}
|
||||
Extract and parse all highest-level valid JSON objects from a string.
|
||||
Returns the last valid top-level JSON object found (dict, list, etc).
|
||||
Raises ValueError if no valid JSON objects are found. Does not clean up syntax.
|
||||
"""
|
||||
if not isinstance(s, str):
|
||||
raise TypeError(f"Expected string input, got {type(s).__name__}")
|
||||
found = []
|
||||
i = 0
|
||||
while i < len(s):
|
||||
if s[i] in '{[':
|
||||
start = i
|
||||
stack = [s[i]]
|
||||
for j in range(i+1, len(s)):
|
||||
if s[j] in '{[':
|
||||
stack.append(s[j])
|
||||
elif s[j] in '}]':
|
||||
if not stack:
|
||||
break
|
||||
open_bracket = stack.pop()
|
||||
if (open_bracket == '{' and s[j] != '}') or (open_bracket == '[' and s[j] != ']'):
|
||||
break
|
||||
if not stack:
|
||||
candidate = s[start:j+1]
|
||||
# Only consider top-level objects (not nested)
|
||||
# Check if start is not inside another object
|
||||
# This is guaranteed by stack being empty only at top-level
|
||||
try:
|
||||
obj = json.loads(candidate)
|
||||
found.append(obj)
|
||||
except Exception:
|
||||
pass
|
||||
i = j # Move past this top-level object
|
||||
break
|
||||
else:
|
||||
pass
|
||||
i += 1
|
||||
if found:
|
||||
return found[-1]
|
||||
raise ValueError(f"No valid JSON objects found in string: {s[:100]}...")
|
||||
|
||||
|
||||
reimbursement_strings = [ # These are for `method='keyword'`
|
||||
@@ -259,6 +230,9 @@ reimbursement_strings = [ # These are for `method='keyword'`
|
||||
"$",
|
||||
"percent",
|
||||
"billed charges",
|
||||
"compensation",
|
||||
"reimbursement",
|
||||
"fee schedule"
|
||||
]
|
||||
reimb_regex = r"(?<![$%])(?:\$\d+|\d+[$%])(?![$%])"
|
||||
|
||||
@@ -361,7 +335,11 @@ def is_empty(value: str | list | pd.Series, pd_mask: bool = True) -> bool | pd.S
|
||||
)
|
||||
)
|
||||
else:
|
||||
return value.isna().all()
|
||||
# If the Series is empty, it's considered empty
|
||||
if value.empty:
|
||||
return True
|
||||
# Use the string-version check for each element in the Series
|
||||
return all(is_empty(str(v)) for v in value)
|
||||
|
||||
if pd.isna(value):
|
||||
return True
|
||||
|
||||
@@ -1,511 +1,203 @@
|
||||
import numpy as np
|
||||
import unittest
|
||||
import json
|
||||
import pandas as pd
|
||||
import pytest
|
||||
import src.utils as utils
|
||||
import src.utils.llm_utils as llm_utils
|
||||
import numpy as np
|
||||
from src.utils import string_utils
|
||||
from constants.delimiters import Delimiter
|
||||
from src.utils.string_utils import (universal_json_load,
|
||||
contains_reimbursement,
|
||||
count_reimbursements_in_exhibit,
|
||||
extract_signature_page,
|
||||
extract_text_from_delimiters, is_empty,
|
||||
json_parsing_search, page_key_sort)
|
||||
|
||||
class TestStringUtils(unittest.TestCase):
|
||||
def test_extract_text_from_delimiters_pipe(self):
|
||||
s = "foo |bar| baz |qux|"
|
||||
self.assertEqual(string_utils.extract_text_from_delimiters(s, Delimiter.PIPE), "qux")
|
||||
self.assertEqual(string_utils.extract_text_from_delimiters(s, Delimiter.PIPE, 0), "bar")
|
||||
self.assertEqual(string_utils.extract_text_from_delimiters("no delimiters", Delimiter.PIPE), "N/A")
|
||||
with self.assertRaises(TypeError):
|
||||
string_utils.extract_text_from_delimiters(123, Delimiter.PIPE)
|
||||
with self.assertRaises(TypeError):
|
||||
string_utils.extract_text_from_delimiters(s, "notadelimiter")
|
||||
|
||||
def test_extract_text_from_delimiters_backtick(self):
|
||||
s = "foo `bar` baz `qux`"
|
||||
self.assertEqual(string_utils.extract_text_from_delimiters(s, Delimiter.BACKTICK), "qux")
|
||||
|
||||
def test_extract_text_from_delimiters_triple_backtick(self):
|
||||
s = "foo ```bar``` baz ```qux```"
|
||||
self.assertEqual(string_utils.extract_text_from_delimiters(s, Delimiter.TRIPLE_BACKTICK), "qux")
|
||||
|
||||
def test_page_key_sort(self):
|
||||
keys = ["10", "2", "A", "1", "B", "1.5"]
|
||||
sorted_keys = sorted(keys, key=string_utils.page_key_sort)
|
||||
self.assertEqual(sorted_keys, ["1", "1.5", "2", "10", "A", "B"])
|
||||
|
||||
def test_json_parsing_search(self):
|
||||
s = '{"FIELD1": "value1", "FIELD2": "value2"}'
|
||||
fields = ["FIELD1", "FIELD2"]
|
||||
result = string_utils.json_parsing_search(s, fields)
|
||||
self.assertEqual(result, {"FIELD1": "value1", "FIELD2": "value2"})
|
||||
|
||||
def test_universal_json_load_valid(self):
|
||||
# Simple dict
|
||||
s = '{"a": 1, "b": 2}'
|
||||
self.assertEqual(string_utils.universal_json_load(s), {"a": 1, "b": 2})
|
||||
# Simple list
|
||||
s = '[1, 2, 3]'
|
||||
self.assertEqual(string_utils.universal_json_load(s), [1, 2, 3])
|
||||
# Nested dict
|
||||
s = '{"a": {"b": [1, 2, {"c": "d"}]}}'
|
||||
self.assertEqual(string_utils.universal_json_load(s), {"a": {"b": [1, 2, {"c": "d"}]}})
|
||||
# List of dicts
|
||||
s = '[{"x": 1}, {"y": 2}]'
|
||||
self.assertEqual(string_utils.universal_json_load(s), [{"x": 1}, {"y": 2}])
|
||||
# Dict of lists
|
||||
s = '{"a": [1, 2], "b": [3, 4]}'
|
||||
self.assertEqual(string_utils.universal_json_load(s), {"a": [1, 2], "b": [3, 4]})
|
||||
# List of strings
|
||||
s = '["foo", "bar", "baz"]'
|
||||
self.assertEqual(string_utils.universal_json_load(s), ["foo", "bar", "baz"])
|
||||
# JSON with text before/after
|
||||
s = 'start {"field": "value"} end'
|
||||
self.assertEqual(string_utils.universal_json_load(s), {"field": "value"})
|
||||
s = 'before [1, 2, 3] after'
|
||||
self.assertEqual(string_utils.universal_json_load(s), [1, 2, 3])
|
||||
# Multiple objects, last wins
|
||||
s = 'first {"a": 1} middle {"b": 2} last {"c": 3}'
|
||||
self.assertEqual(string_utils.universal_json_load(s), {"c": 3})
|
||||
# Multiple arrays, last wins
|
||||
s = 'first [1] middle [2] last [3]'
|
||||
self.assertEqual(string_utils.universal_json_load(s), [3])
|
||||
# Nested JSON with text 1
|
||||
s = 'junk {"outer": {"inner": [1, 2, {"deep": true}]}} more junk'
|
||||
self.assertEqual(string_utils.universal_json_load(s), {"outer": {"inner": [1, 2, {"deep": True}]}})
|
||||
# Nested JSON with text 2
|
||||
s = 'junk [{"dict1_key": "dict1_val"}, {"dict2_key" : "dict2_val"}] more junk'
|
||||
self.assertEqual(string_utils.universal_json_load(s), [{"dict1_key": "dict1_val"}, {"dict2_key" : "dict2_val"}])
|
||||
# List of dicts
|
||||
s = 'junk```json[{"a1": "value_a1","b1": "value_b1"},{"a2": "value_a2","b2": "value_b2"}]```'
|
||||
self.assertEqual(string_utils.universal_json_load(s), [{"a1": "value_a1","b1": "value_b1"},{"a2": "value_a2","b2": "value_b2"}])
|
||||
# Boolean and null values
|
||||
s = '{"ok": true, "val": null}'
|
||||
self.assertEqual(string_utils.universal_json_load(s), {"ok": True, "val": None})
|
||||
|
||||
|
||||
class TestStringUtils:
|
||||
@pytest.mark.parametrize(
|
||||
"raw_text, delimiter, match_index, expected",
|
||||
[
|
||||
(
|
||||
"Here is the answer |this is a pipe answer| and |more text.|",
|
||||
Delimiter.PIPE,
|
||||
0,
|
||||
"this is a pipe answer",
|
||||
),
|
||||
(
|
||||
"Here is the answer |this is a pipe answer| and |more text.|",
|
||||
Delimiter.PIPE,
|
||||
-1,
|
||||
"more text.",
|
||||
),
|
||||
(
|
||||
"|Here| is the answer |this is a pipe answer| and |more text.|",
|
||||
Delimiter.PIPE,
|
||||
1,
|
||||
"this is a pipe answer",
|
||||
),
|
||||
(
|
||||
"Here is the answer `this is a backtick answer` and `more text.`",
|
||||
Delimiter.BACKTICK,
|
||||
0,
|
||||
"this is a backtick answer",
|
||||
),
|
||||
(
|
||||
"Here is the answer `this is a backtick answer` and `more text.`",
|
||||
Delimiter.BACKTICK,
|
||||
-1,
|
||||
"more text.",
|
||||
),
|
||||
(
|
||||
"`Here` is the answer `this` is a `backtick answer` and `more text.`",
|
||||
Delimiter.BACKTICK,
|
||||
2,
|
||||
"backtick answer",
|
||||
),
|
||||
(
|
||||
"Here is the answer ```this is a triple backtick answer``` and more text.",
|
||||
Delimiter.TRIPLE_BACKTICK,
|
||||
0,
|
||||
"this is a triple backtick answer",
|
||||
),
|
||||
(
|
||||
"Here is the answer ```this``` is a ```triple``` backtick ```answer``` and more text.",
|
||||
Delimiter.TRIPLE_BACKTICK,
|
||||
-1,
|
||||
"answer",
|
||||
),
|
||||
(
|
||||
"```Here``` is the answer ```this is a triple backtick answer``` and ```more text.```",
|
||||
Delimiter.TRIPLE_BACKTICK,
|
||||
1,
|
||||
"this is a triple backtick answer",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_extract_text_from_delimiters(
|
||||
self, raw_text, delimiter, match_index, expected
|
||||
):
|
||||
result = extract_text_from_delimiters(raw_text, delimiter, match_index)
|
||||
assert result == expected
|
||||
def test_universal_json_load_invalid(self):
|
||||
# No valid JSON
|
||||
s = 'no json here'
|
||||
with self.assertRaises(ValueError):
|
||||
string_utils.universal_json_load(s)
|
||||
# Broken JSON
|
||||
s = '{broken: "json"}'
|
||||
with self.assertRaises(ValueError):
|
||||
string_utils.universal_json_load(s)
|
||||
# Non-string input
|
||||
with self.assertRaises(TypeError):
|
||||
string_utils.universal_json_load(123)
|
||||
# Unbalanced brackets
|
||||
s = '{"a": [1, 2, 3}'
|
||||
with self.assertRaises(ValueError):
|
||||
string_utils.universal_json_load(s)
|
||||
# Empty string
|
||||
s = ''
|
||||
with self.assertRaises(ValueError):
|
||||
string_utils.universal_json_load(s)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"response_text, field_list, expected",
|
||||
[
|
||||
(
|
||||
'{"name": "John", "age": "30", "city": "New York"}',
|
||||
["name", "age", "city"],
|
||||
{"name": "John", "age": "30", "city": "New York"},
|
||||
),
|
||||
(
|
||||
'{"name": "John", "city": "New York"}',
|
||||
["name", "age", "city"],
|
||||
{"name": "John", "city": "New York"},
|
||||
),
|
||||
("", ["name", "age", "city"], {}),
|
||||
('{"name": "John", "age": "30", "city": "New York"}', [], {}),
|
||||
(
|
||||
'{"name": "", "age": "", "city": ""}',
|
||||
["name", "age", "city"],
|
||||
{"name": "", "age": "", "city": ""},
|
||||
),
|
||||
(
|
||||
'{\n"name": "John",\n"age": "30",\n"city": "New York"\n}',
|
||||
["name", "age", "city"],
|
||||
{"name": "John", "age": "30", "city": "New York"},
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_json_parsing_search(self, response_text, field_list, expected):
|
||||
assert json_parsing_search(response_text, field_list) == expected
|
||||
def test_universal_json_load_edge_cases(self):
|
||||
# Empty object
|
||||
s = 'text {} more text'
|
||||
self.assertEqual(string_utils.universal_json_load(s), {})
|
||||
# Empty array
|
||||
s = 'text [] more text'
|
||||
self.assertEqual(string_utils.universal_json_load(s), [])
|
||||
# Large object
|
||||
large = {f"k{i}": i for i in range(100)}
|
||||
s = f'before {json.dumps(large)} after'
|
||||
self.assertEqual(string_utils.universal_json_load(s), large)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"raw_text, expected",
|
||||
[
|
||||
(
|
||||
'Reasoning ... ```json [{"SERVICE_TERM": "Medicare Covered Services", "REIMB_TERM": "Billed Charges"}, {"SERVICE_TERM": "Medicaid Covered Services", "REIMB_TERM": "120% of Medicaid fee schedule"}]``` ... More text.',
|
||||
[{"SERVICE_TERM": "Medicare Covered Services", "REIMB_TERM": "Billed Charges"}, {"SERVICE_TERM": "Medicaid Covered Services", "REIMB_TERM": "120% of Medicaid fee schedule"}],
|
||||
),
|
||||
(
|
||||
'Reasoning ... answer {"SERVICE_TERM": "Medicare Covered Services", "REIMB_TERM": "Billed Charges"} ... More text.',
|
||||
{"SERVICE_TERM": "Medicare Covered Services", "REIMB_TERM": "Billed Charges"},
|
||||
),
|
||||
(
|
||||
'Reasoning ... answer list [{"SERVICE_TERM": "Medicare Covered Services", "REIMB_TERM": "Billed Charges"}, {"SERVICE_TERM": "Medicaid Covered Services", "REIMB_TERM": "120% of Medicaid fee schedule"}]',
|
||||
[{"SERVICE_TERM": "Medicare Covered Services", "REIMB_TERM": "Billed Charges"}, {"SERVICE_TERM": "Medicaid Covered Services", "REIMB_TERM": "120% of Medicaid fee schedule"}],
|
||||
),
|
||||
(
|
||||
'Reasoning ... single answer {"SERVICE_TERM": "Medicare Covered Services", "REIMB_TERM": "Billed Charges"}',
|
||||
{"SERVICE_TERM": "Medicare Covered Services", "REIMB_TERM": "Billed Charges"},
|
||||
),
|
||||
(
|
||||
'Reasoning ...{"a":"b"}... ```json [{"SERVICE_TERM": "Medicare Covered Services", "REIMB_TERM": "Billed Charges"}, {"SERVICE_TERM": "Medicaid Covered Services", "REIMB_TERM": "120% of Medicaid fee schedule"}]```',
|
||||
[{"SERVICE_TERM": "Medicare Covered Services", "REIMB_TERM": "Billed Charges"}, {"SERVICE_TERM": "Medicaid Covered Services", "REIMB_TERM": "120% of Medicaid fee schedule"}],
|
||||
),
|
||||
(
|
||||
'Reasoning ...{"a":"b"}... answer {"SERVICE_TERM": "Medicare Covered Services", "REIMB_TERM": "Billed Charges"}',
|
||||
{"SERVICE_TERM": "Medicare Covered Services", "REIMB_TERM": "Billed Charges"},
|
||||
),
|
||||
(
|
||||
'Reasoning ...{"a":"b"}... ```json [{"SERVICE_TERM": [1,2,3], "REIMB_TERM": "Billed Charges"}, {"SERVICE_TERM": "Medicaid Covered Services", "REIMB_TERM": "120% of Medicaid fee schedule"}]```',
|
||||
[{"SERVICE_TERM": [1,2,3], "REIMB_TERM": "Billed Charges"}, {"SERVICE_TERM": "Medicaid Covered Services", "REIMB_TERM": "120% of Medicaid fee schedule"}],
|
||||
),
|
||||
(
|
||||
'Reasoning ...{"a":"b"}... answer {"SERVICE_TERM": [1,2,3], "REIMB_TERM": "Billed Charges"}',
|
||||
{"SERVICE_TERM": [1,2,3], "REIMB_TERM": "Billed Charges"},
|
||||
),
|
||||
# following test case with nested dict will fail (because we are using non greedy regex for dict) - kept for reference
|
||||
# (
|
||||
# 'Reasoning ... answer {"SERVICE_TERM": [1,2,3], "REIMB_TERM": {"1":"Billed Charges"}}',
|
||||
# {"SERVICE_TERM": [1,2,3], "REIMB_TERM": {"1":"Billed Charges"}},
|
||||
# ),
|
||||
# following test case with separate lists will fail (because we are using greedy regex for list) - kept for reference
|
||||
# (
|
||||
# 'Reasoning ...[]... answer {"SERVICE_TERM": [1,2,3], "REIMB_TERM": "Billed Charges"}',
|
||||
# {"SERVICE_TERM": [1,2,3], "REIMB_TERM": {"1":"Billed Charges"}},
|
||||
# ),
|
||||
],
|
||||
)
|
||||
def test_universal_json_load(
|
||||
self, raw_text, expected
|
||||
):
|
||||
result = universal_json_load(raw_text)
|
||||
assert result == expected
|
||||
def test_contains_reimbursement(self):
|
||||
s = "90% reimbursement of billed charges"
|
||||
self.assertTrue(string_utils.contains_reimbursement(s, method="keyword"))
|
||||
s = "Payment of $500 per visit"
|
||||
self.assertTrue(string_utils.contains_reimbursement(s, method="regex"))
|
||||
s = "No reimbursement here"
|
||||
self.assertTrue(string_utils.contains_reimbursement(s, method="keyword"))
|
||||
with self.assertRaises(ValueError):
|
||||
string_utils.contains_reimbursement(s, method="invalid")
|
||||
|
||||
@pytest.fixture
|
||||
def mock_invoke_claude(self, mocker):
|
||||
return mocker.patch.object(llm_utils, "invoke_claude")
|
||||
def test_is_empty_string(self):
|
||||
self.assertTrue(string_utils.is_empty(""))
|
||||
self.assertTrue(string_utils.is_empty("N/A"))
|
||||
self.assertFalse(string_utils.is_empty("valid"))
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"text, page, expected",
|
||||
[
|
||||
(
|
||||
{
|
||||
"1": "This page contains a 10% reimbursement schedule.",
|
||||
"2": "No relevant content here.",
|
||||
},
|
||||
"1",
|
||||
True,
|
||||
),
|
||||
(
|
||||
{
|
||||
"1": "This page has no relevant content.",
|
||||
"2": "Still no relevant content here.",
|
||||
},
|
||||
"1",
|
||||
False,
|
||||
),
|
||||
(
|
||||
{
|
||||
"1": "This page contains a 10% reimbursement schedule.",
|
||||
"2": "No relevant content here.",
|
||||
},
|
||||
"invalid_page",
|
||||
False,
|
||||
),
|
||||
("This text contains a $100 reimbursement schedule.", "1", True),
|
||||
("This text has no relevant content.", "1", False),
|
||||
],
|
||||
)
|
||||
def test_contains_reimbursement(self, text, page, expected):
|
||||
result = contains_reimbursement(text, page)
|
||||
assert result == expected
|
||||
def test_is_empty_list(self):
|
||||
self.assertTrue(string_utils.is_empty([]))
|
||||
self.assertTrue(string_utils.is_empty([None, "", "N/A"]))
|
||||
self.assertFalse(string_utils.is_empty(["valid"]))
|
||||
|
||||
def test_invalid_input_type(self, caplog):
|
||||
text = ["This is a list, not a dictionary or string."]
|
||||
page = "1"
|
||||
result = contains_reimbursement(text, page)
|
||||
assert result is False
|
||||
assert "contains_reimbursement - Invalid data type" in caplog.text
|
||||
def test_is_empty_pandas_series(self):
|
||||
s = pd.Series(["valid", "", None, "N/A"])
|
||||
mask = string_utils.is_empty(s, pd_mask=True)
|
||||
self.assertTrue(mask.iloc[1])
|
||||
self.assertTrue(mask.iloc[2])
|
||||
self.assertTrue(mask.iloc[3])
|
||||
self.assertFalse(mask.iloc[0])
|
||||
self.assertFalse(string_utils.is_empty(pd.Series(["valid", None]), pd_mask=False))
|
||||
self.assertTrue(string_utils.is_empty(pd.Series([None, "", "N/A"]), pd_mask=False))
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value, expected",
|
||||
[
|
||||
(None, True),
|
||||
("", True),
|
||||
("N/A", True),
|
||||
("null", True),
|
||||
("none", True),
|
||||
("NaN", True),
|
||||
(np.nan, True),
|
||||
(pd.NA, True),
|
||||
("This is not empty.", False),
|
||||
(42, False),
|
||||
(" ", True), # Whitespace-only string
|
||||
("\t", True), # Tab character
|
||||
("\n", True), # Newline character
|
||||
("n/a", True), # Lowercase N/A
|
||||
("na", True), # Lowercase NA
|
||||
("NULL", True), # Uppercase NULL
|
||||
("NONE", True), # Uppercase NONE
|
||||
("None", True), # Mixed case None (already in empty_values)
|
||||
("UNKNOWN", False), # Should NOT be empty
|
||||
("unknown", False), # Should NOT be empty
|
||||
(" N/A ", True), # N/A with whitespace
|
||||
],
|
||||
)
|
||||
def test_is_empty(self, value, expected):
|
||||
result = is_empty(value)
|
||||
assert result == expected
|
||||
def test_datetime_str(self):
|
||||
result = string_utils.datetime_str()
|
||||
self.assertTrue(result.startswith("["))
|
||||
self.assertTrue(result.endswith("]"))
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"text_dict, filename, expected",
|
||||
[
|
||||
(
|
||||
{
|
||||
"1": "This page has 2 signatures.",
|
||||
"2": "None here.",
|
||||
"3": "Signature of the party (but ignored)",
|
||||
},
|
||||
"test_file.txt",
|
||||
{"1": "This page has 2 signatures."},
|
||||
),
|
||||
(
|
||||
{"1": "No relevant content here.", "2": "Nor over here."},
|
||||
"test_file.txt",
|
||||
{},
|
||||
),
|
||||
(
|
||||
{
|
||||
"1": "This page has a signature.",
|
||||
"2": "signature page follow.",
|
||||
"3": "Signature of the party (but ignored)",
|
||||
},
|
||||
"test_file.txt",
|
||||
{
|
||||
"2": "signature page follow.",
|
||||
"3": "Signature of the party (but ignored)",
|
||||
},
|
||||
),
|
||||
(
|
||||
{
|
||||
"1": "This page has 1 signature.",
|
||||
"2": "No relevant content.",
|
||||
"3": "Signature block here. But ignored bc of the first textract-like page",
|
||||
},
|
||||
"test_file.txt",
|
||||
{"1": "This page has 1 signature."},
|
||||
),
|
||||
(
|
||||
{"1": "No signed names at all.", "2": "Just some random text."},
|
||||
"test_file.txt",
|
||||
{},
|
||||
),
|
||||
(
|
||||
{
|
||||
"1": "some random text",
|
||||
"2": "Signature authorization",
|
||||
"3": "Signature of the party",
|
||||
},
|
||||
"test_file.txt",
|
||||
{"2": "Signature authorization", "3": "Signature of the party"},
|
||||
),
|
||||
({}, "test_file.txt", {}),
|
||||
],
|
||||
)
|
||||
def test_extract_signature_page(self, text_dict, filename, expected):
|
||||
result = extract_signature_page(text_dict, filename)
|
||||
assert result == expected
|
||||
def test_extract_signature_page(self):
|
||||
text_dict = {"1": "Contract page 1", "2": "This page has 2 signatures at the bottom", "3": "Final page"}
|
||||
result = string_utils.extract_signature_page(text_dict, "test.pdf")
|
||||
self.assertIn("2", result)
|
||||
self.assertEqual(result["2"], "This page has 2 signatures at the bottom")
|
||||
text_dict = {"1": "Contract page 1", "2": "Signature page follows this section", "3": "Final signature authorization page"}
|
||||
result = string_utils.extract_signature_page(text_dict, "test.pdf")
|
||||
self.assertIn("2", result)
|
||||
self.assertIn("3", result)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"page_key, expected",
|
||||
[
|
||||
("1", (0, 1)), # Numeric key
|
||||
("10", (0, 10)), # Larger numeric key
|
||||
("A", (1, "A")), # Alphabetic key
|
||||
("Z", (1, "Z")), # Another alphabetic key
|
||||
("1A", (1, "1A")), # Alphanumeric key
|
||||
("", (1, "")), # Empty string
|
||||
("001", (0, 1)), # Numeric key with leading zeros
|
||||
("B2", (1, "B2")), # Alphanumeric key with letter first
|
||||
("-5", (0, -5)), # Negative numeric
|
||||
(" ", (1, " ")), # Space character
|
||||
],
|
||||
)
|
||||
def test_page_key_sort(self, page_key, expected):
|
||||
result = page_key_sort(page_key)
|
||||
assert result == expected
|
||||
def test_extract_effective_date_pages(self):
|
||||
text_dict = {"1": "Contract effective date: January 1, 2024", "2": "General terms and conditions", "3": "This page has 1 signature"}
|
||||
result = string_utils.extract_effective_date_pages(text_dict, "test.pdf")
|
||||
self.assertIn("1", result)
|
||||
self.assertIn("3", result)
|
||||
self.assertEqual(len(result), 2)
|
||||
|
||||
def test_page_key_sort_with_sorting(self):
|
||||
keys = ["10", "IV", "2", "A", "1", "B", "Z"]
|
||||
expected_sorted_keys = ["1", "2", "10", "A", "B", "IV", "Z"]
|
||||
sorted_keys = sorted(keys, key=page_key_sort)
|
||||
assert sorted_keys == expected_sorted_keys
|
||||
def test_normalize_state_to_abbreviation(self):
|
||||
self.assertEqual(string_utils.normalize_state_to_abbreviation("California"), "CA")
|
||||
self.assertEqual(string_utils.normalize_state_to_abbreviation("ca"), "CA")
|
||||
self.assertEqual(string_utils.normalize_state_to_abbreviation("N.Y."), "NY")
|
||||
self.assertEqual(string_utils.normalize_state_to_abbreviation("Unknown State"), "Unknown State")
|
||||
self.assertEqual(string_utils.normalize_state_to_abbreviation("N/A"), "N/A")
|
||||
|
||||
def test_flatten_to_strings(self):
|
||||
self.assertEqual(string_utils.flatten_to_strings(["a", ["b", "c"], "d"]), ["a", "b", "c", "d"])
|
||||
self.assertEqual(string_utils.flatten_to_strings("single"), ["single"])
|
||||
self.assertEqual(string_utils.flatten_to_strings([1, [2, 3], "text"]), ["1", "2", "3", "text"])
|
||||
self.assertEqual(string_utils.flatten_to_strings([" text ", [" nested "]]), ["text", "nested"])
|
||||
def test_universal_json_load_multiple_invalid(self):
|
||||
# Multiple invalid JSON objects
|
||||
s = '{broken: true} some text {not: valid} more text'
|
||||
with self.assertRaises(ValueError):
|
||||
string_utils.universal_json_load(s)
|
||||
|
||||
def test_universal_json_load_one_invalid_multiple_valid(self):
|
||||
# One invalid, multiple valid (should take the last valid)
|
||||
s = '{broken: true} before {"a": 1} middle {"b": 2} end'
|
||||
self.assertEqual(string_utils.universal_json_load(s), {"b": 2})
|
||||
|
||||
def test_universal_json_load_first_nested(self):
|
||||
# Multiple, with the first one being nested
|
||||
s = '{"outer": {"inner": [1, 2]}} middle {"simple": true} end'
|
||||
self.assertEqual(string_utils.universal_json_load(s), {"simple": True})
|
||||
|
||||
def test_universal_json_load_last_nested(self):
|
||||
# Multiple, with the last one being nested
|
||||
s = '{"simple": true} middle {"outer": {"inner": [1, 2]}} end'
|
||||
self.assertEqual(string_utils.universal_json_load(s), {"outer": {"inner": [1, 2]}})
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
|
||||
class TestCountReimbursementsInExhibit:
|
||||
@pytest.mark.parametrize(
|
||||
"exhibit_text, expected_count",
|
||||
[
|
||||
("This exhibit includes a 10% reimbursement and a $100 reimbursement.", 2),
|
||||
("No reimbursements mentioned here.", 0),
|
||||
("Reimbursement of 50% and another reimbursement of $200.", 2),
|
||||
("100 percent reimbursement and fifty dollars reimbursement.", 0),
|
||||
("", 0),
|
||||
("Reimbursement: 20% and $300.", 2),
|
||||
("Reimbursement: 20% and $300. Another 10% reimbursement.", 3),
|
||||
],
|
||||
)
|
||||
def test_count_reimbursements_in_exhibit(self, exhibit_text, expected_count):
|
||||
result = count_reimbursements_in_exhibit(exhibit_text)
|
||||
assert result == expected_count
|
||||
|
||||
|
||||
class TestNormalizeStateToAbbreviation:
|
||||
@pytest.mark.parametrize(
|
||||
"state_value, expected",
|
||||
[
|
||||
# Special values should be preserved
|
||||
("N/A", "N/A"),
|
||||
("UNKNOWN", "UNKNOWN"),
|
||||
("", ""),
|
||||
(None, None),
|
||||
# Valid 2-letter abbreviations should be uppercased
|
||||
("ca", "CA"),
|
||||
("CA", "CA"),
|
||||
("ny", "NY"),
|
||||
("TX", "TX"),
|
||||
("fl", "FL"),
|
||||
# Full state names should be converted to abbreviations
|
||||
("california", "CA"),
|
||||
("California", "CA"),
|
||||
("CALIFORNIA", "CA"),
|
||||
("new york", "NY"),
|
||||
("New York", "NY"),
|
||||
("NEW YORK", "NY"),
|
||||
("texas", "TX"),
|
||||
("Texas", "TX"),
|
||||
("florida", "FL"),
|
||||
("Florida", "FL"),
|
||||
# Common variations should be handled
|
||||
("calif", "CA"),
|
||||
("cal", "CA"),
|
||||
("fla", "FL"),
|
||||
("ny", "NY"),
|
||||
("n.y.", "NY"),
|
||||
("tex", "TX"),
|
||||
("penn", "PA"),
|
||||
("mass", "MA"),
|
||||
("wash", "WA"),
|
||||
("washington state", "WA"),
|
||||
("d.c.", "DC"),
|
||||
("dc", "DC"),
|
||||
("washington dc", "DC"),
|
||||
("district of columbia", "DC"),
|
||||
# Edge cases with whitespace
|
||||
(" california ", "CA"),
|
||||
(" NY ", "NY"),
|
||||
(" texas ", "TX"), # tabs
|
||||
# Invalid/unrecognized values should be returned as-is
|
||||
("invalid_state", "invalid_state"),
|
||||
("123", "123"),
|
||||
("XY", "XY"), # Not a valid state abbreviation
|
||||
("some random text", "some random text"),
|
||||
# All 50 states + DC (spot check a few more)
|
||||
("alabama", "AL"),
|
||||
("alaska", "AK"),
|
||||
("arizona", "AZ"),
|
||||
("wyoming", "WY"),
|
||||
("hawaii", "HI"),
|
||||
("vermont", "VT"),
|
||||
],
|
||||
)
|
||||
def test_normalize_state_to_abbreviation(self, state_value, expected):
|
||||
from src.utils.string_utils import normalize_state_to_abbreviation
|
||||
|
||||
result = normalize_state_to_abbreviation(state_value)
|
||||
assert result == expected
|
||||
|
||||
def test_all_valid_abbreviations_preserved(self):
|
||||
"""Test that all valid 2-letter state abbreviations are preserved when uppercase"""
|
||||
from src.utils.string_utils import normalize_state_to_abbreviation
|
||||
|
||||
valid_abbreviations = [
|
||||
"AL",
|
||||
"AK",
|
||||
"AZ",
|
||||
"AR",
|
||||
"CA",
|
||||
"CO",
|
||||
"CT",
|
||||
"DE",
|
||||
"FL",
|
||||
"GA",
|
||||
"HI",
|
||||
"ID",
|
||||
"IL",
|
||||
"IN",
|
||||
"IA",
|
||||
"KS",
|
||||
"KY",
|
||||
"LA",
|
||||
"ME",
|
||||
"MD",
|
||||
"MA",
|
||||
"MI",
|
||||
"MN",
|
||||
"MS",
|
||||
"MO",
|
||||
"MT",
|
||||
"NE",
|
||||
"NV",
|
||||
"NH",
|
||||
"NJ",
|
||||
"NM",
|
||||
"NY",
|
||||
"NC",
|
||||
"ND",
|
||||
"OH",
|
||||
"OK",
|
||||
"OR",
|
||||
"PA",
|
||||
"RI",
|
||||
"SC",
|
||||
"SD",
|
||||
"TN",
|
||||
"TX",
|
||||
"UT",
|
||||
"VT",
|
||||
"VA",
|
||||
"WA",
|
||||
"WV",
|
||||
"WI",
|
||||
"WY",
|
||||
"DC",
|
||||
]
|
||||
|
||||
for abbrev in valid_abbreviations:
|
||||
result = normalize_state_to_abbreviation(abbrev)
|
||||
assert result == abbrev, f"Failed for abbreviation: {abbrev}"
|
||||
|
||||
# Also test lowercase versions
|
||||
result_lower = normalize_state_to_abbreviation(abbrev.lower())
|
||||
assert (
|
||||
result_lower == abbrev
|
||||
), f"Failed for lowercase abbreviation: {abbrev.lower()}"
|
||||
|
||||
def test_case_insensitivity(self):
|
||||
"""Test that the function handles various case combinations"""
|
||||
from src.utils.string_utils import normalize_state_to_abbreviation
|
||||
|
||||
test_cases = [
|
||||
("california", "CA"),
|
||||
("California", "CA"),
|
||||
("CALIFORNIA", "CA"),
|
||||
("CaLiFoRnIa", "CA"),
|
||||
("new york", "NY"),
|
||||
("New York", "NY"),
|
||||
("NEW YORK", "NY"),
|
||||
("NeW yOrK", "NY"),
|
||||
]
|
||||
|
||||
for input_state, expected in test_cases:
|
||||
result = normalize_state_to_abbreviation(input_state)
|
||||
assert result == expected, f"Failed for: {input_state}"
|
||||
|
||||
def test_edge_cases(self):
|
||||
"""Test edge cases and boundary conditions"""
|
||||
from src.utils.string_utils import normalize_state_to_abbreviation
|
||||
|
||||
# Test with just whitespace
|
||||
assert normalize_state_to_abbreviation(" ") == " "
|
||||
|
||||
# Test with mixed valid/invalid content
|
||||
assert normalize_state_to_abbreviation("california123") == "california123"
|
||||
|
||||
# Test very long strings
|
||||
long_string = "this is a very long string that is not a state"
|
||||
assert normalize_state_to_abbreviation(long_string) == long_string
|
||||
|
||||
# Test single characters
|
||||
assert normalize_state_to_abbreviation("C") == "C"
|
||||
assert normalize_state_to_abbreviation("A") == "A"
|
||||
|
||||
Reference in New Issue
Block a user