cmautobuy/client/test/test_update_ui_event.py

"""设置页认证更新配置、自动检查和后台线程测试。"""

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()