svfs/tests/test_client_logic.py

#!/usr/bin/env python3
"""svfs_client 纯标准库层的测试：manifest / paths / registry / installer / sample。

不 import Qt——这五个模块必须能在没有 PySide6 的环境里跑通（分层约定见
src/svfs_client/__init__.py 的文档字符串）。
"""

import json
import os
import sys
import unittest

from helpers import VFSTestCase, make_tree, quiet, temp_dir, write_file

from sqlite_vfs.core import SQLiteVFS
from sqlite_vfs.publish import export_objects

from svfs_client import PLUGIN_API_VERSION, installer, manifest as manifest_mod
from svfs_client import diagnostics, paths, registry, sample, search, settings

# helpers.py 会把 src/ 加进 sys.path；这里只需要保证它是测试的先决导入
assert sys.modules.get("helpers") is not None


# --------------------------------------------------------------------------
# 测试用的插件包构造
# --------------------------------------------------------------------------

BASE_MANIFEST = {
    "id": "com.example.demo",
    "name": "演示插件",
    "version": "1.0.0",
    "api_version": PLUGIN_API_VERSION,
    "entry": "index.html",
    "icon": "icon.svg",
    "permissions": ["storage"],
    "description": "测试用",
}


def build_plugin(out_db, spec, version="1.0.0", overrides=None):
    """把一棵目录树打成插件包（沿用 installer 的打包路径）。"""
    tree = dict(spec)
    data = dict(BASE_MANIFEST)
    data["version"] = version
    data.update(overrides or {})
    tree["plugin.json"] = json.dumps(data, ensure_ascii=False).encode("utf-8")
    with temp_dir() as src:
        make_tree(src, tree)
        installer._pack_tree(out_db, src, version)
    return out_db


V1_SPEC = {
    "index.html": b"<html>v1</html>",
    "style.css": b"body { color: #eee; }",
    "icon.svg": b"<svg/>",
    "app.js": b"console.log('v1');",
}

V2_SPEC = {
    "index.html": b"<html>v1</html>",        # 未变 → 应共享 blob
    "style.css": b"body { color: #eee; }",   # 未变 → 应共享 blob
    "icon.svg": b"<svg/>",                   # 未变 → 应共享 blob
    "app.js": b"console.log('v2 with more code');",
    "changelog.txt": b"v2: new feature",
}


