fix(engine): validate processor plugin impls (#609)

* fix(engine): validate processor plugin impls

Add the processor implementation base to assert_valid_plugin so
processor plugins are checked against Processor instead of only the
generic config contract. Keep plugin type validation table-driven and
raise explicit AssertionError messages so checks are not skipped under
optimized Python.

Signed-off-by: Johnny Greco <jogreco@nvidia.com>

* test(engine): require plugin base map coverage

---------

Signed-off-by: Johnny Greco <jogreco@nvidia.com>

Author: johnnygreco

packages/data-designer-engine/src/data_designer/engine/testing/utils.py

@@ -5,16 +5,31 @@
 
 from data_designer.config.base import ConfigBase
 from data_designer.engine.configurable_task import ConfigurableTask
+from data_designer.engine.processing.processors.base import Processor
 from data_designer.engine.resources.seed_reader import SeedReader
 from data_designer.plugins.plugin import Plugin, PluginType
 
+_PLUGIN_IMPLEMENTATION_BASES: dict[PluginType, type[object]] = {
+    PluginType.COLUMN_GENERATOR: ConfigurableTask,
+    PluginType.SEED_READER: SeedReader,
+    PluginType.PROCESSOR: Processor,
+}
+if set(_PLUGIN_IMPLEMENTATION_BASES) != set(PluginType):
+    raise AssertionError("Plugin implementation base map must cover all plugin types")
+
+
+def _assert_subclass(cls: type[object], base_cls: type[object], message: str) -> None:
+    if not issubclass(cls, base_cls):
+        raise AssertionError(message)
+
 
 def assert_valid_plugin(plugin: Plugin) -> None:
-    assert issubclass(plugin.config_cls, ConfigBase), "Plugin config class is not a subclass of ConfigBase"
-
-    if plugin.plugin_type == PluginType.COLUMN_GENERATOR:
-        assert issubclass(plugin.impl_cls, ConfigurableTask), (
-            "Column generator plugin impl class must be a subclass of ConfigurableTask"
-        )
-    elif plugin.plugin_type == PluginType.SEED_READER:
-        assert issubclass(plugin.impl_cls, SeedReader), "Seed reader plugin impl class must be a subclass of SeedReader"
+    _assert_subclass(plugin.config_cls, ConfigBase, "Plugin config class is not a subclass of ConfigBase")
+
+    implementation_base = _PLUGIN_IMPLEMENTATION_BASES[plugin.plugin_type]
+    _assert_subclass(
+        plugin.impl_cls,
+        implementation_base,
+        f"{plugin.plugin_type.display_name.capitalize()} plugin impl class must be a subclass of "
+        f"{implementation_base.__name__}",
+    )

packages/data-designer-engine/tests/engine/testing/test_plugin_testing_utils.py

@@ -0,0 +1,132 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+from typing import Any, Literal
+
+import pytest
+
+from data_designer.config.base import ConfigBase, ProcessorConfig, SingleColumnConfig
+from data_designer.engine.configurable_task import ConfigurableTask
+from data_designer.engine.processing.processors.base import Processor
+from data_designer.engine.resources.seed_reader import SeedReader
+from data_designer.engine.testing.utils import assert_valid_plugin
+from data_designer.plugins.plugin import Plugin, PluginType
+
+MODULE_NAME = __name__
+
+
+class ValidColumnGeneratorConfig(SingleColumnConfig):
+    name: str = "valid-column-generator"
+    column_type: Literal["valid-column-generator"] = "valid-column-generator"
+
+    @property
+    def required_columns(self) -> list[str]:
+        return []
+
+    @property
+    def side_effect_columns(self) -> list[str]:
+        return []
+
+
+class ValidColumnGenerator(ConfigurableTask[ValidColumnGeneratorConfig]):
+    pass
+
+
+class ValidSeedReaderConfig(ConfigBase):
+    seed_type: Literal["valid-seed-reader"] = "valid-seed-reader"
+
+
+class ValidSeedReader(SeedReader):
+    def get_dataset_uri(self) -> str:
+        return "unused"
+
+    def create_duckdb_connection(self) -> Any:
+        raise NotImplementedError
+
+
+class ValidProcessorConfig(ProcessorConfig):
+    name: str = "valid-processor"
+    processor_type: Literal["valid-processor"] = "valid-processor"
+
+
+class ValidProcessor(Processor[ValidProcessorConfig]):
+    def process_before_batch(self, data: dict[str, Any]) -> dict[str, Any]:
+        return data
+
+
+class NonProcessor:
+    pass
+
+
+class TaskButNotProcessor(ConfigurableTask[ValidProcessorConfig]):
+    pass
+
+
+@pytest.mark.parametrize(
+    ("plugin_type", "config_class_name", "implementation_class_name"),
+    [
+        (PluginType.COLUMN_GENERATOR, "ValidColumnGeneratorConfig", "ValidColumnGenerator"),
+        (PluginType.SEED_READER, "ValidSeedReaderConfig", "ValidSeedReader"),
+        (PluginType.PROCESSOR, "ValidProcessorConfig", "ValidProcessor"),
+    ],
+)
+def test_assert_valid_plugin_accepts_supported_plugin_types(
+    plugin_type: PluginType,
+    config_class_name: str,
+    implementation_class_name: str,
+) -> None:
+    plugin = Plugin(
+        config_qualified_name=f"{MODULE_NAME}.{config_class_name}",
+        impl_qualified_name=f"{MODULE_NAME}.{implementation_class_name}",
+        plugin_type=plugin_type,
+    )
+
+    assert_valid_plugin(plugin)
+
+
+@pytest.mark.parametrize(
+    ("plugin_type", "config_class_name", "expected_message"),
+    [
+        (
+            PluginType.COLUMN_GENERATOR,
+            "ValidColumnGeneratorConfig",
+            "Column generator plugin impl class must be a subclass of ConfigurableTask",
+        ),
+        (
+            PluginType.SEED_READER,
+            "ValidSeedReaderConfig",
+            "Seed reader plugin impl class must be a subclass of SeedReader",
+        ),
+        (
+            PluginType.PROCESSOR,
+            "ValidProcessorConfig",
+            "Processor plugin impl class must be a subclass of Processor",
+        ),
+    ],
+)
+def test_assert_valid_plugin_rejects_invalid_impl_for_supported_plugin_types(
+    plugin_type: PluginType,
+    config_class_name: str,
+    expected_message: str,
+) -> None:
+    plugin = Plugin(
+        config_qualified_name=f"{MODULE_NAME}.{config_class_name}",
+        impl_qualified_name=f"{MODULE_NAME}.NonProcessor",
+        plugin_type=plugin_type,
+    )
+
+    with pytest.raises(AssertionError, match=expected_message):
+        assert_valid_plugin(plugin)
+
+
+def test_assert_valid_plugin_rejects_processor_plugin_with_configurable_task_impl() -> None:
+    plugin = Plugin(
+        config_qualified_name=f"{MODULE_NAME}.ValidProcessorConfig",
+        impl_qualified_name=f"{MODULE_NAME}.TaskButNotProcessor",
+        plugin_type=PluginType.PROCESSOR,
+    )
+
+    with pytest.raises(AssertionError, match="Processor plugin impl class must be a subclass of Processor"):
+        assert_valid_plugin(plugin)