[feat] Webhook hooks: normalize allow/deny arguments, add the ValidationWebhook main and test loaders

deckhouse/tests.py

@@ -3,9 +3,16 @@
 # Copyright 2024 Flant JSC Licensed under Apache License 2.0
 #
 
+import copy
+import sys
+import types
 import unittest
 import typing
+from pathlib import Path
 
+from dotmap import DotMap
+
+from . import hook, utils, validations
 from .hook import Output
 
 
@@ -121,3 +128,88 @@ def assert_conversion(t: unittest.TestCase, o: Output, res: typing.Union[dict, t
 
 
     t.assertEqual(d[0]["convertedObjects"], expected)
+
+
+def load_validation_webhook(*paths: typing.Union[str, Path]) -> types.ModuleType:
+    """
+        Returns the hook webhook-operator renders for a ValidationWebhook as a module, so
+        hook.testrun can run its main.
+
+        Args:
+            paths: the files of handler.python, in the order the ValidationWebhook joins them
+    """
+    paths = [Path(p) for p in paths]
+    # The names the template has in scope for handler.python.
+    module = _new_module(paths, {"Optional": typing.Optional, "DotMap": DotMap, "hook": hook, "validations": validations})
+
+    # The template defines main before handler.python, so a handler could override it.
+    module.main = lambda ctx: validations.handle(ctx, module.validate)
+    for path in paths:
+        exec(compile(path.read_text(), str(path), "exec"), module.__dict__)
+    template_main = module.main
+
+    def main(ctx: hook.Context):
+        # The template sets keepFullObjectsInMemory: false on every context binding, so the hook
+        # gets the jqFilter result only. A snapshot item a test hands over with the full object
+        # must not let a handler that reads `object` pass here and fail in the cluster.
+        ctx.binding_context = _without_full_objects(ctx.binding_context)
+        ctx.snapshots = ctx.binding_context.get("snapshots", {})
+        return template_main(ctx)
+
+    module.main = main
+    return module
+
+
+def load_conversion_webhook(crd: str, conversions: typing.Iterable[typing.Tuple[str, str]], *paths: typing.Union[str, Path]) -> types.ModuleType:
+    """
+        Returns the hook webhook-operator renders for a ConversionWebhook as a module, so
+        hook.testrun can run its main with module.config.
+
+        Args:
+            crd: the name of the CRD, nodegroups.deckhouse.io
+            conversions: the (from, to) version pairs of the resource, ("v1alpha1", "v1")
+            paths: the files of handler.python, in the order of the resource
+    """
+    paths = [Path(p) for p in paths]
+    # The names the template has in scope for handler.python.
+    module = _new_module(paths, {"typing": typing, "DotMap": DotMap, "hook": hook, "utils": utils})
+
+    # The template pastes the handlers into a utils.BaseConversionHook subclass, indented by four
+    # spaces, every line including blank ones and lines inside multi-line strings.
+    handlers = "\n".join("\n".join("    " + line for line in p.read_text().split("\n")) for p in paths)
+    code = (
+        "class Conversion(utils.BaseConversionHook):\n"
+        "    def __init__(self, ctx: hook.Context):\n"
+        "        super().__init__(ctx)\n"
+        + handlers
+        + "\n\ndef main(ctx: hook.Context):\n    Conversion(ctx).run()\n"
+    )
+    exec(compile(code, str(paths[0]), "exec"), module.__dict__)
+
+    group = crd.split(".", 1)[1]
+    module.config = "configVersion: v1\nkubernetesCustomResourceConversion:\n" + "".join(
+        f"  - name: {source}_to_{target}\n"
+        f"    crdName: {crd}\n"
+        f"    conversions:\n"
+        f"    - fromVersion: {group}/{source}\n"
+        f"      toVersion: {group}/{target}\n"
+        for source, target in conversions
+    )
+    return module
+
+
+def _new_module(paths: typing.List[Path], names: dict) -> types.ModuleType:
+    name = "_".join(p.stem for p in paths) + "_hook"
+    module = types.ModuleType(name)
+    module.__dict__.update(names)
+    sys.modules[name] = module
+    return module
+
+
+def _without_full_objects(binding_context):
+    plain = copy.deepcopy(binding_context.toDict() if isinstance(binding_context, DotMap) else binding_context)
+    for snapshot in (plain.get("snapshots") or {}).values():
+        for item in snapshot or []:
+            if isinstance(item, dict):
+                item.pop("object", None)
+    return plain

deckhouse/validations.py

@@ -6,6 +6,22 @@
 
 import typing
 
+from dotmap import DotMap
+
+# What a webhook hands over as warnings or a message: a string, a list of them, or nothing.
+Messages = typing.Union[str, typing.Iterable[str], None]
+
+
+def _strings(values) -> tuple:
+    """The non-empty strings of `values`, with lists and tuples among them unpacked."""
+    result = []
+    for value in values:
+        if isinstance(value, (list, tuple)):
+            result.extend(_strings(value))
+        elif value:
+            result.append(value)
+    return tuple(result)
+
 
 class ValidationsCollector:
     """
@@ -30,17 +46,39 @@ def data(self):
         """
         return self._data
 
-    def allow(self, *warnings: str):
+    def allow(self, *warnings: Messages):
+        """Allows the request. Each warning is a string or a list of them; empty ones are dropped."""
+        warnings = _strings(warnings)
         response = {"allowed": True}
-        if len(warnings) > 0:
+        if warnings:
             response["warnings"] = warnings
         self.collect(response)
 
-    def deny(self, message: typing.Union[str, None] = None):
+    def deny(self, message: Messages = None):
+        """Denies the request. A list of messages becomes one message, joined with "; "."""
+        messages = _strings([message])
         response = {"allowed": False}
-        if message is not None:
-            response["message"] = message
+        if messages:
+            response["message"] = "; ".join(messages)
         self.collect(response)
 
     def error(self, message: str):
         self.collect({"allowed": False, "message": f"Internal webhook error: {message}"})
+
+
+def handle(ctx, validate: typing.Callable[[DotMap], typing.Tuple[Messages, bool]]):
+    """Answers the admission request of the hook context `ctx` with the result of `validate`.
+
+    This is the main of the hook webhook-operator renders for a ValidationWebhook.
+    validate(binding_context) gets the binding context as a DotMap and returns (message, allowed).
+    The message is a string, a list of them or None: warnings for an allowed request, the reason
+    for a denied one. An exception becomes an internal webhook error.
+    """
+    try:
+        message, allowed = validate(DotMap(ctx.binding_context))
+        if allowed:
+            ctx.output.validations.allow(message)
+        else:
+            ctx.output.validations.deny(message)
+    except Exception as e:
+        ctx.output.validations.error(str(e))

pyproject.toml

@@ -11,12 +11,12 @@ packages = [{ include = "deckhouse" }]
 [tool.poetry.dependencies]
 python = "^3.8"
 dictdiffer = "^0.9.0"
+dotmap = "^1.3.30"
 
 [tool.poetry.group.dev.dependencies]
 black = "^22.12.0"
 pylint = "^2.15.10"
 pytest = "^7.2.0"
-dotmap = "^1.3.30"
 
 [build-system]
 requires = ["poetry-core"]