Ability to parse string expressions

.github/workflows/test-pr.yaml

@@ -18,10 +18,10 @@ jobs:
       - name: Check pyproject.toml
         run: ./scripts/check-pyproject.sh
 
-      - name: Set up Python 3.10
+      - name: Set up Python 3.11
         uses: actions/setup-python@v4
         with:
-          python-version: "3.10"
+          python-version: "3.11"
 
       - name: Install poetry
         run: python3 -m pip install poetry

generic_k8s_webhook/config_parser/entrypoint.py

@@ -1,5 +1,6 @@
 import copy
 
+import generic_k8s_webhook.config_parser.expr_parser as expr_parser
 import generic_k8s_webhook.config_parser.operator_parser as op_parser
 from generic_k8s_webhook import utils
 from generic_k8s_webhook.config_parser.action_parser import ActionParserV1
@@ -57,18 +58,20 @@ def __init__(self, raw_config: dict) -> None:
         # Select the correct parsing method according to the api version, since different api versions
         # expect different schemas
         if self.apiversion == "v1alpha1":
-            self.list_webhook_config = self._parse_alpha1v1(raw_list_webhook_config)
+            self.list_webhook_config = self._parse_v1alpha1(raw_list_webhook_config)
+        elif self.apiversion == "v1beta1":
+            self.list_webhook_config = self._parse_v1beta1(raw_list_webhook_config)
         else:
             raise ValueError(f"The api version {self.apiversion} is not supported")
 
         if len(raw_config) > 0:
             raise ValueError(f"Invalid fields at the manifest level: {raw_config}")
 