class ManifestTest(unittest.TestCase):
    def test_valid_minimal(self):
        data = manifest_mod.validate(dict(BASE_MANIFEST, icon=None))
        self.assertEqual(data["permissions"], ["storage"])

    def test_net_permission_accepted(self):
        data = manifest_mod.validate(
            dict(BASE_MANIFEST, permissions=["net", "storage", "net"])
        )
        self.assertEqual(data["permissions"], ["net", "storage"])

    def test_missing_field(self):
        for field in ("id", "name", "version", "entry", "api_version"):
            broken = {k: v for k, v in BASE_MANIFEST.items() if k != field}
            with self.assertRaisesRegex(manifest_mod.ManifestError, field):
                manifest_mod.validate(broken)

    def test_bad_id(self):
        for bad in ("demo", "Demo.App", "com/example", "-com.example", "", 123,
                    "a" * 64 + ".x"):
            with self.assertRaises(manifest_mod.ManifestError, msg=repr(bad)):
                manifest_mod.validate(dict(BASE_MANIFEST, id=bad))

    def test_wrong_api_version(self):
        with self.assertRaisesRegex(manifest_mod.ManifestError, "api_version"):
            manifest_mod.validate(dict(BASE_MANIFEST, api_version=99))

    def test_older_api_version_accepted(self):
        # 旧清单跑在新宿主上没问题：api_version 是下限匹配，不是相等匹配
        data = manifest_mod.validate(
            dict(BASE_MANIFEST, api_version=PLUGIN_API_VERSION - 1)
        )
        self.assertEqual(data["api_version"], PLUGIN_API_VERSION - 1)

    def test_api_version_must_be_positive(self):
        for bad in (0, -1):
            with self.assertRaisesRegex(manifest_mod.ManifestError, "api_version"):
                manifest_mod.validate(dict(BASE_MANIFEST, api_version=bad))

    def test_permissions_normalized(self):
        data = manifest_mod.validate(dict(BASE_MANIFEST, permissions=["notify", "storage", "notify"]))
        self.assertEqual(data["permissions"], ["notify", "storage"])

    def test_net_hosts_optional_and_normalized(self):
        # 不写 = 不限制（空列表）
        self.assertEqual(manifest_mod.validate(BASE_MANIFEST)["net_hosts"], [])
        data = manifest_mod.validate({
            **BASE_MANIFEST, "permissions": ["net"],
            "net_hosts": ["API.Example.com", "*.Corp.Example.", "api.example.com"],
        })
        self.assertEqual(data["net_hosts"],
                         ["*.corp.example", "api.example.com"])

    def test_net_hosts_rejects_bad_input(self):
        with self.assertRaisesRegex(manifest_mod.ManifestError, "net_hosts"):
            manifest_mod.validate({**BASE_MANIFEST, "permissions": ["net"],
                                   "net_hosts": []})
        with self.assertRaisesRegex(manifest_mod.ManifestError, "net_hosts"):
            manifest_mod.validate({**BASE_MANIFEST, "permissions": ["net"],
                                   "net_hosts": ["http://x:8000/api"]})
        # 声明了白名单却没有 net 权限：几乎肯定写错了，安装时就报出来
        with self.assertRaisesRegex(manifest_mod.ManifestError, "net 权限"):
            manifest_mod.validate({**BASE_MANIFEST, "permissions": ["storage"],
                                   "net_hosts": ["api.example.com"]})

    def test_host_allowed(self):
        # 空 = 不限制
        self.assertTrue(manifest_mod.host_allowed("any.example", []))
        self.assertTrue(manifest_mod.host_allowed("api.example.com",
                                                  ["api.example.com"]))
        self.assertFalse(manifest_mod.host_allowed("evil.example",
                                                   ["api.example.com"]))
        # *. 后缀匹配：子域与其本身都算，但不吃 "notcorp.example"
        patterns = ["*.corp.example"]
        self.assertTrue(manifest_mod.host_allowed("a.corp.example", patterns))
        self.assertTrue(manifest_mod.host_allowed("corp.example", patterns))
        self.assertFalse(manifest_mod.host_allowed("notcorp.example", patterns))
        # 大小写不敏感
        self.assertTrue(manifest_mod.host_allowed("API.Example.com",
                                                  ["api.example.com"]))

    def test_bad_permission(self):
        with self.assertRaises(manifest_mod.ManifestError):
            manifest_mod.validate(dict(BASE_MANIFEST, permissions=["filesystem"]))

    def test_from_json_bom(self):
        text = b"\xef\xbb\xbf" + json.dumps(BASE_MANIFEST).encode("utf-8")
        data = manifest_mod.from_json(text.decode("utf-8"))
        self.assertEqual(data["id"], BASE_MANIFEST["id"])

    def test_from_json_broken(self):
        with self.assertRaises(manifest_mod.ManifestError):
            manifest_mod.from_json("{oops")


class PathsTest(VFSTestCase):
    def test_explicit_data_dir(self):
        self.assertEqual(paths.data_dir("D:/ somewhere/x"), os.path.abspath("D:/ somewhere/x"))

    def test_layout_and_id_paths(self):
        base = paths.ensure_layout(self.path("data"))
        for sub in ("plugins", "storage", "web"):
            self.assertTrue(os.path.isdir(os.path.join(base, sub)))
        pkg = paths.plugin_package(base, "com.example.demo")
        self.assertTrue(pkg.endswith(os.path.join("plugins", "com.example.demo") + os.sep + "plugin.svfs"))
        self.assertEqual(
            paths.storage_file(base, "com.example.demo"),
            os.path.join(base, "storage", "com.example.demo.json"),
        )

    def test_id_with_slash_rejected(self):
        with self.assertRaises(ValueError):
            paths.plugin_package(self.tmp, "../evil")

    def test_dir_size_and_clear(self):
        make_tree(self.path("cache"), {"a.bin": b"x" * 10, "sub/b.bin": b"y" * 5})
        self.assertEqual(paths.dir_size(self.path("cache")), 15)
        self.assertEqual(paths.clear_dir_contents(self.path("cache")), 2)
        self.assertTrue(os.path.isdir(self.path("cache")))  # 目录本身保留
        self.assertEqual(paths.dir_size(self.path("cache")), 0)

    def test_dir_helpers_tolerate_missing_dir(self):
        self.assertEqual(paths.dir_size(self.path("nope")), 0)
        self.assertEqual(paths.clear_dir_contents(self.path("nope")), 0)

    def test_object_cache_dir_is_absolute(self):
        self.assertTrue(os.path.isabs(paths.object_cache_dir()))

    def test_data_dir_pointer_file(self):
        # 默认：无指针 → 默认目录
        self.assertEqual(paths.resolve_data_dir(), paths.data_dir())
        # 写指针（目标目录必须真实存在）→ 生效
        target = paths.ensure_layout(self.path("alt-data"))
        self.assertIsNone(paths.set_data_dir(target))
        self.assertEqual(paths.resolve_data_dir(), os.path.abspath(target))
        # --data 命令行参数最优先
        self.assertEqual(
            paths.resolve_data_dir(self.path("cmdline")),
            os.path.abspath(self.path("cmdline")),
        )
        # 指向不存在的目录 → 回退默认（坏指针不让客户端起不来）
        os.unlink(paths.data_dir_file())
        self.assertEqual(paths.resolve_data_dir(), paths.data_dir())

    def test_set_data_dir_clear_and_invalid(self):
        target = paths.ensure_layout(self.path("alt-data"))
        paths.set_data_dir(target)
        # 空串 = 清除重定向
        self.assertIsNone(paths.set_data_dir(""))
        self.assertEqual(paths.resolve_data_dir(), paths.data_dir())
        # 目录不存在 → 报错且不写指针
        error = paths.set_data_dir(self.path("nope"))
        self.assertIn("不存在", error)
        self.assertEqual(paths.resolve_data_dir(), paths.data_dir())


