[feat] Add test fixtures for conversion and validation webhooks and add base class for realisation conversion webhooks (#8)

Link (#7)

Signed-off-by: Nikolay Mitrofanov <nikolay.mitrofanov@flant.com>
Co-authored-by: Nikolay Mitrofanov <30695496+name212@users.noreply.github.com>

Author: yalosev

deckhouse/tests.py

@@ -0,0 +1,123 @@
+#!/usr/bin/env python3
+#
+# Copyright 2024 Flant JSC Licensed under Apache License 2.0
+#
+
+import unittest
+import typing
+
+from .hook import Output
+
+
+# msg: typing.Tuple[str, ...] | str | None
+def __assert_validation(t: unittest.TestCase, o: Output, allowed: bool, msg: typing.Union[typing.Tuple[str, ...], str, None]):
+    v = o.validations
+
+    t.assertEqual(len(v.data), 1)
+
+    if allowed:
+        t.assertTrue(v.data[0]["allowed"])
+
+        if not msg:
+            return
+
+        if isinstance(msg, str):
+            t.assertEqual(len(v.data[0]["warnings"]), 1)
+            t.assertEqual(v.data[0]["warnings"][0], msg)
+        elif isinstance(msg, tuple):
+            t.assertEqual(v.data[0]["warnings"], msg)
+        else:
+            t.fail("Incorrect msg type")
+    else:
+        if not isinstance(msg, str):
+            t.fail("Incorrect msg type")
+
+        t.assertIsNotNone(msg)
+        t.assertIsInstance(msg, str)
+        t.assertFalse(v.data[0]["allowed"])
+        t.assertEqual(v.data[0]["message"], msg)
+
+
+# msg: typing.Tuple[str, ...] | str | None
+def assert_validation_allowed(t: unittest.TestCase, o: Output, msg: typing.Union[typing.Tuple[str, ...], str, None]):
+    """
+        Assert that validation webhook returns "allowed" result
+
+        Args:
+            t (unittest.TestCase): unit test context (self in Test class method)
+            o (hook.Output): output from hook.testrun
+            msg (any): tuple or str or None, warnings for output, tuple for multiple warnings, str for one warning, None without warnings
+    """
+    __assert_validation(t, o, True, msg)
+
+
+def assert_validation_deny(t: unittest.TestCase, o: Output, msg: str):
+    """
+        Assert that validation webhook returns "deny" result
+
+        Args:
+            t (unittest.TestCase): unit test context (self in Test class method)
+            o (hook.Output): output from hook.testrun
+            msg (str): failed message
+    """
+    __assert_validation(t, o, False, msg)
+
+
+def assert_common_resource_fields(t: unittest.TestCase, obj: dict, api_version: str, name: str, namespace: str = ""):
+    """
+        Assert for object represented as dict api version name and namespace
+        This fixture may be useful for conversion webhook tests for checking
+            that conversion webhook did not change name and namespace and set valid api version
+
+        Args:
+            t (unittest.TestCase): unit test context (self in Test class method)
+            obj (hook.Output): output from hook.testrun
+            api_version (str): API version for expected object
+            name (str): name of expected object
+            namespace (str): namespace of expected object
+    """
+
+    t.assertIn("apiVersion", obj)
+    t.assertEqual(obj["apiVersion"], api_version)
+
+    t.assertIn("metadata", obj)
+
+    t.assertIn("name", obj["metadata"])
+    t.assertEqual(obj["metadata"]["name"], name)
+
+    if namespace:
+        t.assertIn("namespace", obj["metadata"])
+        t.assertEqual(obj["metadata"]["namespace"], namespace)
+
+# res: dict | typing.List[dict] | typing.Callable[[unittest.TestCase, typing.List[dict]], None]
+def assert_conversion(t: unittest.TestCase, o: Output, res: typing.Union[dict, typing.List[dict], typing.Callable[[unittest.TestCase, typing.List[dict]], None]], failed_msg: str):
+    """
+        Assert result of conversion webhook
+
+        Args:
+            t (unittest.TestCase): unit test context (self in Test class method)
+            o (hook.Output): output from hook.testrun
+            res (any): Can be: dict - for one resource convertion, list of dicts for conversion multiple objects per request
+                       or function callable[ (unittest.TestCase, typing.List[dict]) -> None ] for assert objects for your manner
+            failed_msg (str | None): should present for asserting failed result of webhook
+    """
+
+    d = o.conversions.data
+
+    t.assertEqual(len(d), 1)
+
+    if not failed_msg is None:
+        t.assertEqual(len(d[0]), 1)
+        t.assertEqual(d[0]["failedMessage"], failed_msg)
+        return
+
+    if callable(res):
+        res(t, d[0]["convertedObjects"])
+        return
+
+    expected = res
+    if isinstance(res, dict):
+        expected = [res]
+
+
+    t.assertEqual(d[0]["convertedObjects"], expected)

deckhouse/utils.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+#
+# Copyright 2024 Flant JSC Licensed under Apache License 2.0
+#
+
+from .hook import Context
+
+
+class BaseConversionHook:
+    """
+        Base class for convertion webhook realisation.
+        Usage.
+        Create class realisation with methods which named as kubernetesCustomResourceConversion[*].name and
+            which get dict resource for conversion and returns tuple (string|None, dict) with result
+            if string is not None conversion webhook will return error.
+
+        For example. We have next conversion webhook declaration:
+            configVersion: v1
+            kubernetesCustomResourceConversion:
+            - name: alpha1_to_alpha2
+              crdName: nodegroups.deckhouse.io
+            conversions:
+            - fromVersion: deckhouse.io/v1alpha1
+              toVersion: deckhouse.io/v1alpha2
+
+        Then we can create next class for this conversion:
+
+            class NodeGroupConversion(ConversionDispatcher):
+                def __init__(self, ctx: Context):
+                    super().__init__(ctx)
+
+                def alpha1_to_alpha2(self, o: dict) -> typing.Tuple[str | None, dict]:
+                    o["apiVersion"] = "deckhouse.io/v1alpha2"
+                    return None, o
+
+        We added method alpha1_to_alpha2 (named as binding name for conversion), get dict for conversion and returns a tuple.
+
+        And in hook file we can use this class in the next way:
+            def main(ctx: hook.Context):
+                NodeGroupConversion(ctx).run()
+
+            if __name__ == "__main__":
+                hook.run(main, config=config)
+    """
+    def __init__(self, ctx: Context):
+        self._binding_context = ctx.binding_context
+        self._snapshots = ctx.snapshots
+        self.__ctx = ctx
+
+
+    def run(self):
+        binding_name = self._binding_context["binding"]
+
+        try:
+            action = getattr(self, binding_name)
+        except AttributeError:
+            self.__ctx.output.conversions.error("Internal error. Handler for binding {} not found".format(binding_name))
+            return
+
+        try:
+            errors = []
+            from_version = self._binding_context["fromVersion"]
+            to_version = self._binding_context["toVersion"]
+            for obj in self._binding_context["review"]["request"]["objects"]:
+                if from_version != obj["apiVersion"]:
+                    self.__ctx.output.conversions.collect(obj)
+                    continue
+
+                error_msg, res_obj = action(obj)
+                if error_msg is not None:
+                    errors.append(error_msg)
+                    continue
+
+                assert res_obj["apiVersion"] == to_version
+
+                self.__ctx.output.conversions.collect(res_obj)
+            if errors:
+                err_msg = ";".join(errors)
+                self.__ctx.output.conversions.error(err_msg)
+        except Exception as e:
+            self.__ctx.output.conversions.error("Internal error: {}".format(str(e)))
+            return
+