Ability to use expressions in a json patch

Introducing the JsonPatchParserV2. This new json patch parser is
backwards compatible with the already supported json patch operations.
On top of that, it adds the ability to use expressions to generate new
values on a json patch operation. For example, we can add a prefix to
the pod name:

```
op: expr
path: .metadata.name
value: '"prefix-" ++ .metadata.name'
```

generic_k8s_webhook/config_parser/action_parser.py

@@ -45,7 +45,7 @@ def parse(self, raw_config: dict, path_action: str) -> Action:
         condition = self.meta_op_parser.parse(raw_condition, f"{path_action}.condition")
 
         raw_patch = raw_config.pop("patch", [])
-        patch = self.json_patch_parser.parse(raw_patch)
+        patch = self.json_patch_parser.parse(raw_patch, f"{path_action}.patch")
 
         # By default, we always accept the payload
         accept = raw_config.pop("accept", True)

generic_k8s_webhook/config_parser/entrypoint.py

@@ -4,7 +4,7 @@
 from generic_k8s_webhook import utils
 from generic_k8s_webhook.config_parser import expr_parser
 from generic_k8s_webhook.config_parser.action_parser import ActionParserV1
-from generic_k8s_webhook.config_parser.jsonpatch_parser import JsonPatchParserV1
+from generic_k8s_webhook.config_parser.jsonpatch_parser import JsonPatchParserV1, JsonPatchParserV2
 from generic_k8s_webhook.config_parser.webhook_parser import WebhookParserV1
 from generic_k8s_webhook.webhook import Webhook
 
@@ -95,29 +95,30 @@ def _parse_v1alpha1(self, raw_list_webhook_config: dict) -> list[Webhook]:
         return list_webhook_config
 
     def _parse_v1beta1(self, raw_list_webhook_config: dict) -> list[Webhook]:
