Files
svfs/tests/run_tests.py
chakcy e0d5bad7b2 feat: 消息模块(MQTT 推送 + 统一总线),并补退出登录/顺序编排/设置分页签
一、消息模块:服务端主动推送 + 插件定向分发
* topics.py:主题方案 <prefix>/<org>/user/<user>/message、.../user/<user>/plugin/<id>、
  .../broadcast/{message,plugin/<id>}。org 是租户位(多租户上线不用改格式,broker
  按 <prefix>/<org>/# 下发 ACL 即可隔离);用户名/插件 id 经**可逆**百分号编码占
  一个层级,客户端与示例服务端共用同一套拼主题函数;
* bus.py:统一消息总线——按**主题**分类(主机消息进消息中心、插件消息不进,仪表盘
  数据那类业务载荷永远不会混进给人看的列表)、归属校验(org/user 不符即丢弃,broker
  ACL 配错也挡得住)、按 id 去重(QoS 1 重投与"推送/补齐重叠"都不会重复显示)、
  插件消息按插件缓存 50 条,等页面打开后一次性补投;
* mqtt.py + mqtt_service.py:paho 适配(可选依赖 [mqtt])+ Qt 适配(paho 的网络
  线程经排队信号把消息交给主线程的总线);重连交给 paho(指数退避),重连成功
  自动重新订阅;没装 paho / 没配 broker 都只是"没有实时推送",HTTP 补齐照常;
* worker.py:登录后**先**补齐(GET /api/messages?since=)再连 MQTT,两条路进同一
  总线;在途结果带身份快照,退出登录/换账号后丢弃;
* 界面:侧栏 ✉ 未读徽标 + 消息中心(级别/标题/时间/来源/广播标记、立即同步、
  全部已读、清空、通道状态灯);warn/error 顺手发系统托盘通知;
* 插件侧:新权限 messages(PLUGIN_API_VERSION→5),页面里 svfs.messages.subscribe();
  宿主经 window.__svfsPush 投递,早到的先排队、订阅时补发(桥调试台已接上演示);
* examples/plugin_server.py:/api/messages、/api/push(kind=message|plugin)、
  /api/broadcast;配了 SVFS_MQTT_HOST 就真发布到 broker,并把 broker 地址随登录
  响应下发(/api/login 与 /api/me 都带),客户端不用手填。

二、退出登录 = 关掉主窗口 + 回到登录界面
* 强制登录下"未登录的主窗口"本就不该存在:重新登录成功用新会话开回主窗口,关掉
  登录窗即退出客户端;退出时断开推送、清掉该账号收到的消息与总线身份(下一个人
  不该看到上一个人的消息),并收掉已打开的插件页面;
* 用户中心的"退出登录"先关掉自己再进登录门,避免模态窗口叠模态。

三、插件顺序编排
* installer.arrange():编排过的在前、未编排的按名称;管理页卡片加 ↑/↓,侧栏/图标栏/
  管理页三者一致;存 settings 的 plugin_order,只影响显示(不动包与注册表);
* "启动时打开"的默认规则改为"显示顺序里第一个启用的插件"(原来是按 id 排);
* 新增「恢复默认顺序」(编排过才出现)。

四、修复
* 收缩侧栏后页面左上角露出一块纯白方块:页面栈对 Qt 是透明的(主题只给
  QMainWindow/QDialog 设了背景),侧栏让出的那条在重绘时序里露白。给 QStackedWidget
  不透明底色,并加像素级回归用例(撤掉修复即 FAIL,实测 1237 个白像素);
* build_opener() 每次都要隐式枚举 Windows 证书库:全量测试跑几百次后以断点异常
  崩在证书枚举里(faulthandler 定位在 _load_windows_store_certs)。改成进程内只建
  一次 SSL context、启动时在主线程预建;
* 在途的消息补齐结果会喂给新会话(退出登录时用户名已空 → 拼主题抛异常,被 Qt 回调
  静默吞掉):带身份快照,不匹配即丢弃。

五、设置界面按类型分页签(插件 / 服务器 / 存储 / 网络 / 诊断)
* 每页内容在滚动区里(窗口拉小或以后再加项是滚动条,而不是把控件压扁);
* 新增「消息推送(MQTT)」一节:broker(host / host:port / mqtts://host:8883)、组织、
  端口、TLS、主题前缀、broker 账号,保存即重连;状态行显示**生效**配置与来源
  (手配 / 服务端下发)。broker 密码不进界面:默认用登录 token(已加密存储),需要
  单独的密码时由服务端下发——与"凭据不明文落盘"保持一致。

测试与验证
* 新增 tests/test_client_messages.py(主题/封皮/总线,纯层)、tests/test_client_mqtt.py
  (装配层 + 自建最小 broker 的端到端,不依赖 mosquitto)、test_plugin_server.py 的
  消息接口与"/api/push → broker → 客户端总线"闭环;WebEngine 冒烟里加了页面级
  messages 探针;
* **测试入口改为 tests/run_tests.py**:QtWebEngine 在 offscreen 下的解释器退出路径会
  崩(测试全绿但退出码 139,CI 只看退出码就会判失败),runner 按 unittest 的结果
  os._exit 绕开那段清理;测试真崩在中途时退出码照样非 0(用故意失败的用例验证过)。
  README 与 CI 都切到这个入口;
* 全量 720 条通过(6 skip),exit 0。
2026-09-17 01:06:02 +08:00

59 lines
2.3 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""测试入口:跑全部测试,然后按 unittest 的判定结果退出。
为什么需要这一层(而不是直接 ``python -m unittest discover -s tests``):
**QtWebEngine 在 offscreen 下的解释器退出路径会崩**(Windows 实测:测试全绿、
进程却以段错误结束,退出码 139)。CI 只看退出码,于是绿色测试被判成失败;
而且崩在哪一步与用例数量有关,非常难缠。测试的成败本来就该由 unittest 的
``result`` 决定,所以拿到结果后立刻 ``os._exit``,绕开那段我们控制不了的
C++/解释器清理路径。
注意这**不掩盖真实崩溃**:测试跑到一半崩(进程真的挂了)时,unittest 根本
拿不到结果,退出码照样非 0。
用法::
python tests/run_tests.py # 全部
python tests/run_tests.py -p "test_client_*" # 只跑一部分(同 discover 的 -p)
python tests/run_tests.py -v # 逐条打印
"""
import argparse
import os
import sys
import unittest
TESTS_DIR = os.path.dirname(os.path.abspath(__file__))
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("-p", "--pattern", default="test_*.py",
help="测试文件匹配(同 unittest discover 的 -p)")
parser.add_argument("-v", "--verbose", action="store_true", help="逐条打印")
parser.add_argument("-f", "--failfast", action="store_true",
help="第一个失败就停")
args = parser.parse_args(argv)
# 从测试目录里发现:helpers.py 与测试模块都按"平铺导入"写(见 helpers)
sys.path.insert(0, TESTS_DIR)
os.chdir(TESTS_DIR)
suite = unittest.TestLoader().discover(TESTS_DIR, pattern=args.pattern)
runner = unittest.TextTestRunner(verbosity=2 if args.verbose else 1,
failfast=args.failfast)
result = runner.run(suite)
passed = result.wasSuccessful()
summary = ("全部通过" if passed else
f"失败 {len(result.failures)} / 错误 {len(result.errors)}")
sys.stdout.write(f"{summary}\n")
sys.stdout.flush()
sys.stderr.flush()
# 立刻退出:不要让解释器去跑那段会崩的 Qt/Chromium 清理
os._exit(0 if passed else 1)
if __name__ == "__main__":
main()