vdk-core: Verify payload after pre-processing it

Ingestion allows pre-processing payloads. If the payload
is curated during pre-processing, currently, verification and
subsequently job execution would fail.

Verify payload after pre-processing it, since this
pre-processing might be responsible for making it serializable.
Add payload verification functional tests since it can no
longer be tested in send_object_for_ingestion async function.

Tested by: locally run ingestion job which curates json
during pre-processing; functional test added.

Signed-off-by: Yana Zhivkova <[email protected]>

Author: YanaZhivkova

projects/vdk-core/src/vdk/internal/builtin_plugins/ingestion/ingester_base.py

@@ -112,8 +112,6 @@ def send_object_for_ingestion(
         """
         See parent doc
         """
-        self.__verify_payload_format(payload_dict=payload)
-
         if collection_id is None:
             collection_id = "{data_job_name}|{execution_id}".format(
                 data_job_name=self._data_job_name, execution_id=self._op_id
@@ -435,6 +433,11 @@ def _payload_poster_thread(self):
                             metadata=ingestion_metadata,
                         )
 
+                    # Verify payload after pre-processing it, since this preprocessing might be responsible for
+                    # making it serializable
+                    for payload_dict in payload_obj:
+                        self.__verify_payload_format(payload_dict=payload_dict)
+
                     try:
                         ingestion_metadata = self._ingester.ingest_payload(
                             payload=payload_obj,

projects/vdk-core/tests/functional/ingestion/jobs/test-ingest-bad-payload-job/1_ingest_step.py

@@ -0,0 +1,23 @@
+# Copyright 2021 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+import logging
+from datetime import datetime
+
+from vdk.api.job_input import IJobInput
+
+log = logging.getLogger(__name__)
+
+
+def run(job_input: IJobInput):
+    payload_str = job_input.get_arguments()["payload"]
+
+    log.info("Trying to ingest payload: " + payload_str)
+    payload = payload_str
+    if payload_str == "None":
+        payload = None
+    elif payload_str == "date":
+        payload = {"key1": datetime.utcnow()}
+
+    job_input.send_object_for_ingestion(
+        payload=payload, destination_table="object_table", method="memory"
+    )

projects/vdk-core/tests/functional/ingestion/test_payload_verification.py

@@ -0,0 +1,75 @@
+# Copyright 2021 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+"""
+Functional test aiming at verifying that payload is verified
+"""
+import logging
+
+from click.testing import Result
+from vdk.plugin.test_utils.util_funcs import cli_assert_equal
+from vdk.plugin.test_utils.util_funcs import CliEntryBasedTestRunner
+from vdk.plugin.test_utils.util_funcs import jobs_path_from_caller_directory
+from vdk.plugin.test_utils.util_plugins import IngestIntoMemoryPlugin
+
+
+log = logging.getLogger(__name__)
+
+
+def test_payload_verification_none():
+    ingest_plugin = IngestIntoMemoryPlugin()
+    runner = CliEntryBasedTestRunner(ingest_plugin)
+
+    # Use a sample data job, in which bad payload None is passed to
+    # ingestion method call, thus causing a User Error to be raised.
+    result: Result = runner.invoke(
+        [
+            "run",
+            jobs_path_from_caller_directory("test-ingest-bad-payload-job"),
+            "--arguments",
+            '{"payload": "None"}',
+        ]
+    )
+
+    cli_assert_equal(1, result)
+    assert "Payload given to ingestion method should not be empty." in result.stdout
+
+
+def test_payload_verification_bad_type():
+    ingest_plugin = IngestIntoMemoryPlugin()
+    runner = CliEntryBasedTestRunner(ingest_plugin)
+
+    # Use a sample data job, in which bad payload type string is passed to
+    # ingestion method call, thus causing a User Error to be raised.
+    result: Result = runner.invoke(
+        [
+            "run",
+            jobs_path_from_caller_directory("test-ingest-bad-payload-job"),
+            "--arguments",
+            '{"payload": "wrong_type_string"}',
+        ]
+    )
+
+    cli_assert_equal(1, result)
+    assert (
+        "Payload given to ingestion method should be a dictionary, but it is not"
+        in result.stdout
+    )
+
+
+def test_payload_verification_unserializable():
+    ingest_plugin = IngestIntoMemoryPlugin()
+    runner = CliEntryBasedTestRunner(ingest_plugin)
+
+    # Use a sample data job, in which unserializable payload is passed to
+    # ingestion method call, thus causing a User Error to be raised.
+    result: Result = runner.invoke(
+        [
+            "run",
+            jobs_path_from_caller_directory("test-ingest-bad-payload-job"),
+            "--arguments",
+            '{"payload": "date"}',
+        ]
+    )
+
+    cli_assert_equal(1, result)
+    assert "JSON Serialization Error. Payload is not json serializable" in result.stdout

projects/vdk-core/tests/vdk/internal/builtin_plugins/ingestion/test_ingester_base.py

@@ -1,6 +1,5 @@
 # Copyright 2021 VMware, Inc.
 # SPDX-License-Identifier: Apache-2.0
-from datetime import datetime
 from unittest.mock import call
 from unittest.mock import MagicMock
 
@@ -49,38 +48,6 @@ def create_ingester_base(kwargs=None, config_dict=None, ingester=None) -> Ingest
     )
 
 
-def test_send_object_for_ingestion():
-    test_unserializable_payload = {"key1": 42, "key2": datetime.utcnow()}
-    ingester_base = create_ingester_base()
-
-    with pytest.raises(errors.UserCodeError) as exc_info:
-        ingester_base.send_object_for_ingestion(
-            payload=None,
-            destination_table=shared_test_values.get("destination_table1"),
-            method=shared_test_values.get("method"),
-            target=shared_test_values.get("target"),
-        )
-    assert exc_info.type == errors.UserCodeError
-
-    with pytest.raises(errors.UserCodeError) as exc_info:
-        ingester_base.send_object_for_ingestion(
-            payload="wrong_payload_type",
-            destination_table=None,
-            method=shared_test_values.get("method"),
-            target=shared_test_values.get("target"),
-        )
-    assert exc_info.type == errors.UserCodeError
-
-    with pytest.raises(errors.UserCodeError) as exc_info:
-        ingester_base.send_object_for_ingestion(
-            payload=test_unserializable_payload,
-            destination_table=shared_test_values.get("destination_table1"),
-            method=shared_test_values.get("method"),
-            target=shared_test_values.get("target"),
-        )
-    assert exc_info.type == errors.UserCodeError
-
-
 def test_send_object_for_ingestion_send_to_wait():
     ingester_base = create_ingester_base(
         config_dict={"ingester_wait_to_finish_after_every_send": True}