338 lines
11 KiB
Python
338 lines
11 KiB
Python
"""设置页认证更新配置、自动检查和后台线程测试。"""
|
|
|
|
import os
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
from PyQt5.QtCore import QTimer
|
|
from PyQt5.QtTest import QTest
|
|
from PyQt5.QtWidgets import QApplication, QLineEdit
|
|
|
|
from src.mock_admin_gateway import MockAdminGateway
|
|
from src.settings_repository import SettingsRepository
|
|
from src.settings_ui import SettingsPage
|
|
from src.update_service import (
|
|
DEFAULT_UPDATE_MANIFEST_URL,
|
|
DEFAULT_UPDATE_USERNAME,
|
|
LEGACY_DEFAULT_UPDATE_MANIFEST_URL,
|
|
UPDATE_MANIFEST_SETTING,
|
|
UPDATE_USERNAME_SETTING,
|
|
UpdateCheckResult,
|
|
UpdateInfo,
|
|
)
|
|
|
|
|
|
MANIFEST_URL = "https://updates.example.test/autobuy%E2%80%94%E2%80%94manifest.json"
|
|
TEST_USERNAME = "release-reader"
|
|
TEST_PASSWORD = "unit-test-password"
|
|
|
|
|
|
class FakeCredentialStore:
|
|
def __init__(self, saved=None):
|
|
self.saved = saved
|
|
self.save_count = 0
|
|
|
|
def read(self):
|
|
return self.saved
|
|
|
|
def save(self, username, password):
|
|
self.saved = (username, password)
|
|
self.save_count += 1
|
|
|
|
|
|
class FakeUpdateService:
|
|
def __init__(self, result=None, delay=0.0, download_error=None):
|
|
self.result = result or UpdateCheckResult("0.1.0", "0.1.0", False)
|
|
self.delay = delay
|
|
self.download_error = download_error
|
|
self.check_count = 0
|
|
self.download_count = 0
|
|
self.check_credentials = None
|
|
self.download_credentials = None
|
|
|
|
def check(
|
|
self,
|
|
manifest_url,
|
|
current_version="0.1.0",
|
|
is_cancelled=None,
|
|
credentials=None,
|
|
):
|
|
self.check_count += 1
|
|
self.manifest_url = manifest_url
|
|
self.check_credentials = credentials
|
|
if self.delay:
|
|
time.sleep(self.delay)
|
|
return self.result
|
|
|
|
def download_and_stage(
|
|
self,
|
|
update,
|
|
is_cancelled=None,
|
|
on_progress=None,
|
|
credentials=None,
|
|
):
|
|
self.download_count += 1
|
|
self.download_credentials = credentials
|
|
if on_progress is not None:
|
|
on_progress(50)
|
|
if self.download_error is not None:
|
|
raise self.download_error
|
|
if on_progress is not None:
|
|
on_progress(100)
|
|
return Path("app.new")
|
|
|
|
|
|
class FakeButton:
|
|
def setText(self, _text):
|
|
pass
|
|
|
|
def setFocus(self):
|
|
pass
|
|
|
|
|
|
class AcceptDownloadMessageBox:
|
|
def __init__(self, _title, _content, _parent):
|
|
self.yesButton = FakeButton()
|
|
self.cancelButton = FakeButton()
|
|
|
|
def exec(self):
|
|
return True
|
|
|
|
|
|
class UpdateUiEventTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.app = QApplication.instance() or QApplication([])
|
|
|
|
def setUp(self):
|
|
self.temporary_directory = tempfile.TemporaryDirectory()
|
|
self.db_path = Path(self.temporary_directory.name) / "client.db"
|
|
self.repository = SettingsRepository(self.db_path)
|
|
self.credential_store = FakeCredentialStore()
|
|
|
|
def tearDown(self):
|
|
self.temporary_directory.cleanup()
|
|
|
|
def _page(self, service, credential_store=None):
|
|
return SettingsPage(
|
|
settings_repository=self.repository,
|
|
admin_gateway=MockAdminGateway(),
|
|
update_service=service,
|
|
update_credential_store=credential_store or self.credential_store,
|
|
)
|
|
|
|
def _wait_until(self, predicate, timeout_ms=2000):
|
|
elapsed = 0
|
|
while not predicate() and elapsed < timeout_ms:
|
|
QTest.qWait(10)
|
|
elapsed += 10
|
|
self.assertTrue(predicate(), "等待在线更新线程超时")
|
|
|
|
def _save_configuration(self, page):
|
|
page.updateManifestUrlInput.setText(MANIFEST_URL)
|
|
page.updateUsernameInput.setText(TEST_USERNAME)
|
|
page.updatePasswordInput.setText(TEST_PASSWORD)
|
|
page.updateSaveButton.click()
|
|
|
|
def test_update_card_uses_safe_defaults_and_password_input(self):
|
|
page = self._page(FakeUpdateService())
|
|
|
|
self.assertEqual(page.currentVersionLabel.text(), "0.1.0")
|
|
self.assertEqual(
|
|
page.updateManifestUrlInput.text(), DEFAULT_UPDATE_MANIFEST_URL
|
|
)
|
|
self.assertEqual(page.updateUsernameInput.text(), DEFAULT_UPDATE_USERNAME)
|
|
self.assertEqual(page.updatePasswordInput.text(), "")
|
|
self.assertEqual(page.updatePasswordInput.echoMode(), QLineEdit.Password)
|
|
self.assertTrue(page.updateSaveButton.isEnabled())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_save_uses_sqlite_for_non_secret_and_credential_store_for_password(self):
|
|
page = self._page(FakeUpdateService())
|
|
|
|
self._save_configuration(page)
|
|
|
|
self.assertEqual(
|
|
self.repository.get(UPDATE_MANIFEST_SETTING), MANIFEST_URL
|
|
)
|
|
self.assertEqual(self.repository.get(UPDATE_USERNAME_SETTING), TEST_USERNAME)
|
|
self.assertEqual(
|
|
self.credential_store.saved,
|
|
(TEST_USERNAME, TEST_PASSWORD),
|
|
)
|
|
self.assertNotIn(TEST_PASSWORD.encode(), self.db_path.read_bytes())
|
|
self.assertEqual(page.updatePasswordInput.text(), "")
|
|
self.assertIn("已安全保存", page.updatePasswordInput.placeholderText())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_blank_password_preserves_existing_credential(self):
|
|
self.repository.set_many(
|
|
{
|
|
UPDATE_MANIFEST_SETTING: MANIFEST_URL,
|
|
UPDATE_USERNAME_SETTING: TEST_USERNAME,
|
|
}
|
|
)
|
|
store = FakeCredentialStore((TEST_USERNAME, TEST_PASSWORD))
|
|
page = self._page(FakeUpdateService(), store)
|
|
|
|
page.updateSaveButton.click()
|
|
|
|
self.assertEqual(store.save_count, 0)
|
|
self.assertEqual(store.saved, (TEST_USERNAME, TEST_PASSWORD))
|
|
self.assertIn("安全保存", page.updateStatusLabel.text())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_missing_password_stops_save_and_check(self):
|
|
service = FakeUpdateService()
|
|
page = self._page(service)
|
|
|
|
page.updateSaveButton.click()
|
|
page.updateCheckButton.click()
|
|
|
|
self.assertIn("密码", page.updateStatusLabel.text())
|
|
self.assertEqual(service.check_count, 0)
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_unapproved_http_url_is_rejected_when_saving(self):
|
|
service = FakeUpdateService()
|
|
page = self._page(service)
|
|
page.updateManifestUrlInput.setText(
|
|
"http://updates.example.test/manifest.json"
|
|
)
|
|
page.updatePasswordInput.setText(TEST_PASSWORD)
|
|
|
|
page.updateSaveButton.click()
|
|
|
|
self.assertIn("HTTP 只允许固定发布服务器", page.updateStatusLabel.text())
|
|
self.assertEqual(service.check_count, 0)
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_slow_check_does_not_block_or_start_twice(self):
|
|
service = FakeUpdateService(delay=0.08)
|
|
page = self._page(service)
|
|
self._save_configuration(page)
|
|
timer_fired = []
|
|
QTimer.singleShot(10, lambda: timer_fired.append(True))
|
|
|
|
page.updateCheckButton.click()
|
|
page.eventBinder.updateEventBinder.request_check()
|
|
self._wait_until(lambda: bool(timer_fired), timeout_ms=500)
|
|
self._wait_until(
|
|
lambda: page.eventBinder.updateEventBinder._check_thread is None
|
|
)
|
|
|
|
self.assertEqual(service.check_count, 1)
|
|
self.assertEqual(service.check_credentials.username, TEST_USERNAME)
|
|
self.assertEqual(service.check_credentials.password, TEST_PASSWORD)
|
|
self.assertIn("当前已是最新版本", page.updateStatusLabel.text())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_saved_configuration_checks_automatically_after_startup(self):
|
|
self.repository.set_many(
|
|
{
|
|
UPDATE_MANIFEST_SETTING: MANIFEST_URL,
|
|
UPDATE_USERNAME_SETTING: TEST_USERNAME,
|
|
}
|
|
)
|
|
store = FakeCredentialStore((TEST_USERNAME, TEST_PASSWORD))
|
|
service = FakeUpdateService()
|
|
|
|
page = self._page(service, store)
|
|
self._wait_until(lambda: service.check_count == 1)
|
|
self._wait_until(
|
|
lambda: page.eventBinder.updateEventBinder._check_thread is None
|
|
)
|
|
|
|
self.assertIn("当前已是最新版本", page.updateStatusLabel.text())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_legacy_default_manifest_url_is_migrated_before_auto_check(self):
|
|
self.repository.set_many(
|
|
{
|
|
UPDATE_MANIFEST_SETTING: LEGACY_DEFAULT_UPDATE_MANIFEST_URL,
|
|
UPDATE_USERNAME_SETTING: TEST_USERNAME,
|
|
}
|
|
)
|
|
store = FakeCredentialStore((TEST_USERNAME, TEST_PASSWORD))
|
|
service = FakeUpdateService()
|
|
|
|
page = self._page(service, store)
|
|
self._wait_until(lambda: service.check_count == 1)
|
|
self._wait_until(
|
|
lambda: page.eventBinder.updateEventBinder._check_thread is None
|
|
)
|
|
|
|
self.assertEqual(
|
|
self.repository.get(UPDATE_MANIFEST_SETTING),
|
|
DEFAULT_UPDATE_MANIFEST_URL,
|
|
)
|
|
self.assertEqual(service.manifest_url, DEFAULT_UPDATE_MANIFEST_URL)
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_custom_manifest_url_is_not_migrated(self):
|
|
self.repository.set_many(
|
|
{
|
|
UPDATE_MANIFEST_SETTING: MANIFEST_URL,
|
|
UPDATE_USERNAME_SETTING: TEST_USERNAME,
|
|
}
|
|
)
|
|
store = FakeCredentialStore((TEST_USERNAME, TEST_PASSWORD))
|
|
service = FakeUpdateService()
|
|
|
|
page = self._page(service, store)
|
|
self._wait_until(lambda: service.check_count == 1)
|
|
self._wait_until(
|
|
lambda: page.eventBinder.updateEventBinder._check_thread is None
|
|
)
|
|
|
|
self.assertEqual(self.repository.get(UPDATE_MANIFEST_SETTING), MANIFEST_URL)
|
|
self.assertEqual(service.manifest_url, MANIFEST_URL)
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
def test_confirmed_new_version_downloads_with_saved_credentials(self):
|
|
update = UpdateInfo(
|
|
version="0.3.0",
|
|
manifest_url=MANIFEST_URL,
|
|
update_url="https://updates.example.test/CMAutoBuy-0.3.0-update.zip",
|
|
file_name="CMAutoBuy-0.3.0-update.zip",
|
|
size=1024,
|
|
sha256="0" * 64,
|
|
)
|
|
service = FakeUpdateService(
|
|
UpdateCheckResult("0.1.0", "0.3.0", True, update)
|
|
)
|
|
page = self._page(service)
|
|
self._save_configuration(page)
|
|
|
|
with patch("src.update_ui_event.MessageBox", AcceptDownloadMessageBox):
|
|
page.updateCheckButton.click()
|
|
self._wait_until(
|
|
lambda: page.eventBinder.updateEventBinder._check_thread is None
|
|
and page.eventBinder.updateEventBinder._download_thread is None
|
|
)
|
|
|
|
self.assertEqual(service.download_count, 1)
|
|
self.assertEqual(service.download_credentials.username, TEST_USERNAME)
|
|
self.assertEqual(service.download_credentials.password, TEST_PASSWORD)
|
|
self.assertIn("已准备好", page.updateStatusLabel.text())
|
|
page.eventBinder.shutdown()
|
|
page.deleteLater()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|