mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-07 07:36:37 +00:00
ChuckandClaude Opus 5.5
7e066174d9
Whole-tree audit. Every symbol was checked against core, the plugin monorepo and all eight third-party plugins in plugins.json first. - Deprecate (removal 3.10.0) plugin-facing methods nothing calls: LogoDownloader bulk download, ConfigManager backup/secret wrappers, APIHelper extras, BackgroundDataService poll API, PluginManager / PluginStateManager info readers, and a few CacheManager, FontManager, BaseOddsManager, DynamicTeamResolver methods and PluginTestCase. plugin_api_usage.py learns their receiver names; DEPRECATIONS doc regenerated. - Remove core-internal dead code: CacheMetrics, Vegas status/stats plumbing, sync "new cycle" message (followers ignore unknown types), unused operation types, test-only PluginCatalog readers, IPC to_dict and ping, _parse_form_value, CacheStrategyProtocol, ErrorAggregator callbacks, duplicate web response helpers. - Web UI: drop never-mounted json-file-manager.js, the example widget, utils/error_handler.js, four uncalled PluginAPI methods, and 29 escapeHtml shims (call window.LEDEscape directly). Public globals, BaseWidget and widget names unchanged. - Remove six one-off scripts (owner decision) and the unused markupsafe and pytest-mock pins. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
"""
|
|
Tests for plugin operation queue.
|
|
"""
|
|
|
|
import unittest
|
|
import time
|
|
from src.plugin_system.operation_queue import PluginOperationQueue
|
|
from src.plugin_system.operation_types import OperationType, OperationStatus
|
|
|
|
|
|
class TestPluginOperationQueue(unittest.TestCase):
|
|
"""Test plugin operation queue."""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures."""
|
|
self.queue = PluginOperationQueue(max_history=10)
|
|
|
|
def tearDown(self):
|
|
"""Clean up test fixtures."""
|
|
self.queue.shutdown()
|
|
|
|
def test_enqueue_operation(self):
|
|
"""Test enqueuing an operation."""
|
|
operation_id = self.queue.enqueue_operation(
|
|
OperationType.INSTALL,
|
|
"test-plugin",
|
|
{"version": "1.0.0"}
|
|
)
|
|
|
|
self.assertIsNotNone(operation_id)
|
|
|
|
# Check operation status
|
|
operation = self.queue.get_operation_status(operation_id)
|
|
self.assertIsNotNone(operation)
|
|
self.assertEqual(operation.operation_type, OperationType.INSTALL)
|
|
self.assertEqual(operation.plugin_id, "test-plugin")
|
|
|
|
def test_prevent_concurrent_operations(self):
|
|
"""Test that concurrent operations on same plugin are prevented."""
|
|
# Enqueue first operation
|
|
op1_id = self.queue.enqueue_operation(
|
|
OperationType.INSTALL,
|
|
"test-plugin"
|
|
)
|
|
|
|
# Get the operation and ensure it's in PENDING status
|
|
op1 = self.queue.get_operation_status(op1_id)
|
|
self.assertIsNotNone(op1)
|
|
# The operation should be in PENDING status by default
|
|
|
|
# Try to enqueue second operation for same plugin
|
|
# This should fail if the first operation is still pending/running
|
|
# Note: Operations are processed asynchronously, so we need to check
|
|
# if the operation is still active. If it's already completed, the test
|
|
# behavior may differ. For this test, we'll verify the mechanism exists.
|
|
try:
|
|
self.queue.enqueue_operation(
|
|
OperationType.UNINSTALL,
|
|
"test-plugin"
|
|
)
|
|
# If no exception, the first operation may have completed
|
|
# This is acceptable behavior - the check only prevents truly concurrent operations
|
|
except ValueError:
|
|
# Expected behavior - concurrent operation prevented
|
|
pass
|
|
|
|
def test_operation_history(self):
|
|
"""Test operation history tracking."""
|
|
# Enqueue and complete an operation
|
|
operation_id = self.queue.enqueue_operation(
|
|
OperationType.INSTALL,
|
|
"test-plugin",
|
|
operation_callback=lambda op: {"success": True}
|
|
)
|
|
|
|
# Wait for operation to complete
|
|
time.sleep(0.5)
|
|
|
|
# Check history
|
|
history = self.queue._operation_history
|
|
self.assertGreater(len(history), 0)
|
|
|
|
# Find our operation in history
|
|
op_in_history = next(
|
|
(op for op in history if op.operation_id == operation_id),
|
|
None
|
|
)
|
|
self.assertIsNotNone(op_in_history)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|
|
|