svfs/tests/test_client_ui.py

#!/usr/bin/env python3
"""svfs_client Qt 层的界面测试（offscreen 渲染）。

约定与 test_studio_ui.py 相同：``QT_QPA_PLATFORM=offscreen`` 必须在任何
PySide6 导入之前设置；没装含 WebEngine 的 PySide6 时整个模块干净跳过。

WebEngine 冒烟测试会真正创建 QWebEngineView 并加载示例插件——部分环境
（无显卡/容器）可能起不来，那种情况跳过而不是误报失败。
"""

import os

os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
# WebEngine 在无显示环境下的兜底（可用环境变量覆盖）
os.environ.setdefault("QTWEBENGINE_CHROMIUM_FLAGS", "--no-sandbox --disable-gpu")

import json
import sys
import tempfile
import threading
import time
import unittest
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from helpers import VFSTestCase, make_tree, quiet

from sqlite_vfs.core import SQLiteVFS

try:
    from PySide6.QtCore import QSettings, Qt, QUrl
    from PySide6.QtWidgets import QApplication
    from PySide6.QtWebEngineWidgets import QWebEngineView  # noqa: F401 - WebEngine 探测

    HAVE_QT = True
except ImportError:
    HAVE_QT = False

if HAVE_QT:
    from svfs_client import PLUGIN_API_VERSION, paths as client_paths, settings as client_settings
    from svfs_client.app import create_app
    from svfs_client.bridge import SvfsBridge
    from svfs_client.scheme import (
        SvfsSchemeHandler,
        looks_like_navigation,
        make_lookup,
        plugin_url,
    )
    from svfs_client.worker import Client

    from svfs_client import sample as client_sample
    from svfs_client.installer import install_from_file
    from svfs_client.shell import MainWindow
    from svfs_studio import theme as client_theme


class _ApiHandler(BaseHTTPRequestHandler):
    """net 测试用的最小 API 服务：/api/ping → 200，/<id>/manifest.json → 清单，
    /api/private → 恒 401，其它 → 404。``manifest_version`` 非 None 时用来模拟
    服务器上有新版本；``ping_status`` 用来模拟"服务器在线但拒绝凭据"。"""

    manifest_version = None
    ping_status = 200
    client_version = ""

    def _reply(self, code, body):
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path == "/api/ping":
            if _ApiHandler.ping_status != 200:
                self._reply(_ApiHandler.ping_status, b'{"detail": "unauthorized"}')
                return
            self._reply(200, b'{"pong": true}')
            return
        if self.path == "/api/private":
            # 恒定 401：登录失效链路（桥的信号 + 状态灯）测试用
            self._reply(401, b'{"detail": "token expired"}')
            return
        if self.path == "/client/manifest.json":
            if not _ApiHandler.client_version:
                self._reply(404, b'{"error": "no client manifest"}')
                return
            body = json.dumps({
                "version": _ApiHandler.client_version,
                "url": "https://example.invalid/SVFS-Client.zip",
                "notes": "测试用新版本",
            }).encode("utf-8")
            self._reply(200, body)
            return
        if (_ApiHandler.manifest_version
                and self.path.endswith("/manifest.json")):
            plugin_id = self.path.strip("/").split("/")[0]
            body = json.dumps({
                "format": "svfs-objects", "version": 1, "name": plugin_id,
                "snapshot": {"name": _ApiHandler.manifest_version,
                             "digest": "deadbeef"},
                "entries": {},
            }).encode("utf-8")
            self._reply(200, body)
            return
        self._reply(404, b'{"error": "not found"}')

    def log_message(self, *args):  # 静音
        pass


