449 lines
13 KiB
Python
449 lines
13 KiB
Python
"""
|
|
Integration tests for plugin lifecycle.
|
|
|
|
Tests cover:
|
|
- Plugin discovery and loading
|
|
- Plugin initialization and shutdown
|
|
- Plugin inter-communication via event bus
|
|
- Plugin settings persistence
|
|
"""
|
|
import sys
|
|
import unittest
|
|
import tempfile
|
|
import shutil
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
# Add project root to path
|
|
project_root = Path(__file__).parent.parent.parent
|
|
if str(project_root) not in sys.path:
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
# Reset singletons before importing
|
|
from core.event_bus import reset_event_bus
|
|
reset_event_bus()
|
|
|
|
from core.plugin_manager import PluginManager
|
|
from core.plugin_api import PluginAPI, get_api
|
|
from core.event_bus import get_event_bus
|
|
|
|
|
|
class MockOverlay:
|
|
"""Mock overlay window for testing."""
|
|
def __init__(self):
|
|
self.widgets = []
|
|
|
|
def add_widget(self, widget):
|
|
self.widgets.append(widget)
|
|
|
|
def remove_widget(self, widget):
|
|
if widget in self.widgets:
|
|
self.widgets.remove(widget)
|
|
|
|
|
|
class TestPluginLifecycle(unittest.TestCase):
|
|
"""Test complete plugin lifecycle."""
|
|
|
|
def setUp(self):
|
|
"""Set up test environment."""
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.overlay = MockOverlay()
|
|
|
|
# Reset singletons
|
|
reset_event_bus()
|
|
PluginAPI._instance = None
|
|
|
|
def tearDown(self):
|
|
"""Clean up test environment."""
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
# Reset singletons
|
|
reset_event_bus()
|
|
PluginAPI._instance = None
|
|
|
|
def test_plugin_manager_initialization(self):
|
|
"""Test plugin manager initializes correctly."""
|
|
pm = PluginManager(self.overlay)
|
|
|
|
self.assertEqual(pm.overlay, self.overlay)
|
|
self.assertEqual(pm.plugins, {})
|
|
self.assertIsInstance(pm.config, dict)
|
|
|
|
def test_plugin_discovery_finds_plugins(self):
|
|
"""Test that plugin discovery finds available plugins."""
|
|
pm = PluginManager(self.overlay)
|
|
|
|
discovered = pm.discover_plugins()
|
|
|
|
# Should find at least some plugins (depends on plugins directory)
|
|
self.assertIsInstance(discovered, list)
|
|
|
|
def test_plugin_class_registration(self):
|
|
"""Test that plugin classes are registered during discovery."""
|
|
pm = PluginManager(self.overlay)
|
|
|
|
discovered = pm.discover_plugins()
|
|
|
|
for plugin_class in discovered:
|
|
plugin_id = f"{plugin_class.__module__}.{plugin_class.__name__}"
|
|
self.assertIn(plugin_id, pm.plugin_classes)
|
|
|
|
|
|
class TestPluginEventBusIntegration(unittest.TestCase):
|
|
"""Test plugin integration with event bus."""
|
|
|
|
def setUp(self):
|
|
"""Set up test environment."""
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.overlay = MockOverlay()
|
|
|
|
# Reset singletons
|
|
reset_event_bus()
|
|
PluginAPI._instance = None
|
|
|
|
self.event_bus = get_event_bus()
|
|
self.api = get_api()
|
|
|
|
def tearDown(self):
|
|
"""Clean up test environment."""
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
# Reset singletons
|
|
reset_event_bus()
|
|
PluginAPI._instance = None
|
|
|
|
def test_api_singleton(self):
|
|
"""Test that PluginAPI is a singleton."""
|
|
api1 = get_api()
|
|
api2 = get_api()
|
|
|
|
self.assertIs(api1, api2)
|
|
|
|
def test_event_bus_singleton(self):
|
|
"""Test that EventBus is a singleton."""
|
|
bus1 = get_event_bus()
|
|
bus2 = get_event_bus()
|
|
|
|
self.assertIs(bus1, bus2)
|
|
|
|
def test_api_publishes_events(self):
|
|
"""Test that API can publish events."""
|
|
from core.event_bus import SkillGainEvent
|
|
|
|
received_events = []
|
|
|
|
def handler(event):
|
|
received_events.append(event)
|
|
|
|
self.event_bus.subscribe_typed(SkillGainEvent, handler)
|
|
|
|
event = SkillGainEvent(
|
|
skill_name="Rifle",
|
|
skill_value=25.0,
|
|
gain_amount=0.01
|
|
)
|
|
|
|
self.api.publish_typed(event)
|
|
|
|
# Note: Event bus is async, so events might not be received immediately
|
|
# For testing, we use publish_sync
|
|
self.event_bus.publish_sync(event)
|
|
|
|
self.assertEqual(len(received_events), 1)
|
|
self.assertEqual(received_events[0].skill_name, "Rifle")
|
|
|
|
def test_api_shared_data(self):
|
|
"""Test API shared data functionality."""
|
|
self.api.set_data("test_key", "test_value")
|
|
|
|
result = self.api.get_data("test_key")
|
|
|
|
self.assertEqual(result, "test_value")
|
|
|
|
def test_api_shared_data_default(self):
|
|
"""Test API shared data with default."""
|
|
result = self.api.get_data("nonexistent_key", "default")
|
|
|
|
self.assertEqual(result, "default")
|
|
|
|
|
|
class TestPluginAPIRegistration(unittest.TestCase):
|
|
"""Test plugin API registration."""
|
|
|
|
def setUp(self):
|
|
"""Set up test environment."""
|
|
# Reset singletons
|
|
reset_event_bus()
|
|
PluginAPI._instance = None
|
|
|
|
self.api = get_api()
|
|
|
|
def tearDown(self):
|
|
"""Clean up test environment."""
|
|
# Reset singletons
|
|
reset_event_bus()
|
|
PluginAPI._instance = None
|
|
|
|
def test_register_api_endpoint(self):
|
|
"""Test registering an API endpoint."""
|
|
from core.plugin_api import APIEndpoint, APIType
|
|
|
|
def test_handler():
|
|
return "test_result"
|
|
|
|
endpoint = APIEndpoint(
|
|
name="test_api",
|
|
api_type=APIType.UTILITY,
|
|
description="Test API",
|
|
handler=test_handler,
|
|
plugin_id="test_plugin"
|
|
)
|
|
|
|
result = self.api.register_api(endpoint)
|
|
|
|
self.assertTrue(result)
|
|
self.assertIn("test_plugin:test_api", self.api.apis)
|
|
|
|
def test_call_registered_api(self):
|
|
"""Test calling a registered API."""
|
|
from core.plugin_api import APIEndpoint, APIType
|
|
|
|
def test_handler(arg1, arg2):
|
|
return f"Result: {arg1}, {arg2}"
|
|
|
|
endpoint = APIEndpoint(
|
|
name="test_api",
|
|
api_type=APIType.UTILITY,
|
|
description="Test API",
|
|
handler=test_handler,
|
|
plugin_id="test_plugin"
|
|
)
|
|
|
|
self.api.register_api(endpoint)
|
|
result = self.api.call_api("test_plugin", "test_api", "hello", "world")
|
|
|
|
self.assertEqual(result, "Result: hello, world")
|
|
|
|
def test_unregister_api(self):
|
|
"""Test unregistering an API endpoint."""
|
|
from core.plugin_api import APIEndpoint, APIType
|
|
|
|
endpoint = APIEndpoint(
|
|
name="test_api",
|
|
api_type=APIType.UTILITY,
|
|
description="Test API",
|
|
handler=lambda: None,
|
|
plugin_id="test_plugin"
|
|
)
|
|
|
|
self.api.register_api(endpoint)
|
|
self.assertIn("test_plugin:test_api", self.api.apis)
|
|
|
|
self.api.unregister_api("test_plugin", "test_api")
|
|
|
|
self.assertNotIn("test_plugin:test_api", self.api.apis)
|
|
|
|
def test_find_apis_by_type(self):
|
|
"""Test finding APIs by type."""
|
|
from core.plugin_api import APIEndpoint, APIType
|
|
|
|
endpoint1 = APIEndpoint(
|
|
name="ocr_api",
|
|
api_type=APIType.OCR,
|
|
description="OCR API",
|
|
handler=lambda: None,
|
|
plugin_id="plugin1"
|
|
)
|
|
|
|
endpoint2 = APIEndpoint(
|
|
name="log_api",
|
|
api_type=APIType.LOG,
|
|
description="Log API",
|
|
handler=lambda: None,
|
|
plugin_id="plugin2"
|
|
)
|
|
|
|
self.api.register_api(endpoint1)
|
|
self.api.register_api(endpoint2)
|
|
|
|
ocr_apis = self.api.find_apis(APIType.OCR)
|
|
|
|
self.assertEqual(len(ocr_apis), 1)
|
|
self.assertEqual(ocr_apis[0].name, "ocr_api")
|
|
|
|
|
|
class TestPluginDataStoreIntegration(unittest.TestCase):
|
|
"""Test plugin integration with data store."""
|
|
|
|
def setUp(self):
|
|
"""Set up test environment."""
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
|
|
# Reset singletons
|
|
from core.data_store import DataStore
|
|
DataStore._instance = None
|
|
|
|
def tearDown(self):
|
|
"""Clean up test environment."""
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
# Reset singletons
|
|
from core.data_store import DataStore
|
|
DataStore._instance = None
|
|
|
|
def test_data_store_save_and_load(self):
|
|
"""Test saving and loading data from data store."""
|
|
from core.data_store import DataStore
|
|
|
|
ds = DataStore(data_dir=self.temp_dir)
|
|
|
|
# Save data
|
|
result = ds.save("test_plugin", "key1", {"value": 42})
|
|
self.assertTrue(result)
|
|
|
|
# Load data
|
|
loaded = ds.load("test_plugin", "key1")
|
|
self.assertEqual(loaded, {"value": 42})
|
|
|
|
def test_data_store_plugin_isolation(self):
|
|
"""Test that plugins have isolated data stores."""
|
|
from core.data_store import DataStore
|
|
|
|
ds = DataStore(data_dir=self.temp_dir)
|
|
|
|
ds.save("plugin1", "shared_key", "plugin1_value")
|
|
ds.save("plugin2", "shared_key", "plugin2_value")
|
|
|
|
value1 = ds.load("plugin1", "shared_key")
|
|
value2 = ds.load("plugin2", "shared_key")
|
|
|
|
self.assertEqual(value1, "plugin1_value")
|
|
self.assertEqual(value2, "plugin2_value")
|
|
|
|
def test_data_store_get_all_keys(self):
|
|
"""Test getting all keys for a plugin."""
|
|
from core.data_store import DataStore
|
|
|
|
ds = DataStore(data_dir=self.temp_dir)
|
|
|
|
ds.save("test_plugin", "key1", "value1")
|
|
ds.save("test_plugin", "key2", "value2")
|
|
ds.save("test_plugin", "key3", "value3")
|
|
|
|
keys = ds.get_all_keys("test_plugin")
|
|
|
|
self.assertEqual(len(keys), 3)
|
|
self.assertIn("key1", keys)
|
|
self.assertIn("key2", keys)
|
|
self.assertIn("key3", keys)
|
|
|
|
|
|
class TestPluginSettingsIntegration(unittest.TestCase):
|
|
"""Test plugin integration with settings."""
|
|
|
|
def setUp(self):
|
|
"""Set up test environment."""
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.config_file = Path(self.temp_dir) / "settings.json"
|
|
|
|
def tearDown(self):
|
|
"""Clean up test environment."""
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_settings_enable_disable_plugin(self):
|
|
"""Test enabling and disabling plugins via settings."""
|
|
from core.settings import Settings
|
|
|
|
settings = Settings(config_file=str(self.config_file))
|
|
|
|
# Initially not enabled
|
|
self.assertFalse(settings.is_plugin_enabled("test_plugin"))
|
|
|
|
# Enable
|
|
settings.enable_plugin("test_plugin")
|
|
self.assertTrue(settings.is_plugin_enabled("test_plugin"))
|
|
|
|
# Disable
|
|
settings.disable_plugin("test_plugin")
|
|
self.assertFalse(settings.is_plugin_enabled("test_plugin"))
|
|
|
|
def test_settings_persistence(self):
|
|
"""Test that settings persist to file."""
|
|
from core.settings import Settings
|
|
|
|
# Create and modify settings
|
|
settings1 = Settings(config_file=str(self.config_file))
|
|
settings1.enable_plugin("test_plugin")
|
|
settings1.set("custom_key", "custom_value")
|
|
|
|
# Load settings again
|
|
settings2 = Settings(config_file=str(self.config_file))
|
|
|
|
self.assertTrue(settings2.is_plugin_enabled("test_plugin"))
|
|
self.assertEqual(settings2.get("custom_key"), "custom_value")
|
|
|
|
|
|
class TestEndToEndPluginWorkflow(unittest.TestCase):
|
|
"""Test end-to-end plugin workflows."""
|
|
|
|
def setUp(self):
|
|
"""Set up test environment."""
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.overlay = MockOverlay()
|
|
|
|
# Reset all singletons
|
|
reset_event_bus()
|
|
PluginAPI._instance = None
|
|
from core.data_store import DataStore
|
|
DataStore._instance = None
|
|
|
|
def tearDown(self):
|
|
"""Clean up test environment."""
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
# Reset all singletons
|
|
reset_event_bus()
|
|
PluginAPI._instance = None
|
|
from core.data_store import DataStore
|
|
DataStore._instance = None
|
|
|
|
def test_full_plugin_workflow(self):
|
|
"""Test complete plugin workflow from discovery to event handling."""
|
|
# Create plugin manager
|
|
pm = PluginManager(self.overlay)
|
|
|
|
# Discover plugins
|
|
discovered = pm.discover_plugins()
|
|
|
|
# Get API
|
|
api = get_api()
|
|
|
|
# Verify all systems are connected
|
|
self.assertIsNotNone(pm)
|
|
self.assertIsNotNone(api)
|
|
self.assertIsNotNone(get_event_bus())
|
|
|
|
# Test that we can register an API from a "plugin"
|
|
from core.plugin_api import APIEndpoint, APIType
|
|
|
|
def utility_function():
|
|
return "utility_result"
|
|
|
|
endpoint = APIEndpoint(
|
|
name="utility",
|
|
api_type=APIType.UTILITY,
|
|
description="Utility API",
|
|
handler=utility_function,
|
|
plugin_id="test_plugin"
|
|
)
|
|
|
|
api.register_api(endpoint)
|
|
|
|
# Verify we can call it
|
|
result = api.call_api("test_plugin", "utility")
|
|
self.assertEqual(result, "utility_result")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|