#!/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 try: # 自建最小 broker（见 test_client_mqtt）——端到端用例要一个真 broker， # 但测试不应该依赖装 mosquitto from test_client_mqtt import _MiniBroker, wait_until HAVE_BROKER = True except Exception: HAVE_BROKER = False _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_message_push_and_sync(self): """服务端主动推消息：进该用户的待收列表，客户端登录后能补齐。""" token = self._login().json()["token"] headers = {"Authorization": f"Bearer {token}"} response = self.client.post("/api/push", headers=headers, json={ "kind": "message", "title": "备份完成", "body": "3 个包", "level": "warn"}) self.assertEqual(response.status_code, 200, response.text) body = response.json() # 主题按方案拼（组织位在第一段之后，多租户就靠它隔离） self.assertEqual(body["topic"], "svfs/default/user/admin/message") self.assertEqual(body["message"]["title"], "备份完成") self.assertTrue(body["message"]["id"]) self.assertTrue(body["stored"]) # 没配 broker：如实说明是跳过了，不是假装发成功 self.assertIn("skipped", body["mqtt"]) # 补齐：默认全给；给了 since 只给更新的 response = self.client.get("/api/messages", headers=headers) self.assertEqual(len(response.json()["messages"]), 1) newest = response.json()["messages"][0]["time"] response = self.client.get(f"/api/messages?since={newest}", headers=headers) self.assertEqual(response.json()["messages"], []) # 发给别人的消息进别人的待收列表（自己的列表不受影响） response = self.client.post("/api/push", headers=headers, json={"title": "给别人的", "user": "bob"}) self.assertEqual(response.json()["topic"], "svfs/default/user/bob/message") self.assertEqual( len(self.client.get("/api/messages", headers=headers) .json()["messages"]), 1) def test_plugin_push_goes_to_plugin_topic_and_not_backlog(self): """插件消息走插件主题、**不进**待收列表（仪表盘数据补历史没意义）。""" token = self._login().json()["token"] headers = {"Authorization": f"Bearer {token}"} response = self.client.post("/api/push", headers=headers, json={ "kind": "plugin", "plugin": "com.svfs.demo", "data": {"cpu": 0.42}}) self.assertEqual(response.status_code, 200, response.text) body = response.json() self.assertEqual(body["topic"], "svfs/default/user/admin/plugin/com.svfs.demo") self.assertFalse(body["stored"]) self.assertEqual( self.client.get("/api/messages", headers=headers).json()["messages"], []) # 少了 plugin 字段：400，别猜 response = self.client.post("/api/push", headers=headers, json={"kind": "plugin"}) self.assertEqual(response.status_code, 400) def test_broadcast_and_org_announcement(self): """广播走组织主题；登录响应下发组织标识（多租户预留位）。""" body = self._login().json() self.assertEqual(body["org"], "default") self.assertNotIn("mqtt", body) # 没配 broker 就不下发 headers = {"Authorization": f"Bearer {body['token']}"} response = self.client.post("/api/broadcast", headers=headers, json={"title": "全员通知", "body": "维护"}) self.assertEqual(response.status_code, 200, response.text) self.assertEqual(response.json()["topic"], "svfs/default/broadcast/message") def test_org_and_broker_announcement_follows_env(self): """服务端配了组织/broker：登录与 /api/me 都下发（客户端据此订阅）。""" keys = ("SVFS_ORG", "SVFS_MQTT_HOST", "SVFS_MQTT_PORT", "SVFS_MQTT_TLS") old = {key: os.environ.get(key) for key in keys} def restore(): for key, value in old.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value self.addCleanup(restore) os.environ.update({"SVFS_ORG": "acme", "SVFS_MQTT_HOST": "broker.acme", "SVFS_MQTT_PORT": "8883", "SVFS_MQTT_TLS": "true"}) body = self._login().json() self.assertEqual(body["org"], "acme") self.assertEqual(body["mqtt"], {"host": "broker.acme", "tls": True, "port": 8883, "prefix": "svfs", "org": "acme"}) headers = {"Authorization": f"Bearer {body['token']}"} self.assertEqual( self.client.get("/api/me", headers=headers).json()["org"], "acme") # 主题里的组织段跟着变（多租户隔离就靠它） response = self.client.post("/api/push", headers=headers, json={"title": "x"}) self.assertEqual(response.json()["topic"], "svfs/acme/user/admin/message") def test_push_requires_auth(self): self.assertEqual( self.client.post("/api/push", json={"title": "x"}).status_code, 401) self.assertEqual(self.client.get("/api/messages").status_code, 401) @unittest.skipUnless(HAVE_BROKER, "需要 paho-mqtt 与自建 broker 助手") def test_server_push_reaches_client_over_mqtt(self): """端到端：``/api/push`` → broker → 客户端 MQTT 通道 → 消息中心。 这条链路把三方接在一起验证：**broker 地址由登录响应下发**（客户端 不用手配）、服务端按主题方案发布、客户端用同一套主题方案订阅并让 消息落进总线。 """ from svfs_client import settings as client_settings from svfs_client.bus import MessageBus from svfs_client.mqtt import MqttTransport broker = _MiniBroker() self.addCleanup(broker.close) keys = ("SVFS_MQTT_HOST", "SVFS_MQTT_PORT", "SVFS_ORG") old = {key: os.environ.get(key) for key in keys} def restore(): for key, value in old.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value self.addCleanup(restore) os.environ.update({"SVFS_MQTT_HOST": "127.0.0.1", "SVFS_MQTT_PORT": str(broker.port), "SVFS_ORG": "acme"}) session = self._login().json() self.assertEqual(session["mqtt"]["host"], "127.0.0.1") self.assertEqual(session["org"], "acme") # 客户端侧：与服务端同一套主题方案 + 同一个解析器 base = client_paths.ensure_layout(self.path("data")) config = client_settings.mqtt_config(base, session) self.assertTrue(config["enabled"]) self.assertEqual(config["org"], "acme") bus = MessageBus(org=config["org"], user=session["username"]) transport = MqttTransport(config, on_message=bus.deliver) self.addCleanup(transport.stop) transport.start() self.assertTrue(wait_until(lambda: transport.is_connected()), "客户端没有连上 broker") # 服务端推一条 → 经 broker 到客户端 headers = {"Authorization": f"Bearer {session['token']}"} response = self.client.post("/api/push", headers=headers, json={ "kind": "message", "title": "备份完成", "body": "3 个包"}) self.assertEqual(response.json()["mqtt"], "published") self.assertTrue(wait_until(lambda: len(bus.log) == 1), "推送没有到达客户端，消息中心是空的") message = bus.log.items()[0] self.assertEqual(message["title"], "备份完成") self.assertEqual(message["scope"], "user") # 插件消息走另一条通道：不进消息中心，只进该插件的队列 response = self.client.post("/api/push", headers=headers, json={ "kind": "plugin", "plugin": "com.svfs.demo", "data": {"cpu": 0.5}}) self.assertEqual(response.json()["mqtt"], "published") self.assertTrue( wait_until(lambda: bus.pending("com.svfs.demo") == 1), "插件消息没有到达插件队列") self.assertEqual(len(bus.log), 1) # 组织广播：进消息中心并带 broadcast 标记 response = self.client.post("/api/broadcast", headers=headers, json={"title": "全员通知"}) self.assertEqual(response.json()["mqtt"], "published") self.assertTrue(wait_until(lambda: len(bus.log) == 2)) self.assertEqual(bus.log.items()[1]["scope"], "broadcast") 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()