class RegistryTest(VFSTestCase):
    def test_empty_load(self):
        data = registry.load(self.tmp)
        self.assertEqual(data["plugins"], {})

    def test_roundtrip(self):
        data = registry.load(self.tmp)
        registry.put(data, {"id": "com.a.b", "name": "A", "enabled": True})
        registry.save(self.tmp, data)
        again = registry.load(self.tmp)
        self.assertEqual(again["plugins"]["com.a.b"]["name"], "A")
        self.assertEqual(registry.get(again, "com.a.b")["enabled"], True)
        self.assertIsNone(registry.remove(again, "nope"))
        self.assertEqual(registry.remove(again, "com.a.b")["id"], "com.a.b")


class SettingsTest(VFSTestCase):
    """base_url / 登录凭据的规范化、持久化与地址解析。"""

    def test_normalize(self):
        self.assertEqual(settings.normalize_base_url(" http://srv:8000/ "), "http://srv:8000")
        self.assertEqual(settings.normalize_base_url("https://a.b/c/"), "https://a.b/c")
        self.assertEqual(settings.normalize_base_url(None), "")
        self.assertEqual(settings.normalize_base_url(""), "")
        for bad in ("ftp://x", "srv:8000", "//host", "http:/miss"):
            with self.assertRaisesRegex(ValueError, "http", msg=bad):
                settings.normalize_base_url(bad)

    def test_load_defaults_and_roundtrip(self):
        data = settings.load(self.tmp)
        # 缺文件 → 默认（含登录凭据/启动直显/缓存目录/网络/日志）
        self.assertEqual(
            data,
            {"base_url": "", "auth_token": "", "auth_user": "",
             "startup_plugin": "", "plugin_order": [], "cache_dir": "",
             "net_timeout": settings.NET_TIMEOUT_DEFAULT,
             "net_retries": settings.NET_RETRIES_DEFAULT,
             "proxy": "", "ca_bundle": "", "log_to_file": True,
             "org": "", "mqtt_host": "",
             "mqtt_port": settings.MQTT_PORT_DEFAULT, "mqtt_tls": False,
             "mqtt_prefix": "svfs", "mqtt_username": "", "mqtt_password": ""},
        )
        data["base_url"] = "http://10.0.0.5:9000"
        settings.save(self.tmp, data)
        self.assertEqual(settings.load(self.tmp)["base_url"], "http://10.0.0.5:9000")

    def test_broken_settings_file_is_ignored(self):
        write_file(paths.settings_file(self.tmp), b"{broken")
        self.assertEqual(
            settings.load(self.tmp),
            {"base_url": "", "auth_token": "", "auth_user": "",
             "startup_plugin": "", "plugin_order": [], "cache_dir": "",
             "net_timeout": settings.NET_TIMEOUT_DEFAULT,
             "net_retries": settings.NET_RETRIES_DEFAULT,
             "proxy": "", "ca_bundle": "", "log_to_file": True,
             "org": "", "mqtt_host": "",
             "mqtt_port": settings.MQTT_PORT_DEFAULT, "mqtt_tls": False,
             "mqtt_prefix": "svfs", "mqtt_username": "", "mqtt_password": ""},
        )

    def test_resolve_cache_dir(self):
        # 未配置 → 系统默认
        self.assertEqual(settings.resolve_cache_dir(self.tmp),
                         paths.object_cache_dir())
        # 配置了（支持 ~ 与相对展开）→ 生效
        data = settings.load(self.tmp)
        data["cache_dir"] = "~/svfs-cache"
        settings.save(self.tmp, data)
        resolved = settings.resolve_cache_dir(self.tmp)
        self.assertTrue(os.path.isabs(resolved))
        self.assertIn("svfs-cache", resolved)
        self.assertNotIn("~", resolved)

    def test_resolve_manifest_url(self):
        base = "http://srv:8000"
        # 完整地址原样通过
        self.assertEqual(
            settings.resolve_manifest_url(base, "http://other/x/manifest.json"),
            "http://other/x/manifest.json",
        )
        # 插件 id → 约定子目录
        self.assertEqual(
            settings.resolve_manifest_url(base, "com.example.clock"),
            "http://srv:8000/com.example.clock/manifest.json",
        )
        # 相对路径
        self.assertEqual(
            settings.resolve_manifest_url(base, "/clock/stable/manifest.json"),
            "http://srv:8000/clock/stable/manifest.json",
        )
        # 未配置服务器时相对地址无从解析
        with self.assertRaisesRegex(ValueError, "服务器"):
            settings.resolve_manifest_url("", "com.example.clock")
        with self.assertRaisesRegex(ValueError, "请填写"):
            settings.resolve_manifest_url(base, "  ")

    def test_resolve_net_url(self):
        base = "http://srv:8000"
        self.assertEqual(
            settings.resolve_net_url(base, url="https://api.x/v1"), "https://api.x/v1"
        )
        self.assertEqual(
            settings.resolve_net_url(base, path="/api/ping"), "http://srv:8000/api/ping"
        )
        with self.assertRaises(ValueError):
            settings.resolve_net_url("", path="/api/ping")
        with self.assertRaises(ValueError):
            settings.resolve_net_url(base)
        with self.assertRaisesRegex(ValueError, "http"):
            settings.resolve_net_url(base, url="file:///etc/passwd")