def start_api_server():
    server = ThreadingHTTPServer(("127.0.0.1", 0), _ApiHandler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server, f"http://127.0.0.1:{server.server_address[1]}"


def pump(app, predicate=None, timeout=15.0, interval=0.02, settle=0.08):
    """跑事件循环：predicate 为 None 时跑 settle 秒；否则等到谓词成立或超时。"""
    deadline = time.monotonic() + timeout
    if predicate is None:
        end = time.monotonic() + settle
        while time.monotonic() < end:
            app.processEvents()
            time.sleep(interval)
        return True
    while time.monotonic() < deadline:
        app.processEvents()
        time.sleep(interval)
        if predicate():
            end = time.monotonic() + settle
            while time.monotonic() < end:
                app.processEvents()
                time.sleep(interval)
            return True
    return False


if HAVE_QT:

    _APP = None

    def setUpModule():
        global _APP
        # QSettings 重定向到临时目录，测试不碰真实注册表/配置
        QSettings.setDefaultFormat(QSettings.Format.IniFormat)
        QSettings.setPath(
            QSettings.Format.IniFormat, QSettings.Scope.UserScope,
            tempfile.mkdtemp(prefix="svfs_client_settings_"),
        )
        # create_app 内部会先 register_scheme() 再建 QApplication（顺序是硬要求）
        _APP = QApplication.instance() or create_app([sys.argv[0]])
        _APP.setQuitOnLastWindowClosed(False)

    class BridgeTest(VFSTestCase):
        """桥的权限模型与存储隔离（不经过 WebView）。"""

        def bridge(self, permissions):
            return SvfsBridge(
                "com.example.demo", permissions,
                client_paths.storage_file(self.path("data"), "com.example.demo"),
            )

        def test_storage_roundtrip(self):
            bridge = self.bridge(["storage"])
            self.assertTrue(bridge.storageSet("visits", "42"))
            self.assertEqual(json.loads(bridge.storageGet("visits")), 42)
            self.assertEqual(bridge.storageKeys(), ["visits"])
            self.assertTrue(bridge.storageRemove("visits"))
            self.assertIsNone(bridge.storageGet("visits"))

        def test_storage_persists_to_isolated_file(self):
            self.bridge(["storage"]).storageSet("k", '"v"')
            self.assertTrue(
                client_paths.storage_file(self.path("data"), "com.example.demo").endswith(
                    "com.example.demo.json"
                )
            )
            with open(client_paths.storage_file(self.path("data"), "com.example.demo"),
                      encoding="utf-8") as f:
                self.assertEqual(json.load(f), {"k": "v"})

        def test_permission_denied(self):
            bridge = self.bridge([])  # 什么都没声明
            self.assertFalse(bridge.storageSet("k", "1"))
            self.assertIsNone(bridge.storageGet("k"))
            self.assertFalse(bridge.notify("hi"))
            self.assertFalse(bridge.setTitle("t"))
            self.assertEqual(bridge.storageKeys(), [])

        def test_bad_values_rejected(self):
            bridge = self.bridge(["storage"])
            self.assertFalse(bridge.storageSet("k", "not-json"))
            self.assertFalse(bridge.storageSet("", "1"))
            self.assertFalse(bridge.storageSet("x" * 300, "1"))

        def test_notify_and_title_signals(self):
            bridge = self.bridge(["notify", "title"])
            seen = []
            bridge.notifyRequested.connect(seen.append)
            bridge.titleRequested.connect(lambda t: seen.append("T:" + t))
            self.assertTrue(bridge.notify("你好"))
            self.assertTrue(bridge.setTitle("新标题"))
            pump(_APP, settle=0.05)
            self.assertIn("你好", seen)
            self.assertIn("T:新标题", seen)

        def test_host_info_needs_no_permission(self):
            info = self.bridge([]).hostInfo()
            self.assertEqual(info["pluginId"], "com.example.demo")
            self.assertEqual(info["apiVersion"], PLUGIN_API_VERSION)
            self.assertEqual(info["baseUrl"], "")  # 没给 provider

        def test_host_info_carries_base_url(self):
            bridge = SvfsBridge(
                "com.example.demo", [],
                client_paths.storage_file(self.path("data"), "com.example.demo"),
                base_url_provider=lambda: " http://srv:8000/ ",  # 故意带斜杠和空格
            )
            info = bridge.hostInfo()
            self.assertEqual(info["baseUrl"], "http://srv:8000")

        def test_dialog_needs_permission(self):
            bridge = SvfsBridge(
                "com.example.demo", ["net"],
                client_paths.storage_file(self.path("data"), "com.example.demo"),
            )
            self.addCleanup(bridge.shutdown)
            result = json.loads(
                bridge.uiDialog(json.dumps({"kind": "info", "text": "hi"}))
            )
            self.assertFalse(result["ok"])
            self.assertIn("权限", result["error"])

        def test_dialog_rejects_bad_payload(self):
            bridge = SvfsBridge(
                "com.example.demo", ["dialog"],
                client_paths.storage_file(self.path("data"), "com.example.demo"),
            )
            self.addCleanup(bridge.shutdown)
            result = json.loads(bridge.uiDialog("not json"))
            self.assertFalse(result["ok"])

        def test_dialog_shows_and_returns(self):
            # 真弹原生 QMessageBox：定时器在嵌套事件循环里把它自动关掉
            # （等价于用户点掉了对话框），slot 才会返回。
            bridge = SvfsBridge(
                "com.example.demo", ["dialog"],
                client_paths.storage_file(self.path("data"), "com.example.demo"),
            )
            self.addCleanup(bridge.shutdown)
            from PySide6.QtCore import QTimer

            def close_modal():
                modal = QApplication.activeModalWidget()
                if modal is not None:
                    modal.close()

            # confirm（Ok/Cancel 两个按钮）：直接关闭 = 取消 → accepted False
            QTimer.singleShot(150, close_modal)
            result = json.loads(
                bridge.uiDialog(json.dumps(
                    {"kind": "confirm", "title": "调试", "text": "自动关闭"}
                ))
            )
            self.assertTrue(result["ok"])
            self.assertFalse(result["accepted"])

            # message info（只有 Ok 一个按钮）：直接关闭 = 默认按钮 Ok → True
            QTimer.singleShot(150, close_modal)
            result = json.loads(
                bridge.uiDialog(json.dumps(
                    {"kind": "info", "title": "调试", "text": "自动关闭"}
                ))
            )
            self.assertTrue(result["ok"])
            self.assertTrue(result["accepted"])

        def test_net_request_roundtrip(self):
            server, base = start_api_server()
            self.addCleanup(server.shutdown)
            bridge = SvfsBridge(
                "com.example.demo", ["net"],
                client_paths.storage_file(self.path("data"), "com.example.demo"),
                base_url_provider=lambda: base,
            )
            self.addCleanup(bridge.shutdown)
            seen = []
            bridge.netResult.connect(seen.append)
            req_id = bridge.netRequest(json.dumps({"path": "/api/ping"}))
            self.assertTrue(req_id)
            self.assertTrue(pump(_APP, predicate=lambda: bool(seen), timeout=15))
            msg = json.loads(seen[-1])
            self.assertEqual(msg["reqId"], req_id)
            self.assertTrue(msg["ok"])
            self.assertEqual(msg["status"], 200)
            self.assertIn("pong", msg["bodyText"])

        def test_net_http_error_and_missing_permission(self):
            server, base = start_api_server()
            self.addCleanup(server.shutdown)
            bridge = SvfsBridge(
                "com.example.demo", ["net"],
                client_paths.storage_file(self.path("data"), "com.example.demo"),
                base_url_provider=lambda: base,
            )
            self.addCleanup(bridge.shutdown)
            seen = []
            bridge.netResult.connect(seen.append)
            # 404 走 HTTPError 分支：请求送达，状态码交给插件处理
            bridge.netRequest(json.dumps({"path": "/api/nope"}))
            self.assertTrue(pump(_APP, predicate=lambda: bool(seen), timeout=15))
            msg = json.loads(seen[-1])
            self.assertFalse(msg["ok"])
            self.assertEqual(msg["status"], 404)
            # 无效方法被拒
            before = len(seen)
            bridge.netRequest(json.dumps({"path": "/api/ping", "method": "TRACE"}))
            self.assertTrue(pump(_APP, predicate=lambda: len(seen) > before, timeout=15))
            self.assertIn("error", json.loads(seen[-1]))

        def test_net_denied_without_permission(self):
            bridge = self.bridge([])  # 未声明 net
            seen = []
            bridge.netResult.connect(seen.append)
            self.assertEqual(bridge.netRequest(json.dumps({"path": "/api/ping"})), "")
            pump(_APP, settle=0.2)
            self.assertEqual(seen, [])

        def test_net_host_allowlist_blocks_other_hosts(self):
            """清单声明了 net_hosts：白名单外的主机被拒（请求根本发不出去）。

            拒绝也要回一个明确错误给插件（不是静默超时），否则插件作者
            只会看到"请求中…"。
            """
            bridge = SvfsBridge(
                "com.example.demo", ["net"],
                client_paths.storage_file(self.path("data"), "com.example.demo"),
                base_url_provider=lambda: "http://127.0.0.1:1",
                allowed_hosts=["api.example.com"],
            )
            self.addCleanup(bridge.shutdown)
            seen = []
            bridge.netResult.connect(seen.append)
            bridge.netRequest(json.dumps(
                {"url": "http://evil.example/collect", "method": "GET"}))
            self.assertTrue(pump(_APP, predicate=lambda: bool(seen), timeout=10))
            msg = json.loads(seen[-1])
            self.assertFalse(msg["ok"])
            self.assertIn("net_hosts", msg["error"])
            self.assertIn("evil.example", msg["error"])

        def test_net_host_allowlist_allows_listed_host(self):
            server, base = start_api_server()
            self.addCleanup(server.shutdown)
            self.addCleanup(server.server_close)
            bridge = SvfsBridge(
                "com.example.demo", ["net"],
                client_paths.storage_file(self.path("data"), "com.example.demo"),
                base_url_provider=lambda: base,
                allowed_hosts=["127.0.0.1"],  # 不带端口：按主机名匹配
            )
            self.addCleanup(bridge.shutdown)
            seen = []
            bridge.netResult.connect(seen.append)
            self.assertTrue(bridge.netRequest(json.dumps({"path": "/api/ping"})))
            self.assertTrue(pump(_APP, predicate=lambda: bool(seen), timeout=15))
            self.assertEqual(json.loads(seen[-1])["status"], 200)

        def test_net_401_reports_session_rejection(self):
            """宿主注入的 token 被 401 拒绝 → 通知壳层引导重新登录。

            插件拿到的应答不变（401 仍照常交付），只是宿主多知道一件事。
            """
            server, base = start_api_server()
            self.addCleanup(server.shutdown)
            self.addCleanup(server.server_close)
            bridge = SvfsBridge(
                "com.example.demo", ["net"],
                client_paths.storage_file(self.path("data"), "com.example.demo"),
                base_url_provider=lambda: base,
                session_provider=lambda: {"token": "tok", "username": "admin"},
            )
            self.addCleanup(bridge.shutdown)
            seen, rejected = [], []
            bridge.netResult.connect(seen.append)
            bridge.sessionRejected.connect(rejected.append)
            self.assertTrue(bridge.netRequest(json.dumps({"path": "/api/private"})))
            self.assertTrue(pump(_APP, predicate=lambda: bool(seen), timeout=15))
            msg = json.loads(seen[-1])
            self.assertFalse(msg["ok"])
            self.assertEqual(msg["status"], 401)
            # 信号从工作线程发出，排队到主线程 → 再跑一次事件循环
            self.assertTrue(pump(_APP, predicate=lambda: bool(rejected), timeout=5),
                            "401 没有触发 sessionRejected")
            self.assertIn("401", rejected[0])

        def test_net_401_without_host_token_is_silent(self):
            """没有宿主注入的 token（未登录 / 插件自带 Authorization）时，
            401 与宿主会话无关，不该惊动用户。"""
            server, base = start_api_server()
            self.addCleanup(server.shutdown)
            self.addCleanup(server.server_close)

            for provider, extra in (
                (None, {}),  # 未登录：没有可注入的凭据
                (lambda: {"token": "tok", "username": "admin"},
                 {"headers": {"Authorization": "Bearer plugin-own"}}),  # 插件自带
            ):
                bridge = SvfsBridge(
                    "com.example.demo", ["net"],
                    client_paths.storage_file(self.path("data"),
                                              "com.example.demo"),
                    base_url_provider=lambda: base,
                    session_provider=provider,
                )
                self.addCleanup(bridge.shutdown)
                seen, rejected = [], []
                bridge.netResult.connect(seen.append)
                bridge.sessionRejected.connect(rejected.append)
                payload = {"path": "/api/private", **extra}
                self.assertTrue(bridge.netRequest(json.dumps(payload)))
                self.assertTrue(pump(_APP, predicate=lambda: bool(seen), timeout=15))
                self.assertEqual(json.loads(seen[-1])["status"], 401)
                pump(_APP, settle=0.2)
                self.assertEqual(rejected, [])

    class SchemeHandlerTest(VFSTestCase):
        """协议层的路径解析、查找闭包与句柄缓存。"""

        def setUp(self):
            super().setUp()
            self.base = client_paths.ensure_layout(self.path("data"))
            package = self.path("plugin.svfs")
            with quiet():
                client_sample.build(package)
                install_from_file(package, base=self.base)
            self.pid = client_sample.PLUGIN_ID

        def test_make_lookup_respects_registry(self):
            lookup = make_lookup(self.base)
            entry = lookup(self.pid)
            self.assertIsNotNone(entry)
            self.assertEqual(entry["entry"], "index.html")
            self.assertTrue(entry["package_path"].endswith("plugin.svfs"))
            # 未安装的插件拿不到
            self.assertIsNone(lookup("com.nope.app"))
            # 禁用后拿不到（隔离由 lookup 把守）
            from svfs_client import installer

            installer.set_enabled(self.base, self.pid, False)
            self.assertIsNone(lookup(self.pid))

        def test_package_handle_reads_bytes(self):
            lookup = make_lookup(self.base)
            handle = SvfsSchemeHandler(lookup)._handle_for(lookup(self.pid))
            data, ok = handle.read("index.html")
            self.assertTrue(ok)
            self.assertIn(b"<!doctype html>", data)
            # 桥的用法在 app.js 里（页面里引用了它）
            data, ok = handle.read("app.js")
            self.assertTrue(ok)
            self.assertIn(b"svfsReady", data)
            data, ok = handle.read("no/such/file.js")
            self.assertFalse(ok)
            self.assertEqual(data, b"")
            handle.close()

        def test_normalize_rejects_climbing(self):
            normalize = SvfsSchemeHandler._normalize
            self.assertEqual(normalize("/"), "")
            self.assertEqual(normalize("/app.js"), "app.js")
            # 相对段被收敛后落回包内：/a/../b → b（仍在包根层级，安全）
            self.assertEqual(normalize("/a/../b"), "b")
            # 绝对路径穿出顶层会被收敛回根下，同样安全
            self.assertEqual(normalize("/../x"), "x")
            self.assertEqual(normalize("/a/../../etc/passwd"), "etc/passwd")
            # 收敛不掉的相对段（非 QUrl 常态，纯防御）直接拒绝
            self.assertIsNone(normalize("../../etc/passwd"))

        def test_spa_heuristic(self):
            self.assertTrue(looks_like_navigation("settings"))
            self.assertTrue(looks_like_navigation("user/42"))
            self.assertTrue(looks_like_navigation("assets/logo"))  # 无扩展名按路由对待（与 fastapi_static 同一启发式）
            self.assertTrue(looks_like_navigation("images/"))      # 目录路径同理（先试目录索引，再回退 SPA）
            self.assertFalse(looks_like_navigation("app.js"))

        def test_invalidate_closes_handles(self):
            lookup = make_lookup(self.base)
            handler = SvfsSchemeHandler(lookup)
            entry = lookup(self.pid)
            handle = handler._handle_for(entry)
            self.assertIsNotNone(handle.read("index.html")[0])
            handler.invalidate(self.pid)
            # 句柄被关闭：缓存为空，老句柄再也读不到内容
            self.assertEqual(len(handler._cache), 0)
            data, ok = handle.read("index.html")
            self.assertFalse(ok)
            self.assertEqual(data, b"")

    class MainWindowTest(VFSTestCase):
        """主窗口渲染与工作线程流程（不创建 WebEngine 视图）。"""

        def setUp(self):
            super().setUp()
            self.base = client_paths.ensure_layout(self.path("data"))
            # 侧栏收缩/最近打开经 QSettings 跨用例持久化（真实行为就是记住），
            # 测试之间必须复位，否则前一个用例的状态会带进下一个窗口
            from svfs_client.shell import _ui_settings

            _ui_settings().remove("sidebar/collapsed")
            _ui_settings().remove("ui/last_plugin")
            self.window = MainWindow(self.base)
            self.window.show()
            pump(_APP)
            # 等第一次列表到达（启动直显只消费第一次列表），测试才有确定状态
            self.assertTrue(pump(_APP, predicate=lambda: self.window._startup_resolved))

        def tearDown(self):
            self.window._thread.quit()
            self.window._thread.wait(5000)
            self.window.deleteLater()
            pump(_APP)
            super().tearDown()

        def _submit_and_wait(self, op, payload, timeout=20.0):
            done, failed = [], []
            self.window.client.submit(
                op, payload, on_done=lambda p: done.append(p),
                on_fail=lambda t, d: failed.append((t, d)),
            )
            self.assertTrue(pump(_APP, predicate=lambda: done or failed,
                                timeout=timeout),
                            f"{op} 没有在 {timeout}s 内完成")
            self.assertFalse(failed, f"{op} 失败: {failed}")
            return done[0]

        def test_empty_render(self):
            self.assertIs(self.window.stack.currentWidget(), self.window.manager)
            self.assertTrue(self.window.manager.empty.isVisibleTo(self.window.manager))
            self.assertEqual(self.window.plugin_list.count(), 0)

        def test_install_sample_flow(self):
            payload = self._submit_and_wait("install_sample", {})
            self.window.apply_payload(payload)
            pump(_APP)
            pid = client_sample.PLUGIN_ID
            self.assertIn(pid, self.window.manager.cards)
            card = self.window.manager.cards[pid]
            self.assertTrue(card.enabled_check.isChecked())
            self.assertTrue(card.open_button.isEnabled())
            self.assertFalse(self.window.manager.empty.isVisibleTo(self.window.manager))
            # 注册表里也是启用状态
            from svfs_client import registry

            self.assertTrue(
                registry.load(self.base)["plugins"][pid]["enabled"]
            )

        def test_install_rejects_non_plugin_package(self):
            # 打一个没有 plugin.json 的普通包
            src = make_tree(self.path("bare"), {"readme.txt": b"not a plugin"})
            bare = self.path("bare.svfs")
            packer = SQLiteVFS(bare, compress=True)
            try:
                from sqlite_vfs.folder_packer import FolderPacker

                FolderPacker(packer).pack_folder(src, incremental=True)
            finally:
                packer.close()
            failed = []
            self.window.client.submit(
                "install_file", {"path": bare},
                on_done=lambda p: None,
                on_fail=lambda t, d: failed.append((t, d)),
            )
            self.assertTrue(pump(_APP, predicate=lambda: bool(failed)))
            title, detail = failed[0]
            self.assertIn("plugin.json", detail)

        def test_disable_removes_from_sidebar(self):
            payload = self._submit_and_wait("install_sample", {})
            self.window.apply_payload(payload)
            pump(_APP)
            pid = client_sample.PLUGIN_ID
            payload = self._submit_and_wait(
                "set_enabled", {"id": pid, "enabled": False}
            )
            self.window.apply_payload(payload)
            pump(_APP)
            self.assertEqual(self.window.plugin_list.count(), 0)
            card = self.window.manager.cards[pid]
            self.assertEqual(card.state_badge.text(), "已禁用")
            self.assertFalse(card.open_button.isEnabled())

        def test_version_switch_updates_registry(self):
            with quiet():
                client_sample.build(self.path("sample.svfs"))
                install_from_file(self.path("sample.svfs"), base=self.base)
                client_sample.build(self.path("sample.svfs"))  # 升到 2.0.0
                install_from_file(self.path("sample.svfs"), base=self.base)
            payload = self._submit_and_wait("list_plugins", {})
            self.window.apply_payload(payload)
            pump(_APP)
            pid = client_sample.PLUGIN_ID
            entry = self.window.entry_of(pid)
            self.assertEqual(len(entry["snapshots"]), 2)
            old_id = entry["snapshots"][0]["id"]
            payload = self._submit_and_wait("use_version",
                                            {"id": pid, "snapshot_id": old_id})
            self.window.apply_payload(payload)
            pump(_APP)
            self.assertEqual(
                self.window.entry_of(pid)["version"],
                entry["snapshots"][0]["name"],
            )

        def test_verify_reports_ok(self):
            with quiet():
                client_sample.build(self.path("sample.svfs"))
                install_from_file(self.path("sample.svfs"), base=self.base)
            done = []
            self.window.client.submit(
                "verify", {"id": client_sample.PLUGIN_ID, "check_content": True},
                on_done=done.append,
            )
            self.assertTrue(pump(_APP, predicate=lambda: bool(done)))
            self.assertTrue(done[0]["verify"]["ok"], done[0]["verify"]["errors"])

        def test_server_config_flow(self):
            # 初始未配置
            self.assertIn("未配置", self.window.manager.server_url_label.full_text())
            # 设置（故意带尾部斜杠，应被规范化）
            self.assertIsNone(self.window.set_base_url(" http://10.0.0.5:9000/ "))
            self.assertEqual(
                self.window.manager.server_url_label.full_text(), "http://10.0.0.5:9000"
            )
            self.assertEqual(
                client_settings.load(self.base)["base_url"], "http://10.0.0.5:9000"
            )
            self.assertEqual(self.window.status_label.text(), "服务器地址已更新")
            # 非法地址被拒绝，且不落盘
            error = self.window.set_base_url("ftp://nope")
            self.assertIn("http", error)
            self.assertEqual(
                client_settings.load(self.base)["base_url"], "http://10.0.0.5:9000"
            )
            # 清除
            self.assertIsNone(self.window.set_base_url("  "))
            self.assertEqual(client_settings.load(self.base)["base_url"], "")
            self.assertIn("未配置", self.window.manager.server_url_label.full_text())

        def _settle_sidebar(self, width, timeout=5.0):
            """等收缩动画结束（宽度是并行的 min/max 属性动画）。"""
            self.assertTrue(
                pump(_APP, predicate=lambda: self.window.sidebar.width() == width,
                     timeout=timeout),
                f"侧栏宽度没有到达 {width}（当前 {self.window.sidebar.width()}）",
            )

        def test_sidebar_collapse_toggle(self):
            from svfs_client.shell import RAIL_WIDTH, SIDEBAR_WIDTH
            from PySide6.QtWidgets import QBoxLayout

            self.assertFalse(self.window._sidebar_collapsed)
            self.window._set_sidebar_collapsed(True)
            self.assertTrue(self.window._sidebar_collapsed)
            self._settle_sidebar(RAIL_WIDTH)
            self.assertEqual(self.window.collapse_button.text(), "»")
            self.assertFalse(self.window.plugin_list.isVisibleTo(self.window.sidebar))
            # 收起后不是"插件不可达"：图标栏接过列表职责
            self.assertTrue(self.window.rail_list.isVisibleTo(self.window.sidebar))
            self.assertEqual(self.window.rail_list.width(), RAIL_WIDTH - 8)
            # 搜索框只在展开态出现
            self.assertFalse(self.window.search_edit.isVisibleTo(self.window.sidebar))
            # 底栏竖排（头像在齿轮上方），头像收成图标
            self.assertEqual(self.window._bottom_box.direction(),
                             QBoxLayout.Direction.TopToBottom)
            self.assertEqual(self.window.user_button.text(), "")
            self.assertEqual(self.window.settings_button.text(), "⚙")

            self.window._set_sidebar_collapsed(False)
            self.assertFalse(self.window._sidebar_collapsed)
            self._settle_sidebar(SIDEBAR_WIDTH)
            self.assertEqual(self.window.collapse_button.text(), "«")
            self.assertTrue(self.window.plugin_list.isVisibleTo(self.window.sidebar))
            self.assertFalse(self.window.rail_list.isVisibleTo(self.window.sidebar))
            self.assertTrue(self.window.search_edit.isVisibleTo(self.window.sidebar))
            self.assertEqual(self.window._bottom_box.direction(),
                             QBoxLayout.Direction.LeftToRight)
            # 齿轮按钮两种状态下都是图标态（账号条右侧的固定方块）
            self.assertEqual(self.window.settings_button.text(), "⚙")
            self.assertEqual(self.window.settings_button.width(), 30)

        def test_sidebar_search_filters_both_lists(self):
            payload = self._submit_and_wait("install_sample", {})
            self.window.apply_payload(payload)
            pid = client_sample.PLUGIN_ID  # com.svfs.demo
            pump(_APP)
            self.assertEqual(self.window.plugin_list.count(), 1)
            self.assertEqual(self.window.rail_list.count(), 1)

            # 按 id 片段（缩写匹配）命中
            self.window.search_edit.setText("csd")
            pump(_APP)
            self.assertEqual(self.window.plugin_list.count(), 1)
            self.assertEqual(self.window.rail_list.count(), 1)
            self.assertFalse(self.window.search_hint.isVisibleTo(self.window.sidebar))

            # 无匹配：两套列表都空 + 显示提示
            self.window.search_edit.setText("zzz-nothing")
            pump(_APP)
            self.assertEqual(self.window.plugin_list.count(), 0)
            self.assertEqual(self.window.rail_list.count(), 0)
            self.assertTrue(self.window.search_hint.isVisibleTo(self.window.sidebar))

            # 清空即恢复全部
            self.window.search_edit.clear()
            pump(_APP)
            self.assertEqual(self.window.plugin_list.count(), 1)
            self.assertEqual(self.window.rail_list.count(), 1)
            self.assertFalse(self.window.search_hint.isVisibleTo(self.window.sidebar))
            self.assertEqual(pid, self.window.plugin_list.item(0).data(
                Qt.ItemDataRole.UserRole))

        def test_rail_click_opens_plugin_and_syncs_selection(self):
            from svfs_client.shell import RAIL_WIDTH

            payload = self._submit_and_wait("install_sample", {})
            self.window.apply_payload(payload)
            pump(_APP)
            self.window._set_sidebar_collapsed(True)
            self._settle_sidebar(RAIL_WIDTH)

            # 图标栏点击 → 打开对应插件，两套列表选中同步
            self.window._on_rail_row(0)
            pump(_APP)
            pid = client_sample.PLUGIN_ID
            self.assertEqual(self.window._current_view_id(), pid)
            self.assertEqual(self.window.plugin_list.currentRow(), 0)
            self.assertEqual(self.window.rail_list.currentRow(), 0)

        def test_sidebar_has_no_manager_or_sample_button(self):
            # 插件管理入口在设置里，示例插件入口已删除；侧栏只剩插件列表
            self.assertFalse(hasattr(self.window, "manage_button"))
            self.assertFalse(hasattr(self.window, "sample_button"))
            self.assertFalse(hasattr(self.window.manager, "sample_button"))

        def _install_extra_plugin(self, plugin_id):
            """装一个非示例插件（最小合法清单 + 入口），用于多插件场景。"""
            src = make_tree(self.path(plugin_id), {
                "plugin.json": json.dumps({
                    "id": plugin_id, "name": "Other", "version": "1.0.0",
                    "api_version": PLUGIN_API_VERSION, "entry": "index.html",
                    "permissions": [],
                }).encode("utf-8"),
                "index.html": b"<html><body>other</body></html>",
            })
            pkg = self.path(plugin_id + ".svfs")
            packer = SQLiteVFS(pkg)
            try:
                from sqlite_vfs.folder_packer import FolderPacker

                FolderPacker(packer).pack_folder(src)
            finally:
                packer.close()
            with quiet():
                install_from_file(pkg, base=self.base)

        def test_startup_plugin_resolution(self):
            from svfs_client.shell import _ui_settings

            # last_plugin 经 QSettings 跨用例持久化（如调试台用例会写它），
            # 先清掉才能假设"无配置、无最近打开"
            _ui_settings().remove("ui/last_plugin")
            payload = self._submit_and_wait("install_sample", {})
            self.window.apply_payload(payload)  # _startup_resolved 已是 True，不会自动打开
            pid = client_sample.PLUGIN_ID
            other = "com.other.app"
            self._install_extra_plugin(other)
            self.window.apply_payload(self._submit_and_wait("list_plugins", {}))

            # 默认规则：无配置、无 last_plugin → 按 id 排序第一个启用的插件
            self.assertEqual(
                self.window._resolve_startup_plugin(), min(pid, other)
            )
            # 配置优先于一切
            self.assertIsNone(self.window.set_startup_plugin(other))
            self.assertEqual(self.window._resolve_startup_plugin(), other)
            self.assertEqual(
                client_settings.load(self.base)["startup_plugin"], other
            )
            # 未安装的插件被拒绝
            self.assertIn("未安装", self.window.set_startup_plugin("com.nope.app"))
            # 清配置 → 上次打开的插件
            self.assertIsNone(self.window.set_startup_plugin(""))
            _ui_settings().setValue("ui/last_plugin", other)
            self.assertEqual(self.window._resolve_startup_plugin(), other)
            # 禁用的插件跳过：禁用 other 后配置它也落空 → 回到第一个启用的
            _ui_settings().remove("ui/last_plugin")
            payload = self._submit_and_wait(
                "set_enabled", {"id": other, "enabled": False}
            )
            self.window.apply_payload(payload)
            self.window.set_startup_plugin(other)
            self.assertEqual(self.window._resolve_startup_plugin(), pid)
            # 收尾清理：不给后续用例留下配置与 last_plugin
            self.assertIsNone(self.window.set_startup_plugin(""))
            _ui_settings().remove("ui/last_plugin")

        def test_settings_dialog_startup_and_cache(self):
            from svfs_client.dialogs import SettingsDialog

            payload = self._submit_and_wait("install_sample", {})
            self.window.apply_payload(payload)
            pid = client_sample.PLUGIN_ID
            dialog = SettingsDialog(
                self.base, apply_base_url=self.window.set_base_url,
                plugins_provider=lambda: [
                    (e["id"], e.get("name") or e["id"]) for e in self.window._plugins
                ],
                startup_provider=lambda: (
                    client_settings.load(self.base)["startup_plugin"]
                ),
                apply_startup=self.window.set_startup_plugin,
            )
            self.addCleanup(dialog.deleteLater)
            # 启动插件下拉含已装插件；选它并保存
            self.assertIsNotNone(dialog.startup_combo.findData(pid))
            dialog.startup_combo.setCurrentIndex(dialog.startup_combo.findData(pid))
            dialog._apply_startup_clicked()
            self.assertEqual(
                client_settings.load(self.base)["startup_plugin"], pid
            )
            self.assertIn("已保存", dialog.startup_status.text())
            # 缓存路径：自定义 → 落盘并显示生效路径；恢复默认 → 清空
            custom = self.path("cache-custom")
            dialog.cache_edit.setText(custom)
            dialog._save_cache_dir()
            self.assertEqual(client_settings.load(self.base)["cache_dir"], custom)
            self.assertIn("cache-custom", dialog.cache_label.full_text())
            dialog._reset_cache_dir()
            self.assertEqual(client_settings.load(self.base)["cache_dir"], "")

        def test_settings_dialog_opens_manager(self):
            from svfs_client.dialogs import SettingsDialog

            dialog = SettingsDialog(
                self.base, apply_base_url=self.window.set_base_url,
                on_open_manager=self.window._show_manager,
            )
            self.addCleanup(dialog.deleteLater)
            dialog._open_manager()
            # 入口先关对话框再切到插件管理页
            self.assertIs(self.window.stack.currentWidget(), self.window.manager)

        def test_settings_dialog_debug_callback(self):
            from svfs_client.dialogs import SettingsDialog

            calls = []
            dialog = SettingsDialog(
                self.base, apply_base_url=self.window.set_base_url,
                on_install_debug=lambda: calls.append(True),
            )
            self.addCleanup(dialog.deleteLater)
            dialog._install_debug()
            self.assertEqual(len(calls), 1)

        # ---- 更新检查 / 服务器状态 / 诊断 ----

        def test_check_updates_marks_cards(self):
            payload = self._submit_and_wait("install_sample", {})
            self.window.apply_payload(payload)
            pid = client_sample.PLUGIN_ID
            card = self.window.manager.cards[pid]

            # 未检查前没有可升级标记
            self.assertFalse(card.update_badge.isVisibleTo(card))
            self.assertFalse(card.upgrade_button.isVisibleTo(card))

            server, base = start_api_server()
            self.addCleanup(server.shutdown)
            self.addCleanup(server.server_close)
            _ApiHandler.manifest_version = "9.9.9"  # 服务器上有"新版本"
            self.addCleanup(setattr, _ApiHandler, "manifest_version", None)
            self.window.set_base_url(base)

            result = self._submit_and_wait("check_updates", {})
            self.assertEqual(result["updates"], {pid: "9.9.9"})
            self.window._on_updates(result)
            pump(_APP)
            self.assertTrue(card.update_badge.isVisibleTo(card))
            self.assertIn("9.9.9", card.update_badge.text())
            self.assertTrue(card.upgrade_button.isVisibleTo(card))

            # 服务器上的版本不比本地新 → 不标记
            _ApiHandler.manifest_version = "0.0.1"
            result = self._submit_and_wait("check_updates", {})
            self.assertEqual(result["updates"], {})
            self.window._on_updates(result)
            pump(_APP)
            self.assertFalse(card.update_badge.isVisibleTo(card))

        def test_upgrade_plugin_resolves_manifest_url(self):
            payload = self._submit_and_wait("install_sample", {})
            self.window.apply_payload(payload)
            pid = client_sample.PLUGIN_ID
            self.window.set_base_url("http://10.0.0.7:9000")

            captured = []
            original = self.window._submit_install
            self.window._submit_install = (
                lambda op, payload, label: captured.append((op, payload, label)))
            try:
                self.window._upgrade_plugin(pid)
            finally:
                self.window._submit_install = original
            self.assertEqual(len(captured), 1)
            op, payload, _label = captured[0]
            self.assertEqual(op, "install_url")
            self.assertEqual(payload["url"],
                             f"http://10.0.0.7:9000/{pid}/manifest.json")

        def test_ping_server_status_light(self):
            server, base = start_api_server()
            self.addCleanup(server.shutdown)
            self.addCleanup(server.server_close)

            # 未配置：灰灯、无文字，不发请求
            self.window._ping_server()
            pump(_APP, settle=0.2)
            self.assertEqual(self.window.manager.server_status_label.text(), "")
            self.assertIn(client_theme.TEXT_FAINT,
                          self.window.manager.server_dot.styleSheet())

            # 可达：绿灯 + 延迟
            self.window.set_base_url(base)
            result = self._submit_and_wait("ping_server", {})
            self.assertTrue(result["online"])
            self.window._on_server_status(result)
            self.assertIn("ms", self.window.manager.server_status_label.text())

            # 不可达：红灯 + 离线
            self.window.set_base_url("http://127.0.0.1:1")
            result = self._submit_and_wait("ping_server", {})
            self.assertFalse(result["online"])
            self.window._on_server_status(result)
            self.assertEqual(self.window.manager.server_status_label.text(), "离线")

        def test_ping_reports_expired_session_as_warn(self):
            """服务器在线但拒绝凭据：状态灯是「登录已失效」（黄），不是「离线」。

            报"离线"会把用户引去查网络，而真相是会话没了——这两件事要分开。
            """
            server, base = start_api_server()
            self.addCleanup(server.shutdown)
            self.addCleanup(server.server_close)
            self.window.set_base_url(base)
            self.window.set_session({"username": "admin", "token": "tok",
                                     "base_url": base})
            _ApiHandler.ping_status = 401
            self.addCleanup(setattr, _ApiHandler, "ping_status", 200)

            result = self._submit_and_wait("ping_server", {})
            self.assertTrue(result["online"])
            self.assertTrue(result["auth_rejected"])
            self.window._on_server_status(result)
            pump(_APP)
            self.assertEqual(self.window.manager.server_status_label.text(),
                             "登录已失效")
            self.assertIn(client_theme.WARN,
                          self.window.manager.server_dot.styleSheet())
            # 同一件事也把会话标成失效（用户中心/侧栏头像跟着变）
            self.assertTrue(self.window._session.get("invalid"))

        def test_session_rejected_prompts_relogin_once(self):
            """插件 net 撞上 401 → 提示一次 + 一键重新登录；重登后恢复。"""
            self.window.set_session({"username": "admin", "token": "tok"})
            self.window._on_session_rejected("服务器返回 401：登录状态已失效")
            pump(_APP)
            self.assertIn("登录已失效",
                          self.window.manager.server_login_badge.text())
            self.assertTrue(self.window.manager.banner.isVisible())
            self.assertEqual(self.window.manager.banner._button.text(), "重新登录…")
            self.assertTrue(self.window.manager.banner._button.isVisible())

            # 每个打开的插件都会撞上同一个 401：提示只发一次
            self.window.manager.clear_banner()
            self.window._on_session_rejected("服务器返回 401：登录状态已失效")
            pump(_APP)
            self.assertFalse(self.window.manager.banner.isVisible())

            # 重新登录成功（新的干净会话）→ 徽章恢复，允许再次提示
            self.window.set_session({"username": "admin", "token": "new"})
            self.window._on_session_rejected("又一次 401")
            pump(_APP)
            self.assertIn("登录已失效",
                          self.window.manager.server_login_badge.text())
            self.assertTrue(self.window.manager.banner.isVisible())

        def test_check_client_update_notifies_and_opens_download(self):
            server, base = start_api_server()
            self.addCleanup(server.shutdown)
            self.addCleanup(server.server_close)
            self.window.set_base_url(base)

            # 服务器没提供 client/manifest.json → 静默（不是错误）
            result = self._submit_and_wait("check_client_update", {})
            self.assertFalse(result["update"])
            self.assertFalse(result["checked"])
            self.window._on_client_update(result)
            pump(_APP)
            self.assertFalse(self.window.manager.banner.isVisible())

            # 提供了更高版本 → 状态栏 + 横幅，动作按钮打开下载地址
            _ApiHandler.client_version = "99.0.0"
            self.addCleanup(setattr, _ApiHandler, "client_version", "")
            result = self._submit_and_wait("check_client_update", {})
            self.assertTrue(result["update"])
            self.assertEqual(result["latest"], "99.0.0")
            self.window._on_client_update(result)
            pump(_APP)
            self.assertIn("99.0.0", self.window.status_label.text())
            self.assertTrue(self.window.manager.banner.isVisible())
            self.assertEqual(self.window.manager.banner._button.text(),
                             "下载新版本…")

            opened = []
            self.window._open_update_url = opened.append
            self.window.manager.banner._button.click()
            pump(_APP)
            self.assertEqual(opened, ["https://example.invalid/SVFS-Client.zip"])

        def test_banner_action_follows_current_message(self):
            """横幅动作按钮只做"当前这条消息"的事（换消息不能沿用上一个动作）。"""
            calls = []
            self.window.manager.show_message("info", "标题", "详情",
                                             action_text="点我",
                                             action=lambda: calls.append("a"))
            self.window.manager.banner._button.click()
            pump(_APP)
            self.assertEqual(calls, ["a"])
            # 新的一条消息没带动作 → 按钮连显示都不该有
            self.window.manager.show_message("ok", "只有文字")
            pump(_APP)
            self.assertFalse(self.window.manager.banner._button.isVisible())

        def test_plugin_notify_reaches_tray_and_status_bar(self):
            """notify 权限名副其实：有托盘就发系统通知，同时界面内也留痕。"""
            class _FakeTray:
                def __init__(self):
                    self.calls = []

                def showMessage(self, title, message, icon, msecs):  # noqa: N802
                    self.calls.append((title, message, msecs))

            fake = _FakeTray()
            original, self.window._tray = self.window._tray, fake
            self.addCleanup(setattr, self.window, "_tray", original)
            self.window._on_plugin_notify("打包完成")
            pump(_APP)
            self.assertEqual(len(fake.calls), 1)
            self.assertEqual(fake.calls[0][1], "打包完成")
            self.assertEqual(self.window.status_label.text(), "打包完成")

        def test_plugin_notify_without_tray_still_shows_in_window(self):
            """没有系统托盘（容器/被策略禁用/offscreen）时降级为界面内提示。"""
            original, self.window._tray = self.window._tray, None
            self.addCleanup(setattr, self.window, "_tray", original)
            self.window._on_plugin_notify("服务端返回 500")
            pump(_APP)
            self.assertEqual(self.window.status_label.text(), "服务端返回 500")
            self.assertTrue(self.window.manager.banner.isVisible())

        def test_diagnostics_log_collects_errors(self):
            from svfs_client import diagnostics

            diagnostics.clear()
            self.window._on_fail_silent("测试错误", "详情")
            self.assertTrue(any("测试错误" in line
                                for line in diagnostics.recent()))
            # 页面控制台消息也进日志（不要求 WebEngine：直接验证日志接口）
            diagnostics.log("console 输出", "console:com.svfs.demo")
            text = diagnostics.report({"数据目录": self.base})
            self.assertIn("console:com.svfs.demo", text)
            diagnostics.clear()

        def test_install_debug_plugin_opens_new_page(self):
            # 设置 → 桥调试台：装完自动打开（新装/升级都一样）
            self.window._install_debug_plugin()
            self.assertTrue(
                pump(_APP, predicate=lambda: (
                    client_sample.PLUGIN_ID in self.window._views
                ), timeout=30),
                "调试台安装后没有自动打开",
            )
            self.assertIs(
                self.window.stack.currentWidget(),
                self.window._views[client_sample.PLUGIN_ID],
            )
            # 再点一次走升级：视图因版本变化被收掉并重建，仍是当前页
            self.window._install_debug_plugin()
            self.assertTrue(pump(_APP, predicate=lambda: (
                self.window.stack.currentWidget() is self.window._views.get(
                    client_sample.PLUGIN_ID)
            ), timeout=30))

        def test_settings_dialog_applies_base_url(self):
            from svfs_client.dialogs import SettingsDialog

            dialog = SettingsDialog(
                self.base, apply_base_url=self.window.set_base_url
            )
            self.addCleanup(dialog.deleteLater)
            # 合法地址：走 MainWindow.set_base_url（校验+落盘+同步界面）
            dialog.server_edit.setText(" http://10.0.0.9:9000/ ")
            dialog._apply_base_url_clicked()
            self.assertFalse(dialog.server_error.isVisibleTo(dialog))
            self.assertEqual(dialog.server_status.text(), "已保存")
            self.assertEqual(
                client_settings.load(self.base)["base_url"], "http://10.0.0.9:9000"
            )
            self.assertEqual(
                self.window.manager.server_url_label.full_text(),
                "http://10.0.0.9:9000",
            )
            # 非法地址：报错、不落盘
            dialog.server_edit.setText("ftp://nope")
            dialog._apply_base_url_clicked()
            self.assertTrue(dialog.server_error.isVisibleTo(dialog))
            self.assertEqual(
                client_settings.load(self.base)["base_url"], "http://10.0.0.9:9000"
            )

        def test_user_center_refresh_and_actions(self):
            from svfs_client.dialogs import UserCenterDialog

            self.window.set_session({
                "token": "tok", "username": "admin",
                "base_url": "http://10.0.0.9:9000",
                "login_time": "2026-09-15T10:00:00",
            })
            dialog = UserCenterDialog(
                self.base,
                session_provider=lambda: self.window._session,
                on_relogin=self.window._relogin,
                on_logout=self.window.logout,
            )
            self.addCleanup(dialog.deleteLater)
            # 已登录态：头像/用户名/徽章/按钮
            self.assertIn("admin", dialog.user_label.text())
            self.assertEqual(dialog.state_badge.text(), "已登录")
            self.assertTrue(dialog.logout_button.isEnabled())
            self.assertTrue(dialog.relogin_button.objectName() != "Primary")
            # 退出登录（真回调）：会话清空、界面回到未登录态
            dialog.logout_button.click()
            self.assertIsNone(self.window._session)
            self.assertEqual(dialog.user_label.text(), "未登录")
            self.assertFalse(dialog.logout_button.isEnabled())
            self.assertEqual(dialog.relogin_button.objectName(), "Primary")
            self.assertEqual(dialog.state_badge.text(), "未登录")
            # 侧栏头像同步回到「未登录」
            self.assertEqual(self.window.user_button.text(), "未登录")


if HAVE_QT and os.environ.get("SVFS_CLIENT_WEBENGINE_SMOKE", "1") != "0":

    class WebEngineSmokeTest(VFSTestCase):
        """端到端冒烟：WebView 经 svfs:// 加载示例插件 + 桥可用。

        WebEngine 对环境挑剔（无显卡的容器等），起不来时跳过而不是误报。
        """

        def test_sample_plugin_loads_and_bridge_works(self):
            base = client_paths.ensure_layout(self.path("data"))
            with quiet():
                client_sample.build(self.path("sample.svfs"))
                install_from_file(self.path("sample.svfs"), base=base)
            window = MainWindow(base)
            window.show()
            try:
                self.assertTrue(pump(_APP))
                window.open_plugin(client_sample.PLUGIN_ID)
                view = window._views.get(client_sample.PLUGIN_ID)
                self.assertIsNotNone(view)
                # 首开白屏修复：页面底色必须是主题底色，而不是 WebEngine
                # 默认的白色
                self.assertEqual(
                    view.page().backgroundColor().name(), client_theme.BG
                )
            except Exception:
                # WebEngine 的 profile/子进程起不来：此环境无法冒烟，跳过
                window._thread.quit()
                window._thread.wait(5000)
                window.deleteLater()
                pump(_APP)
                self.skipTest("此环境无法创建 WebEngine 视图（offscreen 冒烟被跳过）")

            try:
                loaded = {}
                view.page().loadFinished.connect(
                    lambda ok: loaded.update(ok=ok)
                )
                if view.page().url().host():
                    # 可能已经加载完了
                    loaded.setdefault("ok", True)
                self.assertTrue(
                    pump(_APP, predicate=lambda: loaded.get("ok"), timeout=40),
                    "页面在 40s 内没有加载完成——此环境可能无法运行 WebEngine",
                )

                # 桥已被注入且宿主信息渲染进 DOM
                probe = {}
                page = view.page()

                def ask():
                    view.page().runJavaScript(
                        "typeof window.svfs === 'object' && typeof window.svfs.host.info === 'function'",
                        0,
                        lambda r: probe.update(bridge=r),
                    )
                    view.page().runJavaScript(
                        "document.querySelectorAll('#host-info dt').length > 0",
                        0,
                        lambda r: probe.update(dom=r),
                    )
                    return None

                def probe_ready():
                    if probe.get("bridge") is None or probe.get("dom") is None:
                        ask()  # 未就绪就继续探（runJavaScript 异步回调）
                    return (probe.get("bridge") is not None
                            and probe.get("dom") is not None)

                self.assertTrue(
                    pump(_APP, predicate=probe_ready, timeout=15),
                    f"桥探针没有拿到结果: {probe}",
                )
                self.assertTrue(probe.get("bridge"), "window.svfs 桥没有注入成功")
                self.assertTrue(probe.get("dom"), "host.info 的结果没有渲染进 DOM")

                # 滚动条默认样式已注入页面（宿主给所有插件页统一暗色滚动条）
                page.runJavaScript(
                    "!!document.querySelector('style[data-svfs-scrollbar]')",
                    0, lambda r: probe.update(scrollbar_css=r),
                )
                self.assertTrue(
                    pump(_APP, predicate=lambda: probe.get("scrollbar_css") is not None),
                    "滚动条样式探针没有拿到结果",
                )
                self.assertTrue(
                    probe.get("scrollbar_css"),
                    "宿主没有把滚动条默认样式注入插件页面",
                )

                # storage 桥端到端：页面里 set → Python 落盘 → get 回读
                page.runJavaScript(
                    "(function(){ window.__probe='pending';"
                    "  window.svfs.storage.set('manual', 42).then(function(ok){"
                    "    window.svfs.storage.get('manual').then(function(v){"
                    "      window.__probe = JSON.stringify({ok: ok, v: v,"
                    "        count: document.getElementById('count').textContent});"
                    "    }); }); })();",
                    0, lambda r: None,
                )
                storage = {}

                def storage_ready():
                    page.runJavaScript("window.__probe", 0,
                                       lambda r: storage.update(r=r))
                    r = storage.get("r")
                    return r is not None and r != "pending"

                self.assertTrue(
                    pump(_APP, predicate=storage_ready, timeout=15),
                    f"storage 探针没有拿到结果: {storage}",
                )
                result = json.loads(storage["r"])
                self.assertTrue(result["ok"])
                self.assertEqual(result["v"], 42)

                # 示例自己的访问计数是异步链（get → set → 渲染），
                # 轮询到它变为 "1" 为止，避免和上面的探针抢时序
                def count_ready():
                    page.runJavaScript("document.getElementById('count').textContent",
                                       0, lambda r: storage.update(count=r))
                    return storage.get("count") == "1"

                self.assertTrue(
                    pump(_APP, predicate=count_ready, timeout=15),
                    f"访问计数没有变成 1: {storage.get('count')}",
                )

                # net 桥端到端（页面级）：本地假服务器 + 页面里发起请求，
                # Promise 必须落定——这是"接口一直在请求中"的回归探针
                server, base = start_api_server()
                self.addCleanup(server.shutdown)
                self.addCleanup(server.server_close)
                window.set_base_url(base)
                page.runJavaScript(
                    "(function(){ window.__net='pending';"
                    "  if (!window.svfs.net) { window.__net='error:没有net'; return; }"
                    "  window.svfs.net.request({method:'GET', path:'/api/ping'})"
                    "    .then(function(r){ window.__net = JSON.stringify("
                    "      {status: r.status, body: (r.bodyText || '').slice(0, 100)}); })"
                    "    .catch(function(e){ window.__net = 'error:' + e.message; }); })();",
                    0, lambda r: None,
                )
                net_probe = {}

                def net_ready():
                    page.runJavaScript("window.__net", 0,
                                       lambda r: net_probe.update(r=r))
                    value = net_probe.get("r")
                    return value is not None and value != "pending"

                self.assertTrue(
                    pump(_APP, predicate=net_ready, timeout=20),
                    "页面级 net 请求 20s 内没有落定（一直停在请求中）",
                )
                net_result = json.loads(net_probe["r"])
                self.assertEqual(net_result["status"], 200)
                self.assertIn("pong", net_result["body"])
                self.assertIsNone(window.set_base_url(""))

                # 渲染进程崩溃 → 崩溃占位页（而不是永久白屏）
                from svfs_client import diagnostics

                diagnostics.clear()
                view._on_render_terminated(2, 139)  # 2 = 进程崩溃
                crash_probe = {}

                def crash_ready():
                    page.runJavaScript(
                        "document.body ? document.body.innerText : ''", 0,
                        lambda r: crash_probe.update(t=r))
                    text = crash_probe.get("t") or ""
                    return "崩溃" in text

                self.assertTrue(
                    pump(_APP, predicate=crash_ready, timeout=15),
                    f"崩溃占位页没有渲染: {crash_probe}",
                )
                self.assertTrue(
                    any("渲染进程退出" in line for line in diagnostics.recent()),
                    "崩溃事件没有进诊断日志",
                )
                diagnostics.clear()
            finally:
                window._thread.quit()
                window._thread.wait(5000)
                window.deleteLater()
                pump(_APP)


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