-    def _parse_alpha1v1(self, raw_list_webhook_config: dict) -> list[Webhook]:
+    def _parse_v1alpha1(self, raw_list_webhook_config: dict) -> list[Webhook]:
         webhook_parser = WebhookParserV1(
             action_parser=ActionParserV1(
                 meta_op_parser=op_parser.MetaOperatorParser(
-                    [
+                    list_op_parser_classes=[
                         op_parser.AndParser,
                         op_parser.OrParser,
                         op_parser.EqualParser,
@@ -79,7 +82,35 @@ def _parse_alpha1v1(self, raw_list_webhook_config: dict) -> list[Webhook]:
                         op_parser.ContainParser,
                         op_parser.ConstParser,
                         op_parser.GetValueParser,
-                    ]
+                    ],
+                    raw_str_parser=expr_parser.RawStringParserNotImplemented(),
+                ),
+                json_patch_parser=JsonPatchParserV1(),
+            )
+        )
+        list_webhook_config = [
+            webhook_parser.parse(raw_webhook_config, f"webhooks.{i}")
+            for i, raw_webhook_config in enumerate(raw_list_webhook_config)
+        ]
+        return list_webhook_config
+
+    def _parse_v1beta1(self, raw_list_webhook_config: dict) -> list[Webhook]:
+        webhook_parser = WebhookParserV1(
+            action_parser=ActionParserV1(
+                meta_op_parser=op_parser.MetaOperatorParser(
+                    list_op_parser_classes=[
+                        op_parser.AndParser,
+                        op_parser.OrParser,
+                        op_parser.EqualParser,
+                        op_parser.SumParser,
+                        op_parser.NotParser,
+                        op_parser.ListParser,
+                        op_parser.ForEachParser,
+                        op_parser.ContainParser,
+                        op_parser.ConstParser,
+                        op_parser.GetValueParser,
+                    ],
+                    raw_str_parser=expr_parser.RawStringParserV1(),
                 ),
                 json_patch_parser=JsonPatchParserV1(),
             )

generic_k8s_webhook/config_parser/expr_parser.py

@@ -0,0 +1,198 @@
+import abc
+import ast
+
+from lark import Lark, Transformer
+
+import generic_k8s_webhook.operators as op
+from generic_k8s_webhook import utils
+
+GRAMMAR_V1 = r"""
+    ?start: expr
+
+    ?expr: or
+
+    ?or: and
+        | or "||" and       -> orr
+
+    ?and: comp
+        | and "&&" comp     -> andd
+
+    ?comp: sum
+        | sum "==" sum      -> eq
+        | sum "!=" sum      -> ne
+        | sum "<=" sum      -> le
+        | sum ">=" sum      -> ge
+        | sum "<" sum       -> lt
+        | sum ">" sum       -> gt
+
+    ?sum: product
+        | sum "+" product   -> add
+        | sum "-" product   -> sub
+
+    ?product: atom
+        | product "*" atom  -> mul
+        | product "/" atom  -> div
+
+    ?atom: SIGNED_NUMBER    -> number
+        | ESCAPED_STRING    -> const_string
+        | REF               -> ref
+        | BOOL              -> boolean
+        | "(" expr ")"
+
+    BOOL: "true" | "false"
+    REF: "$"? ("."(CNAME|"*"|INT))+
+
+    %import common.CNAME
+    %import common.SIGNED_NUMBER
+    %import common.ESCAPED_STRING
+    %import common.WS_INLINE
+    %import common.INT
+
+    %ignore WS_INLINE
+"""
+
+
+class MyTransformerV1(Transformer):
+    def orr(self, items):
+        return op.Or(op.List(items))
+
+    def andd(self, items):
+        return op.And(op.List(items))
+
+    def eq(self, items):
+        return op.Equal(op.List(items))
+
+    def ne(self, items):
+        return op.NotEqual(op.List(items))
+
+    def le(self, items):
+        return op.LessOrEqual(op.List(items))
+
+    def ge(self, items):
+        return op.GreaterOrEqual(op.List(items))
+
+    def lt(self, items):
+        return op.LessThan(op.List(items))
+
+    def gt(self, items):
+        return op.GreaterThan(op.List(items))
+
+    def add(self, items):
+        return op.Sum(op.List(items))
+
+    def sub(self, items):
+        return op.Sub(op.List(items))
+
+    def mul(self, items):
+        return op.Mul(op.List(items))
+
+    def div(self, items):
+        return op.Div(op.List(items))
+
+    def number(self, items):
+        (elem,) = items
+        try:
+            elem_number = int(elem)
+        except ValueError:
+            elem_number = float(elem)
+        return op.Const(elem_number)
+
+    def const_string(self, items):
+        (elem,) = items
+        # This evaluates the double-quoted string, so the initial and ending double quotes disappear
+        # and any escaped char is also converted. For example, \" -> "
+        elem_str = ast.literal_eval(elem)
+        return op.Const(elem_str)
+
+    def ref(self, items):
+        (elem,) = items
+        return parse_ref(elem)
+
+    def boolean(self, items):
+        (elem,) = items
+        elem_bool = True if elem == "true" else False
+        return op.Const(elem_bool)
+
+
+def parse_ref(ref: str) -> op.GetValue:
+    """Parses a string that is a reference to some element within a json payload
+    and returns a GetValue object.
+
+    Args:
+        ref (str): The reference to a field in a json payload
+    """
+    # Convert, for example, `.foo.bar` into ["foo", "bar"]
+    path = utils.convert_dot_string_path_to_list(ref)
+
+    # Get the id of the context that it will use
+    if path[0] == "":
+        context_id = -1
+    elif path[0] == "$":
+        context_id = 0
+    else:
+        raise ValueError(f"Invalid {path[0]} in {ref}")
+    return op.GetValue(path[1:], context_id)
+
+
+class IRawStringParser(abc.ABC):
+    def __init__(self) -> None:
+        self.parser = Lark(self.get_grammar())
+        self.transformer = self.get_transformer()
+
+    def parse(self, raw_string: str) -> op.Operator:
+        tree = self.parser.parse(raw_string)
+        print(tree.pretty())  # debug mode
+        operator = self.transformer.transform(tree)
+        return operator
+
+    @classmethod
+    @abc.abstractmethod
+    def get_grammar(cls) -> str:
+        pass
+
+    @classmethod
+    @abc.abstractmethod
+    def get_transformer(cls) -> Transformer:
+        pass
+
+
+class RawStringParserNotImplemented(IRawStringParser):
+    def __init__(self) -> None:  # pylint: disable=super-init-not-called
+        # Empty method
+        pass
+
+    def parse(self, raw_string: str) -> op.Operator:
+        return NotImplementedError("Parsing string expressions is not supported")
+
+    @classmethod
+    def get_grammar(cls) -> str:
+        return ""
+
+    @classmethod
+    def get_transformer(cls) -> Transformer:
+        return Transformer()
+
+
+class RawStringParserV1(IRawStringParser):
+    @classmethod
+    def get_grammar(cls) -> str:
+        return GRAMMAR_V1
+
+    @classmethod
+    def get_transformer(cls) -> Transformer:
+        return MyTransformerV1()
+
+
+def main():
+    parser = Lark(GRAMMAR_V1)
+    # print(parser.parse('.key != "some string"').pretty())
+    tree = parser.parse('"true" != "false"')
+    print(tree.pretty())
+    trans = MyTransformerV1()
+    new_op = trans.transform(tree)
+    print(new_op)
+    print(new_op.get_value([]))
+
+
+if __name__ == "__main__":
+    main()

generic_k8s_webhook/config_parser/operator_parser.py

@@ -1,12 +1,13 @@
 import abc
 import inspect
 
+import generic_k8s_webhook.config_parser.expr_parser as expr_parser
 from generic_k8s_webhook import operators, utils
 from generic_k8s_webhook.config_parser.common import ParsingException
 
 
 class MetaOperatorParser:
-    def __init__(self, list_op_parser_classes: list[type]) -> None:
+    def __init__(self, list_op_parser_classes: list[type], raw_str_parser: expr_parser.IRawStringParser) -> None:
         self.dict_op_parser = {}
         for op_parser_class in list_op_parser_classes:
             # Make sure that op_parser_class is a proper "OperatorParser" derived class
@@ -22,7 +23,24 @@ def __init__(self, list_op_parser_classes: list[type]) -> None:
                 raise RuntimeError(f"Duplicated operator parser {op_parser.get_name()}")
             self.dict_op_parser[op_parser.get_name()] = op_parser
 
-    def parse(self, op_spec: dict, path_op: str) -> operators.Operator:
+        self.raw_str_parser = raw_str_parser
+
+    def parse(self, op_spec: dict | str, path_op: str) -> operators.Operator:
+        if isinstance(op_spec, dict):
+            return self._parse_dict(op_spec, path_op)
+        if isinstance(op_spec, str):
+            return self._parse_str(op_spec, path_op)
+        raise RuntimeError(f"Cannot parse the type {type(op_spec)}. It must be dict or str")
+
+    def _parse_dict(self, op_spec: dict, path_op: str) -> operators.Operator:
+        """It's used to parse a structured operator. Example:
+
+        ```yaml
+        sum:
+            - const: 4
+            - const: 5
+        ```
+        """
         if len(op_spec) != 1:
             raise ValueError(f"Expected exactly one key under {path_op}")
         op_name, op_spec = op_spec.popitem()
@@ -33,6 +51,18 @@ def parse(self, op_spec: dict, path_op: str) -> operators.Operator:
 
         return op
 
+    def _parse_str(self, op_spec: str, path_op: str) -> operators.Operator:
+        """It's used to parse an unstructured operator. Example:
+
+        ```yaml
+        "4 + 5"
+        ```
+        """
+        try:
+            return self.raw_str_parser.parse(op_spec)
+        except Exception as e:
+            raise ParsingException(f"Error when parsing {path_op}") from e
+
 
 class OperatorParser(abc.ABC):
     def __init__(self, meta_op_parser: MetaOperatorParser) -> None:
@@ -205,17 +235,7 @@ def get_name(cls) -> str:
     def parse(self, op_inputs: str, path_op: str) -> operators.GetValue:
         if not isinstance(op_inputs, str):
             raise ValueError(f"Expected to find str but got {op_inputs} in {path_op}")
-        path = utils.convert_dot_string_path_to_list(op_inputs)
-
-        # Get the id of the context that it will use
-        if path[0] == "":
-            context_id = -1
-        elif path[0] == "$":
-            context_id = 0
-        else:
-            raise ValueError(f"Invalid {path[0]} in {path_op}")
-
         try:
-            return operators.GetValue(path, context_id)
+            return expr_parser.parse_ref(op_inputs)
         except TypeError as e:
             raise ParsingException(f"Error when parsing {path_op}") from e