class InstallerTest(VFSTestCase):
    def setUp(self):
        super().setUp()
        self.base = paths.ensure_layout(self.path("data"))
        self.pid = BASE_MANIFEST["id"]

    def install(self, spec, version="1.0.0", **kwargs):
        pkg = self.path(f"plugin-{version}.svfs")
        build_plugin(pkg, spec, version=version)
        with quiet():
            return installer.install_from_file(pkg, base=self.base, **kwargs)

    def test_first_install(self):
        entry = self.install(V1_SPEC)
        self.assertTrue(entry["enabled"])
        self.assertEqual(entry["version"], "1.0.0")
        self.assertEqual(entry["source"]["type"], "local")
        target = paths.plugin_package(self.base, self.pid)
        self.assertTrue(os.path.exists(target))
        # 注册表落盘，列表可读
        self.assertEqual(installer.list_installed(self.base)[0]["id"], self.pid)

    def test_reinstall_is_upgrade(self):
        self.install(V1_SPEC)
        entry = self.install(V2_SPEC, version="2.0.0")
        self.assertEqual(entry["version"], "2.0.0")
        self.assertEqual([s["name"] for s in entry["snapshots"]], ["1.0.0", "2.0.0"])
        self.assertEqual(entry["current_snapshot_id"], entry["snapshots"][1]["id"])

        # 快照即版本：新旧内容都在一个包里
        vfs = SQLiteVFS(paths.plugin_package(self.base, self.pid), readonly=True)
        try:
            self.assertEqual(vfs.read_file("app.js"), V2_SPEC["app.js"])
            vfs.use_snapshot(1)
            self.assertEqual(vfs.read_file("app.js"), V1_SPEC["app.js"])
            # 未变的三个文件共享 blob：8 个 blob < 两个快照的条目总数
            self.assertLess(len(vfs.list_blobs()), 5 + 6)
        finally:
            vfs.close()

    def test_upgrade_rollback(self):
        self.install(V1_SPEC)
        self.install(V2_SPEC, version="2.0.0")
        entry = installer.use_version(self.base, self.pid, 1)
        self.assertEqual(entry["version"], "1.0.0")
        self.assertEqual(entry["current_snapshot_id"], 1)
        # 回滚对协议层生效：按注册表指向的快照开包，看到的是旧内容
        registry_data = registry.load(self.base)
        sid = registry_data["plugins"][self.pid]["current_snapshot_id"]
        vfs = SQLiteVFS(paths.plugin_package(self.base, self.pid), readonly=True,
                        snapshot_id=sid)
        try:
            self.assertEqual(vfs.read_file("app.js"), V1_SPEC["app.js"])
        finally:
            vfs.close()

    def test_rollback_to_missing_snapshot(self):
        self.install(V1_SPEC)
        with self.assertRaises(installer.InstallError):
            installer.use_version(self.base, self.pid, 42)

    def test_uninstall(self):
        self.install(V1_SPEC)
        # 留一份键值存储，卸载时应一并删除
        write_file(paths.storage_file(self.base, self.pid), b"{}")
        installer.uninstall(self.base, self.pid)
        self.assertFalse(os.path.exists(paths.plugin_package(self.base, self.pid)))
        self.assertFalse(os.path.exists(paths.storage_file(self.base, self.pid)))
        self.assertEqual(installer.list_installed(self.base), [])

    def test_reject_package_without_manifest(self):
        src = make_tree(self.path("bare"), {"index.html": b"<html/>"})
        pkg = self.path("bare.svfs")
        with quiet():
            vfs = SQLiteVFS(pkg, compress=True)
            try:
                from sqlite_vfs.folder_packer import FolderPacker

                FolderPacker(vfs).pack_folder(src, incremental=True)
            finally:
                vfs.close()
        with quiet(), self.assertRaisesRegex(installer.InstallError, "plugin.json"):
            installer.install_from_file(pkg, base=self.base)

    def test_reject_tampered_package(self):
        pkg = self.path("plugin-1.0.0.svfs")
        build_plugin(pkg, V1_SPEC)
        # 绕过 API 改元数据但不 recount()：摘要对不上，verify() 应当拦住它
        vfs = SQLiteVFS(pkg)
        try:
            vfs.conn.execute(
                "UPDATE files SET file_size = file_size + 1 WHERE is_directory = 0"
            )
            vfs.conn.commit()
        finally:
            vfs.close()
        with quiet(), self.assertRaises(installer.InstallError):
            installer.install_from_file(pkg, base=self.base)

    def test_reject_wrong_api_version(self):
        pkg = self.path("future.svfs")
        build_plugin(pkg, V1_SPEC, overrides={"api_version": PLUGIN_API_VERSION + 1})
        with quiet(), self.assertRaisesRegex(installer.InstallError, "api_version"):
            installer.install_from_file(pkg, base=self.base)

    def test_accept_older_api_version(self):
        # 语义同 manifest：要求 ≤ 宿主版本即可安装（升级宿主不废掉旧插件）
        pkg = self.path("older.svfs")
        build_plugin(pkg, V1_SPEC, overrides={"api_version": PLUGIN_API_VERSION - 1})
        with quiet():
            entry = installer.install_from_file(pkg, base=self.base)
        self.assertEqual(entry["id"], self.pid)

    def test_set_enabled(self):
        self.install(V1_SPEC)
        entry = installer.set_enabled(self.base, self.pid, False)
        self.assertFalse(entry["enabled"])
        self.assertFalse(registry.load(self.base)["plugins"][self.pid]["enabled"])
        with self.assertRaises(installer.InstallError):
            installer.set_enabled(self.base, "nope.app", True)

    def test_install_progress(self):
        pkg = self.path("plugin-1.0.0.svfs")
        build_plugin(pkg, V1_SPEC)
        events = []
        with quiet():
            installer.install_from_file(pkg, base=self.base,
                                        progress=lambda d, t, label: events.append((d, t, label)))
        self.assertEqual(events, [])  # 首次安装是整包复制，不走逐文件进度

        v2 = self.path("plugin-2.0.0.svfs")
        build_plugin(v2, V2_SPEC, version="2.0.0")
        events = []
        with quiet():
            installer.install_from_file(v2, base=self.base,
                                        progress=lambda d, t, label: events.append((d, t, label)))
        self.assertTrue(events)
        self.assertEqual(max(d for d, _, _ in events), max(t for _, t, _ in events))


