update

Author: refeed

src/tirith/providers/common.py

@@ -1,6 +1,6 @@
 import pydash
 
-from typing import Dict
+from typing import Dict, Any
 
 
 def create_result_dict(value=None, meta=None, err=None) -> Dict:
@@ -11,7 +11,7 @@ class PydashPathNotFound:
     pass
 
 
-def _get_path_value_from_dict_internal(splitted_paths, input_data, place_none_if_not_found=False):
+def _get_path_value_from_input_internal(splitted_paths, input_data, place_none_if_not_found=False):
 
     if not splitted_paths:
         return [input_data] if input_data is not PydashPathNotFound else ([None] if place_none_if_not_found else [])
@@ -25,14 +25,14 @@ def _get_path_value_from_dict_internal(splitted_paths, input_data, place_none_if
         if isinstance(input_data, list):
             for item in input_data:
                 if remaining_paths:
-                    results = _get_path_value_from_dict_internal(remaining_paths, item, place_none_if_not_found)
+                    results = _get_path_value_from_input_internal(remaining_paths, item, place_none_if_not_found)
                     final_data.extend(results)
                 else:
                     final_data.append(item)
         elif isinstance(input_data, dict):
             for value in input_data.values():
                 if remaining_paths:
-                    results = _get_path_value_from_dict_internal(remaining_paths, value, place_none_if_not_found)
+                    results = _get_path_value_from_input_internal(remaining_paths, value, place_none_if_not_found)
                     final_data.extend(results)
                 else:
                     final_data.append(value)
@@ -56,17 +56,17 @@ def _get_path_value_from_dict_internal(splitted_paths, input_data, place_none_if
             # Skip the wildcard marker since iteration is implicit for lists
             paths_to_apply = remaining_paths[1:]
             for val in intermediate_val:
-                results = _get_path_value_from_dict_internal(paths_to_apply, val, place_none_if_not_found)
+                results = _get_path_value_from_input_internal(paths_to_apply, val, place_none_if_not_found)
                 final_data.extend(results)
         elif isinstance(intermediate_val, dict) and remaining_paths[0] == "":
             # If it's a dict and next path is a wildcard, iterate over dict values
             # Skip the wildcard marker and apply remaining paths to each value
             for value in intermediate_val.values():
-                results = _get_path_value_from_dict_internal(remaining_paths[1:], value, place_none_if_not_found)
+                results = _get_path_value_from_input_internal(remaining_paths[1:], value, place_none_if_not_found)
                 final_data.extend(results)
         else:
             # For non-wildcard paths, continue traversal without iteration
-            results = _get_path_value_from_dict_internal(remaining_paths, intermediate_val, place_none_if_not_found)
+            results = _get_path_value_from_input_internal(remaining_paths, intermediate_val, place_none_if_not_found)
             final_data.extend(results)
     else:
         # This is the final path segment
@@ -75,15 +75,16 @@ def _get_path_value_from_dict_internal(splitted_paths, input_data, place_none_if
     return final_data
 
 
-def get_path_value_from_dict(key_path: str, input_dict: dict, place_none_if_not_found: bool = False):
+def get_path_value_from_input(key_path: str, input: Any, place_none_if_not_found: bool = False):
     """
-    Retrieve values from a nested dictionary using a path expression with wildcard support.
+    Retrieve values from a nested data structure using a path expression with wildcard support.
 
-    :param key_path: A dot-separated path to traverse the dictionary.
-                     Use ``*.`` for wildcards to match all items at that level.
+    :param key_path: A dot-separated path to traverse the data structure.
+                     Use ``*`` for wildcard to match all items at that level.
+                     Supports nested structures including dictionaries, lists, and primitives.
     :type key_path: str
-    :param input_dict: The input dictionary to search through.
-    :type input_dict: dict
+    :param input: The input data structure to search through (dict, list, or primitive).
+    :type input: Any
     :param place_none_if_not_found: If True, returns [None] when a path is not found.
                                     If False, returns an empty list []. Defaults to False.
     :type place_none_if_not_found: bool
@@ -96,38 +97,57 @@ def get_path_value_from_dict(key_path: str, input_dict: dict, place_none_if_not_
     Basic path traversal::
 
         >>> data = {"user": {"name": "Alice", "age": 30}}
-        >>> get_path_value_from_dict("user.name", data)
+        >>> get_path_value_from_input("user.name", data)
         ["Alice"]
 
     Wildcard with list items::
 
         >>> data = {"users": [{"name": "Alice"}, {"name": "Bob"}]}
-        >>> get_path_value_from_dict("users.*.name", data)
+        >>> get_path_value_from_input("users.*.name", data)
         ["Alice", "Bob"]
 
     Wildcard with dictionary values::
 
         >>> data = {"countries": {"US": {"capital": "Washington"}, "UK": {"capital": "London"}}}
-        >>> get_path_value_from_dict("countries.*.capital", data)
+        >>> get_path_value_from_input("countries.*.capital", data)
         ["Washington", "London"]
 
-    Leading wildcard::
+    Leading wildcard on lists::
 
         >>> data = [{"name": "Alice"}, {"name": "Bob"}]
-        >>> get_path_value_from_dict("*.name", data)
+        >>> get_path_value_from_input("*.name", data)
         ["Alice", "Bob"]
 
+    Wildcard on primitives::
+
+        >>> get_path_value_from_input("*", 42)
+        [42]
+        >>> get_path_value_from_input("*", "hello")
+        ["hello"]
+
+    Multiple wildcards::
+
+        >>> data = {"groups": [[{"id": 1}, {"id": 2}], [{"id": 3}]]}
+        >>> get_path_value_from_input("groups.*.*.id", data)
+        [1, 2, 3]
+
+    Empty path returns input as-is::
+
+        >>> data = {"key": "value"}
+        >>> get_path_value_from_input("", data)
+        [{"key": "value"}]
+
     Path not found behavior::
 
         >>> data = {"user": {"name": "Alice"}}
-        >>> get_path_value_from_dict("missing.path", data)
+        >>> get_path_value_from_input("missing.path", data)
         []
-        >>> get_path_value_from_dict("missing.path", data, place_none_if_not_found=True)
+        >>> get_path_value_from_input("missing.path", data, place_none_if_not_found=True)
         [None]
     """
     # Handle empty path - return the input data as is
     if not key_path:
-        return [input_dict]
+        return [input]
 
     # Split the path by dots and replace '*' with empty string to mark wildcards
     # Empty strings act as markers to iterate over collections (lists or dict values)
@@ -137,7 +157,7 @@ def get_path_value_from_dict(key_path: str, input_dict: dict, place_none_if_not_
     splitted_attribute = key_path.split(".")
     splitted_attribute = ["" if part == "*" else part for part in splitted_attribute]
 
-    return _get_path_value_from_dict_internal(splitted_attribute, input_dict, place_none_if_not_found)
+    return _get_path_value_from_input_internal(splitted_attribute, input, place_none_if_not_found)
 
 
 class ProviderError:

src/tirith/providers/json/handler.py

@@ -1,12 +1,12 @@
 from typing import Callable, Dict, List
-from ..common import create_result_dict, ProviderError, get_path_value_from_dict
+from ..common import create_result_dict, ProviderError, get_path_value_from_input
 
 
 def get_value(provider_args: Dict, input_data: Dict) -> List[dict]:
     # Must be validated first whether the provider args are valid for this op type
     key_path: str = provider_args["key_path"]
 
-    values = get_path_value_from_dict(key_path, input_data)
+    values = get_path_value_from_input(key_path, input_data)
 
     if len(values) == 0:
         severity_value = 2

src/tirith/providers/kubernetes/handler.py

@@ -1,7 +1,7 @@
 import pydash
 
 from typing import Callable, Dict, List
-from ..common import create_result_dict, ProviderError, get_path_value_from_dict
+from ..common import create_result_dict, ProviderError, get_path_value_from_input
 
 
 def get_value(provider_args: Dict, input_data: Dict, outputs: list) -> Dict:
@@ -21,7 +21,7 @@ def get_value(provider_args: Dict, input_data: Dict, outputs: list) -> Dict:
         if resource["kind"] != target_kind:
             continue
         is_kind_found = True
-        values = get_path_value_from_dict(attribute_path, resource, place_none_if_not_found=True)
+        values = get_path_value_from_input(attribute_path, resource, place_none_if_not_found=True)
         if ".*." not in attribute_path:
             # If there's no * in the attribute path, the values always have 1 member
             values = values[0]

tests/providers/kubernetes/test_attribute.py

@@ -1,4 +1,3 @@
-import json
 import os
 
 from tirith.core.core import start_policy_evaluation
@@ -10,4 +9,4 @@ def test_get_value():
     policy_path = os.path.join(test_dir, "policy.json")
 
     result = start_policy_evaluation(policy_path=policy_path, input_path=input_path)
-    assert result["final_result"] == False
+    assert result["final_result"] is False

tests/providers/test_common.py

@@ -1,5 +1,5 @@
 import pytest
-from tirith.providers.common import get_path_value_from_dict
+from tirith.providers.common import get_path_value_from_input
 
 
 # Test data for simple path access
@@ -91,28 +91,28 @@
 @pytest.mark.parametrize("data,path,expected", simple_path_cases)
 def test_simple_path_access(data, path, expected):
     """Test basic path traversal without wildcards"""
-    result = get_path_value_from_dict(path, data)
+    result = get_path_value_from_input(path, data)
     assert result == expected
 
 
 @pytest.mark.parametrize("data,path,expected", wildcard_list_cases)
 def test_wildcard_with_list(data, path, expected):
     """Test wildcard with list of dictionaries"""
-    result = get_path_value_from_dict(path, data)
+    result = get_path_value_from_input(path, data)
     assert result == expected
 
 
 @pytest.mark.parametrize("data,path,expected_set", wildcard_dict_cases)
 def test_wildcard_with_dict(data, path, expected_set):
     """Test wildcard with dictionary values (order-independent)"""
-    result = get_path_value_from_dict(path, data)
+    result = get_path_value_from_input(path, data)
     assert set(result) == expected_set
 
 
 @pytest.mark.parametrize("data,path,expected", multiple_wildcard_cases)
 def test_multiple_wildcards(data, path, expected):
     """Test multiple wildcards in the path"""
-    result = get_path_value_from_dict(path, data)
+    result = get_path_value_from_input(path, data)
     if isinstance(expected, set):
         assert set(result) == expected
     else:
@@ -122,28 +122,28 @@ def test_multiple_wildcards(data, path, expected):
 @pytest.mark.parametrize("data,path,expected", path_not_found_cases)
 def test_path_not_found_default(data, path, expected):
     """Test that non-existent path returns empty list by default"""
-    result = get_path_value_from_dict(path, data)
+    result = get_path_value_from_input(path, data)
     assert result == expected
 
 
 @pytest.mark.parametrize("data,path,flag,expected", path_not_found_with_flag_cases)
 def test_path_not_found_with_flag(data, path, flag, expected):
     """Test that non-existent path returns [None] when flag is True"""
-    result = get_path_value_from_dict(path, data, place_none_if_not_found=flag)
+    result = get_path_value_from_input(path, data, place_none_if_not_found=flag)
     assert result == expected
 
 
 @pytest.mark.parametrize("data,path,expected", special_value_cases)
 def test_special_values(data, path, expected):
     """Test handling of special values like None and empty paths"""
-    result = get_path_value_from_dict(path, data)
+    result = get_path_value_from_input(path, data)
     assert result == expected
 
 
 @pytest.mark.parametrize("data,path,expected", complex_structure_cases)
 def test_complex_nested_structure(data, path, expected):
     """Test complex real-world-like nested structures"""
-    result = get_path_value_from_dict(path, data)
+    result = get_path_value_from_input(path, data)
     assert result == expected
 
 
@@ -171,22 +171,22 @@ def test_complex_nested_structure(data, path, expected):
 @pytest.mark.parametrize("data,path,expected", edge_case_wildcard_cases)
 def test_edge_case_wildcards(data, path, expected):
     """Test edge cases with wildcards like missing intermediate paths and partial matches"""
-    result = get_path_value_from_dict(path, data)
+    result = get_path_value_from_input(path, data)
     assert result == expected
 
 
 @pytest.mark.parametrize("data,path,flag,expected", empty_container_cases)
 def test_empty_containers(data, path, flag, expected):
     """Test empty containers return empty list"""
-    result = get_path_value_from_dict(path, data, place_none_if_not_found=flag)
+    result = get_path_value_from_input(path, data, place_none_if_not_found=flag)
     assert result == expected
 
 
 @pytest.mark.parametrize("data,path,flag,expected", empty_container_with_flag_cases)
 def test_empty_containers_with_flag(data, path, flag, expected):
     """Test empty containers with place_none_if_not_found flag"""
     # Empty containers don't trigger the flag since they exist but are empty
-    result = get_path_value_from_dict(path, data, place_none_if_not_found=flag)
+    result = get_path_value_from_input(path, data, place_none_if_not_found=flag)
     assert result == expected
 
 
@@ -226,19 +226,19 @@ def test_empty_containers_with_flag(data, path, flag, expected):
 @pytest.mark.parametrize("data,path,expected", wildcard_list_no_remaining_cases)
 def test_wildcard_list_no_remaining_paths(data, path, expected):
     """Test wildcard at the end of path with list and no remaining paths - covers lines 31-32"""
-    result = get_path_value_from_dict(path, data)
+    result = get_path_value_from_input(path, data)
     assert result == expected
 
 
 @pytest.mark.parametrize("data,path,expected_set", wildcard_dict_no_remaining_cases)
 def test_wildcard_dict_no_remaining_paths(data, path, expected_set):
     """Test wildcard at the end of path with dict and no remaining paths - covers lines 38-39"""
-    result = get_path_value_from_dict(path, data)
+    result = get_path_value_from_input(path, data)
     assert set(result) == expected_set
 
 
 @pytest.mark.parametrize("data,path,expected", wildcard_primitive_cases)
 def test_wildcard_primitive_no_remaining_paths(data, path, expected):
     """Test wildcard applied to primitive value with no remaining paths - covers lines 42-43"""
-    result = get_path_value_from_dict(path, data)
+    result = get_path_value_from_input(path, data)
     assert result == expected