+        meta_op_parser = op_parser.MetaOperatorParser(
+            list_op_parser_classes=[
+                op_parser.AndParser,
+                op_parser.AllParser,
+                op_parser.OrParser,
+                op_parser.AnyParser,
+                op_parser.EqualParser,
+                op_parser.SumParser,
+                op_parser.StrConcatParser,
+                op_parser.NotParser,
+                op_parser.ListParser,
+                op_parser.ForEachParser,
+                op_parser.MapParser,
+                op_parser.ContainParser,
+                op_parser.FilterParser,
+                op_parser.ConstParser,
+                op_parser.GetValueParser,
+            ],
+            raw_str_parser=expr_parser.RawStringParserV1(),
+        )
         webhook_parser = WebhookParserV1(
             action_parser=ActionParserV1(
-                meta_op_parser=op_parser.MetaOperatorParser(
-                    list_op_parser_classes=[
-                        op_parser.AndParser,
-                        op_parser.AllParser,
-                        op_parser.OrParser,
-                        op_parser.AnyParser,
-                        op_parser.EqualParser,
-                        op_parser.SumParser,
-                        op_parser.StrConcatParser,
-                        op_parser.NotParser,
-                        op_parser.ListParser,
-                        op_parser.ForEachParser,
-                        op_parser.MapParser,
-                        op_parser.ContainParser,
-                        op_parser.FilterParser,
-                        op_parser.ConstParser,
-                        op_parser.GetValueParser,
-                    ],
-                    raw_str_parser=expr_parser.RawStringParserV1(),
-                ),
-                json_patch_parser=JsonPatchParserV1(),
+                meta_op_parser=meta_op_parser,
+                json_patch_parser=JsonPatchParserV2(meta_op_parser),
             )
         )
         list_webhook_config = [

generic_k8s_webhook/config_parser/jsonpatch_parser.py

@@ -1,13 +1,104 @@
 import abc
 
+import generic_k8s_webhook.config_parser.operator_parser as op_parser
 from generic_k8s_webhook import jsonpatch_helpers, utils
+from generic_k8s_webhook.config_parser.common import ParsingException
 
 
-class IJsonPatchParser(abc.ABC):
+class ParserOp(abc.ABC):
     @abc.abstractmethod
-    def parse(self, raw_patch: list) -> list[jsonpatch_helpers.JsonPatchOperator]:
+    def parse(self, raw_elem: dict, path_op: str) -> jsonpatch_helpers.JsonPatchOperator:
         pass
 
+    def _parse_path(self, raw_elem: dict, key: str) -> list[str]:
+        raw_path = utils.must_pop(raw_elem, key, f"Missing key {key} in {raw_elem}")
+        path = utils.convert_dot_string_path_to_list(raw_path)
+        if path[0] != "":
+            raise ValueError(f"The first element of a path in the patch must be '.', not {path[0]}")
+        return path[1:]
+
+
+class ParseAdd(ParserOp):
+    def parse(self, raw_elem: dict, path_op: str) -> jsonpatch_helpers.JsonPatchOperator:
+        path = self._parse_path(raw_elem, "path")
+        value = utils.must_pop(raw_elem, "value", f"Missing key 'value' in {raw_elem}")
+        return jsonpatch_helpers.JsonPatchAdd(path, value)
+
+
+class ParseRemove(ParserOp):
+    def parse(self, raw_elem: dict, path_op: str) -> jsonpatch_helpers.JsonPatchOperator:
+        path = self._parse_path(raw_elem, "path")
+        return jsonpatch_helpers.JsonPatchRemove(path)
+
+
+class ParseReplace(ParserOp):
+    def parse(self, raw_elem: dict, path_op: str) -> jsonpatch_helpers.JsonPatchOperator:
+        path = self._parse_path(raw_elem, "path")
+        value = utils.must_pop(raw_elem, "value", f"Missing key 'value' in {raw_elem}")
+        return jsonpatch_helpers.JsonPatchReplace(path, value)
+
+
+class ParseCopy(ParserOp):
+    def parse(self, raw_elem: dict, path_op: str) -> jsonpatch_helpers.JsonPatchOperator:
+        path = self._parse_path(raw_elem, "path")
+        fromm = self._parse_path(raw_elem, "from")
+        return jsonpatch_helpers.JsonPatchCopy(path, fromm)
+
+
+class ParseMove(ParserOp):
+    def parse(self, raw_elem: dict, path_op: str) -> jsonpatch_helpers.JsonPatchOperator:
+        path = self._parse_path(raw_elem, "path")
+        fromm = self._parse_path(raw_elem, "from")
+        return jsonpatch_helpers.JsonPatchMove(path, fromm)
+
+
+class ParseTest(ParserOp):
+    def parse(self, raw_elem: dict, path_op: str) -> jsonpatch_helpers.JsonPatchOperator:
+        path = self._parse_path(raw_elem, "path")
+        value = utils.must_pop(raw_elem, "value", f"Missing key 'value' in {raw_elem}")
+        return jsonpatch_helpers.JsonPatchTest(path, value)
+
+
+class ParseExpr(ParserOp):
+    def __init__(self, meta_op_parser: op_parser.MetaOperatorParser) -> None:
+        self.meta_op_parser = meta_op_parser
+
+    def parse(self, raw_elem: dict, path_op: str) -> jsonpatch_helpers.JsonPatchOperator:
+        path = self._parse_path(raw_elem, "path")
+        value = utils.must_pop(raw_elem, "value", f"Missing key 'value' in {raw_elem}")
+        operator = self.meta_op_parser.parse(value, f"{path_op}.value")
+        return jsonpatch_helpers.JsonPatchExpr(path, operator)
+
+
+class IJsonPatchParser(abc.ABC):
+    def parse(self, raw_patch: list, path_op: str) -> list[jsonpatch_helpers.JsonPatchOperator]:
+        patch = []
+        dict_parse_op = self._get_dict_parse_op()
+        for i, raw_elem in enumerate(raw_patch):
+            op = utils.must_pop(raw_elem, "op", f"Missing key 'op' in {raw_elem}")
+
+            # Select the appropiate class needed to parse the operation "op"
+            if op not in dict_parse_op:
+                raise ParsingException(f"Unsupported patch operation {op} on {path_op}")
+            parse_op = dict_parse_op[op]
+            try:
+                parsed_elem = parse_op.parse(raw_elem, f"{path_op}.{i}")
+            except Exception as e:
+                raise ParsingException(f"Error when parsing {path_op}") from e
+
+            # Make sure we have extracted all the keys from "raw_elem"
+            if len(raw_elem) > 0:
+                raise ValueError(f"Unexpected keys {raw_elem}")
+            patch.append(parsed_elem)
+
+        return patch
+
+    @abc.abstractmethod
+    def _get_dict_parse_op(self) -> dict[str, ParserOp]:
+        """A dictionary with the classes that can parse the json patch operations
+        supported by this JsonPatchParser
+        """
+
 
 class JsonPatchParserV1(IJsonPatchParser):
     """Class used to parse a json patch spec V1. Example:
@@ -19,45 +110,28 @@ class JsonPatchParserV1(IJsonPatchParser):
     ```
     """
 
-    def parse(self, raw_patch: list) -> list[jsonpatch_helpers.JsonPatchOperator]:
-        patch = []
-        for raw_elem in raw_patch:
-            op = utils.must_pop(raw_elem, "op", f"Missing key 'op' in {raw_elem}")
-            if op == "add":
-                path = self._parse_path(raw_elem, "path")
-                value = utils.must_pop(raw_elem, "value", f"Missing key 'value' in {raw_elem}")
-                parsed_elem = jsonpatch_helpers.JsonPatchAdd(path, value)
-            elif op == "remove":
-                path = self._parse_path(raw_elem, "path")
-                parsed_elem = jsonpatch_helpers.JsonPatchRemove(path)
-            elif op == "replace":
-                path = self._parse_path(raw_elem, "path")
-                value = utils.must_pop(raw_elem, "value", f"Missing key 'value' in {raw_elem}")
-                parsed_elem = jsonpatch_helpers.JsonPatchReplace(path, value)
-            elif op == "copy":
-                path = self._parse_path(raw_elem, "path")
-                fromm = self._parse_path(raw_elem, "from")
-                parsed_elem = jsonpatch_helpers.JsonPatchCopy(path, fromm)
-            elif op == "move":
-                path = self._parse_path(raw_elem, "path")
-                fromm = self._parse_path(raw_elem, "from")
-                parsed_elem = jsonpatch_helpers.JsonPatchMove(path, fromm)
-            elif op == "test":
-                path = self._parse_path(raw_elem, "path")
-                value = utils.must_pop(raw_elem, "value", f"Missing key 'value' in {raw_elem}")
-                parsed_elem = jsonpatch_helpers.JsonPatchTest(path, value)
-            else:
-                raise ValueError(f"Invalid patch operation {raw_elem['op']}")
+    def _get_dict_parse_op(self) -> dict[str, ParserOp]:
+        return {
+            "add": ParseAdd(),
+            "remove": ParseRemove(),
+            "replace": ParseReplace(),
+            "copy": ParseCopy(),
+            "move": ParseMove(),
+            "test": ParseTest(),
+        }
 
-            if len(raw_elem) > 0:
-                raise ValueError(f"Unexpected keys {raw_elem}")
-            patch.append(parsed_elem)
 
-        return patch
+class JsonPatchParserV2(JsonPatchParserV1):
+    """Class used to parse a json patch spec V2. It supports the same actions as the
+    json patch patch spec V1 plus the ability use expressions to create new values
+    """
 
-    def _parse_path(self, raw_elem: dict, key: str) -> list[str]:
-        raw_path = utils.must_pop(raw_elem, key, f"Missing key {key} in {raw_elem}")
-        path = utils.convert_dot_string_path_to_list(raw_path)
-        if path[0] != "":
-            raise ValueError(f"The first element of a path in the patch must be '.', not {path[0]}")
-        return path[1:]
+    def __init__(self, meta_op_parser: op_parser.MetaOperatorParser) -> None:
+        self.meta_op_parser = meta_op_parser
+
+    def _get_dict_parse_op(self) -> dict[str, ParserOp]:
+        dict_parse_op_v1 = super()._get_dict_parse_op()
+        dict_parse_op_v2 = {
+            "expr": ParseExpr(self.meta_op_parser),
+        }
+        return {**dict_parse_op_v1, **dict_parse_op_v2}

generic_k8s_webhook/jsonpatch_helpers.py

@@ -3,6 +3,7 @@
 
 import jsonpatch
 
+from generic_k8s_webhook import operators
 from generic_k8s_webhook.utils import to_number
 
 
@@ -134,3 +135,19 @@ def __init__(self, path: list[str], value: Any) -> None:
     def generate_patch(self, json_to_patch: dict | list) -> jsonpatch.JsonPatch:
         formatted_path = "/" + "/".join(self.path)
         return jsonpatch.JsonPatch([{"op": "test", "path": formatted_path, "value": self.value}])
+
+
+class JsonPatchExpr(JsonPatchOperator):
+    """It's similar to the JsonPatchAdd, but it first dynamically evaluates the actual value
+    expressed under the "value" keyword and then performs a normal "add" operation using
+    this new value
+    """
+
+    def __init__(self, path: list[str], value: operators.Operator) -> None:
+        super().__init__(path)
+        self.value = value
+
+    def generate_patch(self, json_to_patch: dict | list) -> jsonpatch.JsonPatch:
+        actual_value = self.value.get_value([json_to_patch])
+        json_patch_add = JsonPatchAdd(self.path, actual_value)
+        return json_patch_add.generate_patch(json_to_patch)

tests/jsonpatch_test.yaml

@@ -126,3 +126,14 @@ test_suites:
               value: foo
             payload: { spec: {}, metadata: { name: foo } }
             expected_result: { spec: {}, metadata: { name: foo } }
+  - name: EXPR
+    tests:
+      - schemas: [v1beta1]
+        cases:
+          # Add a prefix
+          - patch:
+              op: expr
+              path: .metadata.name
+              value: '"prefix-" ++ .metadata.name'
+            payload: { spec: {}, metadata: { name: foo } }
+            expected_result: { spec: {}, metadata: { name: prefix-foo } }