mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 14:25:08 +00:00
ChuckandClaude Opus 5.5
b518c51679
- load_plugin: an on_enable() that raises unregisters the instance, so the next load retries instead of returning True "already loaded". - get_plugin_info: guard plugin.get_info(); one plugin raising no longer breaks /api/v3/plugins/installed. - plugin_state.json and the operation history are written with atomic_write_text under their lock. - plugin_loader: module-level lock serialises pip installs across the parallel startup loaders. - store_manager._install_via_download: extract dir cleanup moved to finally. - Test doubles: draw_image() warns (DeprecationWarning; the real DisplayManager has none), MockDisplayManager.draw_text accepts the real signature's optional params, VisualTestDisplayManager logs draw errors at WARNING. - Docs/comments: compatibility.py method name, PluginState.LOADED meaning, brittle schema count, why _report_skip_once uses setdefault. - Remove unused PluginOperationQueue.get_active_operations(). Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
391 lines
15 KiB
Python
391 lines
15 KiB
Python
"""
|
|
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()
|
|
|
|
@patch('src.plugin_system.plugin_loader.requirements_are_satisfied', return_value=False)
|
|
def test_install_dependencies_never_runs_pip_concurrently(
|
|
self, mock_satisfied, tmp_plugins_dir
|
|
):
|
|
"""Startup loads plugins on a thread pool; two pip processes writing
|
|
the same site-packages at once can corrupt it, so installs for
|
|
different plugins must run one at a time."""
|
|
import threading
|
|
import time
|
|
|
|
state = {'running': 0, 'peak': 0}
|
|
guard = threading.Lock()
|
|
|
|
def fake_pip(*args, **kwargs):
|
|
with guard:
|
|
state['running'] += 1
|
|
state['peak'] = max(state['peak'], state['running'])
|
|
time.sleep(0.05)
|
|
with guard:
|
|
state['running'] -= 1
|
|
return MagicMock(returncode=0, stderr="")
|
|
|
|
plugin_dirs = []
|
|
for n in range(4):
|
|
d = tmp_plugins_dir / f"plugin_{n}"
|
|
d.mkdir()
|
|
(d / "requirements.txt").write_text("package1==1.0.0\n")
|
|
plugin_dirs.append(d)
|
|
|
|
results = []
|
|
|
|
def install(d):
|
|
results.append(PluginLoader().install_dependencies(
|
|
d, d.name, plugins_dir=tmp_plugins_dir))
|
|
|
|
with patch('subprocess.run', side_effect=fake_pip) as mock_subprocess:
|
|
threads = [threading.Thread(target=install, args=(d,)) for d in plugin_dirs]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
assert results == [True] * 4
|
|
assert mock_subprocess.call_count == 4
|
|
assert state['peak'] == 1
|