fix: remove hardcoded reference to plugins/config in the cpex/framework/isolated/client.py and update tests. remove methods_to_exclude from validator.

Signed-off-by: habeck <habeck@us.ibm.com>

Author: tedhabeck

cpex/framework/isolated/client.py

@@ -26,7 +26,6 @@
 from cpex.framework.errors import PluginError, convert_exception_to_error
 from cpex.framework.hooks.registry import get_hook_registry
 from cpex.framework.isolated.venv_comm import VenvProcessCommunicator
-from cpex.framework.loader.config import ConfigLoader
 from cpex.framework.models import PluginConfig, PluginContext, PluginErrorModel, PluginPayload, PluginResult
 
 logger = logging.getLogger(__name__)
@@ -35,14 +34,12 @@
 class IsolatedVenvPlugin(Plugin):
     """IsolatedVenvPlugin class."""
 
-    def __init__(self, config: PluginConfig) -> None:
+    def __init__(self, config: PluginConfig, plugin_dirs) -> None:
         """Initialize the plugin's venv environment."""
         super().__init__(config)
         self.implementation = "Python"
         self.comm = None
-        tmp = os.environ.get("PLUGINS_CONFIG_FILE", "plugins/config.yaml")
-        plugin_loader_config = ConfigLoader.load_config(Path(tmp).resolve(), use_jinja=False)
-        self.plugin_dirs = plugin_loader_config.plugin_dirs
+        self.plugin_dirs = plugin_dirs
         # use the first plugin dir specified in the plugin configuration file.
         path = Path(self.plugin_dirs[0]).resolve()
         class_root = self.config.config.get("class_name").split(".")[0]
@@ -210,23 +207,23 @@ async def initialize(self) -> None:
             raise FileNotFoundError(f"plugin path not found: {self.plugin_path}")
 
         venv_path = self.plugin_path / ".venv"
-        
+
         # Prevent directory traversal: ensure requirements_file stays within plugin_path
         requirements_file_input = self.config.config["requirements_file"]
-        
+
         # Handle both relative and absolute paths
         if isinstance(requirements_file_input, Path):
             requirements_file = requirements_file_input
         else:
             requirements_file = Path(requirements_file_input)
-        
+
         # If it's a relative path, resolve it relative to plugin_path
         if not requirements_file.is_absolute():
             requirements_file = (self.plugin_path / requirements_file).resolve()
         else:
             # If absolute, resolve it to normalize
             requirements_file = requirements_file.resolve()
-        
+
         # Validate that the resolved path is within plugin_path (security check)
         try:
             requirements_file.relative_to(self.plugin_path.resolve())

cpex/framework/loader/plugin.py

@@ -53,6 +53,7 @@ def __init__(self) -> None:
             {}
         """
         self._plugin_types: dict[str, Type[Plugin]] = {}
+        self.plugin_dirs: list[str] = []
 
     def __get_plugin_type(self, kind: str) -> Type[Plugin]:
         """Import a plugin type from a python module.
@@ -145,7 +146,7 @@ async def load_and_instantiate_plugin(self, config: PluginConfig) -> Plugin | No
         if config.kind == ISOLATED_VENV_PLUGIN_TYPE:
             from cpex.framework.isolated.client import IsolatedVenvPlugin  # pylint: disable=import-outside-toplevel
 
-            plugin: Plugin = IsolatedVenvPlugin(config)
+            plugin: Plugin = IsolatedVenvPlugin(config, plugin_dirs=self.plugin_dirs.copy())
             await plugin.initialize()
             return plugin
 
@@ -167,8 +168,10 @@ def append_to_search_path(self, plugin_dirs: list[str]) -> None:
         """
         for plugin_dir in plugin_dirs:
             resolved = str(Path(plugin_dir).resolve())
-            if resolved not in sys.path and resolved.startswith(tuple(ALLOWED_PLUGIN_DIRS)):
-                sys.path.append(resolved)
+            if resolved.startswith(tuple(ALLOWED_PLUGIN_DIRS)):
+                self.plugin_dirs.append(plugin_dir)
+                if resolved not in sys.path:
+                    sys.path.append(resolved)
 
     async def shutdown(self) -> None:
         """Shutdown and cleanup plugin loader.

cpex/framework/models.py

