Add json helper function json_replace
Function can replace content for a json path string in a dictionary
This commit is contained in:
@@ -8,6 +8,7 @@ requires-python = ">=3.13"
|
||||
dependencies = [
|
||||
"cryptography>=46.0.3",
|
||||
"jmespath>=1.0.1",
|
||||
"jsonpath-ng>=1.7.0",
|
||||
"psutil>=7.0.0",
|
||||
"requests>=2.32.4",
|
||||
]
|
||||
@@ -28,6 +29,7 @@ build-backend = "hatchling.build"
|
||||
|
||||
[dependency-groups]
|
||||
dev = [
|
||||
"deepdiff>=8.6.1",
|
||||
"pytest>=8.4.1",
|
||||
"pytest-cov>=6.2.1",
|
||||
]
|
||||
|
||||
@@ -5,6 +5,8 @@ json encoder for datetime
|
||||
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]
|
||||
|
||||
|
||||
# subclass JSONEncoder
|
||||
@@ -41,4 +43,22 @@ def json_dumps(data: Any):
|
||||
"""
|
||||
return dumps(data, ensure_ascii=False, default=str)
|
||||
|
||||
|
||||
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
|
||||
|
||||
# __END__
|
||||
|
||||
|
||||
# __END__
|
||||
|
||||
52
test-run/json_handling/json_replace.py
Normal file
52
test-run/json_handling/json_replace.py
Normal file
@@ -0,0 +1,52 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
"""
|
||||
JSON content replace tets
|
||||
"""
|
||||
|
||||
from deepdiff import DeepDiff
|
||||
from corelibs.debug_handling.dump_data import dump_data
|
||||
from corelibs.json_handling.json_helper import modify_with_jsonpath
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""
|
||||
Comment
|
||||
"""
|
||||
__data = {
|
||||
'a': 'b',
|
||||
'foobar': [1, 2, 'a'],
|
||||
'bar': {
|
||||
'a': 1,
|
||||
'b': 'c'
|
||||
},
|
||||
'baz': [
|
||||
{
|
||||
'aa': 1,
|
||||
'ab': 'cc'
|
||||
},
|
||||
{
|
||||
'ba': 2,
|
||||
'bb': 'dd'
|
||||
},
|
||||
],
|
||||
'foo': {
|
||||
'a': [1, 2, 3],
|
||||
'b': ['a', 'b', 'c']
|
||||
}
|
||||
}
|
||||
|
||||
# Modify some values using JSONPath
|
||||
__replace_data = modify_with_jsonpath(__data, 'bar.a', 42)
|
||||
__replace_data = modify_with_jsonpath(__replace_data, 'foo.b[1]', 'modified')
|
||||
__replace_data = modify_with_jsonpath(__replace_data, 'baz[0].ab', 'changed')
|
||||
|
||||
print(f"Original Data:\n{dump_data(__data)}\n")
|
||||
print(f"Modified Data:\n{dump_data(__replace_data)}\n")
|
||||
print(f"Differences:\n{dump_data(DeepDiff(__data, __replace_data, verbose_level=2))}\n")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
# __END__
|
||||
3
tests/unit/json_handling/__init__.py
Normal file
3
tests/unit/json_handling/__init__.py
Normal file
@@ -0,0 +1,3 @@
|
||||
"""
|
||||
tests for json_handling module
|
||||
"""
|
||||
698
tests/unit/json_handling/test_json_helper.py
Normal file
698
tests/unit/json_handling/test_json_helper.py
Normal file
@@ -0,0 +1,698 @@
|
||||
"""
|
||||
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,
|
||||
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(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(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(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("string") is None
|
||||
assert default(42) is None
|
||||
assert default(3.14) is None
|
||||
assert default(True) is None
|
||||
assert default(None) is None
|
||||
assert default([1, 2, 3]) is None
|
||||
assert default({"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)
|
||||
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"
|
||||
46
uv.lock
generated
46
uv.lock
generated
@@ -113,12 +113,14 @@ source = { editable = "." }
|
||||
dependencies = [
|
||||
{ name = "cryptography" },
|
||||
{ name = "jmespath" },
|
||||
{ name = "jsonpath-ng" },
|
||||
{ name = "psutil" },
|
||||
{ name = "requests" },
|
||||
]
|
||||
|
||||
[package.dev-dependencies]
|
||||
dev = [
|
||||
{ name = "deepdiff" },
|
||||
{ name = "pytest" },
|
||||
{ name = "pytest-cov" },
|
||||
]
|
||||
@@ -127,12 +129,14 @@ dev = [
|
||||
requires-dist = [
|
||||
{ name = "cryptography", specifier = ">=46.0.3" },
|
||||
{ name = "jmespath", specifier = ">=1.0.1" },
|
||||
{ name = "jsonpath-ng", specifier = ">=1.7.0" },
|
||||
{ name = "psutil", specifier = ">=7.0.0" },
|
||||
{ name = "requests", specifier = ">=2.32.4" },
|
||||
]
|
||||
|
||||
[package.metadata.requires-dev]
|
||||
dev = [
|
||||
{ name = "deepdiff", specifier = ">=8.6.1" },
|
||||
{ name = "pytest", specifier = ">=8.4.1" },
|
||||
{ name = "pytest-cov", specifier = ">=6.2.1" },
|
||||
]
|
||||
@@ -254,6 +258,18 @@ wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/e8/cb/2da4cc83f5edb9c3257d09e1e7ab7b23f049c7962cae8d842bbef0a9cec9/cryptography-46.0.3-cp38-abi3-win_arm64.whl", hash = "sha256:d89c3468de4cdc4f08a57e214384d0471911a3830fcdaf7a8cc587e42a866372", size = 2918740, upload-time = "2025-10-15T23:18:12.277Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "deepdiff"
|
||||
version = "8.6.1"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "orderly-set" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/19/76/36c9aab3d5c19a94091f7c6c6e784efca50d87b124bf026c36e94719f33c/deepdiff-8.6.1.tar.gz", hash = "sha256:ec56d7a769ca80891b5200ec7bd41eec300ced91ebcc7797b41eb2b3f3ff643a", size = 634054, upload-time = "2025-09-03T19:40:41.461Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/f7/e6/efe534ef0952b531b630780e19cabd416e2032697019d5295defc6ef9bd9/deepdiff-8.6.1-py3-none-any.whl", hash = "sha256:ee8708a7f7d37fb273a541fa24ad010ed484192cd0c4ffc0fa0ed5e2d4b9e78b", size = 91378, upload-time = "2025-09-03T19:40:39.679Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "idna"
|
||||
version = "3.11"
|
||||
@@ -281,6 +297,27 @@ wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/31/b4/b9b800c45527aadd64d5b442f9b932b00648617eb5d63d2c7a6587b7cafc/jmespath-1.0.1-py3-none-any.whl", hash = "sha256:02e2e4cc71b5bcab88332eebf907519190dd9e6e82107fa7f83b1003a6252980", size = 20256, upload-time = "2022-06-17T18:00:10.251Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "jsonpath-ng"
|
||||
version = "1.7.0"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "ply" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/6d/86/08646239a313f895186ff0a4573452038eed8c86f54380b3ebac34d32fb2/jsonpath-ng-1.7.0.tar.gz", hash = "sha256:f6f5f7fd4e5ff79c785f1573b394043b39849fb2bb47bcead935d12b00beab3c", size = 37838, upload-time = "2024-10-11T15:41:42.404Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/35/5a/73ecb3d82f8615f32ccdadeb9356726d6cae3a4bbc840b437ceb95708063/jsonpath_ng-1.7.0-py3-none-any.whl", hash = "sha256:f3d7f9e848cba1b6da28c55b1c26ff915dc9e0b1ba7e752a53d6da8d5cbd00b6", size = 30105, upload-time = "2024-11-20T17:58:30.418Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "orderly-set"
|
||||
version = "5.5.0"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/4a/88/39c83c35d5e97cc203e9e77a4f93bf87ec89cf6a22ac4818fdcc65d66584/orderly_set-5.5.0.tar.gz", hash = "sha256:e87185c8e4d8afa64e7f8160ee2c542a475b738bc891dc3f58102e654125e6ce", size = 27414, upload-time = "2025-07-10T20:10:55.885Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/12/27/fb8d7338b4d551900fa3e580acbe7a0cf655d940e164cb5c00ec31961094/orderly_set-5.5.0-py3-none-any.whl", hash = "sha256:46f0b801948e98f427b412fcabb831677194c05c3b699b80de260374baa0b1e7", size = 13068, upload-time = "2025-07-10T20:10:54.377Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "packaging"
|
||||
version = "25.0"
|
||||
@@ -299,6 +336,15 @@ wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ply"
|
||||
version = "3.11"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/e5/69/882ee5c9d017149285cab114ebeab373308ef0f874fcdac9beb90e0ac4da/ply-3.11.tar.gz", hash = "sha256:00c7c1aaa88358b9c765b6d3000c6eec0ba42abca5351b095321aef446081da3", size = 159130, upload-time = "2018-02-15T19:01:31.097Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/a3/58/35da89ee790598a0700ea49b2a66594140f44dec458c07e8e3d4979137fc/ply-3.11-py2.py3-none-any.whl", hash = "sha256:096f9b8350b65ebd2fd1346b12452efe5b9607f7482813ffca50c22722a807ce", size = 49567, upload-time = "2018-02-15T19:01:27.172Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "psutil"
|
||||
version = "7.1.1"
|
||||
|
||||
Reference in New Issue
Block a user