class RemoteInstallTest(VFSTestCase):
    def setUp(self):
        super().setUp()
        self.base = paths.ensure_layout(self.path("data"))
        self.pid = BASE_MANIFEST["id"]

    def test_install_from_local_objects(self):
        src = self.path("plugin-1.0.0.svfs")
        build_plugin(src, V1_SPEC)
        objects = export_objects(src, self.path("public"))

        entry = installer.install_from_url(
            os.path.join(objects["out_dir"], "manifest.json"), base=self.base
        )
        self.assertEqual(entry["version"], "1.0.0")
        self.assertEqual(entry["source"]["type"], "url")
        # 内容逐字节一致
        vfs = SQLiteVFS(paths.plugin_package(self.base, self.pid), readonly=True)
        try:
            self.assertEqual(vfs.read_file("index.html"), V1_SPEC["index.html"])
            self.assertEqual(vfs.read_file("app.js"), V1_SPEC["app.js"])
        finally:
            vfs.close()

    def test_remote_upgrade(self):
        src1 = self.path("plugin-1.0.0.svfs")
        build_plugin(src1, V1_SPEC)
        with quiet():
            installer.install_from_file(src1, base=self.base)

        src2 = self.path("plugin-2.0.0.svfs")
        build_plugin(src2, V2_SPEC, version="2.0.0")
        objects = export_objects(src2, self.path("p2"))
        entry = installer.install_from_url(
            os.path.join(objects["out_dir"], "manifest.json"), base=self.base
        )
        self.assertEqual(entry["version"], "2.0.0")
        self.assertEqual([s["name"] for s in entry["snapshots"]], ["1.0.0", "2.0.0"])