@@ -1329,17 +1329,7 @@ def to_json(self) -> dict[str, Any]:
         """
         # Get the base serialization from Pydantic
         data = self.model_dump(mode="json", exclude_none=False, exclude_unset=False)
-
-        # Explicitly remove any validator methods or callables that might have been included
-        # These are the @model_validator decorated methods that should not be serialized
-        methods_to_exclude = {
-            "_migrate_legacy_modes",
-            "check_url_or_script_filled",
-            "check_config_and_external",
-        }
-
-        # Filter out any methods or callables from the serialized data
-        return {k: v for k, v in data.items() if k not in methods_to_exclude and not callable(v)}
+        return data
 
 
 class PluginManifest(BaseModel):

tests/unit/cpex/framework/isolated/test_client.py

@@ -57,7 +57,7 @@ def mock_config(self, tmp_path):
     @pytest.fixture
     def plugin(self, mock_config, tmp_path):
         """Create an IsolatedVenvPlugin instance."""
-        plugin_instance = IsolatedVenvPlugin(mock_config)
+        plugin_instance = IsolatedVenvPlugin(mock_config, plugin_dirs=[tmp_path])
         # Override plugin_path to use tmp_path for testing
         plugin_instance.plugin_path = tmp_path / "test_plugin"
         return plugin_instance

tests/unit/cpex/framework/isolated/test_integration.py

@@ -13,11 +13,13 @@
 from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 import pytest
+import yaml
 
 from cpex.framework import GlobalContext, PluginManager
 from cpex.framework.hooks.tools import ToolPreInvokePayload
 from cpex.framework.isolated.client import IsolatedVenvPlugin
-from cpex.framework.models import PluginConfig
+from cpex.framework.loader.plugin import ALLOWED_PLUGIN_DIRS
+from cpex.framework.models import Config, PluginConfig
 
 
 class TestIsolatedPluginIntegration:
@@ -26,48 +28,42 @@ class TestIsolatedPluginIntegration:
     @pytest.fixture
     def integration_config_path(self, tmp_path):
         """Create a temporary config file for integration testing."""
-        config_content = """
-plugin_dirs:
-  - "xplugins"
 
