tegwick
096017b93f
Some checks failed
Test Suite / unit-tests (3.11) (push) Has been cancelled
Test Suite / unit-tests (3.12) (push) Has been cancelled
Test Suite / integration-tests (push) Has been cancelled
Test Suite / e2e-tests (push) Has been cancelled
Test Suite / performance-tests (push) Has been cancelled
Test Suite / code-quality (push) Has been cancelled
Test Suite / security-scan (push) Has been cancelled
Test Suite / test-summary (push) Has been cancelled
Separate capability-specific tests from core system tests to establish clear test organization and separation of concerns. ## Test Reorganization: - **markitect-content tests**: Moved 6 tests to capabilities/markitect-content/tests/ - **markitect-finance tests**: Moved 7 tests to markitect/finance/tests/ - **markitect-query tests**: Moved 1 test to markitect/query_paradigms/tests/ - **markitect-graphql tests**: Moved 2 tests to markitect/graphql/tests/ - **markitect-plugins tests**: Moved 2 tests to markitect/plugins/tests/ ## Makefile Updates: - **make test**: Excludes capability tests, runs only core system tests - **make test-capabilities**: Runs all capability tests - **make test-capability-***: Individual capability test targets - Updated all test targets (test-red, test-green, test-ultra-fast, test-perf) - Added capability test targets to help documentation ## Benefits: - Clear separation between core system tests and capability-specific tests - Faster core test execution (capability tests not run by default) - Individual capability testing for focused development - Supports future capability extraction workflow - Maintains capability test independence Test verification: - Core tests: 1291 tests (capability tests excluded) - Finance capability: 143 tests working independently - Content capability: 79 tests working independently 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
855 lines
27 KiB
Python
855 lines
27 KiB
Python
"""
|
|
Tests for Issue #19: Plugin Architecture and Extensions System
|
|
|
|
This module provides comprehensive tests for the MarkiTect plugin system
|
|
including plugin discovery, loading, management, and CLI integration.
|
|
"""
|
|
|
|
import pytest
|
|
import json
|
|
import tempfile
|
|
import os
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
from markitect.plugins import (
|
|
PluginManager,
|
|
BasePlugin,
|
|
ProcessorPlugin,
|
|
FormatterPlugin,
|
|
PluginType,
|
|
PluginMetadata,
|
|
plugin_registry,
|
|
register_plugin
|
|
)
|
|
from markitect.plugins.manager import PluginManager
|
|
from markitect.plugins.registry import PluginRegistry
|
|
|
|
|
|
class TestPluginArchitecture:
|
|
"""Test suite for plugin architecture components."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test environment."""
|
|
# Clear plugin registry for clean tests
|
|
plugin_registry.cleanup_all()
|
|
plugin_registry._plugins.clear()
|
|
plugin_registry._instances.clear()
|
|
plugin_registry._plugins_by_type.clear()
|
|
|
|
def teardown_method(self):
|
|
"""Clean up after tests."""
|
|
plugin_registry.cleanup_all()
|
|
plugin_registry._plugins.clear()
|
|
plugin_registry._instances.clear()
|
|
plugin_registry._plugins_by_type.clear()
|
|
|
|
|
|
class TestPluginBase:
|
|
"""Test base plugin functionality."""
|
|
|
|
def test_plugin_metadata_creation(self):
|
|
"""Test PluginMetadata creation and properties."""
|
|
metadata = PluginMetadata(
|
|
name="test_plugin",
|
|
version="1.0.0",
|
|
description="Test plugin",
|
|
author="Test Author",
|
|
plugin_type=PluginType.PROCESSOR,
|
|
dependencies=["dep1", "dep2"],
|
|
markitect_version=">=0.1.0"
|
|
)
|
|
|
|
assert metadata.name == "test_plugin"
|
|
assert metadata.version == "1.0.0"
|
|
assert metadata.description == "Test plugin"
|
|
assert metadata.author == "Test Author"
|
|
assert metadata.plugin_type == PluginType.PROCESSOR
|
|
assert metadata.dependencies == ["dep1", "dep2"]
|
|
assert metadata.markitect_version == ">=0.1.0"
|
|
|
|
def test_base_plugin_initialization(self):
|
|
"""Test BasePlugin initialization."""
|
|
|
|
class TestPlugin(BasePlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="test",
|
|
version="1.0.0",
|
|
description="Test",
|
|
plugin_type=PluginType.EXTENSION
|
|
)
|
|
|
|
config = {"option1": "value1", "option2": "value2"}
|
|
plugin = TestPlugin(config)
|
|
|
|
assert plugin.config == config
|
|
assert not plugin.is_initialized
|
|
|
|
def test_plugin_initialization_lifecycle(self):
|
|
"""Test plugin initialization and cleanup lifecycle."""
|
|
|
|
class TestPlugin(BasePlugin):
|
|
def __init__(self, config=None):
|
|
super().__init__(config)
|
|
self.initialized = False
|
|
self.cleaned_up = False
|
|
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="test",
|
|
version="1.0.0",
|
|
description="Test",
|
|
plugin_type=PluginType.EXTENSION
|
|
)
|
|
|
|
def _initialize(self):
|
|
self.initialized = True
|
|
|
|
def cleanup(self):
|
|
self.cleaned_up = True
|
|
|
|
plugin = TestPlugin()
|
|
assert not plugin.initialized
|
|
assert not plugin.is_initialized
|
|
|
|
# Test initialization
|
|
result = plugin.initialize()
|
|
assert result is True
|
|
assert plugin.initialized
|
|
assert plugin.is_initialized
|
|
|
|
# Test cleanup
|
|
plugin.cleanup()
|
|
assert plugin.cleaned_up
|
|
|
|
def test_plugin_initialization_failure(self):
|
|
"""Test plugin initialization failure handling."""
|
|
|
|
class FailingPlugin(BasePlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="failing",
|
|
version="1.0.0",
|
|
description="Failing plugin",
|
|
plugin_type=PluginType.EXTENSION
|
|
)
|
|
|
|
def _initialize(self):
|
|
raise Exception("Initialization failed")
|
|
|
|
plugin = FailingPlugin()
|
|
result = plugin.initialize()
|
|
assert result is False
|
|
assert not plugin.is_initialized
|
|
|
|
|
|
class TestProcessorPlugin:
|
|
"""Test processor plugin functionality."""
|
|
|
|
def test_processor_plugin_interface(self):
|
|
"""Test processor plugin interface implementation."""
|
|
|
|
class TestProcessor(ProcessorPlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="test_processor",
|
|
version="1.0.0",
|
|
description="Test processor",
|
|
plugin_type=PluginType.PROCESSOR
|
|
)
|
|
|
|
def process(self, content: str, **kwargs) -> str:
|
|
return content.upper()
|
|
|
|
processor = TestProcessor()
|
|
result = processor.process("hello world")
|
|
assert result == "HELLO WORLD"
|
|
|
|
# Test default can_process implementation
|
|
assert processor.can_process("any content")
|
|
|
|
def test_processor_plugin_with_options(self):
|
|
"""Test processor plugin with processing options."""
|
|
|
|
class ConfigurableProcessor(ProcessorPlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="configurable_processor",
|
|
version="1.0.0",
|
|
description="Configurable processor",
|
|
plugin_type=PluginType.PROCESSOR
|
|
)
|
|
|
|
def process(self, content: str, **kwargs) -> str:
|
|
if kwargs.get('uppercase', False):
|
|
content = content.upper()
|
|
if kwargs.get('reverse', False):
|
|
content = content[::-1]
|
|
return content
|
|
|
|
processor = ConfigurableProcessor()
|
|
|
|
# Test with no options
|
|
result = processor.process("hello")
|
|
assert result == "hello"
|
|
|
|
# Test with uppercase option
|
|
result = processor.process("hello", uppercase=True)
|
|
assert result == "HELLO"
|
|
|
|
# Test with both options
|
|
result = processor.process("hello", uppercase=True, reverse=True)
|
|
assert result == "OLLEH"
|
|
|
|
|
|
class TestFormatterPlugin:
|
|
"""Test formatter plugin functionality."""
|
|
|
|
def test_formatter_plugin_interface(self):
|
|
"""Test formatter plugin interface implementation."""
|
|
|
|
class TestFormatter(FormatterPlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="test_formatter",
|
|
version="1.0.0",
|
|
description="Test formatter",
|
|
plugin_type=PluginType.FORMATTER
|
|
)
|
|
|
|
def format(self, data, **kwargs) -> str:
|
|
return json.dumps(data, indent=kwargs.get('indent', 2))
|
|
|
|
def get_file_extension(self) -> str:
|
|
return '.json'
|
|
|
|
formatter = TestFormatter()
|
|
data = {"key": "value", "number": 42}
|
|
|
|
result = formatter.format(data)
|
|
parsed = json.loads(result)
|
|
assert parsed == data
|
|
|
|
extension = formatter.get_file_extension()
|
|
assert extension == '.json'
|
|
|
|
|
|
class TestPluginRegistry:
|
|
"""Test plugin registry functionality."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test environment."""
|
|
self.registry = PluginRegistry()
|
|
|
|
def test_plugin_registration(self):
|
|
"""Test plugin registration."""
|
|
|
|
class TestPlugin(BasePlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="test",
|
|
version="1.0.0",
|
|
description="Test",
|
|
plugin_type=PluginType.EXTENSION
|
|
)
|
|
|
|
# Test registration
|
|
name = self.registry.register(TestPlugin)
|
|
assert name == "TestPlugin"
|
|
assert "TestPlugin" in self.registry._plugins
|
|
|
|
# Test registration with custom name
|
|
custom_name = self.registry.register(TestPlugin, "custom_name")
|
|
assert custom_name == "custom_name"
|
|
assert "custom_name" in self.registry._plugins
|
|
|
|
def test_plugin_registration_duplicate_name(self):
|
|
"""Test plugin registration with duplicate name."""
|
|
|
|
class TestPlugin(BasePlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="test",
|
|
version="1.0.0",
|
|
description="Test",
|
|
plugin_type=PluginType.EXTENSION
|
|
)
|
|
|
|
self.registry.register(TestPlugin, "test_name")
|
|
|
|
# Should raise error for duplicate name
|
|
with pytest.raises(ValueError, match="already registered"):
|
|
self.registry.register(TestPlugin, "test_name")
|
|
|
|
def test_plugin_retrieval(self):
|
|
"""Test plugin retrieval from registry."""
|
|
|
|
class TestPlugin(BasePlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="test",
|
|
version="1.0.0",
|
|
description="Test",
|
|
plugin_type=PluginType.EXTENSION
|
|
)
|
|
|
|
self.registry.register(TestPlugin, "test_plugin")
|
|
|
|
# Test successful retrieval
|
|
plugin = self.registry.get_plugin("test_plugin")
|
|
assert plugin is not None
|
|
assert isinstance(plugin, TestPlugin)
|
|
|
|
# Test non-existent plugin
|
|
plugin = self.registry.get_plugin("non_existent")
|
|
assert plugin is None
|
|
|
|
def test_plugin_unregistration(self):
|
|
"""Test plugin unregistration."""
|
|
|
|
class TestPlugin(BasePlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="test",
|
|
version="1.0.0",
|
|
description="Test",
|
|
plugin_type=PluginType.EXTENSION
|
|
)
|
|
|
|
self.registry.register(TestPlugin, "test_plugin")
|
|
plugin = self.registry.get_plugin("test_plugin")
|
|
assert plugin is not None
|
|
|
|
# Test unregistration
|
|
result = self.registry.unregister("test_plugin")
|
|
assert result is True
|
|
|
|
# Plugin should no longer be available
|
|
plugin = self.registry.get_plugin("test_plugin")
|
|
assert plugin is None
|
|
|
|
# Test unregistering non-existent plugin
|
|
result = self.registry.unregister("non_existent")
|
|
assert result is False
|
|
|
|
def test_plugins_by_type(self):
|
|
"""Test retrieving plugins by type."""
|
|
|
|
class ProcessorPlugin1(ProcessorPlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="processor1",
|
|
version="1.0.0",
|
|
description="Processor 1",
|
|
plugin_type=PluginType.PROCESSOR
|
|
)
|
|
|
|
def process(self, content, **kwargs):
|
|
return content
|
|
|
|
class FormatterPlugin1(FormatterPlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="formatter1",
|
|
version="1.0.0",
|
|
description="Formatter 1",
|
|
plugin_type=PluginType.FORMATTER
|
|
)
|
|
|
|
def format(self, data, **kwargs):
|
|
return str(data)
|
|
|
|
def get_file_extension(self):
|
|
return '.txt'
|
|
|
|
self.registry.register(ProcessorPlugin1, "processor1")
|
|
self.registry.register(FormatterPlugin1, "formatter1")
|
|
|
|
# Test getting processors
|
|
processors = self.registry.get_plugins_by_type(PluginType.PROCESSOR)
|
|
assert "processor1" in processors
|
|
assert "formatter1" not in processors
|
|
|
|
# Test getting formatters
|
|
formatters = self.registry.get_plugins_by_type(PluginType.FORMATTER)
|
|
assert "formatter1" in formatters
|
|
assert "processor1" not in formatters
|
|
|
|
def test_list_plugins(self):
|
|
"""Test listing all plugins with metadata."""
|
|
|
|
class TestPlugin(BasePlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="test",
|
|
version="1.0.0",
|
|
description="Test plugin",
|
|
author="Test Author",
|
|
plugin_type=PluginType.EXTENSION
|
|
)
|
|
|
|
self.registry.register(TestPlugin, "test_plugin")
|
|
|
|
plugins = self.registry.list_plugins()
|
|
assert "test_plugin" in plugins
|
|
|
|
plugin_info = plugins["test_plugin"]
|
|
assert plugin_info["name"] == "test"
|
|
assert plugin_info["version"] == "1.0.0"
|
|
assert plugin_info["description"] == "Test plugin"
|
|
assert plugin_info["author"] == "Test Author"
|
|
assert plugin_info["type"] == "extension"
|
|
|
|
|
|
class TestPluginManager:
|
|
"""Test plugin manager functionality."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test environment."""
|
|
# Clear plugin registry
|
|
plugin_registry.cleanup_all()
|
|
plugin_registry._plugins.clear()
|
|
plugin_registry._instances.clear()
|
|
plugin_registry._plugins_by_type.clear()
|
|
|
|
def test_plugin_manager_initialization(self):
|
|
"""Test plugin manager initialization."""
|
|
manager = PluginManager()
|
|
assert manager.config is not None
|
|
assert isinstance(manager.plugin_directories, list)
|
|
|
|
def test_plugin_manager_with_config(self):
|
|
"""Test plugin manager with custom configuration."""
|
|
with tempfile.NamedTemporaryFile(mode='w', suffix='.yml', delete=False) as f:
|
|
f.write("""
|
|
plugin_directories:
|
|
- "custom_plugins"
|
|
auto_discover: false
|
|
plugins:
|
|
test_plugin:
|
|
enabled: true
|
|
""")
|
|
config_path = f.name
|
|
|
|
try:
|
|
manager = PluginManager(config_path)
|
|
assert "custom_plugins" in manager.config.get('plugin_directories', [])
|
|
assert manager.config.get('auto_discover') is False
|
|
assert 'test_plugin' in manager.config.get('plugins', {})
|
|
finally:
|
|
os.unlink(config_path)
|
|
|
|
def test_plugin_discovery_empty(self):
|
|
"""Test plugin discovery with no plugins."""
|
|
manager = PluginManager()
|
|
discovered = manager.discover_plugins()
|
|
# Should be a dictionary (empty or with built-ins)
|
|
assert isinstance(discovered, dict)
|
|
|
|
@patch('importlib.import_module')
|
|
def test_load_plugin_success(self, mock_import):
|
|
"""Test successful plugin loading."""
|
|
|
|
class TestPlugin(BasePlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="test",
|
|
version="1.0.0",
|
|
description="Test",
|
|
plugin_type=PluginType.EXTENSION
|
|
)
|
|
|
|
# Mock module with plugin
|
|
mock_module = Mock()
|
|
mock_module.TestPlugin = TestPlugin
|
|
mock_import.return_value = mock_module
|
|
|
|
manager = PluginManager()
|
|
|
|
# Manually add to discovered plugins
|
|
manager._discovered_plugins = {
|
|
"test_plugin": {
|
|
"module_name": "test_module",
|
|
"class_name": "TestPlugin"
|
|
}
|
|
}
|
|
|
|
plugin = manager.load_plugin("test_plugin")
|
|
assert plugin is not None
|
|
assert isinstance(plugin, TestPlugin)
|
|
|
|
def test_load_plugin_not_found(self):
|
|
"""Test loading non-existent plugin."""
|
|
manager = PluginManager()
|
|
plugin = manager.load_plugin("non_existent_plugin")
|
|
assert plugin is None
|
|
|
|
def test_get_plugins_by_type(self):
|
|
"""Test getting plugins by type."""
|
|
|
|
class TestProcessor(ProcessorPlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="test_processor",
|
|
version="1.0.0",
|
|
description="Test processor",
|
|
plugin_type=PluginType.PROCESSOR
|
|
)
|
|
|
|
def process(self, content, **kwargs):
|
|
return content
|
|
|
|
# Register plugin directly
|
|
plugin_registry.register(TestProcessor, "test_processor")
|
|
|
|
manager = PluginManager()
|
|
processors = manager.get_plugins_by_type(PluginType.PROCESSOR)
|
|
|
|
# Should have at least our test processor
|
|
assert len(processors) >= 1
|
|
assert any(isinstance(p, TestProcessor) for p in processors)
|
|
|
|
|
|
class TestPluginDecorator:
|
|
"""Test plugin registration decorator."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test environment."""
|
|
# Clear plugin registry
|
|
plugin_registry.cleanup_all()
|
|
plugin_registry._plugins.clear()
|
|
plugin_registry._instances.clear()
|
|
plugin_registry._plugins_by_type.clear()
|
|
|
|
def test_register_plugin_decorator(self):
|
|
"""Test @register_plugin decorator."""
|
|
|
|
@register_plugin("decorated_plugin")
|
|
class DecoratedPlugin(BasePlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="decorated",
|
|
version="1.0.0",
|
|
description="Decorated plugin",
|
|
plugin_type=PluginType.EXTENSION
|
|
)
|
|
|
|
# Plugin should be automatically registered
|
|
assert "decorated_plugin" in plugin_registry._plugins
|
|
|
|
# Should be able to retrieve it
|
|
plugin = plugin_registry.get_plugin("decorated_plugin")
|
|
assert plugin is not None
|
|
assert isinstance(plugin, DecoratedPlugin)
|
|
|
|
def test_register_plugin_decorator_no_name(self):
|
|
"""Test @register_plugin decorator without name."""
|
|
|
|
@register_plugin()
|
|
class AutoNamedPlugin(BasePlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="auto_named",
|
|
version="1.0.0",
|
|
description="Auto named plugin",
|
|
plugin_type=PluginType.EXTENSION
|
|
)
|
|
|
|
# Should use class name
|
|
assert "AutoNamedPlugin" in plugin_registry._plugins
|
|
|
|
|
|
class TestBuiltinPlugins:
|
|
"""Test built-in plugins."""
|
|
|
|
def test_json_formatter_plugin(self):
|
|
"""Test built-in JSON formatter plugin."""
|
|
from markitect.plugins.builtin.formatters import JsonFormatter
|
|
|
|
formatter = JsonFormatter()
|
|
assert formatter.metadata.plugin_type == PluginType.FORMATTER
|
|
|
|
data = {"key": "value", "number": 42}
|
|
result = formatter.format(data)
|
|
|
|
parsed = json.loads(result)
|
|
assert parsed == data
|
|
|
|
assert formatter.get_file_extension() == '.json'
|
|
|
|
def test_table_formatter_plugin(self):
|
|
"""Test built-in table formatter plugin."""
|
|
from markitect.plugins.builtin.formatters import TableFormatter
|
|
|
|
formatter = TableFormatter()
|
|
assert formatter.metadata.plugin_type == PluginType.FORMATTER
|
|
|
|
# Test with list of dictionaries
|
|
data = [
|
|
{"name": "John", "age": 30},
|
|
{"name": "Jane", "age": 25}
|
|
]
|
|
|
|
result = formatter.format(data)
|
|
assert "John" in result
|
|
assert "Jane" in result
|
|
assert "name" in result
|
|
assert "age" in result
|
|
|
|
assert formatter.get_file_extension() == '.txt'
|
|
|
|
def test_markdown_processor_plugin(self):
|
|
"""Test built-in markdown processor plugin."""
|
|
from markitect.plugins.builtin.processors import MarkdownProcessor
|
|
|
|
processor = MarkdownProcessor()
|
|
assert processor.metadata.plugin_type == PluginType.PROCESSOR
|
|
|
|
# Test basic processing
|
|
content = "# Header\n\nSome content\n"
|
|
result = processor.process(content)
|
|
assert isinstance(result, str)
|
|
|
|
# Test can_process
|
|
assert processor.can_process("# Markdown header")
|
|
assert processor.can_process("Some **bold** text")
|
|
|
|
|
|
class TestPluginCLIIntegration:
|
|
"""Test plugin CLI command integration."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test environment."""
|
|
# Clear plugin registry
|
|
plugin_registry.cleanup_all()
|
|
plugin_registry._plugins.clear()
|
|
plugin_registry._instances.clear()
|
|
plugin_registry._plugins_by_type.clear()
|
|
|
|
def test_plugin_list_command_import(self):
|
|
"""Test that plugin CLI commands can be imported."""
|
|
# This tests that the CLI commands are properly integrated
|
|
from markitect.cli import plugin_list, plugin_load, plugin_info
|
|
|
|
assert callable(plugin_list)
|
|
assert callable(plugin_load)
|
|
assert callable(plugin_info)
|
|
|
|
def test_plugin_type_enum_import(self):
|
|
"""Test that PluginType enum is accessible for CLI."""
|
|
from markitect.plugins.base import PluginType
|
|
|
|
# Test all plugin types are available
|
|
assert PluginType.PROCESSOR
|
|
assert PluginType.FORMATTER
|
|
assert PluginType.VALIDATOR
|
|
assert PluginType.EXPORTER
|
|
assert PluginType.GENERATOR
|
|
assert PluginType.IMPORTER
|
|
assert PluginType.TRANSFORMER
|
|
assert PluginType.EXTENSION
|
|
assert PluginType.BACKEND
|
|
assert PluginType.COMMAND
|
|
|
|
# Test values are strings
|
|
assert isinstance(PluginType.PROCESSOR.value, str)
|
|
|
|
|
|
class TestPluginErrorHandling:
|
|
"""Test plugin error handling and edge cases."""
|
|
|
|
def test_plugin_with_invalid_metadata(self):
|
|
"""Test plugin with invalid metadata."""
|
|
|
|
class BadMetadataPlugin(BasePlugin):
|
|
@property
|
|
def metadata(self):
|
|
# Missing required fields
|
|
return None
|
|
|
|
plugin = BadMetadataPlugin()
|
|
|
|
# Should handle gracefully
|
|
try:
|
|
plugin_registry.register(BadMetadataPlugin, "bad_plugin")
|
|
# Should not crash, might register as extension type
|
|
except Exception:
|
|
# Exception is acceptable for invalid metadata
|
|
pass
|
|
|
|
def test_plugin_initialization_with_bad_config(self):
|
|
"""Test plugin initialization with invalid configuration."""
|
|
|
|
class ConfigValidatingPlugin(BasePlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="config_validator",
|
|
version="1.0.0",
|
|
description="Config validating plugin",
|
|
plugin_type=PluginType.EXTENSION
|
|
)
|
|
|
|
def validate_config(self):
|
|
errors = []
|
|
if 'required_field' not in self.config:
|
|
errors.append("Missing required_field")
|
|
return errors
|
|
|
|
# Test with invalid config
|
|
plugin = ConfigValidatingPlugin({"wrong_field": "value"})
|
|
errors = plugin.validate_config()
|
|
assert len(errors) > 0
|
|
assert "required_field" in errors[0]
|
|
|
|
# Test with valid config
|
|
plugin = ConfigValidatingPlugin({"required_field": "value"})
|
|
errors = plugin.validate_config()
|
|
assert len(errors) == 0
|
|
|
|
def test_plugin_manager_with_invalid_config_file(self):
|
|
"""Test plugin manager with invalid configuration file."""
|
|
with tempfile.NamedTemporaryFile(mode='w', suffix='.yml', delete=False) as f:
|
|
f.write("invalid: yaml: content: [") # Invalid YAML
|
|
config_path = f.name
|
|
|
|
try:
|
|
# Should not crash, should use defaults
|
|
manager = PluginManager(config_path)
|
|
assert manager.config is not None
|
|
# Should fall back to defaults
|
|
assert 'plugin_directories' in manager.config
|
|
finally:
|
|
os.unlink(config_path)
|
|
|
|
|
|
class TestPluginIntegration:
|
|
"""Integration tests for the plugin system."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test environment."""
|
|
# Clear plugin registry
|
|
plugin_registry.cleanup_all()
|
|
plugin_registry._plugins.clear()
|
|
plugin_registry._instances.clear()
|
|
plugin_registry._plugins_by_type.clear()
|
|
|
|
def test_end_to_end_plugin_workflow(self):
|
|
"""Test complete plugin workflow from registration to usage."""
|
|
|
|
# 1. Create a plugin
|
|
@register_plugin("workflow_processor")
|
|
class WorkflowProcessor(ProcessorPlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="workflow_processor",
|
|
version="1.0.0",
|
|
description="End-to-end workflow processor",
|
|
plugin_type=PluginType.PROCESSOR
|
|
)
|
|
|
|
def process(self, content, **kwargs):
|
|
prefix = kwargs.get('prefix', self.config.get('prefix', ''))
|
|
return f"{prefix}{content}"
|
|
|
|
# 2. Verify registration
|
|
assert "workflow_processor" in plugin_registry._plugins
|
|
|
|
# 3. Create manager and load plugin
|
|
manager = PluginManager()
|
|
plugin = manager.load_plugin("workflow_processor", {"prefix": ">> "})
|
|
|
|
# 4. Use plugin
|
|
assert plugin is not None
|
|
result = plugin.process("Hello World")
|
|
assert result == ">> Hello World"
|
|
|
|
# 5. Verify plugin is in registry
|
|
assert plugin_registry.is_loaded("workflow_processor")
|
|
|
|
# 6. Get plugin by type
|
|
processors = manager.get_plugins_by_type(PluginType.PROCESSOR)
|
|
assert any(isinstance(p, WorkflowProcessor) for p in processors)
|
|
|
|
# 7. Unload plugin
|
|
success = manager.unload_plugin("workflow_processor")
|
|
assert success is True
|
|
assert not plugin_registry.is_loaded("workflow_processor")
|
|
|
|
def test_multiple_plugins_interaction(self):
|
|
"""Test interaction between multiple plugins."""
|
|
|
|
# Register multiple plugins
|
|
@register_plugin("upper_processor")
|
|
class UpperProcessor(ProcessorPlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="upper_processor",
|
|
version="1.0.0",
|
|
description="Uppercase processor",
|
|
plugin_type=PluginType.PROCESSOR
|
|
)
|
|
|
|
def process(self, content, **kwargs):
|
|
return content.upper()
|
|
|
|
@register_plugin("json_test_formatter")
|
|
class JsonTestFormatter(FormatterPlugin):
|
|
@property
|
|
def metadata(self):
|
|
return PluginMetadata(
|
|
name="json_test_formatter",
|
|
version="1.0.0",
|
|
description="JSON test formatter",
|
|
plugin_type=PluginType.FORMATTER
|
|
)
|
|
|
|
def format(self, data, **kwargs):
|
|
return json.dumps(data)
|
|
|
|
def get_file_extension(self):
|
|
return '.json'
|
|
|
|
manager = PluginManager()
|
|
|
|
# Load both plugins
|
|
processor = manager.load_plugin("upper_processor")
|
|
formatter = manager.load_plugin("json_test_formatter")
|
|
|
|
assert processor is not None
|
|
assert formatter is not None
|
|
|
|
# Use them together
|
|
processed = processor.process("hello world")
|
|
formatted = formatter.format({"result": processed})
|
|
|
|
data = json.loads(formatted)
|
|
assert data["result"] == "HELLO WORLD"
|
|
|
|
# Verify both are loaded
|
|
assert plugin_registry.is_loaded("upper_processor")
|
|
assert plugin_registry.is_loaded("json_test_formatter")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
pytest.main([__file__]) |