class SampleTest(VFSTestCase):
    def setUp(self):
        super().setUp()
        self.base = paths.ensure_layout(self.path("data"))

    def test_sample_source_is_valid_plugin(self):
        with temp_dir() as src:
            sample.write_source(src, version=1)
            with open(os.path.join(src, "plugin.json"), "rb") as f:
                data = manifest_mod.from_json(f.read().decode("utf-8"))
            self.assertEqual(data["id"], sample.PLUGIN_ID)
            for name in ("index.html", "app.js", "style.css", "icon.svg"):
                self.assertTrue(os.path.exists(os.path.join(src, name)), name)
            with open(os.path.join(src, "app.js"), encoding="utf-8") as f:
                self.assertIn("svfsReady", f.read())

    def test_sample_build_and_upgrade(self):
        out = self.path("sample.svfs")
        _, v = sample.build(out)
        self.assertEqual(v, "1.0.0")
        with quiet():
            installer.install_from_file(out, base=self.base)

        _, v = sample.build(out)
        self.assertEqual(v, "2.0.0")
        with quiet():
            entry = installer.install_from_file(out, base=self.base)
        self.assertEqual(entry["version"], "2.0.0")
        self.assertEqual([s["name"] for s in entry["snapshots"]], ["1.0.0", "2.0.0"])

        # 升级后的包必须能通过权威校验
        with quiet():
            report = installer._verify_content(out)
        self.assertTrue(report["ok"], report["errors"])


class VersionCompareTest(VFSTestCase):
    """更新检查用的版本比较（installer.is_newer_version）。"""

    def test_newer(self):
        self.assertTrue(installer.is_newer_version("2.0.0", "1.0.0"))
        self.assertTrue(installer.is_newer_version("1.10.0", "1.9.0"))
        self.assertTrue(installer.is_newer_version("1.0.1", "1.0.0"))
        self.assertTrue(installer.is_newer_version("2.0.0", "1.0.0-2"))

    def test_not_newer(self):
        for remote, current in (("1.0.0", "1.0.0"), ("1.0.0", "2.0.0"),
                                ("1.0.0", "1.0.0-2")):
            self.assertFalse(installer.is_newer_version(remote, current),
                             f"{remote} vs {current}")

    def test_unparsable_is_conservative(self):
        self.assertFalse(installer.is_newer_version("", "1.0.0"))
        self.assertFalse(installer.is_newer_version(None, "1.0.0"))

    def test_client_update_status(self):
        payload = {"version": "0.9.0", "url": "https://x/SVFS-Client.zip",
                   "notes": "修了几个坑"}
        result = installer.client_update_status(payload, "0.1.0")
        self.assertTrue(result["update"])
        self.assertEqual(result["latest"], "0.9.0")
        self.assertEqual(result["url"], "https://x/SVFS-Client.zip")
        self.assertEqual(result["notes"], "修了几个坑")
        self.assertEqual(result["current"], "0.1.0")
        # 不比当前新 → 不提示
        self.assertFalse(
            installer.client_update_status(payload, "0.9.0")["update"])
        self.assertFalse(
            installer.client_update_status(payload, "1.0.0")["update"])

    def test_client_update_status_tolerates_missing_or_bad_payload(self):
        for payload in (None, [], "x", {}, {"url": "https://x"}):
            result = installer.client_update_status(payload, "0.1.0")
            self.assertFalse(result["update"], payload)
            self.assertEqual(result["latest"], "")
            self.assertEqual(result["current"], "0.1.0")


