vdk-core: fix bug in error classification

projects/vdk-core/src/vdk/internal/builtin_plugins/run/job_input_error_classifier.py

@@ -31,9 +31,7 @@ def whom_to_blame(
     :return: ResolvableBy.PLATFORM_ERROR if exception was recognized as Platform Team responsibility.
              errors.ResolvableBy.USER_ERROR if exception was recognized as User Error.
     """
-    if isinstance(exception, errors.BaseVdkError) or hasattr(
-        exception, "to_be_fixed_by"
-    ):
+    if is_classified(exception):
         return errors.find_whom_to_blame_from_exception(exception)
     if is_user_error(exception, data_job_path):
         return errors.ResolvableBy.USER_ERROR
@@ -90,6 +88,12 @@ def is_user_error(
     )
 
 
+def is_classified(exception: BaseException):
+    return isinstance(exception, errors.BaseVdkError) or hasattr(
+        exception, errors.ATTR_VDK_RESOLVABLE_BY
+    )
+
+
 def _is_memory_limit_exceeded(exception):
     return errors.exception_matches(
         exception,

projects/vdk-core/src/vdk/internal/core/error_classifiers.py

@@ -18,6 +18,9 @@
 
 log = logging.getLogger(__name__)
 
+ATTR_VDK_RESOLVABLE_BY = "_vdk_resolvable_by"
+ATTR_VDK_RESOLVABLE_ACTUAL = "_vdk_resolvable_actual"
+
 # RESOLVABLE CONTEXT
 
 
@@ -141,17 +144,17 @@ def clear_intermediate_errors():
 
 
 def set_exception_resolvable_by(exception: BaseException, resolvable_by: ResolvableBy):
-    setattr(exception, "_vdk_resolvable_by", resolvable_by)
+    setattr(exception, ATTR_VDK_RESOLVABLE_BY, resolvable_by)
     setattr(
         exception,
-        "_vdk_resolvable_actual",
+        ATTR_VDK_RESOLVABLE_ACTUAL,
         __error_type_to_actual_resolver(resolvable_by),
     )
 
 
 def get_exception_resolvable_by(exception: BaseException):
-    if hasattr(exception, "_vdk_resolvable_by"):
-        return getattr(exception, "_vdk_resolvable_by")
+    if hasattr(exception, ATTR_VDK_RESOLVABLE_BY):
+        return getattr(exception, ATTR_VDK_RESOLVABLE_BY)
     else:
         return None
 

projects/vdk-core/tests/functional/run/jobs/pandas-key-error-job/10_pandas_step.py

@@ -0,0 +1,24 @@
+# Copyright 2021-2023 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+import pandas as pd
+from vdk.api.job_input import IJobInput
+
+
+def run(job_input: IJobInput):
+    df1 = pd.DataFrame({"lkey": ["foo", "bar", "baz", "foo"], "value": [1, 2, 3, 5]})
+    df2 = pd.DataFrame({"rkey": ["foo", "bar", "baz", "foo"], "value": [5, 6, 7, 8]})
+    pd.merge(
+        df1,
+        df2,
+        how="inner",
+        on=[
+            "timestamp",
+            "status_code",
+            "response_status",
+            "exception",
+            "region",
+            "region_type",
+            "tenant",
+            "utc_time",
+        ],
+    )

projects/vdk-core/tests/functional/run/test_run_errors.py

@@ -197,7 +197,8 @@ def db_connection_execute_operation(execution_cursor: ExecutionCursor):
 
 
 def _get_job_status(tmp_termination_msg_file):
-    return json.loads(tmp_termination_msg_file.read_text())["status"]
+    result = json.loads(tmp_termination_msg_file.read_text())
+    return result["status"]
 
 
 @mock.patch.dict(os.environ, {VDK_DB_DEFAULT_TYPE: DB_TYPE_SQLITE_MEMORY})
@@ -211,3 +212,14 @@ def test_user_error_handled(tmp_termination_msg_file):
     )
     cli_assert_equal(0, result)
     assert (json.loads(tmp_termination_msg_file.read_text())["status"]) == "Success"
+
+
+def test_error_from_pandas_user_error(tmp_termination_msg_file):
+    errors.resolvable_context().clear()
+    runner = CliEntryBasedTestRunner()
+
+    result: Result = runner.invoke(["run", util.job_path("pandas-key-error-job")])
+    cli_assert_equal(1, result)
+    assert _get_job_status(tmp_termination_msg_file) == "User error"
+    assert '"blamee": "User Error"' in result.output
+    assert '"exception_name": "KeyError"' in result.output

projects/vdk-core/tests/vdk/internal/builtin_plugins/termination_message/test_writer.py

@@ -11,6 +11,7 @@
     TerminationMessageWriterPlugin,
 )
 from vdk.internal.builtin_plugins.version.version import get_version
+from vdk.internal.core import errors
 from vdk.internal.core.config import ConfigurationBuilder
 from vdk.internal.core.context import CoreContext
 from vdk.internal.core.errors import get_blamee_overall
@@ -110,6 +111,7 @@ class WriterTest(unittest.TestCase):
     @patch("builtins.open")
     @patch(f"{get_blamee_overall.__module__}.{get_blamee_overall.__name__}")
     def test_writer(self, get_blamee_overall, mock_open):
+        errors.resolvable_context().clear()
         get_blamee_overall.return_value = None
         mock_file = MagicMock()
         mock_open.return_value = mock_file

projects/vdk-plugins/vdk-csv/tests/functional/test_csv_plugin.py

@@ -151,32 +151,33 @@ def test_export_csv_with_already_existing_file(tmpdir):
             cli_assert_equal(1, result)
 
 
-def test_csv_export_with_nonexistent_table(tmpdir):
-    db_dir = str(tmpdir) + "vdk-sqlite.db"
-    with mock.patch.dict(
-        os.environ,
-        {
-            "VDK_DB_DEFAULT_TYPE": "SQLITE",
-            "VDK_SQLITE_FILE": db_dir,
-        },
-    ):
-        runner = CliEntryBasedTestRunner(sqlite_plugin, csv_plugin)
-        drop_table(runner, "test_table")
-        result = runner.invoke(
-            [
-                "export-csv",
-                "--query",
-                "SELECT * FROM test_table",
-                "--file",
-                "result3.csv",
-            ]
-        )
-        assert isinstance(result.exception, OperationalError)
-        assert hasattr(result.exception, "_vdk_resolvable_actual")
-        assert (
-            getattr(result.exception, "_vdk_resolvable_actual")
-            == ResolvableByActual.PLATFORM
-        )
+# TODO: Uncomment after https://github.com/vmware/versatile-data-kit/pull/2840 is merged
+# def test_csv_export_with_nonexistent_table(tmpdir):
+#     db_dir = str(tmpdir) + "vdk-sqlite.db"
+#     with mock.patch.dict(
+#         os.environ,
+#         {
+#             "VDK_DB_DEFAULT_TYPE": "SQLITE",
+#             "VDK_SQLITE_FILE": db_dir,
+#         },
+#     ):
+#         runner = CliEntryBasedTestRunner(sqlite_plugin, csv_plugin)
+#         drop_table(runner, "test_table")
+#         result = runner.invoke(
+#             [
+#                 "export-csv",
+#                 "--query",
+#                 "SELECT * FROM test_table",
+#                 "--file",
+#                 "result3.csv",
+#             ]
+#         )
+#         assert isinstance(result.exception, OperationalError)
+#         assert hasattr(result.exception, "_vdk_resolvable_actual")
+#         assert (
+#             getattr(result.exception, "_vdk_resolvable_actual")
+#             == ResolvableByActual.USER
+#         )
 
 
 def test_csv_export_with_no_data(tmpdir):