Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 13 additions & 8 deletions converter/converter/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,33 +50,38 @@ def delete_paths(data: Dict[str, Any], paths: List[str]) -> None:
paths: List of dot-separated paths (e.g., "a.b.c")
"""

def delete_recursively(d: Dict[str, Any], keys: List[str]) -> None:
def delete_recursively(
d: Dict[str, Any], keys: List[str], current_path: str
) -> None:
if not keys or not isinstance(d, (dict, list)):
return

key = keys[0]
if key.endswith("[]"): # Handle array notation
key = key[:-2] # Remove the array notation
if key in d and isinstance(d[key], list):
for item in d[key]:
for index, item in enumerate(d[key]):
if isinstance(item, dict):
delete_recursively(item, keys[1:])
new_current_path = f"{current_path}.{key}[{index}]"
delete_recursively(item, keys[1:], new_current_path)
elif len(keys) == 1:
# Delete target key if it exists
if isinstance(d, dict):
logger.info("Deleting key: %s", key)
new_current_path = f"{current_path}.{key}"
logger.info("Removing path: %s", new_current_path)
d.pop(key, None)
else:
# Recurse if intermediate key exists
if key in d:
delete_recursively(d[key], keys[1:])
new_current_path = f"{current_path}.{key}"
delete_recursively(d[key], keys[1:], new_current_path)
# Clean up empty dictionaries or lists
if isinstance(d[key], (dict, list)) and not d[key]:
logger.info("Deleting key: %s", key)
logger.info("Removing path: %s", new_current_path)
d.pop(key)

for path in paths:
delete_recursively(data, path.strip("$.").split("."))
delete_recursively(data, path.strip("$.").split("."), "$")


def add_space_before_uppercase(text):
Expand Down Expand Up @@ -199,7 +204,7 @@ def update_json_value(data, jsonpath_query, new_value):
for match in matches:
old_value = get_field_value(data, str(match.full_path))
logger.info(
"Updating value from %s to %s at path %s",
"Updating value from %s to %s at path $.%s",
old_value,
new_value,
match.full_path,
Expand Down
72 changes: 67 additions & 5 deletions converter/tests/test_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -163,11 +163,73 @@ def test_delete_paths_logs_info(self, mock_logger):
first_call_args, _ = mock_logger.info.call_args_list[0]
second_call_args, _ = mock_logger.info.call_args_list[1]

assert first_call_args[0] == "Deleting key: %s"
assert first_call_args[1] == "a"
assert first_call_args[0] == "Removing path: %s"
assert first_call_args[1] == "$.a"

assert second_call_args[0] == "Deleting key: %s"
assert second_call_args[1] == "c"
assert second_call_args[0] == "Removing path: %s"
assert second_call_args[1] == "$.b.c"

@patch("converter.utils.logger")
def test_delete_array_logs_info(self, mock_logger):
# Arrange
data = {
"a": [
{
"b": 1,
},
{
"b": 2,
"c": 3,
},
],
}

# Act
delete_paths(data, ["a[].b"])

# Assert logger has been called twice
assert mock_logger.info.call_count == 2

# Check the log message content
first_call_args, _ = mock_logger.info.call_args_list[0]
second_call_args, _ = mock_logger.info.call_args_list[1]

assert first_call_args[0] == "Removing path: %s"
assert first_call_args[1] == "$.a[0].b"

assert second_call_args[0] == "Removing path: %s"
assert second_call_args[1] == "$.a[1].b"

@patch("converter.utils.logger")
def test_delete_empty_dict_logs_info(self, mock_logger):
# Arrange
data = {
"a": {
"b": {
"c": 1,
},
}
}

# Act
delete_paths(data, ["a.b.c"])

# Assert logger has been called three times
assert mock_logger.info.call_count == 3

# Check the log message content
first_call_args, _ = mock_logger.info.call_args_list[0]
second_call_args, _ = mock_logger.info.call_args_list[1]
third_call_args, _ = mock_logger.info.call_args_list[2]

assert first_call_args[0] == "Removing path: %s"
assert first_call_args[1] == "$.a.b.c"

assert second_call_args[0] == "Removing path: %s"
assert second_call_args[1] == "$.a.b"

assert third_call_args[0] == "Removing path: %s"
assert third_call_args[1] == "$.a"


def test_translate_keys():
Expand Down Expand Up @@ -356,7 +418,7 @@ def test_update_json_value_logs_info(self, mock_logger):

# Check the log message content
args, kwargs = mock_logger.info.call_args
assert args[0] == "Updating value from %s to %s at path %s"
assert args[0] == "Updating value from %s to %s at path $.%s"
assert args[1] == "P1"
assert args[2] == "P2"
assert str(args[3]) == "qualification.details.priority"
Expand Down
Loading