-plugin_settings:
-  parallel_execution_within_band: true
-  plugin_timeout: 30
-  fail_on_plugin_error: false
-
-plugins:
-  - name: "test_isolated_plugin"
-    kind: "isolated_venv"
-    description: "Test isolated plugin"
-    version: "1.0.0"
-    author: "Test"
-    hooks: ["tool_pre_invoke"]
-    config:
-      class_name: "test_plugin.TestPlugin"
-      requirements_file: "requirements.txt"
-      script_path: "xplugins"
-"""
-        config_file = tmp_path / "test_config.yaml"
+        cfg = Config(plugins=[PluginConfig(name="test_isolated_plugin", kind="isolated_venv",description="Test isolated plugin",version="1.0.0",author="Test",hooks=["tool_pre_invoke"],
+                                     config={
+                                      "class_name": "test_plugin.TestPlugin",
+                                      "requirements_file": "requirements.txt"
+        })],plugin_dirs=[str((tmp_path / "xplugins").resolve())],
+        plugin_settings={
+            "parallel_execution_within_band": True,
+            "plugin_timeout": 30,
+            "fail_on_plugin_error": False
+        })
+        config_file = tmp_path / "xplugins" / "test_config.yaml"
+        class_root = tmp_path / "xplugins" / "test_plugin"
+        class_root.mkdir(parents=True, exist_ok=True)
+        dumped_cfg = cfg.model_dump(mode="json")
+        config_content = yaml.safe_dump(dumped_cfg, default_flow_style=False)
         config_file.write_text(config_content)
         return str(config_file)
 
     @pytest.mark.asyncio
     @patch("cpex.framework.isolated.client.VenvProcessCommunicator")
     @patch.object(IsolatedVenvPlugin, "create_venv")
     async def test_plugin_manager_with_isolated_plugin(
-        self, mock_create_venv, mock_comm_class, integration_config_path
+        self, mock_create_venv, mock_comm_class, integration_config_path, tmp_path
     ):
         """Test PluginManager loading and initializing an isolated plugin."""
         # Setup mocks
         mock_create_venv.return_value = None
         mock_comm = MagicMock()
         mock_comm.install_requirements = MagicMock()
         mock_comm_class.return_value = mock_comm
-        
-        # Create manager
-        manager = PluginManager(integration_config_path)
-        
-        await manager.initialize()
+        with patch('cpex.framework.loader.plugin.ALLOWED_PLUGIN_DIRS', { str((tmp_path / "xplugins" ).resolve())}):
+            # Create manager
+            manager = PluginManager(integration_config_path)
+            
+            await manager.initialize()
 
     @pytest.mark.asyncio
     @patch("cpex.framework.isolated.client.VenvProcessCommunicator")
@@ -96,35 +92,39 @@ async def test_isolated_plugin_full_lifecycle(self, mock_create_venv, mock_comm_
             "config": {
                 "class_name": "test_plugin.TestPlugin",
                 "requirements_file": "requirements.txt",
-                "script_path": "tests/unit/cpex/fixtures/plugins/isolated"
             }
         }
+        resolved_plugin_path = (tmp_path / "xplugins" ).resolve()
+        plugin_root = resolved_plugin_path / "test_plugin"
+        plugin_root.mkdir(parents=True, exist_ok=True)
+        # resolved_plugin_path.mkdir(parents=True, exist_ok=True)
+        with patch('cpex.framework.loader.plugin.ALLOWED_PLUGIN_DIRS', { str(resolved_plugin_path) }):
 
-        config = PluginConfig(**config_dict)
-        
-        # Create and initialize plugin
-        plugin = IsolatedVenvPlugin(config)
-        
-        with patch("cpex.framework.isolated.client.get_hook_registry") as mock_registry:
-            from cpex.framework.hooks.tools import ToolPreInvokeResult
-            mock_reg = MagicMock()
-            mock_reg.get_result_type.return_value = ToolPreInvokeResult
-            mock_reg.json_to_result = MagicMock()
-            mock_reg.json_to_result.return_value = ToolPreInvokeResult(continue_processing=True)
-            mock_registry.return_value = mock_reg
+            config = PluginConfig(**config_dict)
 
-            await plugin.initialize()
+            # Create and initialize plugin
+            plugin = IsolatedVenvPlugin(config, plugin_dirs=[resolved_plugin_path])
 
-            # Invoke hook
-            payload = ToolPreInvokePayload(name="test_tool", args={})
-            global_ctx = GlobalContext(request_id="req-123")
-            from cpex.framework.models import PluginContext
-            context = PluginContext(global_context=global_ctx)
-            
-            result = await plugin.invoke_hook("tool_pre_invoke", payload, context)
-            
-            assert result is not None
-            assert result.continue_processing is True
+            with patch("cpex.framework.isolated.client.get_hook_registry") as mock_registry:
+                from cpex.framework.hooks.tools import ToolPreInvokeResult
+                mock_reg = MagicMock()
+                mock_reg.get_result_type.return_value = ToolPreInvokeResult
+                mock_reg.json_to_result = MagicMock()
+                mock_reg.json_to_result.return_value = ToolPreInvokeResult(continue_processing=True)
+                mock_registry.return_value = mock_reg
+                
+                await plugin.initialize()
+                
+                # Invoke hook
+                payload = ToolPreInvokePayload(name="test_tool", args={})
+                global_ctx = GlobalContext(request_id="req-123")
+                from cpex.framework.models import PluginContext
+                context = PluginContext(global_context=global_ctx)
+                
+                result = await plugin.invoke_hook("tool_pre_invoke", payload, context)
+                
+                assert result is not None
+                assert result.continue_processing is True
 
     @pytest.mark.asyncio
     async def test_isolated_plugin_error_handling(self, tmp_path):
@@ -139,11 +139,15 @@ async def test_isolated_plugin_error_handling(self, tmp_path):
             "config": {
                 "class_name": "test_plugin.TestPlugin",
                 "requirements_file": "requirements.txt",
-                "script_path": "tests/unit/cpex/fixtures/plugins/isolated"
             }
         }
         config = PluginConfig(**config_dict)
-        plugin = IsolatedVenvPlugin(config)
+        resolved_plugin_path = (tmp_path / "xplugins" ).resolve()
+        cache_root = resolved_plugin_path / "test_plugin"
+        cache_root.mkdir(parents=True, exist_ok=True)
+        # resolved_plugin_path.mkdir(parents=True, exist_ok=True)
+
+        plugin = IsolatedVenvPlugin(config, plugin_dirs=[str(resolved_plugin_path)])
 
         # Try to invoke hook without initialization
         from cpex.framework.errors import PluginError
@@ -182,7 +186,11 @@ async def test_isolated_plugin_with_multiple_hooks(
         }
 
         config = PluginConfig(**config_dict)
-        plugin = IsolatedVenvPlugin(config)
+        resolved_plugin_path = (tmp_path / "xplugins" ).resolve()
+        cache_root = resolved_plugin_path / "test_plugin"
+        cache_root.mkdir(parents=True, exist_ok=True)
+
+        plugin = IsolatedVenvPlugin(config, plugin_dirs=[str(resolved_plugin_path)])
 
         await plugin.initialize()
 
@@ -266,11 +274,14 @@ def capture_task(script_path, task_data):
             "config": {
                 "class_name": "test_plugin.TestPlugin",
                 "requirements_file": "requirements.txt",
-                "script_path": "tests/unit/cpex/fixtures/plugins/isolated"
             }
         }
         config = PluginConfig(**config_dict)
-        plugin = IsolatedVenvPlugin(config)
+        resolved_plugin_path = (tmp_path / "xplugins" ).resolve()
+        cache_root = resolved_plugin_path / "test_plugin"
+        cache_root.mkdir(parents=True, exist_ok=True)
+
+        plugin = IsolatedVenvPlugin(config, plugin_dirs=[str(resolved_plugin_path)])
 
         await plugin.initialize()
 
@@ -334,11 +345,14 @@ async def test_isolated_plugin_violation_handling(
             "config": {
                 "class_name": "test_plugin.TestPlugin",
                 "requirements_file": "requirements.txt",
-                "script_path": "tests/unit/cpex/fixtures/plugins/isolated"
             }
         }
         config = PluginConfig(**config_dict)
-        plugin = IsolatedVenvPlugin(config)
+        resolved_plugin_path = (tmp_path / "xplugins" ).resolve()
+        cache_root = resolved_plugin_path / "test_plugin"
+        cache_root.mkdir(parents=True, exist_ok=True)
+
+        plugin = IsolatedVenvPlugin(config, plugin_dirs=[str(resolved_plugin_path)])
 
         await plugin.initialize()