"""
Widget Stress Test Plugin
Tests widget system under load by:
- Creating multiple widgets rapidly
- Testing layout arrangements
- Simulating user interactions
- Measuring performance
This plugin helps identify memory leaks, performance bottlenecks,
and stability issues in the widget system.
"""
import time
import random
from datetime import datetime
from typing import List, Dict
from dataclasses import dataclass
from core.base_plugin import BasePlugin
from core.api.widget_api import get_widget_api, WidgetType, WidgetConfig
@dataclass
class StressTestResult:
"""Result of a stress test operation."""
operation: str
count: int
duration_ms: float
success: bool
error: str = None
class WidgetStressTestPlugin(BasePlugin):
"""
Stress test suite for the WidgetAPI.
Tests include:
- Bulk widget creation
- Rapid show/hide cycles
- Layout stress testing
- Memory pressure simulation
"""
def __init__(self):
super().__init__()
self.widget_api = None
self.results: List[StressTestResult] = []
self.created_widgets: List[str] = []
self.main_widget = None
def initialize(self):
"""Initialize and run stress tests."""
self.widget_api = get_widget_api()
self._create_control_widget()
self._run_stress_tests()
def _create_control_widget(self):
"""Create main control widget for results."""
self.main_widget = self.widget_api.create_widget(
name="widget_stress_control",
title="๐ฅ Widget Stress Test Control",
size=(700, 500),
position=(100, 100),
widget_type=WidgetType.CONTROL
)
self.main_widget.show()
self._update_control_display()
def _update_control_display(self):
"""Update control widget with current results."""
try:
from PyQt6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QLabel,
QPushButton, QTextBrowser, QProgressBar, QGroupBox
)
from PyQt6.QtCore import Qt
container = QWidget()
main_layout = QVBoxLayout(container)
# Title
title = QLabel("๐ฅ Widget Stress Test Suite")
title.setStyleSheet("font-size: 18px; font-weight: bold; color: #ff8c42;")
main_layout.addWidget(title)
# Results display
self.results_display = QTextBrowser()
self.results_display.setHtml(self._generate_results_html())
main_layout.addWidget(self.results_display)
# Control buttons
btn_layout = QHBoxLayout()
self.btn_create = QPushButton("Create 10 Widgets")
self.btn_create.clicked.connect(lambda: self._run_test_create_batch(10))
btn_layout.addWidget(self.btn_create)
self.btn_stress = QPushButton("Run Full Stress Test")
self.btn_stress.clicked.connect(self._run_stress_tests)
btn_layout.addWidget(self.btn_stress)
self.btn_cleanup = QPushButton("Cleanup All")
self.btn_cleanup.clicked.connect(self._cleanup_all)
btn_layout.addWidget(self.btn_cleanup)
main_layout.addLayout(btn_layout)
# Status bar
self.status_label = QLabel("Ready")
main_layout.addWidget(self.status_label)
self.main_widget.set_content(container)
except ImportError as e:
print(f"Widget creation error: {e}")
def _generate_results_html(self) -> str:
"""Generate HTML results display."""
html = """
"""
if not self.results:
html += "No tests run yet. Click 'Run Full Stress Test' to begin.
"
else:
# Summary metrics
total_ops = len(self.results)
total_widgets = sum(r.count for r in self.results if r.operation == "create")
avg_time = sum(r.duration_ms for r in self.results) / len(self.results)
errors = sum(1 for r in self.results if not r.success)
html += f"""
{total_ops} Operations
{total_widgets} Widgets Created
{avg_time:.1f}ms Avg Time
{errors} Errors
| Operation | Count | Duration | Result |
|---|
"""
for result in self.results:
status_class = "success" if result.success else "error"
status_text = "โ" if result.success else "โ"
html += f"""
| {result.operation} |
{result.count} |
{result.duration_ms:.2f}ms |
{status_text} |
"""
html += "
"
return html
def _record_result(self, operation: str, count: int, duration_ms: float,
success: bool, error: str = None):
"""Record a test result."""
result = StressTestResult(
operation=operation,
count=count,
duration_ms=duration_ms,
success=success,
error=error
)
self.results.append(result)
self._update_control_display()
def _run_stress_tests(self):
"""Execute complete stress test suite."""
self.results.clear()
self._update_control_display()
# Test 1: Rapid creation
self._run_test_create_batch(10)
self._run_test_create_batch(20)
# Test 2: Layout stress
self._run_test_layout_stress()
# Test 3: Show/hide cycles
self._run_test_visibility_cycles()
# Test 4: Property modifications
self._run_test_property_modifications()
# Test 5: Concurrent operations
self._run_test_concurrent_operations()
# Final cleanup
self._cleanup_all()
if self.status_label:
self.status_label.setText("โ
All stress tests completed")
def _run_test_create_batch(self, count: int):
"""Test creating multiple widgets rapidly."""
start = time.time()
created = 0
error = None
try:
for i in range(count):
widget_name = f"stress_widget_{int(time.time()*1000)}_{i}"
widget = self.widget_api.create_widget(
name=widget_name,
title=f"Stress Test {i+1}",
size=(200, 150),
position=(random.randint(50, 800), random.randint(50, 600)),
widget_type=random.choice([WidgetType.MINI, WidgetType.CONTROL, WidgetType.CHART])
)
self.created_widgets.append(widget_name)
created += 1
except Exception as e:
error = str(e)
duration = (time.time() - start) * 1000
self._record_result("create", created, duration, error is None, error)
def _run_test_layout_stress(self):
"""Test layout arrangements under load."""
start = time.time()
error = None
try:
# Create some widgets first
for i in range(15):
widget = self.widget_api.create_widget(
name=f"layout_stress_{i}",
title=f"Layout {i}",
size=(180, 120)
)
self.created_widgets.append(f"layout_stress_{i}")
widget.show()
# Test different layouts rapidly
layouts = ["grid", "horizontal", "vertical", "cascade"]
for layout in layouts:
self.widget_api.arrange_widgets(layout=layout, spacing=random.randint(5, 20))
time.sleep(0.1)
# Test grid snapping
self.widget_api.snap_to_grid(grid_size=20)
except Exception as e:
error = str(e)
duration = (time.time() - start) * 1000
self._record_result("layout_stress", 15, duration, error is None, error)
def _run_test_visibility_cycles(self):
"""Test rapid show/hide cycles."""
start = time.time()
error = None
cycles = 0
try:
# Create test widgets
for i in range(5):
name = f"vis_cycle_{i}"
if not self.widget_api.widget_exists(name):
widget = self.widget_api.create_widget(
name=name,
title=f"Cycle {i}",
size=(150, 100)
)
self.created_widgets.append(name)
# Rapid show/hide cycles
for _ in range(10):
self.widget_api.hide_all_widgets()
time.sleep(0.05)
self.widget_api.show_all_widgets()
time.sleep(0.05)
cycles += 1
except Exception as e:
error = str(e)
duration = (time.time() - start) * 1000
self._record_result("visibility_cycles", cycles, duration, error is None, error)
def _run_test_property_modifications(self):
"""Test rapid property changes."""
start = time.time()
error = None
changes = 0
try:
# Create test widget
widget = self.widget_api.create_widget(
name="prop_test",
title="Property Test",
size=(250, 200)
)
self.created_widgets.append("prop_test")
widget.show()
# Rapid property changes
for i in range(20):
widget.set_opacity(random.uniform(0.3, 1.0))
widget.move(random.randint(100, 500), random.randint(100, 400))
widget.resize(random.randint(200, 400), random.randint(150, 300))
widget.set_locked(random.choice([True, False]))
widget.set_title(f"Property Test {i}")
changes += 5
except Exception as e:
error = str(e)
duration = (time.time() - start) * 1000
self._record_result("property_modifications", changes, duration, error is None, error)
def _run_test_concurrent_operations(self):
"""Test multiple operations in quick succession."""
start = time.time()
error = None
ops = 0
try:
# Create widgets while modifying others
for i in range(10):
# Create new
name = f"concurrent_{i}"
widget = self.widget_api.create_widget(
name=name,
title=f"Concurrent {i}",
size=(150, 100)
)
self.created_widgets.append(name)
widget.show()
ops += 1
# Modify existing
if self.widget_api.widget_exists("prop_test"):
w = self.widget_api.get_widget("prop_test")
w.set_opacity(random.uniform(0.5, 1.0))
ops += 1
# Get all widgets
all_widgets = self.widget_api.get_all_widgets()
ops += 1
except Exception as e:
error = str(e)
duration = (time.time() - start) * 1000
self._record_result("concurrent_operations", ops, duration, error is None, error)
def _cleanup_all(self):
"""Clean up all created widgets."""
start = time.time()
error = None
try:
# Close all widgets we created
for name in self.created_widgets:
try:
if self.widget_api.widget_exists(name):
self.widget_api.close_widget(name)
except:
pass
self.created_widgets.clear()
except Exception as e:
error = str(e)
duration = (time.time() - start) * 1000
self._record_result("cleanup", 0, duration, error is None, error)
if self.status_label:
self.status_label.setText(f"๐งน Cleanup completed in {duration:.1f}ms")
def shutdown(self):
"""Clean up on shutdown."""
self._cleanup_all()
if self.main_widget:
self.main_widget.close()
# Plugin entry point
plugin_class = WidgetStressTestPlugin