cmshoppe/tests/test_update_check.py

import socket
import os
import sys
import unittest

sys.path.insert(0, os.path.dirname(__file__))
from _helpers import REPO_ROOT

from app import update_check

assert REPO_ROOT


class UpdateCheckTests(unittest.TestCase):
    def test_compare_versions_uses_numeric_segments(self):
        self.assertGreater(update_check.compare_versions("0.10.0", "0.2.0"), 0)
        self.assertEqual(0, update_check.compare_versions("1.2", "1.2.0"))
        self.assertLess(update_check.compare_versions("v1.2.3", "1.2.4"), 0)

    def test_forced_update_by_min_supported_version(self):
        info = update_check.UpdateInfo(
            latest_version="1.2.0",
            min_supported_version="1.1.0",
            force_update=False,
        )

        self.assertTrue(update_check.is_forced_update(info, "1.0.9"))
        self.assertFalse(update_check.is_forced_update(info, "1.1.0"))

    def test_forced_update_by_force_flag_and_latest_version(self):
        forced = update_check.UpdateInfo(latest_version="1.2.0", force_update=True)
        optional = update_check.UpdateInfo(latest_version="1.2.0", force_update=False)

        self.assertTrue(update_check.is_forced_update(forced, "1.1.9"))
        self.assertFalse(update_check.is_forced_update(optional, "1.1.9"))
        self.assertFalse(update_check.is_forced_update(forced, "1.2.0"))

    def test_check_for_update_forced_response(self):
        def fetcher(_url, _timeout):
            return {
                "latest_version": "1.2.0",
                "min_supported_version": "1.1.0",
                "force_update": True,
                "download_url": "https://example.test/cmshopee.zip",
                "sha256": "abc",
                "size_bytes": 123,
                "package_format": "cmshopee-portable-v1",
                "updater_protocol": 1,
                "min_updater_protocol": 1,
                "signature_algorithm": "",
                "manifest_signature": "",
                "message": "请升级后继续使用",
            }

        result = update_check.check_for_update(
            current_version="1.0.0",
            url="https://example.test/version.json",
            fetcher=fetcher,
        )

        self.assertTrue(result.checked)
        self.assertTrue(result.forced)
        self.assertFalse(result.can_enter)
        self.assertEqual("1.2.0", result.latest_version)
        self.assertEqual("https://example.test/cmshopee.zip", result.download_url)
        self.assertEqual(123, result.size_bytes)
        self.assertEqual("cmshopee-portable-v1", result.package_format)
        self.assertEqual(1, result.min_updater_protocol)

    def test_check_for_update_accepts_release_wrapper_response(self):
        def fetcher(_url, _timeout):
            return {
                "platform": "windows",
                "release": {
                    "version": "0.1.1",
                    "download_url": "https://example.test/cmshopee-0.1.1.zip",
                    "sha256": "abc",
                    "release_notes": "新版说明",
                },
            }

        result = update_check.check_for_update(
            current_version="0.1.0",
            url="https://example.test/releases/latest?platform=windows",
            fetcher=fetcher,
        )

        self.assertTrue(result.checked)
        self.assertFalse(result.forced)
        self.assertEqual("0.1.1", result.latest_version)
        self.assertEqual("https://example.test/cmshopee-0.1.1.zip", result.download_url)
        self.assertEqual("abc", result.sha256)
        self.assertEqual("新版说明", result.message)

    def test_check_for_update_accepts_cmhub_release_contract(self):
        def fetcher(_url, _timeout):
            return {
                "platform": "windows",
                "release": {
                    "version": "0.1.6",
                    "download_url": "http://185.216.248.75:24521/down/release.zip",
                    "sha256": "a" * 64,
                    "release_notes": "启动时联网获取更新信息",
                    "force_update": True,
                    "size_bytes": 53318076,
                    "published_at": "2026-07-13T17:10:59.660653+08:00",
                },
            }

        result = update_check.check_for_update(
            current_version="0.1.5",
            url="https://cm.example.test/api/v1/client/releases/latest?platform=windows",
            fetcher=fetcher,
        )

        self.assertTrue(result.forced)
        self.assertFalse(result.can_enter)
        self.assertEqual("0.1.6", result.latest_version)
        self.assertEqual(53318076, result.size_bytes)
        self.assertEqual("", result.package_format)
        self.assertEqual(0, result.updater_protocol)

    def test_check_for_update_parses_client_policy_without_affecting_release(self):
        result = update_check.check_for_update(
            current_version="0.1.5",
            url="https://cm.example.test/api/v1/client/releases/latest?platform=windows",
            fetcher=lambda _url, _timeout: {
                "platform": "windows",
                "release": {
                    "version": "0.1.6",
                    "force_update": False,
                },
                "client_policy": {
                    "policy_version": 1,
                    "subscription_check_enabled": True,
                    "subscription_enforcement_enabled": True,
                    "updated_at": "2026-07-28T15:19:19+08:00",
                },
            },
        )

        self.assertEqual("0.1.6", result.latest_version)
        self.assertEqual("enforce", result.client_policy.mode)
        self.assertEqual("", result.client_policy_error)

    def test_invalid_client_policy_does_not_break_update_detection(self):
        result = update_check.check_for_update(
            current_version="0.1.5",
            url="https://cm.example.test/version.json",
            fetcher=lambda _url, _timeout: {
                "release": {"version": "0.1.6"},
                "client_policy": {
                    "policy_version": 1,
                    "subscription_check_enabled": "false",
                    "subscription_enforcement_enabled": True,
                    "updated_at": "2026-07-28T15:19:19+08:00",
                },
            },
        )

        self.assertEqual("0.1.6", result.latest_version)
        self.assertIsNone(result.client_policy)
        self.assertIn("布尔值", result.client_policy_error)
        self.assertEqual("", result.error)

    def test_release_can_be_null_when_policy_is_available(self):
        result = update_check.check_for_update(
            current_version="0.1.5",
            url="https://cm.example.test/version.json",
            fetcher=lambda _url, _timeout: {
                "platform": "windows",
                "release": None,
                "client_policy": {
                    "policy_version": 1,
                    "subscription_check_enabled": False,
                    "subscription_enforcement_enabled": False,
                    "updated_at": "2026-07-28T15:19:19+08:00",
                },
            },
        )

        self.assertTrue(result.checked)
        self.assertFalse(result.forced)
        self.assertEqual("", result.error)
        self.assertEqual("off", result.client_policy.mode)

    def test_network_failure_allows_entry(self):
        def fetcher(_url, _timeout):
            raise socket.timeout("timeout")

        result = update_check.check_for_update(
            current_version="1.0.0",
            url="https://example.test/version.json",
            fetcher=fetcher,
        )

        self.assertTrue(result.checked)
        self.assertFalse(result.forced)
        self.assertTrue(result.can_enter)
        self.assertIn("启动版本检查失败", result.error)

    def test_invalid_response_allows_entry(self):
        result = update_check.check_for_update(
            current_version="1.0.0",
            url="https://example.test/version.json",
            fetcher=lambda _url, _timeout: {"message": "missing versions"},
        )

        self.assertTrue(result.checked)
        self.assertFalse(result.forced)
        self.assertTrue(result.can_enter)
        self.assertIn("缺少 latest_version", result.error)

    def test_empty_update_url_skips_check(self):
        result = update_check.check_for_update(current_version="1.0.0", url="")

        self.assertFalse(result.checked)
        self.assertFalse(result.forced)
        self.assertEqual("", result.error)


if __name__ == "__main__":
    unittest.main()