Refactor the jsonpatch tests

tests/conditions_test.py

@@ -2,30 +2,22 @@
 
 import pytest
 import yaml
+from test_utils import expand_schemas
 
 from generic_k8s_webhook.config_parser.entrypoint import GenericWebhookConfigManifest
 
 SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
 CONDITIONS_YAML = os.path.join(SCRIPT_DIR, "conditions_test.yaml")
 
 
-def _expand_schemas(schemas_subsets: dict[str, list[str]], list_schemas: list[str]) -> list[str]:
-    final_schemas = set()
-    for schema in list_schemas:
-        final_schemas.add(schema)
-        for schemas_superset in schemas_subsets.get(schema, []):
-            final_schemas.add(schemas_superset)
-    return sorted(list(final_schemas))
-
-
 def _parse_tests() -> list[tuple]:
     with open(CONDITIONS_YAML, "r") as f:
         raw_tests = yaml.safe_load(f)
 
     parsed_tests = []
     for test_suite in raw_tests["test_suites"]:
         for test in test_suite["tests"]:
-            for schema in _expand_schemas(raw_tests["schemas_subsets"], test["schemas"]):
+            for schema in expand_schemas(raw_tests["schemas_subsets"], test["schemas"]):
                 for i, case in enumerate(test["cases"]):
                     parsed_tests.append(
                         (

tests/jsonpatch_test.py

@@ -0,0 +1,45 @@
+import os
+
+import pytest
+import yaml
+from test_utils import expand_schemas
+
+from generic_k8s_webhook.config_parser.entrypoint import GenericWebhookConfigManifest
+
+SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+JSONPATCH_YAML = os.path.join(SCRIPT_DIR, "jsonpatch_test.yaml")
+
+
+def _parse_tests() -> list[tuple]:
+    with open(JSONPATCH_YAML, "r") as f:
+        raw_tests = yaml.safe_load(f)
+
+    parsed_tests = []
+    for test_suite in raw_tests["test_suites"]:
+        for test in test_suite["tests"]:
+            for schema in expand_schemas(raw_tests["schemas_subsets"], test["schemas"]):
+                for i, case in enumerate(test["cases"]):
+                    parsed_tests.append(
+                        (
+                            f"{test_suite['name']}_{i}",  # name
+                            schema,  # schema
+                            case["patch"],  # patch
+                            case["payload"],  # payload
+                            case["expected_result"],  # expected_result
+                        )
+                    )
+    return parsed_tests
+
+
+@pytest.mark.parametrize(("name", "schema", "patch", "payload", "expected_result"), _parse_tests())
+def test_all(name, schema, patch, payload, expected_result):
+    raw_config = {
+        "apiVersion": f"generic-webhook/{schema}",
+        "kind": "GenericWebhookConfig",
+        "webhooks": [{"name": "test-webhook", "path": "test-path", "actions": [{"patch": [patch]}]}],
+    }
+    gwcm = GenericWebhookConfigManifest(raw_config)
+    action = gwcm.list_webhook_config[0].list_actions[0]
+    json_patch = action.get_patches(payload)
+    result = json_patch.apply(payload)
+    assert result == expected_result

tests/jsonpatch_test.yaml

@@ -0,0 +1,128 @@
+schemas_subsets:
+  # v1alpha1 is a subset of the features from v1beta1
+  # For this reason, any test that uses v1alpha1 will also be executed for v1beta1
+  v1alpha1:
+    - v1beta1
+
+# Each suite stresses a specific jsonpatch operator.
+# Each test specifies a patch operation to be applied on a json payload
+# and the expected result.
+test_suites:
+  - name: ADD
+    tests:
+      - schemas: [v1alpha1]
+        cases:
+          # Change the value of a simple key
+          - patch:
+              op: add
+              path: .spec
+              value: foo
+            payload: { spec: {}, metadata: {} }
+            expected_result: { spec: foo, metadata: {} }
+          # Add a subkey that doesn't exist
+          - patch:
+              op: add
+              path: .spec.subkey
+              value: foo
+            payload: { spec: {}, metadata: {} }
+            expected_result: { spec: { subkey: foo }, metadata: {} }
+          # Add a 2 subkeys that don't exist
+          - patch:
+              op: add
+              path: .spec.subkey1.subkey2
+              value: foo
+            payload: { spec: {}, metadata: {} }
+            expected_result:
+              { spec: { subkey1: { subkey2: foo } }, metadata: {} }
+          # Add an element to an existing empty list
+          - patch:
+              op: add
+              path: .spec.containers.-
+              value: { name: main }
+            payload: { spec: { containers: [] }, metadata: {} }
+            expected_result:
+              { spec: { containers: [{ name: main }] }, metadata: {} }
+          # Add an element to an existing non-empty list
+          - patch:
+              op: add
+              path: .spec.containers.-
+              value: { name: main }
+            payload: { spec: { containers: [{ name: sidecar }] }, metadata: {} }
+            expected_result:
+              spec: { containers: [{ name: sidecar }, { name: main }] }
+              metadata: {}
+          # Add an element to a non-existing list
+          - patch:
+              op: add
+              path: .spec.containers.-
+              value: { name: main }
+            payload: { spec: {}, metadata: {} }
+            expected_result:
+              { spec: { containers: [{ name: main }] }, metadata: {} }
+          # Add a new entry on the second element of the list
+          - patch:
+              op: add
+              path: .spec.containers.0.metadata
+              value: {}
+            payload: { spec: { containers: [{ name: main }] }, metadata: {} }
+            expected_result:
+              spec: { containers: [{ name: main, metadata: {} }] }
+              metadata: {}
+  - name: REMOVE
+    tests:
+      - schemas: [v1alpha1]
+        cases:
+          # Remove the value of a simple key
+          - patch:
+              op: remove
+              path: .spec
+            payload: { spec: {}, metadata: {} }
+            expected_result: { metadata: {} }
+  - name: REPLACE
+    tests:
+      - schemas: [v1alpha1]
+        cases:
+          # Replace the value of a simple key
+          - patch:
+              op: replace
+              path: .metadata.name
+              value: bar
+            payload: { spec: {}, metadata: { name: foo } }
+            expected_result: { spec: {}, metadata: { name: bar } }
+  - name: COPY
+    tests:
+      - schemas: [v1alpha1]
+        cases:
+          # Copy the value from a simple key to another
+          - patch:
+              op: copy
+              path: .metadata.name
+              from: .spec.containers.0.name
+            payload:
+              { spec: { containers: [{ name: bar }] }, metadata: { name: foo } }
+            expected_result:
+              { spec: { containers: [{ name: bar }] }, metadata: { name: bar } }
+  - name: MOVE
+    tests:
+      - schemas: [v1alpha1]
+        cases:
+          # Move the value from a simple key to another
+          - patch:
+              op: move
+              path: .metadata.name
+              from: .spec.containers.0.name
+            payload:
+              { spec: { containers: [{ name: bar }] }, metadata: { name: foo } }
+            expected_result:
+              { spec: { containers: [{}] }, metadata: { name: bar } }
+  - name: TEST
+    tests:
+      - schemas: [v1alpha1]
+        cases:
+          # Test the value of a simple key
+          - patch:
+              op: test
+              path: .metadata.name
+              value: foo
+            payload: { spec: {}, metadata: { name: foo } }
+            expected_result: { spec: {}, metadata: { name: foo } }

tests/test_utils.py

@@ -58,3 +58,12 @@ def wait_for_server_ready(port: int, tls: bool = False) -> None:
     response = requests.get(url, verify=False, timeout=1)
     if response.status_code != 200:
         raise RuntimeError(f"Error when doing the health check to the server. Error {response.status_code}")
+
+
+def expand_schemas(schemas_subsets: dict[str, list[str]], list_schemas: list[str]) -> list[str]:
+    final_schemas = set()
+    for schema in list_schemas:
+        final_schemas.add(schema)
+        for schemas_superset in schemas_subsets.get(schema, []):
+            final_schemas.add(schemas_superset)
+    return sorted(list(final_schemas))