chakcy
f62704ca0e
一个插件 = 一个完整的 .svfs 包：客户端把插件前端经 svfs://<插件id>/ 直接加载 进 WebView（不落盘、不起端口、不解包），每个插件一个独立 origin（浏览器按 origin 隔离 localStorage）与独立键值存储；快照即版本——升级 = 派生快照的增量 打包，回滚 = 把注册表指针切回旧快照，包文件一个字节不用动。svfs_client 复用 svfs_studio 的主题与基础部件，故两个包一并入库。 按评估结论补齐的 P0 三块： * 会话：桥在"宿主注入的 token 被 401 拒绝"时通知壳层，提示一次并给出一键 「重新登录…」；服务器状态灯把「在线但登录失效」（黄）与「离线」（红）分开， 不再把凭据失效误报成网络不可达； * 凭据：新增 credentials.py——token 不再明文进 settings.json。Windows 走 DPAPI （密文只有当前用户在本机解得开），其它平台走系统凭据库（可选依赖 keyring）， 都没有才明文兜底并在「设置 → 登录凭据」如实提示；令牌与签发它的 base_url 绑定（换服务器不会把旧令牌发出去）；老字段 auth_token 启动时自动迁移； * 强登录链路保持：启动动画 → 静默校验 → 登录窗 → 主窗口，不可达时离线宽限。 P1/P2：net 超时与重试（只对幂等方法，避免重发 POST 变两单）可配、代理与 自签 CA 可配（loopback 永远直连）；诊断日志滚动落盘（1MB×3，可关）——崩溃后 还能回看现场；客户端自更新检查（静态 <base_url>/client/manifest.json，缺失 即静默）；插件清单可选 net_hosts 白名单；notify 改走系统托盘（无托盘降级状态栏）； 设置界面新增网络 / 登录凭据 / 诊断日志三块。 examples/：示例服务端（FastAPI，含登录、上传即 publish、客户端版本接口）、 Vue 3 插件模板（带浏览器版宿主模拟器）、svfs-plugins-root/ 为模板配套的演示 插件实体。测试：tests/test_client_*.py（纯层 + Qt 层 + 一次真实 WebEngine 冒烟）、 tests/test_studio_*.py、tests/test_plugin_server.py。
745 lines
32 KiB
Python
745 lines
32 KiB
Python
#!/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": "", "cache_dir": "",
|
|
"net_timeout": settings.NET_TIMEOUT_DEFAULT,
|
|
"net_retries": settings.NET_RETRIES_DEFAULT,
|
|
"proxy": "", "ca_bundle": "", "log_to_file": True},
|
|
)
|
|
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": "", "cache_dir": "",
|
|
"net_timeout": settings.NET_TIMEOUT_DEFAULT,
|
|
"net_retries": settings.NET_RETRIES_DEFAULT,
|
|
"proxy": "", "ca_bundle": "", "log_to_file": True},
|
|
)
|
|
|
|
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 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()
|