webdrop-bridge/test_update_no_hang.py

#!/usr/bin/env python
"""Test script to verify the update feature no longer hangs the UI.

This script demonstrates that the update download happens in a background
thread and doesn't block the UI thread.
"""

import asyncio
import logging
from pathlib import Path
from unittest.mock import MagicMock, Mock, patch

from PySide6.QtCore import QCoreApplication, QThread, QTimer

from webdrop_bridge.config import Config
from webdrop_bridge.core.updater import Release, UpdateManager
from webdrop_bridge.ui.main_window import MainWindow, UpdateDownloadWorker

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)


def test_update_download_runs_in_background():
    """Verify that update download runs in a background thread."""
    print("\n=== Testing Update Download Background Thread ===\n")
    
    app = QCoreApplication.instance() or QCoreApplication([])
    
    # Create a mock release
    release = Release(
        tag_name="v0.0.2",
        name="Release 0.0.2",
        version="0.0.2",
        body="Test release notes",
        assets=[{"name": "installer.msi", "browser_download_url": "http://example.com/installer.msi"}],
        published_at="2026-01-30T00:00:00Z"
    )
    
    # Create a mock update manager
    manager = Mock(spec=UpdateManager)
    
    # Track if download_update was called
    download_called = False
    download_thread_id = None
    
    async def mock_download(rel):
        nonlocal download_called, download_thread_id
        download_called = True
        download_thread_id = QThread.currentThreadId()
        # Simulate network operation
        await asyncio.sleep(0.1)
        return Path("/tmp/fake_installer.msi")
    
    async def mock_verify(file_path, rel):
        nonlocal download_thread_id
        await asyncio.sleep(0.1)
        return True
    
    manager.download_update = mock_download
    manager.verify_checksum = mock_verify
    
    # Create the worker
    worker = UpdateDownloadWorker(manager, release, "0.0.1")
    
    # Track signals
    signals_emitted = []
    worker.download_complete.connect(lambda p: signals_emitted.append(("complete", p)))
    worker.download_failed.connect(lambda e: signals_emitted.append(("failed", e)))
    worker.finished.connect(lambda: signals_emitted.append(("finished",)))
    
    # Create a thread and move worker to it
    thread = QThread()
    worker.moveToThread(thread)
    
    # Track if worker runs in different thread
    main_thread_id = QThread.currentThreadId()
    worker_thread_id = None
    
    def on_worker_run_started():
        nonlocal worker_thread_id
        worker_thread_id = QThread.currentThreadId()
        logger.info(f"Worker running in thread: {worker_thread_id}")
        logger.info(f"Main thread: {main_thread_id}")
    
    thread.started.connect(on_worker_run_started)
    thread.started.connect(worker.run)
    
    # Start the thread and process events until done
    thread.start()
    
    # Wait for completion with timeout
    start_time = asyncio.get_event_loop().time() if hasattr(asyncio.get_event_loop(), 'time') else 0
    while not download_called and len(signals_emitted) < 3:
        app.processEvents()
        QTimer.singleShot(10, app.quit)
        app.exec()
        if len(signals_emitted) >= 3:
            break
    
    # Cleanup
    thread.quit()
    thread.wait()
    
    # Verify results
    print(f"\n✓ Download called: {download_called}")
    print(f"✓ Signals emitted: {len(signals_emitted)}")
    
    # Check if completion signal was emitted (shows async operations completed)
    has_complete_or_failed = any(sig[0] in ("complete", "failed") for sig in signals_emitted)
    has_finished = any(sig[0] == "finished" for sig in signals_emitted)
    
    print(f"✓ Completion/Failed signal emitted: {has_complete_or_failed}")
    print(f"✓ Finished signal emitted: {has_finished}")
    
    if has_complete_or_failed and has_finished:
        print("\n✅ SUCCESS: Update download runs asynchronously without blocking UI!")
        return True
    else:
        print("\n❌ FAILED: Signals not emitted properly")
        print(f"   Signals: {signals_emitted}")
        return False


def test_update_download_worker_exists():
    """Verify that UpdateDownloadWorker class exists and has correct signals."""
    print("\n=== Testing UpdateDownloadWorker Class ===\n")
    
    # Check class exists
    assert hasattr(UpdateDownloadWorker, '__init__'), "UpdateDownloadWorker missing __init__"
    print("✓ UpdateDownloadWorker class exists")
    
    # Check signals
    required_signals = ['download_complete', 'download_failed', 'update_status', 'finished']
    for signal_name in required_signals:
        assert hasattr(UpdateDownloadWorker, signal_name), f"Missing signal: {signal_name}"
        print(f"✓ Signal '{signal_name}' defined")
    
    # Check methods
    assert hasattr(UpdateDownloadWorker, 'run'), "UpdateDownloadWorker missing run method"
    print("✓ Method 'run' defined")
    
    print("\n✅ SUCCESS: UpdateDownloadWorker properly implemented!")
    return True


def test_main_window_uses_async_download():
    """Verify that MainWindow uses async download instead of blocking."""
    print("\n=== Testing MainWindow Async Download Integration ===\n")
    
    # Check that _perform_update_async exists (new async version)
    assert hasattr(MainWindow, '_perform_update_async'), "MainWindow missing _perform_update_async"
    print("✓ Method '_perform_update_async' exists (new async version)")
    
    # Check that old blocking _perform_update is gone
    assert not hasattr(MainWindow, '_perform_update'), \
        "MainWindow still has old blocking _perform_update method"
    print("✓ Old blocking '_perform_update' method removed")
    
    # Check download/failed handlers exist
    assert hasattr(MainWindow, '_on_download_complete'), "MainWindow missing _on_download_complete"
    assert hasattr(MainWindow, '_on_download_failed'), "MainWindow missing _on_download_failed"
    print("✓ Download completion handlers exist")
    
    print("\n✅ SUCCESS: MainWindow properly integrated with async download!")
    return True


if __name__ == "__main__":
    print("\n" + "="*60)
    print("UPDATE FEATURE FIX VERIFICATION")
    print("="*60)
    
    try:
        # Test 1: Worker exists
        test1 = test_update_download_worker_exists()
        
        # Test 2: MainWindow integration
        test2 = test_main_window_uses_async_download()
        
        # Test 3: Async operation
        test3 = test_update_download_runs_in_background()
        
        print("\n" + "="*60)
        if test1 and test2 and test3:
            print("✅ ALL TESTS PASSED - UPDATE FEATURE HANG FIXED!")
            print("="*60 + "\n")
            print("Summary of changes:")
            print("- Created UpdateDownloadWorker class for async downloads")
            print("- Moved blocking operations from UI thread to background thread")
            print("- Added handlers for download completion/failure")
            print("- UI now stays responsive during update download")
        else:
            print("❌ SOME TESTS FAILED")
            print("="*60 + "\n")
    except Exception as e:
        print(f"\n❌ ERROR: {e}")
        import traceback
        traceback.print_exc()