135 lines
3.9 KiB
Python
Executable File
135 lines
3.9 KiB
Python
Executable File
#!/usr/bin/env python
|
|
"""
|
|
Test to verify async loading doesn't block the main thread.
|
|
|
|
This test demonstrates that the UI remains responsive during image loading.
|
|
"""
|
|
|
|
import time
|
|
import sys
|
|
from pathlib import Path
|
|
from PyQt6.QtWidgets import QApplication
|
|
from PyQt6.QtCore import QTimer
|
|
|
|
from pyPhotoAlbum.async_backend import AsyncImageLoader, ImageCache, LoadPriority
|
|
|
|
|
|
def test_nonblocking_load():
|
|
"""Test that async image loading doesn't block the main thread"""
|
|
|
|
print("Testing non-blocking async image loading...")
|
|
|
|
# Track if main thread stays responsive
|
|
main_thread_ticks = []
|
|
|
|
def main_thread_tick():
|
|
"""This should continue running during async loads"""
|
|
main_thread_ticks.append(time.time())
|
|
print(f"✓ Main thread tick {len(main_thread_ticks)} (responsive!)")
|
|
|
|
# Create Qt application
|
|
app = QApplication(sys.argv)
|
|
|
|
# Create async loader
|
|
cache = ImageCache(max_memory_mb=128)
|
|
loader = AsyncImageLoader(cache=cache, max_workers=2)
|
|
|
|
# Track loaded images
|
|
loaded_images = []
|
|
|
|
def on_image_loaded(path, image, user_data):
|
|
loaded_images.append(path)
|
|
print(f"✓ Loaded: {path} (size: {image.size})")
|
|
|
|
def on_load_failed(path, error_msg, user_data):
|
|
print(f"✗ Failed: {path} - {error_msg}")
|
|
|
|
loader.image_loaded.connect(on_image_loaded)
|
|
loader.load_failed.connect(on_load_failed)
|
|
|
|
# Start the async loader
|
|
loader.start()
|
|
print("✓ Async loader started")
|
|
|
|
# Request some image loads (these would normally block for 50-500ms each)
|
|
test_images = [
|
|
Path("assets/sample1.jpg"),
|
|
Path("assets/sample2.jpg"),
|
|
Path("assets/sample3.jpg"),
|
|
]
|
|
|
|
print(f"\nRequesting {len(test_images)} image loads...")
|
|
for img_path in test_images:
|
|
loader.request_load(img_path, priority=LoadPriority.HIGH)
|
|
print(f" → Queued: {img_path}")
|
|
|
|
print("\nMain thread should remain responsive while images load in background...")
|
|
|
|
# Setup main thread ticker (should run continuously)
|
|
ticker = QTimer()
|
|
ticker.timeout.connect(main_thread_tick)
|
|
ticker.start(100) # Tick every 100ms
|
|
|
|
# Setup test timeout
|
|
def check_completion():
|
|
elapsed = time.time() - start_time
|
|
|
|
if len(loaded_images) >= len(test_images):
|
|
print(f"\n✓ All images loaded in {elapsed:.2f}s")
|
|
print(f"✓ Main thread ticked {len(main_thread_ticks)} times during loading")
|
|
|
|
if len(main_thread_ticks) >= 3:
|
|
print("✓ SUCCESS: Main thread remained responsive!")
|
|
else:
|
|
print("✗ FAIL: Main thread was blocked!")
|
|
|
|
# Cleanup
|
|
ticker.stop()
|
|
loader.stop()
|
|
app.quit()
|
|
|
|
elif elapsed > 10.0:
|
|
print(f"\n✗ Timeout: Only loaded {len(loaded_images)}/{len(test_images)} images")
|
|
ticker.stop()
|
|
loader.stop()
|
|
app.quit()
|
|
|
|
# Check completion every 200ms
|
|
completion_timer = QTimer()
|
|
completion_timer.timeout.connect(check_completion)
|
|
completion_timer.start(200)
|
|
|
|
start_time = time.time()
|
|
|
|
# Run Qt event loop (this should NOT block)
|
|
app.exec()
|
|
|
|
print("\nTest completed!")
|
|
|
|
# Report results
|
|
print(f"\nResults:")
|
|
print(f" Images loaded: {len(loaded_images)}/{len(test_images)}")
|
|
print(f" Main thread ticks: {len(main_thread_ticks)}")
|
|
print(f" Cache stats: {cache.get_stats()}")
|
|
|
|
return len(main_thread_ticks) >= 3 # Success if main thread ticked at least 3 times
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("=" * 60)
|
|
print("Async Non-Blocking Test")
|
|
print("=" * 60)
|
|
print()
|
|
|
|
success = test_nonblocking_load()
|
|
|
|
print()
|
|
print("=" * 60)
|
|
if success:
|
|
print("✓ TEST PASSED: Async loading is non-blocking")
|
|
else:
|
|
print("✗ TEST FAILED: Main thread was blocked")
|
|
print("=" * 60)
|
|
|
|
sys.exit(0 if success else 1)
|