""" Tests for PluginLoader. Tests plugin directory discovery, module loading, and class instantiation. """ import subprocess import pytest from unittest.mock import MagicMock, patch from src.plugin_system.plugin_loader import PluginLoader from src.exceptions import PluginError class TestPluginLoader: """Test PluginLoader functionality.""" @pytest.fixture def plugin_loader(self): """Create a PluginLoader instance.""" return PluginLoader() @pytest.fixture def tmp_plugins_dir(self, tmp_path): """Create a temporary plugins directory.""" plugins_dir = tmp_path / "plugins" plugins_dir.mkdir() return plugins_dir def test_init(self): """Test PluginLoader initialization.""" loader = PluginLoader() assert loader.logger is not None assert loader._loaded_modules == {} def test_find_plugin_directory_direct_path(self, plugin_loader, tmp_plugins_dir): """Test finding plugin directory by direct path.""" plugin_dir = tmp_plugins_dir / "test_plugin" plugin_dir.mkdir() result = plugin_loader.find_plugin_directory( "test_plugin", tmp_plugins_dir ) assert result == plugin_dir def test_find_plugin_directory_with_prefix(self, plugin_loader, tmp_plugins_dir): """Test finding plugin directory with ledmatrix- prefix.""" plugin_dir = tmp_plugins_dir / "ledmatrix-test_plugin" plugin_dir.mkdir() result = plugin_loader.find_plugin_directory( "test_plugin", tmp_plugins_dir ) assert result == plugin_dir def test_find_plugin_directory_from_mapping(self, plugin_loader, tmp_plugins_dir): """Test finding plugin directory from provided mapping.""" plugin_dir = tmp_plugins_dir / "custom_plugin_name" plugin_dir.mkdir() plugin_directories = { "test_plugin": plugin_dir } result = plugin_loader.find_plugin_directory( "test_plugin", tmp_plugins_dir, plugin_directories=plugin_directories ) assert result == plugin_dir def test_find_plugin_directory_not_found(self, plugin_loader, tmp_plugins_dir): """Test finding non-existent plugin directory.""" result = plugin_loader.find_plugin_directory( "nonexistent_plugin", tmp_plugins_dir ) assert result is None @patch('importlib.util.spec_from_file_location') @patch('importlib.util.module_from_spec') def test_load_module(self, mock_module_from_spec, mock_spec_from_file, plugin_loader, tmp_plugins_dir): """Test loading a plugin module.""" plugin_dir = tmp_plugins_dir / "test_plugin" plugin_dir.mkdir() plugin_file = plugin_dir / "manager.py" plugin_file.write_text("# Plugin code") mock_spec = MagicMock() mock_spec.loader = MagicMock() mock_spec_from_file.return_value = mock_spec mock_module = MagicMock() mock_module_from_spec.return_value = mock_module result = plugin_loader.load_module("test_plugin", plugin_dir, "manager.py") assert result == mock_module mock_spec_from_file.assert_called_once() mock_module_from_spec.assert_called_once_with(mock_spec) def test_load_module_invalid_file(self, plugin_loader, tmp_plugins_dir): """Test loading invalid plugin module.""" plugin_dir = tmp_plugins_dir / "test_plugin" plugin_dir.mkdir() # Don't create the entry file with pytest.raises(PluginError, match="Entry point file not found"): plugin_loader.load_module("test_plugin", plugin_dir, "nonexistent.py") def test_get_plugin_class(self, plugin_loader): """Test getting plugin class from module.""" # Create a real class for testing class TestPlugin: pass mock_module = MagicMock() mock_module.Plugin = TestPlugin result = plugin_loader.get_plugin_class("test_plugin", mock_module, "Plugin") assert result == TestPlugin def test_get_plugin_class_not_found(self, plugin_loader): """Test getting non-existent plugin class from module.""" mock_module = MagicMock() mock_module.__name__ = "test_module" # Use delattr to properly remove the attribute if hasattr(mock_module, 'Plugin'): delattr(mock_module, 'Plugin') with pytest.raises(PluginError, match="Class.*not found"): plugin_loader.get_plugin_class("test_plugin", mock_module, "Plugin") def test_instantiate_plugin(self, plugin_loader): """Test instantiating a plugin class.""" mock_class = MagicMock() mock_instance = MagicMock() mock_class.return_value = mock_instance config = {"test": "config"} display_manager = MagicMock() cache_manager = MagicMock() plugin_manager = MagicMock() result = plugin_loader.instantiate_plugin( "test_plugin", mock_class, config, display_manager, cache_manager, plugin_manager ) assert result == mock_instance # Plugin class is called with keyword arguments mock_class.assert_called_once_with( plugin_id="test_plugin", config=config, display_manager=display_manager, cache_manager=cache_manager, plugin_manager=plugin_manager ) def test_instantiate_plugin_error(self, plugin_loader): """Test error handling when instantiating plugin class.""" mock_class = MagicMock() mock_class.side_effect = Exception("Instantiation error") with pytest.raises(PluginError, match="Failed to instantiate"): plugin_loader.instantiate_plugin( "test_plugin", mock_class, {}, MagicMock(), MagicMock(), MagicMock() ) @patch('subprocess.run') def test_install_dependencies(self, mock_subprocess, plugin_loader, tmp_plugins_dir): """Test installing plugin dependencies.""" plugin_dir = tmp_plugins_dir / "test_plugin" plugin_dir.mkdir() requirements_file = plugin_dir / "requirements.txt" requirements_file.write_text("package1==1.0.0\npackage2>=2.0.0\n") mock_subprocess.return_value = MagicMock(returncode=0) result = plugin_loader.install_dependencies(plugin_dir, "test_plugin", plugins_dir=tmp_plugins_dir) assert result is True mock_subprocess.assert_called_once() @patch('subprocess.run') def test_install_dependencies_no_requirements(self, mock_subprocess, plugin_loader, tmp_plugins_dir): """Test when no requirements.txt exists.""" plugin_dir = tmp_plugins_dir / "test_plugin" plugin_dir.mkdir() result = plugin_loader.install_dependencies(plugin_dir, "test_plugin", plugins_dir=tmp_plugins_dir) assert result is True mock_subprocess.assert_not_called() @patch('subprocess.run') def test_install_dependencies_failure(self, mock_subprocess, plugin_loader, tmp_plugins_dir): """Test handling dependency installation failure.""" plugin_dir = tmp_plugins_dir / "test_plugin" plugin_dir.mkdir() requirements_file = plugin_dir / "requirements.txt" requirements_file.write_text("package1==1.0.0\n") mock_subprocess.return_value = MagicMock(returncode=1) result = plugin_loader.install_dependencies(plugin_dir, "test_plugin", plugins_dir=tmp_plugins_dir) assert result is False @patch('src.plugin_system.plugin_loader.requirements_are_satisfied', return_value=False) @patch('subprocess.run') def test_install_dependencies_retries_with_ignore_installed_on_apt_conflict( self, mock_subprocess, mock_satisfied, plugin_loader, tmp_plugins_dir ): """An apt-managed package with no pip RECORD file triggers a retry with --ignore-installed rather than silently assuming the old version satisfies the requirement. requirements_are_satisfied() is mocked False here because this scenario is exactly the case where the installed (apt) version does NOT satisfy the pin — that's why pip attempts a reinstall in the first place.""" plugin_dir = tmp_plugins_dir / "test_plugin" plugin_dir.mkdir() requirements_file = plugin_dir / "requirements.txt" requirements_file.write_text("requests>=2.33.0,<3.0.0\n") first_attempt = MagicMock( returncode=1, stderr="ERROR: Cannot uninstall requests 2.32.3\nuninstall-no-record-file" ) retry_attempt = MagicMock(returncode=0, stderr="") mock_subprocess.side_effect = [first_attempt, retry_attempt] result = plugin_loader.install_dependencies(plugin_dir, "test_plugin", plugins_dir=tmp_plugins_dir) assert result is True assert mock_subprocess.call_count == 2 retry_cmd = mock_subprocess.call_args_list[1][0][0] assert "--ignore-installed" in retry_cmd @patch('src.plugin_system.plugin_loader.requirements_are_satisfied', return_value=False) @patch('subprocess.run') def test_install_dependencies_apt_conflict_retry_also_fails( self, mock_subprocess, mock_satisfied, plugin_loader, tmp_plugins_dir ): """Still tolerates the failure (returns True) if the --ignore-installed retry itself fails, matching the prior soft-fallback behavior.""" plugin_dir = tmp_plugins_dir / "test_plugin" plugin_dir.mkdir() requirements_file = plugin_dir / "requirements.txt" requirements_file.write_text("requests>=2.33.0,<3.0.0\n") first_attempt = MagicMock( returncode=1, stderr="ERROR: Cannot uninstall requests 2.32.3\nuninstall-no-record-file" ) retry_attempt = MagicMock(returncode=1, stderr="some other pip error") mock_subprocess.side_effect = [first_attempt, retry_attempt] result = plugin_loader.install_dependencies(plugin_dir, "test_plugin", plugins_dir=tmp_plugins_dir) assert result is True assert mock_subprocess.call_count == 2 @patch('src.plugin_system.plugin_loader.requirements_are_satisfied', return_value=False) @patch('subprocess.run') def test_install_dependencies_apt_conflict_retry_times_out( self, mock_subprocess, mock_satisfied, plugin_loader, tmp_plugins_dir ): """A retry timeout must be tolerated the same way as a retry failure (return True), not propagate to the outer TimeoutExpired handler and return False.""" plugin_dir = tmp_plugins_dir / "test_plugin" plugin_dir.mkdir() requirements_file = plugin_dir / "requirements.txt" requirements_file.write_text("requests>=2.33.0,<3.0.0\n") first_attempt = MagicMock( returncode=1, stderr="ERROR: Cannot uninstall requests 2.32.3\nuninstall-no-record-file" ) mock_subprocess.side_effect = [ first_attempt, subprocess.TimeoutExpired(cmd="pip", timeout=300), ] result = plugin_loader.install_dependencies(plugin_dir, "test_plugin", plugins_dir=tmp_plugins_dir) assert result is True assert mock_subprocess.call_count == 2 @patch('subprocess.run') def test_install_dependencies_already_satisfied_skips_pip(self, mock_subprocess, plugin_loader, tmp_plugins_dir): """A requirement already satisfied in the current environment shouldn't invoke pip.""" plugin_dir = tmp_plugins_dir / "test_plugin" plugin_dir.mkdir() requirements_file = plugin_dir / "requirements.txt" requirements_file.write_text("pytest>=1.0\n") result = plugin_loader.install_dependencies(plugin_dir, "test_plugin", plugins_dir=tmp_plugins_dir) assert result is True mock_subprocess.assert_not_called() def test_install_dependencies_requires_plugins_dir(self, plugin_loader, tmp_plugins_dir): """plugins_dir is a required argument, not an optional trust-me flag -- calling without it must fail loudly (TypeError) rather than silently falling back to trusting plugin_dir unchecked.""" plugin_dir = tmp_plugins_dir / "test_plugin" plugin_dir.mkdir() with pytest.raises(TypeError): plugin_loader.install_dependencies(plugin_dir, "test_plugin") @patch('subprocess.run') def test_install_dependencies_rejects_path_outside_plugins_dir( self, mock_subprocess, plugin_loader, tmp_path, tmp_plugins_dir ): """A plugin_dir that doesn't actually live inside plugins_dir (e.g. a manifest-derived id crafted to traverse elsewhere) must be rejected rather than read from -- this is the path-injection containment check CodeQL flagged as missing.""" outside_dir = tmp_path / "outside" outside_dir.mkdir() (outside_dir / "requirements.txt").write_text("requests>=2.0\n") result = plugin_loader.install_dependencies(outside_dir, "evil_plugin", plugins_dir=tmp_plugins_dir) assert result is False mock_subprocess.assert_not_called()