109 lines
3.1 KiB
Python
109 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
InstallUpdateThread 콜백 수정 테스트
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
|
|
# 경로 추가
|
|
current_dir = os.path.dirname(os.path.abspath(__file__))
|
|
if current_dir not in sys.path:
|
|
sys.path.append(current_dir)
|
|
|
|
from PySide6.QtCore import QThread, Signal
|
|
from src.img_module.image_processor_manager import ImageProcessorManager, ProcessorInfo
|
|
|
|
|
|
class TestInstallUpdateThread(QThread):
|
|
"""설치/업데이트를 별도 스레드에서 실행 (테스트용)"""
|
|
install_completed = Signal(bool, str)
|
|
progress_updated = Signal(str, int)
|
|
|
|
def __init__(self, manager, processor_info):
|
|
super().__init__()
|
|
self.manager = manager
|
|
self.processor_info = processor_info
|
|
|
|
def run(self):
|
|
"""설치/업데이트 실행"""
|
|
try:
|
|
# progress_callback을 위한 래퍼 함수 생성
|
|
def progress_callback(message: str, progress: int):
|
|
print(f"[PROGRESS] {progress}% - {message}")
|
|
self.progress_updated.emit(message, progress)
|
|
|
|
success = self.manager.download_and_install(
|
|
self.processor_info,
|
|
progress_callback=progress_callback
|
|
)
|
|
message = "설치가 완료되었습니다." if success else "설치에 실패했습니다."
|
|
self.install_completed.emit(success, message)
|
|
except Exception as e:
|
|
print(f"[ERROR] 설치 중 오류 발생: {e}")
|
|
self.install_completed.emit(False, f"설치 중 오류 발생: {e}")
|
|
|
|
|
|
def test_callback_fix():
|
|
"""콜백 수정 테스트"""
|
|
print("=== 콜백 수정 테스트 시작 ===")
|
|
|
|
# 더미 ProcessorInfo 생성
|
|
processor_info = ProcessorInfo(
|
|
id="test",
|
|
program_id="image_server",
|
|
version="1.0.0",
|
|
update_level="patch",
|
|
download_url="https://example.com/test.zip",
|
|
release_notes="Test release",
|
|
release_date=None,
|
|
is_mandatory=False,
|
|
created_at=None,
|
|
is_stable=True
|
|
)
|
|
|
|
# 매니저 생성 (실제로는 더 많은 파라미터가 필요하지만 테스트용)
|
|
try:
|
|
manager = ImageProcessorManager(
|
|
logger=None,
|
|
base_dir='.',
|
|
toggle_states={},
|
|
unwanted_words={},
|
|
authenticated_by_admin=False,
|
|
image_worker_fatal=None,
|
|
supabase_manager=None
|
|
)
|
|
|
|
# 스레드 생성 및 실행
|
|
thread = TestInstallUpdateThread(manager, processor_info)
|
|
|
|
# 시그널 연결
|
|
def on_progress(message, progress):
|
|
print(f"Progress received: {progress}% - {message}")
|
|
|
|
def on_completed(success, message):
|
|
print(f"Completed: {success} - {message}")
|
|
|
|
thread.progress_updated.connect(on_progress)
|
|
thread.install_completed.connect(on_completed)
|
|
|
|
print("스레드 시작...")
|
|
thread.run() # 직접 실행 (테스트용)
|
|
|
|
print("=== 콜백 수정 테스트 완료 ===")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"[ERROR] 테스트 실패: {e}")
|
|
return False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_callback_fix()
|
|
|
|
|
|
|
|
|
|
|