Ability to use expressions in a json patch

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 } }