class PluginOrderTest(VFSTestCase):
    """插件顺序编排（纯函数 + 设置持久化）。"""

    @staticmethod
    def _entry(plugin_id, name):
        return {"id": plugin_id, "name": name}

    def test_arrange_without_order_keeps_name_order(self):
        entries = [self._entry("com.b", "Bravo"), self._entry("com.a", "Alpha")]
        arranged = installer.arrange(entries, [])
        self.assertEqual([e["id"] for e in arranged], ["com.a", "com.b"])
        # 空 order 与 None 等价（"从没编排过"这两种写法都可能出现）
        self.assertEqual(installer.arrange(entries, None), arranged)
        # 不改动传入的列表（原地排序会污染调用方持有的 payload）
        self.assertEqual([e["id"] for e in entries], ["com.b", "com.a"])

    def test_arrange_puts_ordered_ids_first(self):
        entries = [self._entry("com.a", "Alpha"), self._entry("com.b", "Bravo"),
                   self._entry("com.c", "Charlie")]
        arranged = installer.arrange(entries, ["com.c", "com.a"])
        self.assertEqual([e["id"] for e in arranged],
                         ["com.c", "com.a", "com.b"])  # 未编排的按名称跟后面

    def test_arrange_ignores_uninstalled_ids(self):
        entries = [self._entry("com.a", "Alpha"), self._entry("com.b", "Bravo")]
        # com.gone 已卸载：当没看见，剩下两个仍按给定顺序排
        arranged = installer.arrange(entries, ["com.gone", "com.b"])
        self.assertEqual([e["id"] for e in arranged], ["com.b", "com.a"])

    def test_arrange_is_stable_for_new_plugins(self):
        """新装的插件（不在 order 里）落在"未编排"那一档，不干扰已编排的。"""
        entries = [self._entry("com.a", "Alpha"), self._entry("com.new", "Zulu"),
                   self._entry("com.b", "Bravo")]
        arranged = installer.arrange(entries, ["com.b", "com.a"])
        self.assertEqual([e["id"] for e in arranged],
                         ["com.b", "com.a", "com.new"])

    def test_plugin_order_roundtrip_and_cleanup(self):
        base = paths.ensure_layout(self.path("data"))
        self.assertEqual(settings.plugin_order(base), [])
        # 去重 + 保序 + 丢掉空值
        self.assertEqual(
            settings.set_plugin_order(base, ["com.b", "com.a", "com.b", "", None]),
            ["com.b", "com.a"],
        )
        self.assertEqual(settings.plugin_order(base), ["com.b", "com.a"])
        # 复位
        settings.set_plugin_order(base, [])
        self.assertEqual(settings.plugin_order(base), [])

    def test_plugin_order_survives_bad_values(self):
        base = paths.ensure_layout(self.path("data"))
        data = settings.load(base)
        data["plugin_order"] = "com.a"  # 手改坏成字符串
        settings.save(base, data)
        self.assertEqual(settings.plugin_order(base), [])
        # 返回的是副本：改它不影响模块级默认值（否则后面每次读取都被污染）
        first = settings.load(base)["plugin_order"]
        first.append("com.polluted")
        self.assertEqual(settings.load(base)["plugin_order"], [])
        self.assertEqual(settings.DEFAULTS["plugin_order"], [])


class DiagnosticsTest(VFSTestCase):
    """诊断日志：环形缓冲、报告、自身绝不抛异常。"""

    def setUp(self):
        super().setUp()
        diagnostics.clear()
        # 落盘是全局开关：默认关掉（否则会把日志写进别的用例的临时目录）
        diagnostics.configure(None)

    def tearDown(self):
        diagnostics.clear()
        diagnostics.configure(None)
        super().tearDown()

    def test_log_and_recent(self):
        diagnostics.log("第一条", "worker")
        diagnostics.log("第二条", "net")
        lines = diagnostics.recent()
        self.assertEqual(len(lines), 2)
        self.assertIn("[worker] 第一条", lines[0])
        self.assertIn("[net] 第二条", lines[1])
        self.assertEqual(diagnostics.recent(1), [lines[-1]])

    def test_ring_limit(self):
        for i in range(diagnostics.LOG_LIMIT + 30):
            diagnostics.log(f"第 {i} 条")
        lines = diagnostics.recent()
        self.assertEqual(len(lines), diagnostics.LOG_LIMIT)
        # 最旧的被挤掉，最新的一定在
        self.assertIn(f"第 {diagnostics.LOG_LIMIT + 29} 条", lines[-1])
        self.assertNotIn("第 0 条", lines[0])

    def test_clear(self):
        diagnostics.log("x")
        diagnostics.clear()
        self.assertEqual(diagnostics.recent(), [])

    def test_log_never_raises(self):
        # 日志服务于排障，不能反过来把调用方搞挂
        diagnostics.log(None)          # 非字符串 message
        diagnostics.log("ok", object())  # 非字符串 source
        self.assertTrue(diagnostics.recent())

    def test_report_contains_env_and_logs(self):
        diagnostics.log("页面报错: boom", "console:com.svfs.demo")
        text = diagnostics.report({"数据目录": "C:/x", "Qt 版本": "6.6"})
        self.assertIn("SVFS Client 诊断报告", text)
        self.assertIn("数据目录: C:/x", text)
        self.assertIn("Qt 版本: 6.6", text)
        self.assertIn("页面报错: boom", text)
        self.assertIn("console:com.svfs.demo", text)


