324 lines
9.0 KiB
Python
324 lines
9.0 KiB
Python
"""
|
|
Integration tests for new EU-Utility features.
|
|
"""
|
|
|
|
import sys
|
|
import time
|
|
import tempfile
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
# Add parent to path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from core.plugin_api import PluginAPI
|
|
|
|
|
|
def test_auto_updater():
|
|
"""Test AutoUpdater plugin."""
|
|
print("\n" + "="*60)
|
|
print("TEST: AutoUpdater Plugin")
|
|
print("="*60)
|
|
|
|
from plugins.auto_updater import AutoUpdaterPlugin, VersionInfo
|
|
|
|
# Test version comparison
|
|
v1 = VersionInfo.from_string("1.2.3")
|
|
v2 = VersionInfo.from_string("1.2.4")
|
|
assert v1 < v2, "Version comparison failed"
|
|
|
|
v3 = VersionInfo.from_string("2.0.0-beta")
|
|
assert v1 < v3, "Prerelease version comparison failed"
|
|
|
|
# Test plugin instantiation
|
|
plugin = AutoUpdaterPlugin()
|
|
assert plugin.name == "auto_updater"
|
|
assert plugin.version == "1.0.0"
|
|
|
|
# Test configuration
|
|
plugin.set_config({"channel": "beta"})
|
|
assert plugin.get_config()["channel"] == "beta"
|
|
|
|
print("✓ AutoUpdater tests passed")
|
|
return True
|
|
|
|
|
|
def test_plugin_marketplace():
|
|
"""Test PluginMarketplace plugin."""
|
|
print("\n" + "="*60)
|
|
print("TEST: PluginMarketplace Plugin")
|
|
print("="*60)
|
|
|
|
from plugins.plugin_marketplace import PluginMarketplacePlugin, MarketplacePlugin
|
|
|
|
# Test plugin instantiation
|
|
plugin = PluginMarketplacePlugin()
|
|
assert plugin.name == "plugin_marketplace"
|
|
|
|
# Test category retrieval (will use cache)
|
|
# Note: This will fail without network, but tests the structure
|
|
try:
|
|
categories = plugin.get_categories()
|
|
print(f" Categories: {categories}")
|
|
except Exception as e:
|
|
print(f" (Network unavailable for live test: {e})")
|
|
|
|
# Test installed plugin tracking
|
|
assert plugin.get_installed_plugins() == []
|
|
|
|
print("✓ PluginMarketplace tests passed")
|
|
return True
|
|
|
|
|
|
def test_cloud_sync():
|
|
"""Test CloudSync plugin."""
|
|
print("\n" + "="*60)
|
|
print("TEST: CloudSync Plugin")
|
|
print("="*60)
|
|
|
|
from plugins.cloud_sync import CloudSyncPlugin, SyncConfig, CloudProvider
|
|
|
|
# Test plugin instantiation
|
|
plugin = CloudSyncPlugin()
|
|
assert plugin.name == "cloud_sync"
|
|
|
|
# Test configuration
|
|
config = SyncConfig(
|
|
enabled=True,
|
|
provider="custom",
|
|
encrypt_data=True,
|
|
)
|
|
plugin.set_sync_config(config)
|
|
assert plugin.get_sync_config().enabled == True
|
|
|
|
# Test provider config
|
|
plugin.set_provider_config(CloudProvider.CUSTOM, {
|
|
"upload_url": "https://test.com/upload",
|
|
"api_key": "test_key",
|
|
})
|
|
|
|
print("✓ CloudSync tests passed")
|
|
return True
|
|
|
|
|
|
def test_stats_dashboard():
|
|
"""Test StatsDashboard plugin."""
|
|
print("\n" + "="*60)
|
|
print("TEST: StatsDashboard Plugin")
|
|
print("="*60)
|
|
|
|
from plugins.stats_dashboard import StatsDashboardPlugin
|
|
|
|
# Create temp directory for test data
|
|
temp_dir = tempfile.mkdtemp()
|
|
|
|
try:
|
|
# Test plugin instantiation
|
|
plugin = StatsDashboardPlugin()
|
|
assert plugin.name == "stats_dashboard"
|
|
|
|
# Override data dir
|
|
plugin._config["data_dir"] = temp_dir
|
|
plugin._data_dir = Path(temp_dir)
|
|
|
|
# Start plugin
|
|
plugin.on_start()
|
|
|
|
# Test metric recording
|
|
plugin.record_counter("test_counter", 5)
|
|
assert plugin.get_counter("test_counter") == 5
|
|
|
|
plugin.record_gauge("test_gauge", 42.5)
|
|
assert plugin.get_gauge("test_gauge") == 42.5
|
|
|
|
plugin.record_histogram("test_histogram", 100)
|
|
plugin.record_histogram("test_histogram", 200)
|
|
stats = plugin.get_histogram_stats("test_histogram")
|
|
assert stats["count"] == 2
|
|
|
|
# Test event recording
|
|
plugin.record_event("test", "test_event", {"key": "value"})
|
|
events = plugin.get_events(source="test")
|
|
assert len(events) == 1
|
|
|
|
# Test timing
|
|
with plugin.time_operation("test_operation"):
|
|
time.sleep(0.01)
|
|
|
|
# Test report generation
|
|
report = plugin.generate_report()
|
|
assert "system_health" in report
|
|
|
|
# Test dashboard summary
|
|
summary = plugin.get_dashboard_summary()
|
|
assert "uptime" in summary
|
|
|
|
# Stop plugin
|
|
plugin.on_stop()
|
|
|
|
print("✓ StatsDashboard tests passed")
|
|
return True
|
|
|
|
finally:
|
|
shutil.rmtree(temp_dir, ignore_errors=True)
|
|
|
|
|
|
def test_import_export():
|
|
"""Test ImportExport plugin."""
|
|
print("\n" + "="*60)
|
|
print("TEST: ImportExport Plugin")
|
|
print("="*60)
|
|
|
|
from plugins.import_export import (
|
|
ImportExportPlugin, ExportProfile, ExportFormat, ImportMode
|
|
)
|
|
|
|
# Create temp directories
|
|
temp_dir = tempfile.mkdtemp()
|
|
data_dir = Path(temp_dir) / "data"
|
|
data_dir.mkdir()
|
|
|
|
try:
|
|
# Test plugin instantiation
|
|
plugin = ImportExportPlugin()
|
|
assert plugin.name == "import_export"
|
|
|
|
# Override directories
|
|
plugin._config["export_dir"] = str(data_dir / "exports")
|
|
plugin._config["import_dir"] = str(data_dir / "imports")
|
|
plugin._config["temp_dir"] = str(data_dir / "temp")
|
|
plugin._data_dir = data_dir
|
|
|
|
# Create some test data
|
|
(data_dir / "test_config.json").write_text('{"setting": "value"}')
|
|
|
|
# Start plugin
|
|
plugin.on_start()
|
|
|
|
# Test export profiles
|
|
profiles = plugin.get_export_profiles()
|
|
assert "full" in profiles
|
|
assert "minimal" in profiles
|
|
|
|
# Test custom profile creation
|
|
custom = plugin.create_custom_profile(
|
|
name="custom",
|
|
include_settings=True,
|
|
include_plugins=False,
|
|
)
|
|
assert custom.name == "custom"
|
|
|
|
# Test export (minimal profile to avoid dependency on other plugins)
|
|
result = plugin.export_data(
|
|
profile="minimal",
|
|
format=ExportFormat.JSON,
|
|
)
|
|
assert result.success
|
|
assert Path(result.filepath).exists()
|
|
|
|
# Test validation
|
|
validation = plugin.validate_import_file(result.filepath)
|
|
assert validation["valid"]
|
|
|
|
# Test backup creation (use backup_ prefix so list_backups finds it)
|
|
backup = plugin.create_backup("backup_test")
|
|
assert backup.success
|
|
|
|
# Test listing backups
|
|
backups = plugin.list_backups()
|
|
assert len(backups) >= 1
|
|
|
|
print("✓ ImportExport tests passed")
|
|
return True
|
|
|
|
finally:
|
|
shutil.rmtree(temp_dir, ignore_errors=True)
|
|
|
|
|
|
def test_plugin_integration():
|
|
"""Test plugins work with PluginAPI."""
|
|
print("\n" + "="*60)
|
|
print("TEST: Plugin Integration with PluginAPI")
|
|
print("="*60)
|
|
|
|
api = PluginAPI()
|
|
|
|
# Load all new plugins
|
|
from plugins.auto_updater import AutoUpdaterPlugin
|
|
from plugins.plugin_marketplace import PluginMarketplacePlugin
|
|
from plugins.cloud_sync import CloudSyncPlugin
|
|
from plugins.stats_dashboard import StatsDashboardPlugin
|
|
from plugins.import_export import ImportExportPlugin
|
|
|
|
# Load plugins
|
|
updater = api.load_plugin(AutoUpdaterPlugin)
|
|
marketplace = api.load_plugin(PluginMarketplacePlugin)
|
|
sync = api.load_plugin(CloudSyncPlugin)
|
|
stats = api.load_plugin(StatsDashboardPlugin)
|
|
ie = api.load_plugin(ImportExportPlugin)
|
|
|
|
# Verify all loaded
|
|
assert len(api.get_all_plugins()) == 5
|
|
|
|
# Test plugin info
|
|
info = api.get_plugin_info()
|
|
plugin_names = [p["name"] for p in info]
|
|
assert "auto_updater" in plugin_names
|
|
assert "plugin_marketplace" in plugin_names
|
|
assert "cloud_sync" in plugin_names
|
|
assert "stats_dashboard" in plugin_names
|
|
assert "import_export" in plugin_names
|
|
|
|
# Start all
|
|
api.start_all()
|
|
|
|
# Verify all initialized
|
|
for plugin in api.get_all_plugins():
|
|
assert plugin.is_initialized()
|
|
|
|
# Stop all
|
|
api.stop_all()
|
|
|
|
print("✓ Plugin integration tests passed")
|
|
return True
|
|
|
|
|
|
def run_all_tests():
|
|
"""Run all integration tests."""
|
|
print("\n" + "="*60)
|
|
print("EU-UTILITY FEATURE INTEGRATION TESTS")
|
|
print("="*60)
|
|
|
|
tests = [
|
|
("AutoUpdater", test_auto_updater),
|
|
("PluginMarketplace", test_plugin_marketplace),
|
|
("CloudSync", test_cloud_sync),
|
|
("StatsDashboard", test_stats_dashboard),
|
|
("ImportExport", test_import_export),
|
|
("Plugin Integration", test_plugin_integration),
|
|
]
|
|
|
|
passed = 0
|
|
failed = 0
|
|
|
|
for name, test_func in tests:
|
|
try:
|
|
if test_func():
|
|
passed += 1
|
|
except Exception as e:
|
|
print(f"✗ {name} test failed: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
failed += 1
|
|
|
|
print("\n" + "="*60)
|
|
print(f"RESULTS: {passed} passed, {failed} failed")
|
|
print("="*60)
|
|
|
|
return failed == 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
success = run_all_tests()
|
|
sys.exit(0 if success else 1)
|