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