class DiagnosticsFileTest(VFSTestCase):
    """诊断日志落盘与轮转：崩了/重启之后还能回看现场。"""

    def setUp(self):
        super().setUp()
        diagnostics.clear()
        self.logs = self.path("logs")
        self.addCleanup(diagnostics.configure, None)

    def _log_file(self):
        return os.path.join(self.logs, diagnostics.LOG_FILE_NAME)

    def test_log_is_appended_to_file(self):
        diagnostics.configure(self.logs)
        self.assertEqual(diagnostics.log_path(), self._log_file())
        diagnostics.log("插件页面报错", "console:com.svfs.demo")
        with open(self._log_file(), encoding="utf-8") as handle:
            text = handle.read()
        self.assertIn("[console:com.svfs.demo] 插件页面报错", text)
        # 报告里也带上日志文件位置（贴给开发者时知道去哪儿找全量日志）
        self.assertIn(diagnostics.log_path(), diagnostics.report({}))

    def test_configure_none_stops_writing(self):
        diagnostics.configure(self.logs)
        diagnostics.log("第一条")
        diagnostics.configure(None)
        self.assertEqual(diagnostics.log_path(), "")
        diagnostics.log("第二条（不落盘）")
        with open(self._log_file(), encoding="utf-8") as handle:
            text = handle.read()
        self.assertIn("第一条", text)
        self.assertNotIn("不落盘", text)
        # 内存那份照旧（导出报告仍然有它）
        self.assertIn("第二条（不落盘）", diagnostics.recent()[-1])

    def test_rotation_keeps_size_bounded(self):
        diagnostics.configure(self.logs, max_bytes=1024, backups=2)
        for index in range(200):
            diagnostics.log(f"第 {index} 行，凑一点长度好触发轮转")
        files = sorted(os.listdir(self.logs))
        self.assertIn(diagnostics.LOG_FILE_NAME, files)
        self.assertIn(diagnostics.LOG_FILE_NAME + ".1", files)
        # 份数封顶：最多 1 个当前文件 + backups 个历史文件
        self.assertLessEqual(len(files), 1 + 2)
        # 最新的日志一定还在当前文件里；总量有界（不会无限长）
        with open(self._log_file(), encoding="utf-8") as handle:
            self.assertIn("第 199 行", handle.read())
        total = sum(os.path.getsize(os.path.join(self.logs, name))
                    for name in files)
        self.assertLess(total, 1024 * (1 + 2) * 4)

    def test_unusable_log_dir_falls_back_to_memory_only(self):
        # 目标是个文件（不是目录）：建不出来 → 只留内存，不能抛
        blocker = self.path("not-a-dir")
        write_file(blocker, b"x")
        diagnostics.configure(blocker)
        self.assertEqual(diagnostics.log_path(), "")
        diagnostics.log("还能记")
        self.assertIn("还能记", diagnostics.recent()[-1])


class SearchTest(unittest.TestCase):
    """侧栏搜索的模糊匹配（纯标准库）。"""

    def test_empty_query_matches_everything(self):
        self.assertEqual(search.score("", "任意"), 0.0)
        self.assertTrue(search.matches("   ", "任意"))

    def test_contains_ignores_case(self):
        self.assertTrue(search.matches("svfs", "SVFS 桥调试台", "com.svfs.demo"))
        self.assertTrue(search.matches("SVFS", "svfs demo"))
        self.assertTrue(search.matches("桥", "SVFS 桥调试台", "com.svfs.demo"))

    def test_contains_beats_acronym(self):
        # "clock" 包含匹配；"clk" 只能缩写匹配 —— 包含分应更高
        contains = search.score("clo", "com.example.clock")
        acronym = search.score("clo", "c-l-o")
        self.assertIsNotNone(contains)
        self.assertIsNotNone(acronym)
        self.assertGreater(contains, acronym)

    def test_acronym_matches_subsequence(self):
        self.assertTrue(search.matches("csd", "SVFS 桥调试台", "com.svfs.demo"))
        self.assertTrue(search.matches("cec", "clock", "com.example.clock"))

    def test_position_matters(self):
        early = search.score("ab", "ab-xyz")
        late = search.score("ab", "xyz-abcdef")
        self.assertGreater(early, late)

    def test_no_match_returns_none(self):
        self.assertIsNone(search.score("zzz", "SVFS 桥调试台", "com.svfs.demo"))
        # 字符顺序不对 → 不算缩写匹配
        self.assertIsNone(search.score("dcs", "SVFS", "com.svfs.demo"))

    def test_matches_id_or_name(self):
        # 显示名不匹配但 id 匹配
        self.assertTrue(search.matches("demo", "调试台", "com.svfs.demo"))


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