"""设置页认证更新配置、自动检查和后台线程测试。""" 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_successful_check_exposes_versions_and_availability(self): update = UpdateInfo( version="0.3.0", manifest_url=MANIFEST_URL, update_url="https://updates.example.test/CMAutoBuy_0.3.0.zip", file_name="CMAutoBuy_0.3.0.zip", size=1024, sha256="0" * 64, ) page = self._page(FakeUpdateService()) changes = [] page.eventBinder.updateEventBinder.updateAvailabilityChanged.connect( lambda available, current, latest: changes.append( (available, current, latest) ) ) with patch("src.update_ui_event.MessageBox.exec", return_value=False): page.eventBinder.updateEventBinder._on_check_succeeded( UpdateCheckResult("0.1.0", "0.3.0", True, update) ) self.assertEqual(changes, [(True, "0.1.0", "0.3.0")]) self.assertEqual( page.currentVersionLabel.text(), "本地 0.1.0；线上 0.3.0" ) 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()