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。
234 lines
9.3 KiB
Python
234 lines
9.3 KiB
Python
#!/usr/bin/env python3
|
||
"""examples/plugin_server.py 的最小验证（需要 fastapi + httpx，缺则整模块跳过）。
|
||
|
||
覆盖两条链路：
|
||
|
||
* API 本身：登录换 token、/api/ping（需 Bearer token）、上传 .svfs →
|
||
服务端 publish → 清单/对象可被 GET；
|
||
* **完整闭环**：经 HTTP 上传后，起一个真实 HTTP 服务吐插件目录，
|
||
用客户端的 ``installer.install_from_url`` 从服务端装回本地。
|
||
"""
|
||
|
||
import importlib.util
|
||
import json
|
||
import os
|
||
import threading
|
||
import unittest
|
||
from functools import partial
|
||
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
||
|
||
import helpers
|
||
from helpers import VFSTestCase, quiet
|
||
|
||
try:
|
||
from fastapi.testclient import TestClient
|
||
|
||
HAVE_FASTAPI = True
|
||
except ImportError:
|
||
HAVE_FASTAPI = False
|
||
|
||
EXAMPLE_PATH = os.path.join(helpers.REPO_ROOT, "examples", "plugin_server.py")
|
||
|
||
if HAVE_FASTAPI:
|
||
from svfs_client import installer, registry
|
||
from svfs_client import paths as client_paths
|
||
from svfs_client import sample as client_sample
|
||
|
||
_load_seq = [0]
|
||
|
||
def _load_example(plugins_dir):
|
||
"""以指定插件目录加载示例模块（StaticFiles 在 import 时建 mount）。"""
|
||
_load_seq[0] += 1
|
||
os.environ["SVFS_PLUGINS_DIR"] = plugins_dir
|
||
spec = importlib.util.spec_from_file_location(
|
||
f"plugin_server_example_{_load_seq[0]}", EXAMPLE_PATH
|
||
)
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
return module
|
||
|
||
|
||
if HAVE_FASTAPI:
|
||
|
||
class PluginServerTest(VFSTestCase):
|
||
def setUp(self):
|
||
super().setUp()
|
||
self.plugins_dir = self.path("plugins-root")
|
||
self.mod = _load_example(self.plugins_dir)
|
||
self.client = TestClient(self.mod.app)
|
||
|
||
def _build_sample(self, name="sample.svfs"):
|
||
package = self.path(name)
|
||
with quiet():
|
||
client_sample.build(package)
|
||
return package
|
||
|
||
def _upload(self, plugin_id, pkg_path, token=None):
|
||
with open(pkg_path, "rb") as handle:
|
||
headers = {"X-SVFS-Token": token} if token is not None else {}
|
||
return self.client.post(
|
||
f"/plugins/{plugin_id}",
|
||
files={"file": ("plugin.svfs", handle.read())},
|
||
headers=headers,
|
||
)
|
||
|
||
def _login(self, username="admin", password="admin123"):
|
||
return self.client.post(
|
||
"/api/login", json={"username": username, "password": password}
|
||
)
|
||
|
||
def test_index_open_ping_requires_auth(self):
|
||
# /api/* 需要登录；服务索引保持开放
|
||
response = self.client.get("/api/ping")
|
||
self.assertEqual(response.status_code, 401)
|
||
|
||
response = self.client.get("/")
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertEqual(response.json()["plugins"], [])
|
||
|
||
def test_login_then_ping(self):
|
||
response = self._login()
|
||
self.assertEqual(response.status_code, 200, response.text)
|
||
body = response.json()
|
||
self.assertEqual(body["username"], "admin")
|
||
self.assertTrue(body["token"])
|
||
|
||
headers = {"Authorization": f"Bearer {body['token']}"}
|
||
response = self.client.get("/api/ping", headers=headers)
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertTrue(response.json()["pong"])
|
||
self.assertEqual(response.json()["user"], "admin")
|
||
|
||
def test_login_rejects_bad_credentials(self):
|
||
response = self._login(password="wrong")
|
||
self.assertEqual(response.status_code, 401)
|
||
self.assertIn("不正确", response.json()["detail"])
|
||
|
||
def test_me_and_logout(self):
|
||
token = self._login().json()["token"]
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
response = self.client.get("/api/me", headers=headers)
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertEqual(response.json()["username"], "admin")
|
||
|
||
response = self.client.post("/api/logout", headers=headers)
|
||
self.assertEqual(response.status_code, 200)
|
||
|
||
# token 已作废
|
||
response = self.client.get("/api/me", headers=headers)
|
||
self.assertEqual(response.status_code, 401)
|
||
|
||
def test_custom_users_env(self):
|
||
os.environ["SVFS_USERS"] = "alice:pw1, bob: pw2"
|
||
try:
|
||
response = self._login() # 默认演示账号被关闭
|
||
self.assertEqual(response.status_code, 401)
|
||
response = self._login("alice", "pw1")
|
||
self.assertEqual(response.status_code, 200)
|
||
response = self._login("bob", "pw2") # 值两侧空白被容忍
|
||
self.assertEqual(response.status_code, 200)
|
||
finally:
|
||
del os.environ["SVFS_USERS"]
|
||
|
||
def test_upload_then_serve(self):
|
||
package = self._build_sample()
|
||
response = self._upload(client_sample.PLUGIN_ID, package)
|
||
self.assertEqual(response.status_code, 200, response.text)
|
||
body = response.json()
|
||
self.assertTrue(body["ok"])
|
||
self.assertEqual(body["id"], client_sample.PLUGIN_ID)
|
||
self.assertEqual(body["version"], "1.0.0")
|
||
|
||
# 服务列表与 manifest 可取
|
||
self.assertEqual(self.client.get("/").json()["plugins"],
|
||
[client_sample.PLUGIN_ID])
|
||
response = self.client.get(f"/{client_sample.PLUGIN_ID}/manifest.json")
|
||
self.assertEqual(response.status_code, 200)
|
||
manifest = response.json()
|
||
self.assertEqual(manifest["format"], "svfs-objects")
|
||
|
||
# 任意一个 blob 也能按 hash 取到
|
||
blob_hash = next(
|
||
entry["hash"] for entry in manifest["entries"].values()
|
||
if "hash" in entry
|
||
)
|
||
response = self.client.get(
|
||
f"/{client_sample.PLUGIN_ID}/blobs/{blob_hash[:2]}/{blob_hash}"
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertTrue(len(response.content) > 0)
|
||
|
||
def test_upload_rejects_mismatched_id(self):
|
||
package = self._build_sample()
|
||
response = self._upload("com.other.app", package)
|
||
self.assertEqual(response.status_code, 400)
|
||
self.assertIn("不一致", response.json()["detail"])
|
||
|
||
def test_upload_rejects_corrupt_package(self):
|
||
junk = self.path("junk.svfs")
|
||
with open(junk, "wb") as handle:
|
||
handle.write(b"not a sqlite file")
|
||
response = self._upload("com.other.app", junk)
|
||
self.assertEqual(response.status_code, 400)
|
||
|
||
def test_upload_token(self):
|
||
package = self._build_sample()
|
||
os.environ["SVFS_UPLOAD_TOKEN"] = "secret"
|
||
try:
|
||
response = self._upload(client_sample.PLUGIN_ID, package)
|
||
self.assertEqual(response.status_code, 403)
|
||
response = self._upload(client_sample.PLUGIN_ID, package,
|
||
token="wrong")
|
||
self.assertEqual(response.status_code, 403)
|
||
response = self._upload(client_sample.PLUGIN_ID, package,
|
||
token="secret")
|
||
self.assertEqual(response.status_code, 200)
|
||
finally:
|
||
del os.environ["SVFS_UPLOAD_TOKEN"]
|
||
|
||
class PluginServerEndToEndTest(VFSTestCase):
|
||
"""上传 → 服务端 publish → 真实 HTTP 服务 → 客户端 install_from_url。"""
|
||
|
||
def test_full_circle(self):
|
||
plugins_dir = self.path("plugins-root")
|
||
mod = _load_example(plugins_dir)
|
||
client = TestClient(mod.app)
|
||
|
||
package = self.path("sample.svfs")
|
||
with quiet():
|
||
client_sample.build(package)
|
||
with open(package, "rb") as handle:
|
||
response = client.post(
|
||
f"/plugins/{client_sample.PLUGIN_ID}",
|
||
files={"file": ("plugin.svfs", handle.read())},
|
||
)
|
||
self.assertEqual(response.status_code, 200, response.text)
|
||
|
||
# 用真实 HTTP 服务吐出插件目录，走客户端完整的远程安装链路
|
||
server = ThreadingHTTPServer(
|
||
("127.0.0.1", 0), partial(SimpleHTTPRequestHandler,
|
||
directory=plugins_dir)
|
||
)
|
||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||
thread.start()
|
||
self.addCleanup(server.shutdown)
|
||
base = f"http://127.0.0.1:{server.server_address[1]}"
|
||
|
||
data_base = client_paths.ensure_layout(self.path("data"))
|
||
with quiet():
|
||
entry = installer.install_from_url(
|
||
f"{base}/{client_sample.PLUGIN_ID}/manifest.json",
|
||
base=data_base,
|
||
)
|
||
self.assertEqual(entry["version"], "1.0.0")
|
||
self.assertEqual(entry["source"]["type"], "url")
|
||
self.assertEqual(
|
||
registry.load(data_base)["plugins"][client_sample.PLUGIN_ID]["version"],
|
||
"1.0.0",
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|