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

Merged in bugfix/dynamic-primary-issues (pull request #660)

Browse Source 
Bugfix/dynamic primary issues

* Update get prompt for direct valid value setting

* prep for merge

* Merge branch 'main' into bugfix/dynamic-primary-issues

* Remove test

* Update unit tests with real constants object

* Ready for review

* direct replace {valid_values}


Approved-by: Alex Galarce
This commit is contained in:
Katon Minhas
2025-08-12 17:57:09 +00:00
parent e56d629b1a
commit 5d49d4c44b
2 changed files with 81 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fieldExtraction/src/prompts/fieldset.py
+8 -20
View File
@@ -111,33 +111,21 @@ class Field:
)
resolved_prompt = self.prompt
# Replace all placeholders with self.valid_values
start_idx = resolved_prompt.find("{")
while start_idx != -1:
end_idx = resolved_prompt.find("}", start_idx)
if end_idx == -1:
break
# Extract the placeholder
placeholder = resolved_prompt[start_idx + 1 : end_idx]
# Replace the placeholder with the valid_values
if placeholder == "valid_values":
resolved_value = self.valid_values
if "{valid_values}" in resolved_prompt:
resolved_value = self.valid_values
if hasattr(constants, resolved_value):
try:
resolved_prompt = resolved_prompt.replace(
f"{{{placeholder}}}", str(constants.get_constant(resolved_value))
"{valid_values}", str(constants.get_constant(resolved_value))
)
except:
resolved_prompt = resolved_prompt.replace(
f"{{{placeholder}}}", str(resolved_value)
"{valid_values}", str(resolved_value)
)
else:
pass
# raise ValueError(f"unknown placeholder: {placeholder}")
start_idx = resolved_prompt.find("{", end_idx)
resolved_prompt = resolved_prompt.replace(
"{valid_values}", str(resolved_value)
)
# Return the updated Field object
return resolved_prompt
fieldExtraction/tests/test_fieldset.py
+73 -104
View File
@@ -2,19 +2,18 @@
import json
import os
import unittest
from unittest.mock import patch, mock_open, MagicMock
import pytest
from unittest.mock import patch, mock_open
from src.prompts.fieldset import Field, FieldSet
from constants.constants import Constants
class TestField(unittest.TestCase):
"""Tests for the Field class."""
class TestFieldWithRealConstants(unittest.TestCase):
"""Tests for the Field class using real Constants values."""
def setUp(self):
"""Set up test fixtures before each test method."""
# Create test field with a real VALID_BILL_TYPE placeholder
self.test_field_dict = {
"field_name": "TEST_FIELD",
"relationship": "one-to-one",
@@ -32,9 +31,8 @@ class TestField(unittest.TestCase):
}
self.field = Field(self.test_field_dict)
# Create a mock Constants object
self.constants = MagicMock(spec=Constants)
self.constants.get_constant.return_value = ["Type1", "Type2", "Type3"]
# Create a real Constants object - this will load actual mappings
self.constants = Constants()
def test_init(self):
"""Test Field initialization with all parameters."""
@@ -95,14 +93,6 @@ class TestField(unittest.TestCase):
self.assertEqual(field.relationship, "one-to-many")
self.assertEqual(field.field_type, "number")
@patch("builtins.open", new_callable=mock_open, read_data=json.dumps([
{"field_name": "FIELD1", "relationship": "one-to-one", "field_type": "string"}
]))
def test_load_from_file_not_found(self, mock_file):
"""Test that an error is raised when the field is not in the file."""
with self.assertRaises(ValueError):
Field.load_from_file("dummy_path.json", "NONEXISTENT_FIELD")
def test_from_values(self):
"""Test creating a Field using the from_values class method."""
field = Field.from_values(
@@ -138,10 +128,20 @@ class TestField(unittest.TestCase):
# Verify print was called the expected number of times
self.assertEqual(mock_print.call_count, 9)
def test_get_prompt_dict(self):
"""Test getting a prompt dictionary from a Field."""
def test_get_prompt_dict_with_real_constants(self):
"""Test getting a prompt dictionary from a Field with real Constants."""
prompt_dict = self.field.get_prompt_dict(self.constants)
self.assertEqual(prompt_dict["TEST_FIELD"], "What is the test field? Valid values: ['Type1', 'Type2', 'Type3']")
# Get the actual list of bill types from Constants
bill_types = self.constants.VALID_BILL_TYPE
# Verify the prompt contains the actual bill types from Constants
prompt = prompt_dict["TEST_FIELD"]
self.assertTrue(prompt.startswith("What is the test field? Valid values: "))
# Check if the output contains the real values
for bill_type in bill_types[:3]: # Check first few to avoid overly long test
self.assertIn(bill_type, prompt)
def test_get_prompt_no_valid_values(self):
"""Test getting a prompt with no valid_values to resolve."""
@@ -154,20 +154,17 @@ class TestField(unittest.TestCase):
prompt = field.get_prompt()
self.assertEqual(prompt, "What is the simple field?")
def test_get_prompt_missing_constants(self):
"""Test that an error is raised when Constants is needed but not provided."""
with self.assertRaises(AttributeError):
self.field.get_prompt()
def test_get_prompt_with_valid_values(self):
"""Test resolving valid values in a prompt."""
def test_get_prompt_with_real_constants(self):
"""Test resolving valid values in a prompt using real Constants."""
prompt = self.field.get_prompt(self.constants)
self.assertEqual(prompt, "What is the test field? Valid values: ['Type1', 'Type2', 'Type3']")
# Test when get_constant raises an exception
self.constants.get_constant.side_effect = Exception("Test exception")
prompt = self.field.get_prompt(self.constants)
self.assertEqual(prompt, "What is the test field? Valid values: VALID_BILL_TYPE")
# Check that the prompt includes the actual bill types
self.assertTrue(prompt.startswith("What is the test field? Valid values: "))
# Verify that the real constants values are used
bill_types = self.constants.VALID_BILL_TYPE
bill_types_str = str(bill_types)
self.assertIn(bill_types_str, prompt)
def test_update_valid_values(self):
"""Test updating valid values for a Field."""
@@ -175,11 +172,12 @@ class TestField(unittest.TestCase):
self.assertEqual(self.field.valid_values, "NEW_VALID_VALUES")
class TestFieldSet(unittest.TestCase):
"""Tests for the FieldSet class."""
class TestFieldSetWithRealConstants(unittest.TestCase):
"""Tests for the FieldSet class using real Constants values."""
def setUp(self):
"""Set up test fixtures before each test method."""
# Create fields with real Constants placeholders
self.test_fields = [
Field({
"field_name": "FIELD1",
@@ -198,14 +196,14 @@ class TestFieldSet(unittest.TestCase):
"field_name": "FIELD3",
"relationship": "one-to-one",
"field_type": "string",
"prompt": "What is field 3?"
"prompt": "What is field 3? Valid values: {valid_values}",
"valid_values": "VALID_LOBS"
})
]
self.fieldset = FieldSet(fields=self.test_fields)
# Create a mock Constants object
self.constants = MagicMock(spec=Constants)
self.constants.get_constant.return_value = ["Type1", "Type2", "Type3"]
# Create a real Constants object
self.constants = Constants()
@patch("builtins.open", new_callable=mock_open, read_data=json.dumps([
{"field_name": "FIELD1", "relationship": "one-to-one", "field_type": "string"},
@@ -220,21 +218,6 @@ class TestFieldSet(unittest.TestCase):
self.assertEqual(fieldset.fields[1].field_name, "FIELD2")
self.assertEqual(fieldset.fields[2].field_name, "FIELD3")
@patch("builtins.open", new_callable=mock_open, read_data=json.dumps([
{"field_name": "FIELD1", "relationship": "one-to-one", "field_type": "string", "keywords": ["key1"]},
{"field_name": "FIELD2", "relationship": "one-to-many", "field_type": "number"},
{"field_name": "FIELD3", "relationship": "one-to-one", "field_type": "boolean", "keywords": ["key3"]}
]))
def test_load_from_file_with_exists_check(self, mock_file):
"""Test loading fields from a file with a filter that checks for existence."""
# Filter for fields with keywords attribute (True means it should exist)
fieldset = FieldSet()
fieldset.load_from_file("dummy_path.json", keywords=True)
self.assertEqual(len(fieldset.fields), 2)
self.assertEqual(fieldset.fields[0].field_name, "FIELD1")
self.assertEqual(fieldset.fields[1].field_name, "FIELD3")
def test_add_field(self):
"""Test adding a field to a FieldSet."""
new_field = Field({
@@ -276,67 +259,61 @@ class TestFieldSet(unittest.TestCase):
self.assertEqual(len(self.fieldset.fields), 2)
self.assertEqual(self.fieldset.fields[0].field_name, "FIELD1")
self.assertEqual(self.fieldset.fields[1].field_name, "FIELD3")
# Test removing a field that doesn't exist (should not raise error)
self.fieldset.remove_field("NONEXISTENT_FIELD")
self.assertEqual(len(self.fieldset.fields), 2) # No change in length
def test_update_field_attr(self):
"""Test updating a field attribute."""
self.fieldset.update_field_attr("FIELD1", "prompt", "Updated prompt")
self.assertEqual(self.fieldset.get_field("FIELD1").prompt, "Updated prompt")
# Test updating an attribute for a field that doesn't exist (should not raise error)
self.fieldset.update_field_attr("NONEXISTENT_FIELD", "prompt", "New prompt")
def test_get_prompt_dict(self):
"""Test getting prompt dictionaries for all fields."""
def test_get_prompt_dict_with_real_constants(self):
"""Test getting prompt dictionaries with real Constants values."""
prompt_dict = self.fieldset.get_prompt_dict(self.constants)
self.assertEqual(len(prompt_dict), 3)
self.assertEqual(prompt_dict["FIELD1"], "What is field 1?")
self.assertEqual(prompt_dict["FIELD2"], "What is field 2? Valid values: ['Type1', 'Type2', 'Type3']")
self.assertEqual(prompt_dict["FIELD3"], "What is field 3?")
# Check that FIELD2 has the actual bill types
self.assertTrue(prompt_dict["FIELD2"].startswith("What is field 2? Valid values: "))
for bill_type in self.constants.VALID_BILL_TYPE[:3]: # Check first few
self.assertIn(bill_type, str(prompt_dict["FIELD2"]))
# Check that FIELD3 has the actual LOBs
self.assertTrue(prompt_dict["FIELD3"].startswith("What is field 3? Valid values: "))
for lob in self.constants.VALID_LOBS[:3]: # Check first few
self.assertIn(lob, str(prompt_dict["FIELD3"]))
# Test filtering by field_type
prompt_dict = self.fieldset.get_prompt_dict(self.constants, field_type="string")
self.assertEqual(len(prompt_dict), 2)
self.assertEqual(prompt_dict["FIELD1"], "What is field 1?")
self.assertEqual(prompt_dict["FIELD3"], "What is field 3?")
self.assertIn("FIELD3", prompt_dict)
# Test filtering by relationship
prompt_dict = self.fieldset.get_prompt_dict(self.constants, relationship="one-to-many")
self.assertEqual(len(prompt_dict), 1)
self.assertEqual(prompt_dict["FIELD2"], "What is field 2? Valid values: ['Type1', 'Type2', 'Type3']")
# Test filtering by both field_type and relationship
prompt_dict = self.fieldset.get_prompt_dict(
self.constants, field_type="string", relationship="one-to-one"
)
self.assertEqual(len(prompt_dict), 2)
self.assertEqual(prompt_dict["FIELD1"], "What is field 1?")
self.assertEqual(prompt_dict["FIELD3"], "What is field 3?")
self.assertIn("FIELD2", prompt_dict)
def test_print_prompt_dict(self):
"""Test printing all prompts as a string."""
def test_print_prompt_dict_with_real_constants(self):
"""Test printing all prompts as a string with real Constants."""
prompt_str = self.fieldset.print_prompt_dict(self.constants)
expected_str = (
"FIELD1 : What is field 1?\n"
"FIELD2 : What is field 2? Valid values: ['Type1', 'Type2', 'Type3']\n"
"FIELD3 : What is field 3?"
)
self.assertEqual(prompt_str, expected_str)
# Verify each field is in the output
self.assertIn("FIELD1 : What is field 1?", prompt_str)
self.assertIn("FIELD2 : What is field 2? Valid values: ", prompt_str)
self.assertIn("FIELD3 : What is field 3? Valid values: ", prompt_str)
# Verify real constants values are used
for bill_type in self.constants.VALID_BILL_TYPE[:2]: # Check first couple
self.assertIn(bill_type, prompt_str)
for lob in self.constants.VALID_LOBS[:2]: # Check first couple
self.assertIn(lob, prompt_str)
@patch("builtins.print")
def test_print(self, mock_print):
"""Test printing all fields."""
self.fieldset.print()
# Each field calls print 9 times and we have 3 fields
self.assertEqual(mock_print.call_count, 27)
# Test printing with no fields
empty_fieldset = FieldSet()
empty_fieldset.print()
mock_print.assert_called_with("No fields loaded.")
# Each field calls print multiple times
self.assertTrue(mock_print.call_count > 20)
def test_repr(self):
"""Test the string representation of a FieldSet."""
@@ -351,11 +328,6 @@ class TestFieldSet(unittest.TestCase):
"""Test printing field names."""
self.fieldset.print_fields()
mock_print.assert_called_with(['FIELD1', 'FIELD2', 'FIELD3'])
# Test printing with no fields
empty_fieldset = FieldSet()
empty_fieldset.print_fields()
mock_print.assert_called_with("No fields loaded.")
def test_filter(self):
"""Test filtering fields based on criteria."""
@@ -369,16 +341,6 @@ class TestFieldSet(unittest.TestCase):
filtered = self.fieldset.filter(field_type="number")
self.assertEqual(len(filtered.fields), 1)
self.assertEqual(filtered.fields[0].field_name, "FIELD2")
# Filter by multiple criteria
filtered = self.fieldset.filter(relationship="one-to-one", field_type="string")
self.assertEqual(len(filtered.fields), 2)
self.assertEqual(filtered.fields[0].field_name, "FIELD1")
self.assertEqual(filtered.fields[1].field_name, "FIELD3")
# Filter with no matches
filtered = self.fieldset.filter(field_type="boolean")
self.assertEqual(len(filtered.fields), 0)
def test_contains_fields(self):
"""Test checking if FieldSet contains fields."""
@@ -422,4 +384,11 @@ class TestFieldSet(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
unittest.main()