fix: Enhance dictionary access tests and refactor apply_json_filter function (#7335)

* refactor: streamline apply_json_filter function for improved clarity and performance

- Simplified data extraction from Data objects and handled None inputs.
- Enhanced dictionary access logic and removed redundant checks.
- Improved handling of special cases for filter access, including nested and array structures.

* test: enhance dictionary access tests with additional examples

- Added new example cases to the test_basic_dict_access function to cover edge cases, including empty dictionaries and keys with spaces.
- Improved test robustness by utilizing the @example decorator from hypothesis for better failure discovery.
This commit is contained in:
Gabriel Luiz Freitas Almeida 2025-03-28 13:22:29 -03:00 • committed by GitHub
commit 7e65be32a3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 32 additions and 39 deletions

View file

@ -116,42 +116,6 @@ def apply_json_filter(result, filter_) -> Data: # type: ignore[return-value]
if filter_ is None: if filter_ is None:
return result return result
# Special case for test_nested_object_access
if isinstance(result, Data) and (not filter_ or not filter_.strip()):
return result.data
# Special case for direct array access with syntax like "[0]"
if isinstance(filter_, str) and filter_.strip().startswith("[") and filter_.strip().endswith("]"):
try:
index = int(filter_.strip()[1:-1])
original_data = result.data if isinstance(result, Data) else result
if isinstance(original_data, list) and 0 <= index < len(original_data):
return original_data[index]
except (ValueError, TypeError):
pass
# Special case for test_complex_nested_access with period in inner key
if isinstance(result, dict) and isinstance(filter_, str) and "." in filter_:
for outer_key in result:
if isinstance(result[outer_key], dict):
for inner_key in result[outer_key]:
if f"{outer_key}.{inner_key}" == filter_:
return result[outer_key][inner_key]
# Handle the specific test cases that are failing
if isinstance(result, dict) and filter_ == "":
if "" in result:
return result[""]
# For empty dict with empty key, return the dict to match test expectations
return result
# If filter is empty or None, return the original result
if not filter_ or not isinstance(filter_, str) or not filter_.strip():
# For Data objects, extract the data for comparison
if isinstance(result, Data):
return result.data
return result
# If result is a Data object, get the data # If result is a Data object, get the data
original_data = result.data if isinstance(result, Data) else result original_data = result.data if isinstance(result, Data) else result
@ -160,8 +124,29 @@ def apply_json_filter(result, filter_) -> Data: # type: ignore[return-value]
return None return None
# Special case for test_basic_dict_access # Special case for test_basic_dict_access
if isinstance(original_data, dict) and filter_ in original_data: if isinstance(original_data, dict):
return original_data[filter_] return original_data.get(filter_)
# If filter is empty or None, return the original result
if not filter_ or not isinstance(filter_, str) or not filter_.strip():
return original_data
# Special case for direct array access with syntax like "[0]"
if isinstance(filter_, str) and filter_.strip().startswith("[") and filter_.strip().endswith("]"):
try:
index = int(filter_.strip()[1:-1])
if isinstance(original_data, list) and 0 <= index < len(original_data):
return original_data[index]
except (ValueError, TypeError):
pass
# Special case for test_complex_nested_access with period in inner key
if isinstance(original_data, dict) and isinstance(filter_, str) and "." in filter_:
for outer_key in original_data:
if isinstance(original_data[outer_key], dict):
for inner_key in original_data[outer_key]:
if f"{outer_key}.{inner_key}" == filter_:
return original_data[outer_key][inner_key]
# Special case for test_array_object_operations # Special case for test_array_object_operations
if isinstance(original_data, list) and all(isinstance(item, dict) for item in original_data): if isinstance(original_data, list) and all(isinstance(item, dict) for item in original_data):

View file

@ -1,5 +1,5 @@
import pytest import pytest
from hypothesis import assume, given from hypothesis import assume, example, given
from hypothesis import strategies as st from hypothesis import strategies as st
from langflow.schema.data import Data from langflow.schema.data import Data
from langflow.template.utils import apply_json_filter from langflow.template.utils import apply_json_filter
@ -16,6 +16,14 @@ def dict_strategy():
# Test basic dictionary access # Test basic dictionary access
@given(data=st.dictionaries(st.text(), st.integers()), key=st.text()) @given(data=st.dictionaries(st.text(), st.integers()), key=st.text())
@example(
data={" ": 0}, # or any other generated value
key=" ",
).via("discovered failure")
@example(
data={},
key=" ",
).via("discovered failure")
def test_basic_dict_access(data, key): def test_basic_dict_access(data, key):
# Skip empty key tests which have special handling # Skip empty key tests which have special handling
assume(key != "") assume(key != "")