Move json_handling to corelibs_json module
This commit is contained in:
Clemens Schwaighofer
@@ -12,6 +12,7 @@ dependencies = [
|
||||
"corelibs-encryption>=1.0.0",
|
||||
"corelibs-enum-base>=1.0.0",
|
||||
"corelibs-file>=1.0.0",
|
||||
"corelibs-json>=1.0.0",
|
||||
"corelibs-regex-checks>=1.0.0",
|
||||
"corelibs-stack-trace>=1.0.0",
|
||||
"corelibs-text-colors>=1.0.0",
|
||||
|
||||
@@ -2,11 +2,12 @@
|
||||
helper functions for jmespath interfaces
|
||||
"""
|
||||
|
||||
from warnings import deprecated
|
||||
from typing import Any
|
||||
import jmespath
|
||||
import jmespath.exceptions
|
||||
from corelibs_json.jmespath_support import jmespath_search as jmespath_search_ng
|
||||
|
||||
|
||||
@deprecated("Use corelibs_json.jmespath_support.jmespath_search instead")
|
||||
def jmespath_search(search_data: dict[Any, Any] | list[Any], search_params: str) -> Any:
|
||||
"""
|
||||
jmespath search wrapper
|
||||
@@ -22,18 +23,6 @@ def jmespath_search(search_data: dict[Any, Any] | list[Any], search_params: str)
|
||||
Returns:
|
||||
Any: dict/list/etc, None if nothing found
|
||||
"""
|
||||
try:
|
||||
search_result = jmespath.search(search_params, search_data)
|
||||
except jmespath.exceptions.LexerError as excp:
|
||||
raise ValueError(f"Compile failed: {search_params}: {excp}") from excp
|
||||
except jmespath.exceptions.ParseError as excp:
|
||||
raise ValueError(f"Parse failed: {search_params}: {excp}") from excp
|
||||
except jmespath.exceptions.JMESPathTypeError as excp:
|
||||
raise ValueError(f"Search failed with JMESPathTypeError: {search_params}: {excp}") from excp
|
||||
except TypeError as excp:
|
||||
raise ValueError(f"Type error for search_params: {excp}") from excp
|
||||
return search_result
|
||||
|
||||
# TODO: compile jmespath setup
|
||||
return jmespath_search_ng(search_data, search_params)
|
||||
|
||||
# __END__
|
||||
|
||||
@@ -2,35 +2,37 @@
|
||||
json encoder for datetime
|
||||
"""
|
||||
|
||||
from warnings import warn, deprecated
|
||||
from typing import Any
|
||||
from json import JSONEncoder, dumps
|
||||
from datetime import datetime, date
|
||||
import copy
|
||||
from jsonpath_ng import parse # pyright: ignore[reportMissingTypeStubs, reportUnknownVariableType]
|
||||
from corelibs_json.json_support import (
|
||||
default_isoformat as default_isoformat_ng,
|
||||
DateTimeEncoder as DateTimeEncoderCoreLibs,
|
||||
json_dumps as json_dumps_ng,
|
||||
modify_with_jsonpath as modify_with_jsonpath_ng,
|
||||
)
|
||||
|
||||
|
||||
# subclass JSONEncoder
|
||||
class DateTimeEncoder(JSONEncoder):
|
||||
class DateTimeEncoder(DateTimeEncoderCoreLibs):
|
||||
"""
|
||||
Override the default method
|
||||
dumps(..., cls=DateTimeEncoder, ...)
|
||||
"""
|
||||
def default(self, o: Any) -> str | None:
|
||||
if isinstance(o, (date, datetime)):
|
||||
return o.isoformat()
|
||||
return None
|
||||
|
||||
|
||||
warn("Use corelibs_json.json_support.DateTimeEncoder instead", DeprecationWarning, stacklevel=2)
|
||||
|
||||
|
||||
@deprecated("Use corelibs_json.json_support.default_isoformat instead")
|
||||
def default_isoformat(obj: Any) -> str | None:
|
||||
"""
|
||||
default override
|
||||
dumps(..., default=default, ...)
|
||||
"""
|
||||
if isinstance(obj, (date, datetime)):
|
||||
return obj.isoformat()
|
||||
return None
|
||||
return default_isoformat_ng(obj)
|
||||
|
||||
|
||||
@deprecated("Use corelibs_json.json_support.json_dumps instead")
|
||||
def json_dumps(data: Any):
|
||||
"""
|
||||
wrapper for json.dumps with sure dump without throwing Exceptions
|
||||
@@ -41,22 +43,15 @@ def json_dumps(data: Any):
|
||||
Returns:
|
||||
_type_ -- _description_
|
||||
"""
|
||||
return dumps(data, ensure_ascii=False, default=str)
|
||||
return json_dumps_ng(data)
|
||||
|
||||
|
||||
@deprecated("Use corelibs_json.json_support.modify_with_jsonpath instead")
|
||||
def modify_with_jsonpath(data: dict[Any, Any], path: str, new_value: Any):
|
||||
"""
|
||||
Modify dictionary using JSONPath (more powerful than JMESPath for modifications)
|
||||
"""
|
||||
result = copy.deepcopy(data)
|
||||
jsonpath_expr = parse(path) # pyright: ignore[reportUnknownVariableType]
|
||||
|
||||
# Find and update all matches
|
||||
matches = jsonpath_expr.find(result) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
|
||||
for match in matches: # pyright: ignore[reportUnknownVariableType]
|
||||
match.full_path.update(result, new_value) # pyright: ignore[reportUnknownMemberType]
|
||||
|
||||
return result
|
||||
return modify_with_jsonpath_ng(data, path, new_value)
|
||||
|
||||
# __END__
|
||||
|
||||
|
||||
@@ -1,3 +0,0 @@
|
||||
"""
|
||||
tests for json_handling module
|
||||
"""
|
||||
@@ -1,869 +0,0 @@
|
||||
"""
|
||||
tests for corelibs.json_handling.jmespath_helper
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
import pytest
|
||||
from corelibs.json_handling.jmespath_helper import jmespath_search
|
||||
|
||||
|
||||
# MARK: jmespath_search tests
|
||||
class TestJmespathSearch:
|
||||
"""Test cases for jmespath_search function"""
|
||||
|
||||
def test_simple_key_lookup(self):
|
||||
"""Test simple key lookup in dictionary"""
|
||||
data = {"name": "John", "age": 30}
|
||||
result = jmespath_search(data, "name")
|
||||
assert result == "John"
|
||||
|
||||
def test_nested_key_lookup(self):
|
||||
"""Test nested key lookup"""
|
||||
data = {
|
||||
"user": {
|
||||
"profile": {
|
||||
"name": "John",
|
||||
"age": 30
|
||||
}
|
||||
}
|
||||
}
|
||||
result = jmespath_search(data, "user.profile.name")
|
||||
assert result == "John"
|
||||
|
||||
def test_array_index_access(self):
|
||||
"""Test accessing array element by index"""
|
||||
data = {
|
||||
"items": [
|
||||
{"id": 1, "name": "Item 1"},
|
||||
{"id": 2, "name": "Item 2"},
|
||||
{"id": 3, "name": "Item 3"}
|
||||
]
|
||||
}
|
||||
result = jmespath_search(data, "items[1].name")
|
||||
assert result == "Item 2"
|
||||
|
||||
def test_array_slice(self):
|
||||
"""Test array slicing"""
|
||||
data = {"numbers": [1, 2, 3, 4, 5]}
|
||||
result = jmespath_search(data, "numbers[1:3]")
|
||||
assert result == [2, 3]
|
||||
|
||||
def test_wildcard_projection(self):
|
||||
"""Test wildcard projection on array"""
|
||||
data = {
|
||||
"users": [
|
||||
{"name": "Alice", "age": 25},
|
||||
{"name": "Bob", "age": 30},
|
||||
{"name": "Charlie", "age": 35}
|
||||
]
|
||||
}
|
||||
result = jmespath_search(data, "users[*].name")
|
||||
assert result == ["Alice", "Bob", "Charlie"]
|
||||
|
||||
def test_filter_expression(self):
|
||||
"""Test filter expression"""
|
||||
data = {
|
||||
"products": [
|
||||
{"name": "Product 1", "price": 100, "stock": 5},
|
||||
{"name": "Product 2", "price": 200, "stock": 0},
|
||||
{"name": "Product 3", "price": 150, "stock": 10}
|
||||
]
|
||||
}
|
||||
result = jmespath_search(data, "products[?stock > `0`].name")
|
||||
assert result == ["Product 1", "Product 3"]
|
||||
|
||||
def test_pipe_expression(self):
|
||||
"""Test pipe expression"""
|
||||
data = {
|
||||
"items": [
|
||||
{"name": "Item 1", "value": 10},
|
||||
{"name": "Item 2", "value": 20},
|
||||
{"name": "Item 3", "value": 30}
|
||||
]
|
||||
}
|
||||
result = jmespath_search(data, "items[*].value | [0]")
|
||||
assert result == 10
|
||||
|
||||
def test_multi_select_hash(self):
|
||||
"""Test multi-select hash"""
|
||||
data = {"name": "John", "age": 30, "city": "New York", "country": "USA"}
|
||||
result = jmespath_search(data, "{name: name, age: age}")
|
||||
assert result == {"name": "John", "age": 30}
|
||||
|
||||
def test_multi_select_list(self):
|
||||
"""Test multi-select list"""
|
||||
data = {"first": "John", "last": "Doe", "age": 30}
|
||||
result = jmespath_search(data, "[first, last]")
|
||||
assert result == ["John", "Doe"]
|
||||
|
||||
def test_flatten_projection(self):
|
||||
"""Test flatten projection"""
|
||||
data = {
|
||||
"groups": [
|
||||
{"items": [1, 2, 3]},
|
||||
{"items": [4, 5, 6]}
|
||||
]
|
||||
}
|
||||
result = jmespath_search(data, "groups[].items[]")
|
||||
assert result == [1, 2, 3, 4, 5, 6]
|
||||
|
||||
def test_function_length(self):
|
||||
"""Test length function"""
|
||||
data = {"items": [1, 2, 3, 4, 5]}
|
||||
result = jmespath_search(data, "length(items)")
|
||||
assert result == 5
|
||||
|
||||
def test_function_max(self):
|
||||
"""Test max function"""
|
||||
data = {"numbers": [10, 5, 20, 15]}
|
||||
result = jmespath_search(data, "max(numbers)")
|
||||
assert result == 20
|
||||
|
||||
def test_function_min(self):
|
||||
"""Test min function"""
|
||||
data = {"numbers": [10, 5, 20, 15]}
|
||||
result = jmespath_search(data, "min(numbers)")
|
||||
assert result == 5
|
||||
|
||||
def test_function_sort(self):
|
||||
"""Test sort function"""
|
||||
data = {"numbers": [3, 1, 4, 1, 5, 9, 2, 6]}
|
||||
result = jmespath_search(data, "sort(numbers)")
|
||||
assert result == [1, 1, 2, 3, 4, 5, 6, 9]
|
||||
|
||||
def test_function_sort_by(self):
|
||||
"""Test sort_by function"""
|
||||
data = {
|
||||
"people": [
|
||||
{"name": "Charlie", "age": 35},
|
||||
{"name": "Alice", "age": 25},
|
||||
{"name": "Bob", "age": 30}
|
||||
]
|
||||
}
|
||||
result = jmespath_search(data, "sort_by(people, &age)[*].name")
|
||||
assert result == ["Alice", "Bob", "Charlie"]
|
||||
|
||||
def test_function_join(self):
|
||||
"""Test join function"""
|
||||
data = {"names": ["Alice", "Bob", "Charlie"]}
|
||||
result = jmespath_search(data, "join(', ', names)")
|
||||
assert result == "Alice, Bob, Charlie"
|
||||
|
||||
def test_function_keys(self):
|
||||
"""Test keys function"""
|
||||
data = {"name": "John", "age": 30, "city": "New York"}
|
||||
result = jmespath_search(data, "keys(@)")
|
||||
assert sorted(result) == ["age", "city", "name"]
|
||||
|
||||
def test_function_values(self):
|
||||
"""Test values function"""
|
||||
data = {"a": 1, "b": 2, "c": 3}
|
||||
result = jmespath_search(data, "values(@)")
|
||||
assert sorted(result) == [1, 2, 3]
|
||||
|
||||
def test_function_type(self):
|
||||
"""Test type function"""
|
||||
data = {"string": "test", "number": 42, "array": [1, 2, 3]}
|
||||
result = jmespath_search(data, "type(string)")
|
||||
assert result == "string"
|
||||
|
||||
def test_function_contains(self):
|
||||
"""Test contains function"""
|
||||
data = {"items": [1, 2, 3, 4, 5]}
|
||||
result = jmespath_search(data, "contains(items, `3`)")
|
||||
assert result is True
|
||||
|
||||
def test_current_node_reference(self):
|
||||
"""Test current node @ reference"""
|
||||
data = [1, 2, 3, 4, 5]
|
||||
result = jmespath_search(data, "@")
|
||||
assert result == [1, 2, 3, 4, 5]
|
||||
|
||||
def test_not_null_expression(self):
|
||||
"""Test not_null expression"""
|
||||
data = {
|
||||
"items": [
|
||||
{"name": "Item 1", "description": "Desc 1"},
|
||||
{"name": "Item 2", "description": None},
|
||||
{"name": "Item 3"}
|
||||
]
|
||||
}
|
||||
result = jmespath_search(data, "items[*].description | [?@ != null]")
|
||||
assert result == ["Desc 1"]
|
||||
|
||||
def test_search_returns_none_for_missing_key(self):
|
||||
"""Test that searching for non-existent key returns None"""
|
||||
data = {"name": "John", "age": 30}
|
||||
result = jmespath_search(data, "nonexistent")
|
||||
assert result is None
|
||||
|
||||
def test_search_with_list_input(self):
|
||||
"""Test search with list as input"""
|
||||
data = [
|
||||
{"name": "Alice", "score": 85},
|
||||
{"name": "Bob", "score": 92},
|
||||
{"name": "Charlie", "score": 78}
|
||||
]
|
||||
result = jmespath_search(data, "[?score > `80`].name")
|
||||
assert result == ["Alice", "Bob"]
|
||||
|
||||
def test_deeply_nested_structure(self):
|
||||
"""Test searching deeply nested structure"""
|
||||
data = {
|
||||
"level1": {
|
||||
"level2": {
|
||||
"level3": {
|
||||
"level4": {
|
||||
"level5": {
|
||||
"value": "deep_value"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
result = jmespath_search(data, "level1.level2.level3.level4.level5.value")
|
||||
assert result == "deep_value"
|
||||
|
||||
def test_complex_filter_expression(self):
|
||||
"""Test complex filter with multiple conditions"""
|
||||
data = {
|
||||
"products": [
|
||||
{"name": "Product 1", "price": 100, "stock": 5, "category": "A"},
|
||||
{"name": "Product 2", "price": 200, "stock": 0, "category": "B"},
|
||||
{"name": "Product 3", "price": 150, "stock": 10, "category": "A"},
|
||||
{"name": "Product 4", "price": 120, "stock": 3, "category": "A"}
|
||||
]
|
||||
}
|
||||
result = jmespath_search(
|
||||
data,
|
||||
"products[?category == 'A' && stock > `0`].name"
|
||||
)
|
||||
assert result == ["Product 1", "Product 3", "Product 4"]
|
||||
|
||||
def test_recursive_descent(self):
|
||||
"""Test recursive descent operator"""
|
||||
data = {
|
||||
"store": {
|
||||
"book": [
|
||||
{"title": "Book 1", "price": 10},
|
||||
{"title": "Book 2", "price": 20}
|
||||
],
|
||||
"bicycle": {
|
||||
"price": 100
|
||||
}
|
||||
}
|
||||
}
|
||||
# Note: JMESPath doesn't have a true recursive descent like JSONPath's '..'
|
||||
# but we can test nested projections
|
||||
result = jmespath_search(data, "store.book[*].price")
|
||||
assert result == [10, 20]
|
||||
|
||||
def test_empty_dict_input(self):
|
||||
"""Test search on empty dictionary"""
|
||||
data: dict[Any, Any] = {}
|
||||
result = jmespath_search(data, "key")
|
||||
assert result is None
|
||||
|
||||
def test_empty_list_input(self):
|
||||
"""Test search on empty list"""
|
||||
data: list[Any] = []
|
||||
result = jmespath_search(data, "[0]")
|
||||
assert result is None
|
||||
|
||||
def test_unicode_keys_and_values(self):
|
||||
"""Test search with unicode keys and values"""
|
||||
data = {
|
||||
"日本語": "テスト",
|
||||
"emoji_🎉": "🚀",
|
||||
"nested": {
|
||||
"中文": "测试"
|
||||
}
|
||||
}
|
||||
# JMESPath requires quoted identifiers for unicode keys
|
||||
result = jmespath_search(data, '"日本語"')
|
||||
assert result == "テスト"
|
||||
|
||||
result2 = jmespath_search(data, 'nested."中文"')
|
||||
assert result2 == "测试"
|
||||
|
||||
def test_numeric_values(self):
|
||||
"""Test search with various numeric values"""
|
||||
data = {
|
||||
"int": 42,
|
||||
"float": 3.14,
|
||||
"negative": -10,
|
||||
"zero": 0,
|
||||
"scientific": 1e10
|
||||
}
|
||||
result = jmespath_search(data, "float")
|
||||
assert result == 3.14
|
||||
|
||||
def test_boolean_values(self):
|
||||
"""Test search with boolean values"""
|
||||
data = {
|
||||
"items": [
|
||||
{"name": "Item 1", "active": True},
|
||||
{"name": "Item 2", "active": False},
|
||||
{"name": "Item 3", "active": True}
|
||||
]
|
||||
}
|
||||
result = jmespath_search(data, "items[?active].name")
|
||||
assert result == ["Item 1", "Item 3"]
|
||||
|
||||
def test_null_values(self):
|
||||
"""Test search with null/None values"""
|
||||
data = {
|
||||
"name": "John",
|
||||
"middle_name": None,
|
||||
"last_name": "Doe"
|
||||
}
|
||||
result = jmespath_search(data, "middle_name")
|
||||
assert result is None
|
||||
|
||||
def test_mixed_types_in_array(self):
|
||||
"""Test search on array with mixed types"""
|
||||
data = {"mixed": [1, "two", 3.0, True, None, {"key": "value"}]}
|
||||
result = jmespath_search(data, "mixed[5].key")
|
||||
assert result == "value"
|
||||
|
||||
def test_expression_with_literals(self):
|
||||
"""Test expression with literal values"""
|
||||
data = {
|
||||
"items": [
|
||||
{"name": "Item 1", "price": 100},
|
||||
{"name": "Item 2", "price": 200}
|
||||
]
|
||||
}
|
||||
result = jmespath_search(data, "items[?price == `100`].name")
|
||||
assert result == ["Item 1"]
|
||||
|
||||
def test_comparison_operators(self):
|
||||
"""Test various comparison operators"""
|
||||
data = {
|
||||
"numbers": [
|
||||
{"value": 10},
|
||||
{"value": 20},
|
||||
{"value": 30},
|
||||
{"value": 40}
|
||||
]
|
||||
}
|
||||
result = jmespath_search(data, "numbers[?value >= `20` && value <= `30`].value")
|
||||
assert result == [20, 30]
|
||||
|
||||
def test_logical_operators(self):
|
||||
"""Test logical operators (and, or, not)"""
|
||||
data = {
|
||||
"items": [
|
||||
{"name": "A", "active": True, "stock": 5},
|
||||
{"name": "B", "active": False, "stock": 0},
|
||||
{"name": "C", "active": True, "stock": 0},
|
||||
{"name": "D", "active": False, "stock": 10}
|
||||
]
|
||||
}
|
||||
result = jmespath_search(data, "items[?active || stock > `0`].name")
|
||||
assert result == ["A", "C", "D"]
|
||||
|
||||
|
||||
# MARK: Error handling tests
|
||||
class TestJmespathSearchErrors:
|
||||
"""Test error handling in jmespath_search function"""
|
||||
|
||||
def test_lexer_error_invalid_syntax(self):
|
||||
"""Test LexerError is converted to ValueError for invalid syntax"""
|
||||
data = {"name": "John"}
|
||||
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
jmespath_search(data, "name[")
|
||||
|
||||
# This actually raises a ParseError, not LexerError
|
||||
assert "Parse failed" in str(exc_info.value)
|
||||
|
||||
def test_lexer_error_unclosed_bracket(self):
|
||||
"""Test LexerError for unclosed bracket"""
|
||||
data = {"items": [1, 2, 3]}
|
||||
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
jmespath_search(data, "items[0")
|
||||
|
||||
# This actually raises a ParseError, not LexerError
|
||||
assert "Parse failed" in str(exc_info.value)
|
||||
|
||||
def test_parse_error_invalid_expression(self):
|
||||
"""Test ParseError is converted to ValueError"""
|
||||
data = {"name": "John"}
|
||||
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
jmespath_search(data, "name..age")
|
||||
|
||||
assert "Parse failed" in str(exc_info.value)
|
||||
|
||||
def test_parse_error_invalid_filter(self):
|
||||
"""Test ParseError for invalid filter syntax"""
|
||||
data = {"items": [1, 2, 3]}
|
||||
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
jmespath_search(data, "items[?@")
|
||||
|
||||
assert "Parse failed" in str(exc_info.value)
|
||||
|
||||
def test_type_error_invalid_function_usage(self):
|
||||
"""Test JMESPathTypeError for invalid function usage"""
|
||||
data = {"name": "John", "age": 30}
|
||||
|
||||
# Trying to use length on a string (in some contexts this might cause type errors)
|
||||
# Note: This might not always raise an error depending on JMESPath version
|
||||
# Using a more reliable example: trying to use max on non-array
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
jmespath_search(data, "max(name)")
|
||||
|
||||
assert "Search failed with JMESPathTypeError" in str(exc_info.value)
|
||||
|
||||
def test_type_error_with_none_search_params(self):
|
||||
"""Test TypeError when search_params is None"""
|
||||
data = {"name": "John"}
|
||||
|
||||
# None or empty string raises EmptyExpressionError from jmespath
|
||||
with pytest.raises(Exception) as exc_info: # Catches any exception
|
||||
jmespath_search(data, None) # type: ignore
|
||||
|
||||
# The error message should indicate an empty expression issue
|
||||
assert "empty" in str(exc_info.value).lower() or "Type error" in str(exc_info.value)
|
||||
|
||||
def test_type_error_with_invalid_search_params_type(self):
|
||||
"""Test TypeError when search_params is not a string"""
|
||||
data = {"name": "John"}
|
||||
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
jmespath_search(data, 123) # type: ignore
|
||||
|
||||
assert "Type error for search_params" in str(exc_info.value)
|
||||
|
||||
def test_type_error_with_dict_search_params(self):
|
||||
"""Test TypeError when search_params is a dict"""
|
||||
data = {"name": "John"}
|
||||
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
jmespath_search(data, {"key": "value"}) # type: ignore
|
||||
|
||||
assert "Type error for search_params" in str(exc_info.value)
|
||||
|
||||
def test_error_message_includes_search_params(self):
|
||||
"""Test that error messages include the search parameters"""
|
||||
data = {"name": "John"}
|
||||
invalid_query = "name["
|
||||
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
jmespath_search(data, invalid_query)
|
||||
|
||||
error_message = str(exc_info.value)
|
||||
assert invalid_query in error_message
|
||||
# This raises ParseError, not LexerError
|
||||
assert "Parse failed" in error_message
|
||||
|
||||
def test_error_message_includes_exception_details(self):
|
||||
"""Test that error messages include original exception details"""
|
||||
data = {"items": [1, 2, 3]}
|
||||
invalid_query = "items[?"
|
||||
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
jmespath_search(data, invalid_query)
|
||||
|
||||
error_message = str(exc_info.value)
|
||||
# Should contain both the query and some indication of what went wrong
|
||||
assert invalid_query in error_message
|
||||
|
||||
|
||||
# MARK: Edge cases
|
||||
class TestJmespathSearchEdgeCases:
|
||||
"""Test edge cases for jmespath_search function"""
|
||||
|
||||
def test_very_large_array(self):
|
||||
"""Test searching large array"""
|
||||
data = {"items": [{"id": i, "value": i * 10} for i in range(1000)]}
|
||||
result = jmespath_search(data, "items[500].value")
|
||||
assert result == 5000
|
||||
|
||||
def test_very_deep_nesting(self):
|
||||
"""Test very deep nesting"""
|
||||
# Create 20-level deep nested structure
|
||||
data: dict[str, Any] = {"level0": {}}
|
||||
current = data["level0"]
|
||||
for i in range(1, 20):
|
||||
current[f"level{i}"] = {}
|
||||
current = current[f"level{i}"]
|
||||
current["value"] = "deep"
|
||||
|
||||
# Build the search path
|
||||
path = ".".join([f"level{i}" for i in range(20)]) + ".value"
|
||||
result = jmespath_search(data, path)
|
||||
assert result == "deep"
|
||||
|
||||
def test_special_characters_in_keys(self):
|
||||
"""Test keys with special characters (requires escaping)"""
|
||||
data = {"my-key": "value", "my.key": "value2"}
|
||||
|
||||
# JMESPath requires quoting for keys with special characters
|
||||
result = jmespath_search(data, '"my-key"')
|
||||
assert result == "value"
|
||||
|
||||
result2 = jmespath_search(data, '"my.key"')
|
||||
assert result2 == "value2"
|
||||
|
||||
def test_numeric_string_keys(self):
|
||||
"""Test keys that look like numbers"""
|
||||
data = {"123": "numeric_key", "456": "another"}
|
||||
result = jmespath_search(data, '"123"')
|
||||
assert result == "numeric_key"
|
||||
|
||||
def test_empty_string_key(self):
|
||||
"""Test empty string as key"""
|
||||
data = {"": "empty_key_value", "normal": "normal_value"}
|
||||
result = jmespath_search(data, '""')
|
||||
assert result == "empty_key_value"
|
||||
|
||||
def test_whitespace_in_keys(self):
|
||||
"""Test keys with whitespace"""
|
||||
data = {"my key": "value", " trimmed ": "value2"}
|
||||
result = jmespath_search(data, '"my key"')
|
||||
assert result == "value"
|
||||
|
||||
def test_array_with_negative_index(self):
|
||||
"""Test negative array indexing"""
|
||||
data = {"items": [1, 2, 3, 4, 5]}
|
||||
# JMESPath actually supports negative indexing
|
||||
result = jmespath_search(data, "items[-1]")
|
||||
assert result == 5
|
||||
|
||||
def test_out_of_bounds_array_index(self):
|
||||
"""Test out of bounds array access"""
|
||||
data = {"items": [1, 2, 3]}
|
||||
result = jmespath_search(data, "items[10]")
|
||||
assert result is None
|
||||
|
||||
def test_chaining_multiple_operations(self):
|
||||
"""Test chaining multiple JMESPath operations"""
|
||||
data: dict[str, Any] = {
|
||||
"users": [
|
||||
{"name": "Alice", "posts": [{"id": 1}, {"id": 2}]},
|
||||
{"name": "Bob", "posts": [{"id": 3}, {"id": 4}, {"id": 5}]},
|
||||
{"name": "Charlie", "posts": []}
|
||||
]
|
||||
}
|
||||
result = jmespath_search(data, "users[*].posts[].id")
|
||||
assert result == [1, 2, 3, 4, 5]
|
||||
|
||||
def test_projection_on_non_array(self):
|
||||
"""Test projection on non-array (should handle gracefully)"""
|
||||
data = {"value": "not_an_array"}
|
||||
result = jmespath_search(data, "value[*]")
|
||||
assert result is None
|
||||
|
||||
def test_filter_on_non_array(self):
|
||||
"""Test filter on non-array"""
|
||||
data = {"value": "string"}
|
||||
result = jmespath_search(data, "value[?@ == 'x']")
|
||||
assert result is None
|
||||
|
||||
def test_combining_filters_and_projections(self):
|
||||
"""Test combining filters with projections"""
|
||||
data = {
|
||||
"products": [
|
||||
{
|
||||
"name": "Product 1",
|
||||
"variants": [
|
||||
{"color": "red", "stock": 5},
|
||||
{"color": "blue", "stock": 0}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "Product 2",
|
||||
"variants": [
|
||||
{"color": "green", "stock": 10},
|
||||
{"color": "yellow", "stock": 3}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
result = jmespath_search(
|
||||
data,
|
||||
"products[*].variants[?stock > `0`].color"
|
||||
)
|
||||
assert result == [["red"], ["green", "yellow"]]
|
||||
|
||||
def test_search_with_root_array(self):
|
||||
"""Test search when root is an array"""
|
||||
data = [
|
||||
{"name": "Alice", "age": 25},
|
||||
{"name": "Bob", "age": 30}
|
||||
]
|
||||
result = jmespath_search(data, "[0].name")
|
||||
assert result == "Alice"
|
||||
|
||||
def test_search_with_primitive_root(self):
|
||||
"""Test search when root is a primitive value"""
|
||||
# When root is primitive, only @ should work
|
||||
data_str = "simple_string"
|
||||
result = jmespath_search(data_str, "@") # type: ignore
|
||||
assert result == "simple_string"
|
||||
|
||||
def test_function_with_empty_array(self):
|
||||
"""Test functions on empty arrays"""
|
||||
data: dict[str, list[Any]] = {"items": []}
|
||||
result = jmespath_search(data, "length(items)")
|
||||
assert result == 0
|
||||
|
||||
def test_nested_multi_select(self):
|
||||
"""Test nested multi-select operations"""
|
||||
data = {
|
||||
"person": {
|
||||
"name": "John",
|
||||
"age": 30,
|
||||
"address": {
|
||||
"city": "New York",
|
||||
"country": "USA"
|
||||
}
|
||||
}
|
||||
}
|
||||
result = jmespath_search(
|
||||
data,
|
||||
"person.{name: name, city: address.city}"
|
||||
)
|
||||
assert result == {"name": "John", "city": "New York"}
|
||||
|
||||
|
||||
# MARK: Integration tests
|
||||
class TestJmespathSearchIntegration:
|
||||
"""Integration tests for complex real-world scenarios"""
|
||||
|
||||
def test_api_response_parsing(self):
|
||||
"""Test parsing typical API response structure"""
|
||||
api_response = {
|
||||
"status": "success",
|
||||
"data": {
|
||||
"users": [
|
||||
{
|
||||
"id": 1,
|
||||
"name": "Alice",
|
||||
"email": "alice@example.com",
|
||||
"active": True,
|
||||
"metadata": {
|
||||
"created_at": "2025-01-01",
|
||||
"last_login": "2025-10-23"
|
||||
}
|
||||
},
|
||||
{
|
||||
"id": 2,
|
||||
"name": "Bob",
|
||||
"email": "bob@example.com",
|
||||
"active": False,
|
||||
"metadata": {
|
||||
"created_at": "2025-02-01",
|
||||
"last_login": "2025-05-15"
|
||||
}
|
||||
},
|
||||
{
|
||||
"id": 3,
|
||||
"name": "Charlie",
|
||||
"email": "charlie@example.com",
|
||||
"active": True,
|
||||
"metadata": {
|
||||
"created_at": "2025-03-01",
|
||||
"last_login": "2025-10-20"
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
"metadata": {
|
||||
"total": 3,
|
||||
"page": 1
|
||||
}
|
||||
}
|
||||
|
||||
# Get all active user emails
|
||||
result = jmespath_search(api_response, "data.users[?active].email")
|
||||
assert result == ["alice@example.com", "charlie@example.com"]
|
||||
|
||||
# Get user names and creation dates
|
||||
result2 = jmespath_search(
|
||||
api_response,
|
||||
"data.users[*].{name: name, created: metadata.created_at}"
|
||||
)
|
||||
assert len(result2) == 3
|
||||
assert result2[0]["name"] == "Alice"
|
||||
assert result2[0]["created"] == "2025-01-01"
|
||||
|
||||
def test_config_file_parsing(self):
|
||||
"""Test parsing configuration-like structure"""
|
||||
config = {
|
||||
"version": "1.0",
|
||||
"environments": {
|
||||
"development": {
|
||||
"database": {
|
||||
"host": "localhost",
|
||||
"port": 5432,
|
||||
"name": "dev_db"
|
||||
},
|
||||
"cache": {
|
||||
"enabled": True,
|
||||
"ttl": 300
|
||||
}
|
||||
},
|
||||
"production": {
|
||||
"database": {
|
||||
"host": "prod.example.com",
|
||||
"port": 5432,
|
||||
"name": "prod_db"
|
||||
},
|
||||
"cache": {
|
||||
"enabled": True,
|
||||
"ttl": 3600
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
# Get production database host
|
||||
result = jmespath_search(config, "environments.production.database.host")
|
||||
assert result == "prod.example.com"
|
||||
|
||||
# Get all database names using values() - object wildcard returns an object
|
||||
# Need to convert to list for sorting
|
||||
result2 = jmespath_search(config, "values(environments)[*].database.name")
|
||||
assert result2 is not None
|
||||
assert sorted(result2) == ["dev_db", "prod_db"]
|
||||
|
||||
def test_nested_filtering_and_transformation(self):
|
||||
"""Test complex nested filtering and transformation"""
|
||||
data = {
|
||||
"departments": [
|
||||
{
|
||||
"name": "Engineering",
|
||||
"employees": [
|
||||
{"name": "Alice", "salary": 100000, "level": "Senior"},
|
||||
{"name": "Bob", "salary": 80000, "level": "Mid"},
|
||||
{"name": "Charlie", "salary": 120000, "level": "Senior"}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "Marketing",
|
||||
"employees": [
|
||||
{"name": "Dave", "salary": 70000, "level": "Junior"},
|
||||
{"name": "Eve", "salary": 90000, "level": "Mid"}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
# Get all senior employees with salary > 100k
|
||||
result = jmespath_search(
|
||||
data,
|
||||
"departments[*].employees[?level == 'Senior' && salary > `100000`].name"
|
||||
)
|
||||
# Note: 100000 is not > 100000, so Alice is excluded
|
||||
assert result == [["Charlie"], []]
|
||||
|
||||
# Get flattened list (using >= instead and flatten operator)
|
||||
result2 = jmespath_search(
|
||||
data,
|
||||
"departments[].employees[?level == 'Senior' && salary >= `100000`].name | []"
|
||||
)
|
||||
assert sorted(result2) == ["Alice", "Charlie"]
|
||||
|
||||
def test_working_with_timestamps(self):
|
||||
"""Test searching and filtering timestamp-like data"""
|
||||
data = {
|
||||
"events": [
|
||||
{"name": "Event 1", "timestamp": "2025-10-20T10:00:00"},
|
||||
{"name": "Event 2", "timestamp": "2025-10-21T15:30:00"},
|
||||
{"name": "Event 3", "timestamp": "2025-10-23T08:45:00"},
|
||||
{"name": "Event 4", "timestamp": "2025-10-24T12:00:00"}
|
||||
]
|
||||
}
|
||||
|
||||
# Get events after a certain date (string comparison)
|
||||
result = jmespath_search(
|
||||
data,
|
||||
"events[?timestamp > '2025-10-22'].name"
|
||||
)
|
||||
assert result == ["Event 3", "Event 4"]
|
||||
|
||||
def test_aggregation_operations(self):
|
||||
"""Test aggregation-like operations"""
|
||||
data = {
|
||||
"sales": [
|
||||
{"product": "A", "quantity": 10, "price": 100},
|
||||
{"product": "B", "quantity": 5, "price": 200},
|
||||
{"product": "C", "quantity": 8, "price": 150}
|
||||
]
|
||||
}
|
||||
|
||||
# Get all quantities
|
||||
quantities = jmespath_search(data, "sales[*].quantity")
|
||||
assert quantities == [10, 5, 8]
|
||||
|
||||
# Get max quantity
|
||||
max_quantity = jmespath_search(data, "max(sales[*].quantity)")
|
||||
assert max_quantity == 10
|
||||
|
||||
# Get min price
|
||||
min_price = jmespath_search(data, "min(sales[*].price)")
|
||||
assert min_price == 100
|
||||
|
||||
# Get sorted products by price
|
||||
sorted_products = jmespath_search(
|
||||
data,
|
||||
"sort_by(sales, &price)[*].product"
|
||||
)
|
||||
assert sorted_products == ["A", "C", "B"]
|
||||
|
||||
def test_data_transformation_pipeline(self):
|
||||
"""Test data transformation pipeline"""
|
||||
raw_data = {
|
||||
"response": {
|
||||
"items": [
|
||||
{
|
||||
"id": "item-1",
|
||||
"attributes": {
|
||||
"name": "Product A",
|
||||
"specs": {"weight": 100, "color": "red"}
|
||||
},
|
||||
"available": True
|
||||
},
|
||||
{
|
||||
"id": "item-2",
|
||||
"attributes": {
|
||||
"name": "Product B",
|
||||
"specs": {"weight": 200, "color": "blue"}
|
||||
},
|
||||
"available": False
|
||||
},
|
||||
{
|
||||
"id": "item-3",
|
||||
"attributes": {
|
||||
"name": "Product C",
|
||||
"specs": {"weight": 150, "color": "red"}
|
||||
},
|
||||
"available": True
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
# Get available red products
|
||||
result = jmespath_search(
|
||||
raw_data,
|
||||
"response.items[?available && attributes.specs.color == 'red'].attributes.name"
|
||||
)
|
||||
assert result == ["Product A", "Product C"]
|
||||
|
||||
# Transform to simplified structure
|
||||
result2 = jmespath_search(
|
||||
raw_data,
|
||||
"response.items[*].{id: id, name: attributes.name, weight: attributes.specs.weight}"
|
||||
)
|
||||
assert len(result2) == 3
|
||||
assert result2[0] == {"id": "item-1", "name": "Product A", "weight": 100}
|
||||
|
||||
|
||||
# __END__
|
||||
@@ -1,698 +0,0 @@
|
||||
"""
|
||||
tests for corelibs.json_handling.json_helper
|
||||
"""
|
||||
|
||||
import json
|
||||
from datetime import datetime, date
|
||||
from typing import Any
|
||||
from corelibs.json_handling.json_helper import (
|
||||
DateTimeEncoder,
|
||||
default_isoformat,
|
||||
json_dumps,
|
||||
modify_with_jsonpath
|
||||
)
|
||||
|
||||
|
||||
# MARK: DateTimeEncoder tests
|
||||
class TestDateTimeEncoder:
|
||||
"""Test cases for DateTimeEncoder class"""
|
||||
|
||||
def test_datetime_encoding(self):
|
||||
"""Test encoding datetime objects"""
|
||||
dt = datetime(2025, 10, 23, 15, 30, 45, 123456)
|
||||
data = {"timestamp": dt}
|
||||
|
||||
result = json.dumps(data, cls=DateTimeEncoder)
|
||||
decoded = json.loads(result)
|
||||
|
||||
assert decoded["timestamp"] == "2025-10-23T15:30:45.123456"
|
||||
|
||||
def test_date_encoding(self):
|
||||
"""Test encoding date objects"""
|
||||
d = date(2025, 10, 23)
|
||||
data = {"date": d}
|
||||
|
||||
result = json.dumps(data, cls=DateTimeEncoder)
|
||||
decoded = json.loads(result)
|
||||
|
||||
assert decoded["date"] == "2025-10-23"
|
||||
|
||||
def test_mixed_datetime_date_encoding(self):
|
||||
"""Test encoding mixed datetime and date objects"""
|
||||
dt = datetime(2025, 10, 23, 15, 30, 45)
|
||||
d = date(2025, 10, 23)
|
||||
data = {
|
||||
"timestamp": dt,
|
||||
"date": d,
|
||||
"name": "test"
|
||||
}
|
||||
|
||||
result = json.dumps(data, cls=DateTimeEncoder)
|
||||
decoded = json.loads(result)
|
||||
|
||||
assert decoded["timestamp"] == "2025-10-23T15:30:45"
|
||||
assert decoded["date"] == "2025-10-23"
|
||||
assert decoded["name"] == "test"
|
||||
|
||||
def test_nested_datetime_encoding(self):
|
||||
"""Test encoding nested structures with datetime objects"""
|
||||
data = {
|
||||
"event": {
|
||||
"name": "Meeting",
|
||||
"start": datetime(2025, 10, 23, 10, 0, 0),
|
||||
"end": datetime(2025, 10, 23, 11, 0, 0),
|
||||
"participants": [
|
||||
{"name": "Alice", "joined": datetime(2025, 10, 23, 10, 5, 0)},
|
||||
{"name": "Bob", "joined": datetime(2025, 10, 23, 10, 10, 0)}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
result = json.dumps(data, cls=DateTimeEncoder)
|
||||
decoded = json.loads(result)
|
||||
|
||||
assert decoded["event"]["start"] == "2025-10-23T10:00:00"
|
||||
assert decoded["event"]["end"] == "2025-10-23T11:00:00"
|
||||
assert decoded["event"]["participants"][0]["joined"] == "2025-10-23T10:05:00"
|
||||
assert decoded["event"]["participants"][1]["joined"] == "2025-10-23T10:10:00"
|
||||
|
||||
def test_list_of_datetimes(self):
|
||||
"""Test encoding list of datetime objects"""
|
||||
data = {
|
||||
"timestamps": [
|
||||
datetime(2025, 10, 23, 10, 0, 0),
|
||||
datetime(2025, 10, 23, 11, 0, 0),
|
||||
datetime(2025, 10, 23, 12, 0, 0)
|
||||
]
|
||||
}
|
||||
|
||||
result = json.dumps(data, cls=DateTimeEncoder)
|
||||
decoded = json.loads(result)
|
||||
|
||||
assert decoded["timestamps"][0] == "2025-10-23T10:00:00"
|
||||
assert decoded["timestamps"][1] == "2025-10-23T11:00:00"
|
||||
assert decoded["timestamps"][2] == "2025-10-23T12:00:00"
|
||||
|
||||
def test_encoder_with_normal_types(self):
|
||||
"""Test that encoder works with standard JSON types"""
|
||||
data = {
|
||||
"string": "test",
|
||||
"number": 42,
|
||||
"float": 3.14,
|
||||
"boolean": True,
|
||||
"null": None,
|
||||
"list": [1, 2, 3],
|
||||
"dict": {"key": "value"}
|
||||
}
|
||||
|
||||
result = json.dumps(data, cls=DateTimeEncoder)
|
||||
decoded = json.loads(result)
|
||||
|
||||
assert decoded == data
|
||||
|
||||
def test_encoder_returns_none_for_unsupported_types(self):
|
||||
"""Test that encoder default method returns None for unsupported types"""
|
||||
encoder = DateTimeEncoder()
|
||||
|
||||
# The default method should return None for non-date/datetime objects
|
||||
result = encoder.default("string")
|
||||
assert result is None
|
||||
|
||||
result = encoder.default(42)
|
||||
assert result is None
|
||||
|
||||
result = encoder.default([1, 2, 3])
|
||||
assert result is None
|
||||
|
||||
|
||||
# MARK: default function tests
|
||||
class TestDefaultFunction:
|
||||
"""Test cases for the default function"""
|
||||
|
||||
def test_default_datetime(self):
|
||||
"""Test default function with datetime"""
|
||||
dt = datetime(2025, 10, 23, 15, 30, 45)
|
||||
result = default_isoformat(dt)
|
||||
assert result == "2025-10-23T15:30:45"
|
||||
|
||||
def test_default_date(self):
|
||||
"""Test default function with date"""
|
||||
d = date(2025, 10, 23)
|
||||
result = default_isoformat(d)
|
||||
assert result == "2025-10-23"
|
||||
|
||||
def test_default_with_microseconds(self):
|
||||
"""Test default function with datetime including microseconds"""
|
||||
dt = datetime(2025, 10, 23, 15, 30, 45, 123456)
|
||||
result = default_isoformat(dt)
|
||||
assert result == "2025-10-23T15:30:45.123456"
|
||||
|
||||
def test_default_returns_none_for_other_types(self):
|
||||
"""Test that default returns None for non-date/datetime objects"""
|
||||
assert default_isoformat("string") is None
|
||||
assert default_isoformat(42) is None
|
||||
assert default_isoformat(3.14) is None
|
||||
assert default_isoformat(True) is None
|
||||
assert default_isoformat(None) is None
|
||||
assert default_isoformat([1, 2, 3]) is None
|
||||
assert default_isoformat({"key": "value"}) is None
|
||||
|
||||
def test_default_as_json_default_parameter(self):
|
||||
"""Test using default function as default parameter in json.dumps"""
|
||||
data = {
|
||||
"timestamp": datetime(2025, 10, 23, 15, 30, 45),
|
||||
"date": date(2025, 10, 23),
|
||||
"name": "test"
|
||||
}
|
||||
|
||||
result = json.dumps(data, default=default_isoformat)
|
||||
decoded = json.loads(result)
|
||||
|
||||
assert decoded["timestamp"] == "2025-10-23T15:30:45"
|
||||
assert decoded["date"] == "2025-10-23"
|
||||
assert decoded["name"] == "test"
|
||||
|
||||
|
||||
# MARK: json_dumps tests
|
||||
class TestJsonDumps:
|
||||
"""Test cases for json_dumps function"""
|
||||
|
||||
def test_basic_dict(self):
|
||||
"""Test json_dumps with basic dictionary"""
|
||||
data = {"name": "test", "value": 42}
|
||||
result = json_dumps(data)
|
||||
decoded = json.loads(result)
|
||||
assert decoded == data
|
||||
|
||||
def test_unicode_characters(self):
|
||||
"""Test json_dumps preserves unicode characters (ensure_ascii=False)"""
|
||||
data = {"name": "テスト", "emoji": "🎉", "chinese": "测试"}
|
||||
result = json_dumps(data)
|
||||
|
||||
# ensure_ascii=False means unicode characters should be preserved
|
||||
assert "テスト" in result
|
||||
assert "🎉" in result
|
||||
assert "测试" in result
|
||||
|
||||
decoded = json.loads(result)
|
||||
assert decoded == data
|
||||
|
||||
def test_datetime_objects_as_string(self):
|
||||
"""Test json_dumps converts datetime to string (default=str)"""
|
||||
dt = datetime(2025, 10, 23, 15, 30, 45)
|
||||
data = {"timestamp": dt}
|
||||
|
||||
result = json_dumps(data)
|
||||
decoded = json.loads(result)
|
||||
|
||||
# default=str will convert datetime to its string representation
|
||||
assert isinstance(decoded["timestamp"], str)
|
||||
assert "2025-10-23" in decoded["timestamp"]
|
||||
|
||||
def test_date_objects_as_string(self):
|
||||
"""Test json_dumps converts date to string"""
|
||||
d = date(2025, 10, 23)
|
||||
data = {"date": d}
|
||||
|
||||
result = json_dumps(data)
|
||||
decoded = json.loads(result)
|
||||
|
||||
assert isinstance(decoded["date"], str)
|
||||
assert "2025-10-23" in decoded["date"]
|
||||
|
||||
def test_complex_nested_structure(self):
|
||||
"""Test json_dumps with complex nested structures"""
|
||||
data = {
|
||||
"user": {
|
||||
"name": "John",
|
||||
"age": 30,
|
||||
"active": True,
|
||||
"balance": 100.50,
|
||||
"tags": ["admin", "user"],
|
||||
"metadata": {
|
||||
"created": datetime(2025, 1, 1, 0, 0, 0),
|
||||
"updated": date(2025, 10, 23)
|
||||
}
|
||||
},
|
||||
"items": [
|
||||
{"id": 1, "name": "Item 1"},
|
||||
{"id": 2, "name": "Item 2"}
|
||||
]
|
||||
}
|
||||
|
||||
result = json_dumps(data)
|
||||
decoded = json.loads(result)
|
||||
|
||||
assert decoded["user"]["name"] == "John"
|
||||
assert decoded["user"]["age"] == 30
|
||||
assert decoded["user"]["active"] is True
|
||||
assert decoded["user"]["balance"] == 100.50
|
||||
assert decoded["user"]["tags"] == ["admin", "user"]
|
||||
assert decoded["items"][0]["id"] == 1
|
||||
|
||||
def test_empty_dict(self):
|
||||
"""Test json_dumps with empty dictionary"""
|
||||
data: dict[str, Any] = {}
|
||||
result = json_dumps(data)
|
||||
assert result == "{}"
|
||||
|
||||
def test_empty_list(self):
|
||||
"""Test json_dumps with empty list"""
|
||||
data: list[Any] = []
|
||||
result = json_dumps(data)
|
||||
assert result == "[]"
|
||||
|
||||
def test_list_data(self):
|
||||
"""Test json_dumps with list as root element"""
|
||||
data = [1, 2, 3, "test", True, None]
|
||||
result = json_dumps(data)
|
||||
decoded = json.loads(result)
|
||||
assert decoded == data
|
||||
|
||||
def test_none_value(self):
|
||||
"""Test json_dumps with None"""
|
||||
data = None
|
||||
result = json_dumps(data)
|
||||
assert result == "null"
|
||||
|
||||
def test_boolean_values(self):
|
||||
"""Test json_dumps with boolean values"""
|
||||
data = {"true_val": True, "false_val": False}
|
||||
result = json_dumps(data)
|
||||
decoded = json.loads(result)
|
||||
assert decoded["true_val"] is True
|
||||
assert decoded["false_val"] is False
|
||||
|
||||
def test_numeric_values(self):
|
||||
"""Test json_dumps with various numeric values"""
|
||||
data = {
|
||||
"int": 42,
|
||||
"float": 3.14,
|
||||
"negative": -10,
|
||||
"zero": 0,
|
||||
"scientific": 1e10
|
||||
}
|
||||
result = json_dumps(data)
|
||||
decoded = json.loads(result)
|
||||
assert decoded == data
|
||||
|
||||
def test_custom_object_conversion(self):
|
||||
"""Test json_dumps with custom objects (converted via str)"""
|
||||
class CustomObject:
|
||||
"""test class"""
|
||||
def __str__(self):
|
||||
return "custom_value"
|
||||
|
||||
data = {"custom": CustomObject()}
|
||||
result = json_dumps(data)
|
||||
decoded = json.loads(result)
|
||||
assert decoded["custom"] == "custom_value"
|
||||
|
||||
def test_special_float_values(self):
|
||||
"""Test json_dumps handles special float values"""
|
||||
data = {
|
||||
"infinity": float('inf'),
|
||||
"neg_infinity": float('-inf'),
|
||||
"nan": float('nan')
|
||||
}
|
||||
result = json_dumps(data)
|
||||
# These should be converted to strings via default=str
|
||||
assert "Infinity" in result or "inf" in result.lower()
|
||||
|
||||
|
||||
# MARK: modify_with_jsonpath tests
|
||||
class TestModifyWithJsonpath:
|
||||
"""Test cases for modify_with_jsonpath function"""
|
||||
|
||||
def test_simple_path_modification(self):
|
||||
"""Test modifying a simple path"""
|
||||
data = {"name": "old_name", "age": 30}
|
||||
result = modify_with_jsonpath(data, "$.name", "new_name")
|
||||
|
||||
assert result["name"] == "new_name"
|
||||
assert result["age"] == 30
|
||||
# Original data should not be modified
|
||||
assert data["name"] == "old_name"
|
||||
|
||||
def test_nested_path_modification(self):
|
||||
"""Test modifying nested path"""
|
||||
data = {
|
||||
"user": {
|
||||
"profile": {
|
||||
"name": "John",
|
||||
"age": 30
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
result = modify_with_jsonpath(data, "$.user.profile.name", "Jane")
|
||||
|
||||
assert result["user"]["profile"]["name"] == "Jane"
|
||||
assert result["user"]["profile"]["age"] == 30
|
||||
# Original should be unchanged
|
||||
assert data["user"]["profile"]["name"] == "John"
|
||||
|
||||
def test_array_index_modification(self):
|
||||
"""Test modifying array element by index"""
|
||||
data = {
|
||||
"items": [
|
||||
{"id": 1, "name": "Item 1"},
|
||||
{"id": 2, "name": "Item 2"},
|
||||
{"id": 3, "name": "Item 3"}
|
||||
]
|
||||
}
|
||||
|
||||
result = modify_with_jsonpath(data, "$.items[1].name", "Updated Item 2")
|
||||
|
||||
assert result["items"][1]["name"] == "Updated Item 2"
|
||||
assert result["items"][0]["name"] == "Item 1"
|
||||
assert result["items"][2]["name"] == "Item 3"
|
||||
# Original unchanged
|
||||
assert data["items"][1]["name"] == "Item 2"
|
||||
|
||||
def test_wildcard_modification(self):
|
||||
"""Test modifying multiple elements with wildcard"""
|
||||
data = {
|
||||
"users": [
|
||||
{"name": "Alice", "active": True},
|
||||
{"name": "Bob", "active": True},
|
||||
{"name": "Charlie", "active": True}
|
||||
]
|
||||
}
|
||||
|
||||
result = modify_with_jsonpath(data, "$.users[*].active", False)
|
||||
|
||||
# All active fields should be updated
|
||||
for user in result["users"]:
|
||||
assert user["active"] is False
|
||||
# Original unchanged
|
||||
for user in data["users"]:
|
||||
assert user["active"] is True
|
||||
|
||||
def test_deep_copy_behavior(self):
|
||||
"""Test that modifications don't affect the original data"""
|
||||
original = {
|
||||
"level1": {
|
||||
"level2": {
|
||||
"level3": {
|
||||
"value": "original"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
result = modify_with_jsonpath(original, "$.level1.level2.level3.value", "modified")
|
||||
|
||||
assert result["level1"]["level2"]["level3"]["value"] == "modified"
|
||||
assert original["level1"]["level2"]["level3"]["value"] == "original"
|
||||
|
||||
# Verify deep copy by modifying nested dict in result
|
||||
result["level1"]["level2"]["new_key"] = "new_value"
|
||||
assert "new_key" not in original["level1"]["level2"]
|
||||
|
||||
def test_modify_to_different_type(self):
|
||||
"""Test changing value to different type"""
|
||||
data = {"count": "10"}
|
||||
result = modify_with_jsonpath(data, "$.count", 10)
|
||||
|
||||
assert result["count"] == 10
|
||||
assert isinstance(result["count"], int)
|
||||
assert data["count"] == "10"
|
||||
|
||||
def test_modify_to_complex_object(self):
|
||||
"""Test replacing value with complex object"""
|
||||
data = {"simple": "value"}
|
||||
new_value = {"complex": {"nested": "structure"}}
|
||||
|
||||
result = modify_with_jsonpath(data, "$.simple", new_value)
|
||||
|
||||
assert result["simple"] == new_value
|
||||
assert result["simple"]["complex"]["nested"] == "structure"
|
||||
|
||||
def test_modify_to_list(self):
|
||||
"""Test replacing value with list"""
|
||||
data = {"items": None}
|
||||
result = modify_with_jsonpath(data, "$.items", [1, 2, 3])
|
||||
|
||||
assert result["items"] == [1, 2, 3]
|
||||
assert data["items"] is None
|
||||
|
||||
def test_modify_to_none(self):
|
||||
"""Test setting value to None"""
|
||||
data = {"value": "something"}
|
||||
result = modify_with_jsonpath(data, "$.value", None)
|
||||
|
||||
assert result["value"] is None
|
||||
assert data["value"] == "something"
|
||||
|
||||
def test_recursive_descent(self):
|
||||
"""Test using recursive descent operator"""
|
||||
data: dict[str, Any] = {
|
||||
"store": {
|
||||
"book": [
|
||||
{"title": "Book 1", "price": 10},
|
||||
{"title": "Book 2", "price": 20}
|
||||
],
|
||||
"bicycle": {
|
||||
"price": 100
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
# Update all prices
|
||||
result = modify_with_jsonpath(data, "$..price", 0)
|
||||
|
||||
assert result["store"]["book"][0]["price"] == 0
|
||||
assert result["store"]["book"][1]["price"] == 0
|
||||
assert result["store"]["bicycle"]["price"] == 0
|
||||
# Original unchanged
|
||||
assert data["store"]["book"][0]["price"] == 10
|
||||
|
||||
def test_specific_array_elements(self):
|
||||
"""Test updating specific array elements by index"""
|
||||
data = {
|
||||
"products": [
|
||||
{"name": "Product 1", "price": 100, "stock": 5},
|
||||
{"name": "Product 2", "price": 200, "stock": 0},
|
||||
{"name": "Product 3", "price": 150, "stock": 10}
|
||||
]
|
||||
}
|
||||
|
||||
# Update first product's price
|
||||
result = modify_with_jsonpath(data, "$.products[0].price", 0)
|
||||
|
||||
assert result["products"][0]["price"] == 0
|
||||
assert result["products"][1]["price"] == 200 # not modified
|
||||
assert result["products"][2]["price"] == 150 # not modified
|
||||
|
||||
def test_empty_dict(self):
|
||||
"""Test modifying empty dictionary"""
|
||||
data: dict[str, Any] = {}
|
||||
result = modify_with_jsonpath(data, "$.nonexistent", "value")
|
||||
|
||||
# Should return the original empty dict since path doesn't exist
|
||||
assert result == {}
|
||||
|
||||
def test_complex_real_world_scenario(self):
|
||||
"""Test complex real-world modification scenario"""
|
||||
data: dict[str, Any] = {
|
||||
"api_version": "1.0",
|
||||
"config": {
|
||||
"database": {
|
||||
"host": "localhost",
|
||||
"port": 5432,
|
||||
"credentials": {
|
||||
"username": "admin",
|
||||
"password": "secret"
|
||||
}
|
||||
},
|
||||
"services": [
|
||||
{"name": "auth", "enabled": True, "port": 8001},
|
||||
{"name": "api", "enabled": True, "port": 8002},
|
||||
{"name": "cache", "enabled": False, "port": 8003}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
# Update database port
|
||||
result = modify_with_jsonpath(data, "$.config.database.port", 5433)
|
||||
assert result["config"]["database"]["port"] == 5433
|
||||
|
||||
# Update all service ports
|
||||
result2 = modify_with_jsonpath(result, "$.config.services[*].enabled", True)
|
||||
assert all(service["enabled"] for service in result2["config"]["services"])
|
||||
|
||||
# Original unchanged
|
||||
assert data["config"]["database"]["port"] == 5432
|
||||
assert data["config"]["services"][2]["enabled"] is False
|
||||
|
||||
def test_list_slice_modification(self):
|
||||
"""Test modifying list slice"""
|
||||
data = {"numbers": [1, 2, 3, 4, 5]}
|
||||
|
||||
# Modify first three elements
|
||||
result = modify_with_jsonpath(data, "$.numbers[0:3]", 0)
|
||||
|
||||
assert result["numbers"][0] == 0
|
||||
assert result["numbers"][1] == 0
|
||||
assert result["numbers"][2] == 0
|
||||
assert result["numbers"][3] == 4
|
||||
assert result["numbers"][4] == 5
|
||||
|
||||
def test_modify_with_datetime_value(self):
|
||||
"""Test modifying with datetime value"""
|
||||
data = {"timestamp": "2025-01-01T00:00:00"}
|
||||
new_datetime = datetime(2025, 10, 23, 15, 30, 45)
|
||||
|
||||
result = modify_with_jsonpath(data, "$.timestamp", new_datetime)
|
||||
|
||||
assert result["timestamp"] == new_datetime
|
||||
assert isinstance(result["timestamp"], datetime)
|
||||
|
||||
|
||||
# MARK: Integration tests
|
||||
class TestIntegration:
|
||||
"""Integration tests combining multiple functions"""
|
||||
|
||||
def test_encoder_and_json_dumps_comparison(self):
|
||||
"""Test that DateTimeEncoder and json_dumps handle datetimes differently"""
|
||||
dt = datetime(2025, 10, 23, 15, 30, 45)
|
||||
data = {"timestamp": dt}
|
||||
|
||||
# Using DateTimeEncoder produces ISO format
|
||||
with_encoder = json.dumps(data, cls=DateTimeEncoder)
|
||||
decoded_encoder = json.loads(with_encoder)
|
||||
assert decoded_encoder["timestamp"] == "2025-10-23T15:30:45"
|
||||
|
||||
# Using json_dumps (default=str) produces string representation
|
||||
with_dumps = json_dumps(data)
|
||||
decoded_dumps = json.loads(with_dumps)
|
||||
assert isinstance(decoded_dumps["timestamp"], str)
|
||||
assert "2025-10-23" in decoded_dumps["timestamp"]
|
||||
|
||||
def test_modify_and_serialize(self):
|
||||
"""Test modifying data and then serializing it"""
|
||||
data = {
|
||||
"event": {
|
||||
"name": "Meeting",
|
||||
"date": date(2025, 10, 23),
|
||||
"attendees": [
|
||||
{"name": "Alice", "confirmed": False},
|
||||
{"name": "Bob", "confirmed": False}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
# Modify confirmation status
|
||||
modified = modify_with_jsonpath(data, "$.event.attendees[*].confirmed", True)
|
||||
|
||||
# Serialize with datetime handling
|
||||
serialized = json.dumps(modified, cls=DateTimeEncoder)
|
||||
decoded = json.loads(serialized)
|
||||
|
||||
assert decoded["event"]["date"] == "2025-10-23"
|
||||
assert decoded["event"]["attendees"][0]["confirmed"] is True
|
||||
assert decoded["event"]["attendees"][1]["confirmed"] is True
|
||||
|
||||
def test_round_trip_with_modification(self):
|
||||
"""Test full round trip: serialize -> modify -> serialize"""
|
||||
original = {
|
||||
"config": {
|
||||
"updated": datetime(2025, 10, 23, 15, 30, 45),
|
||||
"version": "1.0"
|
||||
}
|
||||
}
|
||||
|
||||
# Serialize
|
||||
json_str = json.dumps(original, cls=DateTimeEncoder)
|
||||
|
||||
# Deserialize
|
||||
deserialized = json.loads(json_str)
|
||||
|
||||
# Modify
|
||||
modified = modify_with_jsonpath(deserialized, "$.config.version", "2.0")
|
||||
|
||||
# Serialize again
|
||||
final_json = json_dumps(modified)
|
||||
final_data = json.loads(final_json)
|
||||
|
||||
assert final_data["config"]["version"] == "2.0"
|
||||
assert final_data["config"]["updated"] == "2025-10-23T15:30:45"
|
||||
|
||||
|
||||
# MARK: Edge cases
|
||||
class TestEdgeCases:
|
||||
"""Test edge cases and error scenarios"""
|
||||
|
||||
def test_circular_reference_in_modify(self):
|
||||
"""Test that modify_with_jsonpath handles data without circular references"""
|
||||
# Note: JSON doesn't support circular references, so we test normal nested data
|
||||
data = {
|
||||
"a": {
|
||||
"b": {
|
||||
"c": "value"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
result = modify_with_jsonpath(data, "$.a.b.c", "new_value")
|
||||
assert result["a"]["b"]["c"] == "new_value"
|
||||
|
||||
def test_unicode_in_keys_and_values(self):
|
||||
"""Test handling unicode in both keys and values"""
|
||||
data = {
|
||||
"日本語": "テスト",
|
||||
"emoji_🎉": "🚀",
|
||||
"normal": "value"
|
||||
}
|
||||
|
||||
result = json_dumps(data)
|
||||
decoded = json.loads(result)
|
||||
|
||||
assert decoded["日本語"] == "テスト"
|
||||
assert decoded["emoji_🎉"] == "🚀"
|
||||
assert decoded["normal"] == "value"
|
||||
|
||||
def test_very_nested_structure(self):
|
||||
"""Test deeply nested structure"""
|
||||
# Create a 10-level deep nested structure
|
||||
data: dict[str, Any] = {"level0": {}}
|
||||
current = data["level0"]
|
||||
for i in range(1, 10):
|
||||
current[f"level{i}"] = {}
|
||||
current = current[f"level{i}"]
|
||||
current["value"] = "deep_value"
|
||||
|
||||
result = modify_with_jsonpath(data, "$..value", "modified_deep_value")
|
||||
|
||||
# Navigate to the deep value
|
||||
current = result["level0"]
|
||||
for i in range(1, 10):
|
||||
current = current[f"level{i}"]
|
||||
assert current["value"] == "modified_deep_value"
|
||||
|
||||
def test_large_list_modification(self):
|
||||
"""Test modifying large list"""
|
||||
data = {"items": [{"id": i, "value": i * 10} for i in range(100)]}
|
||||
|
||||
result = modify_with_jsonpath(data, "$.items[*].value", 0)
|
||||
|
||||
assert all(item["value"] == 0 for item in result["items"])
|
||||
assert len(result["items"]) == 100
|
||||
|
||||
def test_mixed_date_types_encoding(self):
|
||||
"""Test encoding with both date and datetime in same structure"""
|
||||
data = {
|
||||
"created_date": date(2025, 10, 23),
|
||||
"created_datetime": datetime(2025, 10, 23, 15, 30, 45),
|
||||
"updated_date": date(2025, 10, 24),
|
||||
"updated_datetime": datetime(2025, 10, 24, 16, 45, 30)
|
||||
}
|
||||
|
||||
result = json.dumps(data, cls=DateTimeEncoder)
|
||||
decoded = json.loads(result)
|
||||
|
||||
assert decoded["created_date"] == "2025-10-23"
|
||||
assert decoded["created_datetime"] == "2025-10-23T15:30:45"
|
||||
assert decoded["updated_date"] == "2025-10-24"
|
||||
assert decoded["updated_datetime"] == "2025-10-24T16:45:30"
|
||||
Reference in New Issue
Block a user