From 06ec7c38bc08540590e01ea09db16388a30d6d4d Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Wed, 16 Sep 2026 17:15:16 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=BB=9F=E4=B8=80=20svfs=20=E5=AD=90?= =?UTF-8?q?=E5=91=BD=E4=BB=A4=E5=85=A5=E5=8F=A3=EF=BC=8C=E6=89=93=E5=8C=85?= =?UTF-8?q?=E8=BF=9B=E5=BA=A6=E5=9B=9E=E8=B0=83=E4=B8=8E=E5=AF=B9=E8=B1=A1?= =?UTF-8?q?=E7=BC=93=E5=AD=98=E8=B7=AF=E5=BE=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 桌面应用（svfs-studio / svfs-client）要用的前置改动： * 新增 src/sqlite_vfs/cli/main.py：`svfs <子命令>` 统一入口，旧的独立命令 （vfspack / vfsunpack / …）保留为别名，两套跑同一段 execute(args)；各子命令 因此从 sys.exit() 改成返回退出码，便于被复用； * FolderPacker.pack_folder(progress=…)：每处理一个条目（目录也算）回调一次， 供 GUI 进度条使用；只有传了回调时才做"只 stat、不读内容"的预扫描，CLI 的 行为与开销不变；回调抛出的异常一路上抛——GUI 的中断语义就靠它； * publish.default_cache_dir()：对象缓存目录（客户端设置里可改）；load_manifest 对"误把服务器根地址当清单填进来"给出可用插件列表的提示，而不是只甩一个 format=None。 测试：新增 tests/test_cli_main.py、tests/test_pack_progress.py，test_publish.py 补充。 --- Scripts/code_content.py | 2 + src/sqlite_vfs/cli/fetch.py | 35 ++-- src/sqlite_vfs/cli/main.py | 288 ++++++++++++++++++++++++++++++++ src/sqlite_vfs/cli/migrate.py | 38 +++-- src/sqlite_vfs/cli/packer.py | 33 ++-- src/sqlite_vfs/cli/publish.py | 34 ++-- src/sqlite_vfs/cli/unpacker.py | 42 +++-- src/sqlite_vfs/cli/verify.py | 33 ++-- src/sqlite_vfs/folder_packer.py | 41 ++++- src/sqlite_vfs/publish.py | 14 ++ tests/test_cli_main.py | 197 ++++++++++++++++++++++ tests/test_pack_progress.py | 145 ++++++++++++++++ tests/test_publish.py | 20 +++ 13 files changed, 836 insertions(+), 86 deletions(-) create mode 100644 src/sqlite_vfs/cli/main.py create mode 100644 tests/test_cli_main.py create mode 100644 tests/test_pack_progress.py diff --git a/Scripts/code_content.py b/Scripts/code_content.py index eea10a7..31945f2 100644 --- a/Scripts/code_content.py +++ b/Scripts/code_content.py @@ -10,6 +10,8 @@ def generate_markdown_from_py_files(directory, output_file): dirs[:] = [d for d in dirs if d != ".vscode"] dirs[:] = [d for d in dirs if d != "Scripts"] dirs[:] = [d for d in dirs if d != "build"] + dirs[:] = [d for d in dirs if d != "dist"] + dirs[:] = [d for d in dirs if d != "node_modules"] for file in files: if ( file.endswith(".py") diff --git a/src/sqlite_vfs/cli/fetch.py b/src/sqlite_vfs/cli/fetch.py index 453a77b..d1067df 100644 --- a/src/sqlite_vfs/cli/fetch.py +++ b/src/sqlite_vfs/cli/fetch.py @@ -4,6 +4,9 @@ 清单可以是本地路径，也可以是 URL（对象默认从清单同级目录取）。 每个对象都按名字里的 sha256 校验，所以传输被改动或服务端内容不符都会被拒绝， 而且内容不可变，缓存可以长期复用（默认缓存目录见 --help）。 + +结构说明：``add_arguments`` / ``execute`` 供统一的 ``svfs`` 入口复用； +``fetch_objects`` 是独立的命令行入口（vfsfetch），跑的是同一段 ``execute``。 """ import argparse @@ -13,11 +16,10 @@ import time from .. import publish +DESCRIPTION = "按清单从对象存储取回文件（内容按 sha256 校验）" -def fetch_objects(): - parser = argparse.ArgumentParser( - description="按清单从对象存储取回文件（内容按 sha256 校验）" - ) + +def add_arguments(parser): parser.add_argument( "manifest", help="清单路径或 URL（如 https://cdn.example.com/app/manifest.json）", @@ -57,17 +59,18 @@ def fetch_objects(): help="跳过 sha256 校验（默认校验，不推荐关闭）", ) parser.add_argument("--json", action="store_true", help="以 JSON 输出统计") + return parser - args = parser.parse_args() +def execute(args) -> int: try: manifest = publish.load_manifest(args.manifest) except FileNotFoundError: print(f"错误: 清单不存在: {args.manifest}") - sys.exit(1) + return 1 except Exception as e: print(f"错误: 无法读取清单: {e}") - sys.exit(1) + return 1 base_url = args.base_url local_root = None @@ -94,20 +97,20 @@ def fetch_objects(): stats = fetcher.fetch(args.out_dir, paths=args.path) except FileNotFoundError as e: print(f"错误: {e}") - sys.exit(1) + return 1 except ValueError as e: print(f"错误: {e}") - sys.exit(1) + return 1 except Exception as e: print(f"错误: 下载失败: {e}") - sys.exit(1) + return 1 elapsed = time.time() - started if args.json: import json print(json.dumps(stats.as_dict(), ensure_ascii=False, indent=2)) - return + return 0 snapshot = manifest.get("snapshot", {}) print(f"包: {manifest.get('name')} 快照: {snapshot.get('id')} ({snapshot.get('name')})") @@ -119,7 +122,15 @@ def fetch_objects(): print(f" 下载字节: {stats.bytes_downloaded / 1024:.1f} KB") print(f" 耗时: {elapsed:.2f} 秒") print(f" 输出: {os.path.abspath(args.out_dir)}") + return 0 + + +def fetch_objects(argv=None) -> int: + parser = argparse.ArgumentParser(description=DESCRIPTION) + add_arguments(parser) + args = parser.parse_args(argv) + return execute(args) if __name__ == "__main__": - fetch_objects() + sys.exit(fetch_objects()) diff --git a/src/sqlite_vfs/cli/main.py b/src/sqlite_vfs/cli/main.py new file mode 100644 index 0000000..12c6536 --- /dev/null +++ b/src/sqlite_vfs/cli/main.py @@ -0,0 +1,288 @@ +#!/usr/bin/env python3 +"""svfs：统一的命令行入口（git 风格的子命令）。 + + svfs pack 把文件夹打包成单个 .svfs（原 vfspack） + svfs unpack 从 .svfs 导出文件（原 vfsunpack） + svfs verify 校验完整性（摘要 / 内容哈希 / 验签）（原 vfsverify） + svfs sign 给快照摘要做 HMAC 签名（原 vfsverify --sign） + svfs info 查看包的概要（统计 + 快照列表）（原 vfsverify --list-snapshots 的超集） + svfs migrate 迁移旧格式（原 vfsmigrate） + svfs publish 发布成对象 + 清单（原 vfspublish） + svfs fetch 按清单取回（原 vfsfetch） + +旧的六个 ``vfs*`` 命令全部保留为别名——它们与 ``svfs <同名子命令>`` 跑的是同一段 +``execute``，输出与退出码一致，脚本不必迁移。 + +约定（所有子命令一致）： + +* 退出码：0 成功，1 失败（argparse 的用法错误是 2）； +* ``--snapshot ID`` 在所有"选已有快照"的子命令里都是 id；``pack --snapshot NAME`` + 例外，它是给**新建**快照起名字（语义不同，help 里写明了）； +* ``--json`` 的子命令输出机器可读结果，字段名与库的返回值一致。 +""" + +import argparse +import json +import os +import sys + +from ..core import SCHEMA_VERSION, SQLiteVFS, VFSFormatError +from . import fetch as cli_fetch +from . import migrate as cli_migrate +from . import packer as cli_pack +from . import publish as cli_publish +from . import unpacker as cli_unpack +from . import verify as cli_verify + +try: + from importlib.metadata import PackageNotFoundError + from importlib.metadata import version as _version + + __version__ = _version("sqlite-vfs") +except PackageNotFoundError: # pragma: no cover - 未安装元数据时（直接跑源码） + __version__ = "0.0.0" + +DESCRIPTION = f"SVFS 工具集 v{__version__} —— 把文件夹打包成一个 SQLite 文件，以及它的发布、校验与迁移" + + +# ---------------------------------------------------------------------- +# sign（原 vfsverify --sign：签名是写操作，从只读的 verify 里拆出来） +# ---------------------------------------------------------------------- + + +def _add_sign_arguments(parser): + parser.add_argument("db_file", help="SQLite数据库文件路径") + parser.add_argument("--key", required=True, help="HMAC 密钥（不会写入库里）") + parser.add_argument("--key-id", help="密钥标识，便于同一快照有多个签名") + parser.add_argument( + "--snapshot", metavar="ID", help="给指定快照签名（默认取最新快照）" + ) + parser.add_argument("--json", action="store_true", help="以 JSON 输出结果") + return parser + + +def _execute_sign(args) -> int: + if not os.path.exists(args.db_file): + print(f"错误: 数据库文件不存在: {args.db_file}") + return 1 + + try: + vfs = SQLiteVFS(args.db_file, snapshot_id=args.snapshot) + except VFSFormatError as e: + print(f"错误: {e}") + return 1 + except Exception as e: + print(f"错误: 无法打开数据库 {args.db_file}: {e}") + return 1 + + # SQLiteVFS 接受字符串 id；JSON 输出里给回数字更符合"这是个 id"的直觉 + try: + snapshot_for_json = int(args.snapshot) if args.snapshot else None + except ValueError: + snapshot_for_json = args.snapshot + + try: + signature = vfs.sign(args.key, key_id=args.key_id) + except Exception as e: + print(f"错误: 签名失败: {e}") + return 1 + finally: + vfs.close() + + if args.json: + print( + json.dumps( + {"signature": signature, "key_id": args.key_id, "snapshot": snapshot_for_json}, + ensure_ascii=False, + indent=2, + ) + ) + else: + suffix = f"，key_id={args.key_id}" if args.key_id else "" + print(f"已签名（hmac-sha256{suffix}）: {signature}") + print("密钥没有写入库，请自行保管——丢了就验不了签。") + return 0 + + +# ---------------------------------------------------------------------- +# info（包的概要：统计 + 快照列表，只读） +# ---------------------------------------------------------------------- + + +def _add_info_arguments(parser): + parser.add_argument("db_file", help="SQLite数据库文件路径") + parser.add_argument( + "--snapshot", metavar="ID", help="查看指定快照（默认取最新快照）" + ) + parser.add_argument("--json", action="store_true", help="以 JSON 输出结果") + return parser + + +def _execute_info(args) -> int: + if not os.path.exists(args.db_file): + print(f"错误: 数据库文件不存在: {args.db_file}") + return 1 + + try: + vfs = SQLiteVFS(args.db_file, readonly=True, snapshot_id=args.snapshot) + except VFSFormatError as e: + print(f"错误: {e}") + print("提示: 旧格式的包先迁移：svfs migrate （会先留一份 .bak 备份）") + return 1 + except Exception as e: + print(f"错误: 无法打开数据库 {args.db_file}: {e}") + return 1 + + try: + stats = vfs.get_stats() + snapshots = vfs.list_snapshots() + blobs = vfs.list_blobs() + signatures = vfs.list_signatures() + entries = vfs.all_entries() + finally: + vfs.close() + + files = [row for row in entries if not row["is_directory"]] + logical_size = sum(row["file_size"] or 0 for row in files) + unique_bytes = sum(row["size"] or 0 for row in blobs) + stored_bytes = sum(row["stored_size"] or 0 for row in blobs) + db_size = os.path.getsize(args.db_file) + saved = max(logical_size - stored_bytes, 0) + + if args.json: + print( + json.dumps( + { + "path": os.path.abspath(args.db_file), + "schema_version": stats.get("schema_version"), + "name": stats.get("name"), + "snapshot_id": stats.get("snapshot_id"), + "snapshot_name": stats.get("snapshot_name"), + "digest": stats.get("digest"), + "files": len(files), + "directories": len(entries) - len(files), + "logical_size": logical_size, + "unique_contents": len(blobs), + "unique_bytes": unique_bytes, + "stored_bytes": stored_bytes, + "db_file_size": db_size, + "saved_bytes": saved, + "snapshots": [ + { + "id": s["id"], + "name": s["name"], + "parent_id": s["parent_id"], + "total_files": s["total_files"], + "total_size": s["total_size"], + "digest": s["digest"], + } + for s in snapshots + ], + "signatures": len(signatures), + }, + ensure_ascii=False, + indent=2, + default=str, + ) + ) + return 0 + + print(f"包: {stats.get('name') or '-'}（schema v{stats.get('schema_version') or SCHEMA_VERSION}）") + print(f" 路径: {os.path.abspath(args.db_file)}") + print(f" 当前快照: {stats.get('snapshot_name') or '-'}（id={stats.get('snapshot_id')}）") + print( + f" 条目: {len(files)} 个文件，{len(entries) - len(files)} 个目录" + f"，逻辑大小 {logical_size} 字节" + ) + print( + f" 存储: 唯一内容 {len(blobs)} 份（{unique_bytes} 字节），" + f"库内占用 {stored_bytes} 字节，库文件 {db_size} 字节" + ) + if logical_size: + print(f" 去重+压缩共省下: {saved} 字节（{(saved / logical_size) * 100:.0f}%）") + print(f" 签名: {len(signatures)} 条") + if snapshots: + cli_verify._print_snapshots(_SnapshotView(stats.get("snapshot_id"), snapshots)) + else: + print("库里还没有快照，写入时会自动建一个。") + return 0 + + +class _SnapshotView: + """给 ``_print_snapshots`` 用的最小接口：它只需要 _active_snapshot() 和快照列表。""" + + def __init__(self, active_id, snapshots): + self._active_id = active_id + self._snapshots = snapshots + + def _active_snapshot(self): + return self._active_id + + def list_snapshots(self): + return self._snapshots + + +# ---------------------------------------------------------------------- +# 调度 +# ---------------------------------------------------------------------- + +#: 子命令 → (模块/处理器, 一句话说明)。模块为 None 表示本文件内实现。 +COMMANDS = { + "pack": (cli_pack, "把文件夹打包成单个 .svfs（原 vfspack）"), + "unpack": (cli_unpack, "从 .svfs 导出文件（原 vfsunpack）"), + "verify": (cli_verify, "校验完整性：摘要 / 内容哈希 / 验签（原 vfsverify）"), + "sign": (None, "给快照摘要做 HMAC 签名（原 vfsverify --sign）"), + "info": (None, "查看包的概要：统计 + 快照列表"), + "migrate": (cli_migrate, "把旧格式（v1/v2）迁移到当前格式（原 vfsmigrate）"), + "publish": (cli_publish, "发布成内容寻址的对象 + 清单（原 vfspublish）"), + "fetch": (cli_fetch, "按清单从对象存储取回（原 vfsfetch）"), +} + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="svfs", + description=DESCRIPTION, + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog="示例:\n" + " svfs pack ./dist app.svfs --compress\n" + " svfs info app.svfs\n" + " svfs verify app.svfs --check-content\n" + " svfs unpack app.svfs ./restored --snapshot 2\n", + ) + parser.add_argument("--version", action="version", version=f"svfs {__version__}") + subparsers = parser.add_subparsers(dest="command", metavar="<命令>") + + for name, (module, description) in COMMANDS.items(): + sub = subparsers.add_parser(name, description=description, help=description) + if module is not None: + module.add_arguments(sub) + elif name == "sign": + _add_sign_arguments(sub) + elif name == "info": + _add_info_arguments(sub) + + return parser + + +def main(argv=None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + + if not getattr(args, "command", None): + # 不带子命令就打印帮助：比甩一段 usage 错误友好，也不会让人以为是坏了 + parser.print_help() + return 0 + + handler = { + "sign": _execute_sign, + "info": _execute_info, + }.get(args.command) + if handler is None: + module = COMMANDS[args.command][0] + return module.execute(args) + return handler(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/sqlite_vfs/cli/migrate.py b/src/sqlite_vfs/cli/migrate.py index b0be3bf..680f802 100644 --- a/src/sqlite_vfs/cli/migrate.py +++ b/src/sqlite_vfs/cli/migrate.py @@ -3,6 +3,9 @@ 默认原地迁移并留一份 .bak 备份；用 --output 可以输出到新文件， 适合原库位于只读介质、或想先比对再切换的场景。 + +结构说明：``add_arguments`` / ``execute`` 供统一的 ``svfs`` 入口复用； +``main`` 是独立的命令行入口（vfsmigrate），跑的是同一段 ``execute``。 """ import argparse @@ -12,11 +15,10 @@ import sys from .. import migration from ..core import SCHEMA_VERSION +DESCRIPTION = f"把旧格式的虚拟文件系统迁移到 schema v{SCHEMA_VERSION}" -def main(): - parser = argparse.ArgumentParser( - description=f"把旧格式的虚拟文件系统迁移到 schema v{SCHEMA_VERSION}" - ) + +def add_arguments(parser): parser.add_argument("db_file", help="要迁移的数据库文件路径") parser.add_argument( "--output", @@ -29,30 +31,28 @@ def main(): help="原地迁移时不生成 .bak 备份", ) parser.add_argument("--name", help="迁移后默认快照的名字（默认沿用库里的名称）") + return parser - args = parser.parse_args() +def execute(args) -> int: if not os.path.exists(args.db_file): print(f"错误: 数据库文件不存在: {args.db_file}") - sys.exit(1) + return 1 try: version = migration.detect_schema_version(args.db_file) except Exception as e: print(f"错误: 无法读取 {args.db_file}: {e}") - sys.exit(1) + return 1 if version is None: print(f"错误: {args.db_file} 不是虚拟文件系统数据库") - sys.exit(1) + return 1 if version == SCHEMA_VERSION: print(f"无需迁移: 已经是 schema v{SCHEMA_VERSION}") - return - - total = [0] + return 0 def progress(done, total_count): - total[0] = total_count if total_count and (done == total_count or done % 500 == 0): print(f" 迁移进度: {done}/{total_count}", flush=True) @@ -67,10 +67,10 @@ def main(): ) except FileExistsError as e: print(f"错误: {e}") - sys.exit(1) + return 1 except Exception as e: print(f"错误: 迁移失败，原库未被改动: {e}") - sys.exit(1) + return 1 print("迁移完成!") print(f" 条目数: {result['files']}，去重后 blob 数: {result['blobs']}") @@ -81,7 +81,15 @@ def main(): if result["backup"]: print(f" 备份保留: {result['backup']}") print(f" 可以用 `vfsverify \"{result['output']}\" --check-content` 复核内容") + return 0 + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=DESCRIPTION) + add_arguments(parser) + args = parser.parse_args(argv) + return execute(args) if __name__ == "__main__": - main() + sys.exit(main()) diff --git a/src/sqlite_vfs/cli/packer.py b/src/sqlite_vfs/cli/packer.py index 620baba..207105a 100644 --- a/src/sqlite_vfs/cli/packer.py +++ b/src/sqlite_vfs/cli/packer.py @@ -1,6 +1,9 @@ #!/usr/bin/env python3 """ 文件夹打包工具 - 将文件夹打包为SQLite虚拟文件系统 + +结构说明：``add_arguments`` / ``execute`` 供统一的 ``svfs`` 入口复用； +``pack`` 是独立的命令行入口（vfspack），跑的是同一段 ``execute``。 """ import os @@ -11,9 +14,10 @@ from .. import compression from ..folder_packer import FolderPacker from ..core import SQLiteVFS, VFSFormatError +DESCRIPTION = "将文件夹打包为SQLite虚拟文件系统" -def pack(): - parser = argparse.ArgumentParser(description="将文件夹打包为SQLite虚拟文件系统") + +def add_arguments(parser): parser.add_argument("source_folder", help="要打包的源文件夹路径") parser.add_argument("output_db", help="输出的SQLite数据库文件路径") parser.add_argument("--name", default="packed_filesystem", help="文件系统名称") @@ -40,20 +44,21 @@ def pack(): metavar="NAME", help="把这次的改动打包成一个新快照（从当前快照派生），实现版本历史", ) + return parser - args = parser.parse_args() +def execute(args) -> int: if not os.path.exists(args.source_folder): print(f"错误: 源文件夹不存在: {args.source_folder}") - sys.exit(1) + return 1 if not os.path.isdir(args.source_folder): print(f"错误: 源路径不是文件夹: {args.source_folder}") - sys.exit(1) + return 1 output_dir = os.path.dirname(os.path.abspath(args.output_db)) if not os.path.isdir(output_dir): print(f"错误: 输出目录不存在: {output_dir}") - sys.exit(1) + return 1 if os.path.exists(args.output_db): if args.incremental or args.snapshot: @@ -71,7 +76,7 @@ def pack(): ) except (VFSFormatError, sqlite3.Error) as e: print(f"错误: 无法使用输出数据库 {args.output_db}: {e}") - sys.exit(1) + return 1 # 创建打包器 packer = FolderPacker( @@ -91,7 +96,7 @@ def pack(): ) except Exception as e: print(f"错误: 无法创建快照 {args.snapshot}: {e}") - sys.exit(1) + return 1 print( f"已创建快照 {args.snapshot}（id={snapshot_id}" + (f"，派生自 {parent}" if parent else "，库内首个快照") @@ -108,9 +113,15 @@ def pack(): finally: packer.sqlite_vfs.close() - if not success: - sys.exit(1) + return 0 if success else 1 + + +def pack(argv=None) -> int: + parser = argparse.ArgumentParser(description=DESCRIPTION) + add_arguments(parser) + args = parser.parse_args(argv) + return execute(args) if __name__ == "__main__": - pack() + sys.exit(pack()) diff --git a/src/sqlite_vfs/cli/publish.py b/src/sqlite_vfs/cli/publish.py index 6a8fad4..9827f85 100644 --- a/src/sqlite_vfs/cli/publish.py +++ b/src/sqlite_vfs/cli/publish.py @@ -4,6 +4,9 @@ 产出可以直接 `aws s3 sync` / `rclone copy` 到对象存储或静态托管： 对象名就是内容的 sha256，所以客户端的缓存策略可以是"永久"， 而且每个对象都能端到端校验（CDN 换不掉内容）。 + +结构说明：``add_arguments`` / ``execute`` 供统一的 ``svfs`` 入口复用； +``publish_objects`` 是独立的命令行入口（vfspublish），跑的是同一段 ``execute``。 """ import argparse @@ -13,11 +16,10 @@ import sys from .. import publish from ..core import VFSFormatError +DESCRIPTION = "把虚拟文件系统导出成内容寻址的对象目录 + 清单" -def publish_objects(): - parser = argparse.ArgumentParser( - description="把虚拟文件系统导出成内容寻址的对象目录 + 清单" - ) + +def add_arguments(parser): parser.add_argument("db_file", help="SQLite数据库文件路径") parser.add_argument("out_dir", help="输出目录（会创建 blobs/ 与 manifest.json）") parser.add_argument( @@ -39,17 +41,15 @@ def publish_objects(): action="store_true", help="只生成清单，不导出对象（用于只更新清单的场景）", ) + return parser - args = parser.parse_args() +def execute(args) -> int: if not os.path.exists(args.db_file): print(f"错误: 数据库文件不存在: {args.db_file}") - sys.exit(1) - - done = [0] + return 1 def progress(current, total): - done[0] = current if total and (current == total or current % 200 == 0): print(f" 导出对象: {current}/{total}", flush=True) @@ -94,13 +94,13 @@ def publish_objects(): ) except VFSFormatError as e: print(f"错误: {e}") - sys.exit(1) + return 1 except FileExistsError as e: print(f"错误: {e}") - sys.exit(1) + return 1 except Exception as e: print(f"错误: 发布失败: {e}") - sys.exit(1) + return 1 print("发布完成!") print(f" 快照: {result['snapshot_id']} 摘要: {(result['digest'] or '')[:16]}") @@ -116,7 +116,15 @@ def publish_objects(): print(f" 清单: {result['manifest']}（{result['manifest_bytes'] / 1024:.1f} KB）") print(f" 输出目录: {result['out_dir']}") print(" 下一步: 把整个输出目录同步到对象存储/静态托管即可") + return 0 + + +def publish_objects(argv=None) -> int: + parser = argparse.ArgumentParser(description=DESCRIPTION) + add_arguments(parser) + args = parser.parse_args(argv) + return execute(args) if __name__ == "__main__": - publish_objects() + sys.exit(publish_objects()) diff --git a/src/sqlite_vfs/cli/unpacker.py b/src/sqlite_vfs/cli/unpacker.py index 00b4e5f..1878bc9 100644 --- a/src/sqlite_vfs/cli/unpacker.py +++ b/src/sqlite_vfs/cli/unpacker.py @@ -1,6 +1,9 @@ #!/usr/bin/env python3 """ 文件夹解包工具 - 从SQLite虚拟文件系统导出文件 + +结构说明：``add_arguments`` / ``execute`` 供统一的 ``svfs`` 入口复用； +``unpack`` 是独立的命令行入口（vfsunpack），跑的是同一段 ``execute``。 """ import os @@ -10,9 +13,10 @@ import argparse from ..core import SQLiteVFS, VFSFormatError from ..folder_unpacker import FolderUnpacker +DESCRIPTION = "从SQLite虚拟文件系统导出文件" -def unpack(): - parser = argparse.ArgumentParser(description="从SQLite虚拟文件系统导出文件") + +def add_arguments(parser): parser.add_argument("db_file", help="SQLite数据库文件路径") parser.add_argument("export_folder", help="导出目标文件夹路径") parser.add_argument("--path", help="要导出的特定文件或目录路径（默认导出全部）") @@ -35,27 +39,28 @@ def unpack(): action="store_true", help="忽略时间戳错误，只导出文件内容", ) + return parser - args = parser.parse_args() +def execute(args) -> int: if not os.path.exists(args.db_file): print(f"错误: 数据库文件不存在: {args.db_file}") - sys.exit(1) + return 1 if not os.path.isfile(args.db_file): print(f"错误: 数据库路径不是文件: {args.db_file}") - sys.exit(1) + return 1 try: sqlite_vfs = SQLiteVFS(args.db_file, snapshot_id=args.snapshot) except VFSFormatError as e: print(f"错误: {e}") - sys.exit(1) + return 1 except FileNotFoundError as e: print(f"错误: {e}") - sys.exit(1) + return 1 except (sqlite3.Error, ValueError) as e: print(f"错误: 无法打开数据库 {args.db_file}: {e}") - sys.exit(1) + return 1 # 创建解包器 unpacker = FolderUnpacker(sqlite_vfs=sqlite_vfs) @@ -65,7 +70,7 @@ def unpack(): snapshots = unpacker.list_snapshots() if not snapshots: print("库里没有任何快照") - return + return 0 print(f"{'ID':>4s} {'名称':<20s} {'文件数':>6s} {'总大小':>12s} 摘要") for snapshot in snapshots: digest = snapshot["digest"] or "-" @@ -74,7 +79,7 @@ def unpack(): f"{snapshot['total_files'] or 0:>6d} " f"{snapshot['total_size'] or 0:>12d} {digest[:16]}" ) - return + return 0 if args.list_root: # 只列出根目录内容 @@ -83,13 +88,13 @@ def unpack(): for item in items: file_type = "DIR" if item["is_directory"] else "FILE" print(f" {file_type}: {item['name']} (路径: {item['path']})") - return + return 0 if args.path: # 导出特定路径 if not sqlite_vfs.exists(args.path): print(f"错误: 虚拟文件系统中不存在该路径: {args.path}") - sys.exit(1) + return 1 if sqlite_vfs.is_dir(args.path): local_dir = os.path.join( @@ -114,11 +119,20 @@ def unpack(): except RuntimeError as e: print(f"错误: {e}") - sys.exit(1) + return 1 finally: unpacker.sqlite_vfs.close() + return 0 + + +def unpack(argv=None) -> int: + parser = argparse.ArgumentParser(description=DESCRIPTION) + add_arguments(parser) + args = parser.parse_args(argv) + return execute(args) + if __name__ == "__main__": - unpack() + sys.exit(unpack()) diff --git a/src/sqlite_vfs/cli/verify.py b/src/sqlite_vfs/cli/verify.py index 586a257..d852b57 100644 --- a/src/sqlite_vfs/cli/verify.py +++ b/src/sqlite_vfs/cli/verify.py @@ -2,6 +2,9 @@ """vfsverify：校验 .svfs 的完整性、查看快照、签名与验签。 只读操作（除非显式 --sign），所以可以直接对着只读介质上的资源包跑。 + +结构说明：``add_arguments`` / ``execute`` 供统一的 ``svfs`` 入口复用； +``verify`` 是独立的命令行入口（vfsverify），跑的是同一段 ``execute``。 """ import argparse @@ -11,6 +14,8 @@ import sys from ..core import SQLiteVFS, VFSFormatError +DESCRIPTION = "校验虚拟文件系统的完整性（摘要 / 内容哈希 / 签名）" + def _print_snapshots(vfs): snapshots = vfs.list_snapshots() @@ -30,10 +35,7 @@ def _print_snapshots(vfs): print("（* 表示默认快照，即未指定 --snapshot 时使用的那个）") -def verify(): - parser = argparse.ArgumentParser( - description="校验虚拟文件系统的完整性（摘要 / 内容哈希 / 签名）" - ) +def add_arguments(parser): parser.add_argument("db_file", help="SQLite数据库文件路径") parser.add_argument( "--snapshot", metavar="ID", help="指定要校验的快照（默认取最新快照）" @@ -56,28 +58,29 @@ def verify(): parser.add_argument( "--json", action="store_true", help="以 JSON 输出结果，便于交给外部工具处理" ) + return parser - args = parser.parse_args() +def execute(args) -> int: if not os.path.exists(args.db_file): print(f"错误: 数据库文件不存在: {args.db_file}") - sys.exit(1) + return 1 readonly = not args.sign try: vfs = SQLiteVFS(args.db_file, readonly=readonly, snapshot_id=args.snapshot) except VFSFormatError as e: print(f"错误: {e}") - sys.exit(1) + return 1 except Exception as e: print(f"错误: 无法打开数据库 {args.db_file}: {e}") - sys.exit(1) + return 1 exit_code = 0 try: if args.list_snapshots: _print_snapshots(vfs) - return + return 0 report = vfs.verify(check_content=args.check_content) signature_report = None @@ -127,9 +130,15 @@ def verify(): finally: vfs.close() - if exit_code: - sys.exit(exit_code) + return exit_code + + +def verify(argv=None) -> int: + parser = argparse.ArgumentParser(description=DESCRIPTION) + add_arguments(parser) + args = parser.parse_args(argv) + return execute(args) if __name__ == "__main__": - verify() + sys.exit(verify()) diff --git a/src/sqlite_vfs/folder_packer.py b/src/sqlite_vfs/folder_packer.py index 98286e2..927b727 100644 --- a/src/sqlite_vfs/folder_packer.py +++ b/src/sqlite_vfs/folder_packer.py @@ -75,18 +75,32 @@ class FolderPacker: # 入口 # ------------------------------------------------------------------ - def pack_folder(self, source_folder, incremental=False, rehash=False): + def pack_folder(self, source_folder, incremental=False, rehash=False, progress=None): """打包文件夹。 返回是否全部成功（有写入失败的条目时返回 False）。 + + progress: 可选回调 ``progress(done, total, rel_path)``，每处理一个条目 + （目录也算）回调一次，供 GUI 之类需要进度条的调用方使用。 + ``total`` 是要处理的总条目数，需要额外做一次"只 stat、不读内容"的预扫描， + 所以**只在传了回调时**才计算；CLI 不传回调，行为与开销都保持不变。 + 回调抛出的异常会一路向上抛（打包到一半想中断就靠这个）。 """ if incremental: - return self._pack_incremental(source_folder, rehash=rehash) - return self._pack_full(source_folder) + return self._pack_incremental(source_folder, rehash=rehash, progress=progress) + return self._pack_full(source_folder, progress=progress) - def _walk(self, source_folder): + def _count_entries(self, source_folder): + """数一遍待处理条目，给进度条当分母（只 stat，不读文件内容）。 + + 注意它会顺带累加 ``excluded_count``，所以调用方必须在开始统计前把它清零。 + """ + return sum(1 for _ in self._walk(source_folder)) + + def _walk(self, source_folder, progress=None, total=0): """遍历源目录，逐个产出 ('dir'|'file', abs_path, rel_path)。""" db_abspath = os.path.abspath(self.sqlite_vfs.db_path) + done = 0 for root, dirs, files in os.walk(source_folder): # 目录剪枝：被排除的目录整棵跳过 dirs[:] = [ @@ -98,7 +112,11 @@ class FolderPacker: ] for dir_name in dirs: abs_path = os.path.join(root, dir_name) - yield "dir", abs_path, os.path.relpath(abs_path, source_folder) + rel_path = os.path.relpath(abs_path, source_folder) + done += 1 + if progress is not None: + progress(done, total, rel_path) + yield "dir", abs_path, rel_path for file_name in files: abs_path = os.path.join(root, file_name) rel_path = os.path.relpath(abs_path, source_folder) @@ -108,13 +126,17 @@ class FolderPacker: if os.path.abspath(abs_path) == db_abspath: print(f"跳过输出数据库自身: {rel_path}") continue + done += 1 + if progress is not None: + progress(done, total, rel_path) yield "file", abs_path, rel_path # ------------------------------------------------------------------ # 全量 # ------------------------------------------------------------------ - def _pack_full(self, source_folder): + def _pack_full(self, source_folder, progress=None): + total = self._count_entries(source_folder) if progress is not None else 0 print(f"开始打包文件夹（全量）: {source_folder}") start_time = datetime.now() @@ -126,7 +148,7 @@ class FolderPacker: self.excluded_count = 0 failed = [] - for kind, abs_path, rel_path in self._walk(source_folder): + for kind, abs_path, rel_path in self._walk(source_folder, progress, total): if kind == "dir": if not self.sqlite_vfs.add_directory(rel_path, source_folder): failed.append(rel_path) @@ -210,7 +232,8 @@ class FolderPacker: self.sqlite_vfs.conn.commit() return deleted - def _pack_incremental(self, source_folder, rehash=False): + def _pack_incremental(self, source_folder, rehash=False, progress=None): + total = self._count_entries(source_folder) if progress is not None else 0 mode = "sha256 全量比对" if rehash else "mtime+size 比对" print(f"开始打包文件夹（增量，{mode}）: {source_folder}") start_time = datetime.now() @@ -225,7 +248,7 @@ class FolderPacker: self.excluded_count = 0 failed = [] - for kind, abs_path, rel_path in self._walk(source_folder): + for kind, abs_path, rel_path in self._walk(source_folder, progress, total): key = self.sqlite_vfs._normalize(rel_path) seen.add(key) previous = known.get(key) diff --git a/src/sqlite_vfs/publish.py b/src/sqlite_vfs/publish.py index b8f0069..ec09093 100644 --- a/src/sqlite_vfs/publish.py +++ b/src/sqlite_vfs/publish.py @@ -251,6 +251,20 @@ def load_manifest(source, timeout=30): raw = f.read() manifest = json.loads(raw.decode("utf-8")) if manifest.get("format") != MANIFEST_FORMAT: + plugins = manifest.get("plugins") if isinstance(manifest, dict) else None + if isinstance(plugins, list): + # 最常见的误操作：把服务器根地址当清单填进来（服务端索引恰好 + # 也是 JSON）。顺手把可用插件 id 报出来，别只甩一个 format=None + if plugins: + raise ValueError( + "这个地址是服务器索引而不是插件清单（现有插件: " + + "、".join(str(p) for p in plugins[:8]) + + "）。请填插件 id，或 <服务器地址>/<插件id>/manifest.json" + ) + raise ValueError( + "这个地址是服务器索引而不是插件清单（服务器上还没有插件）——" + "请先发布一个插件，再填它的插件 id 或 manifest 地址" + ) raise ValueError( f"不是 {MANIFEST_FORMAT} 清单（format={manifest.get('format')!r}）" ) diff --git a/tests/test_cli_main.py b/tests/test_cli_main.py new file mode 100644 index 0000000..cce5ca1 --- /dev/null +++ b/tests/test_cli_main.py @@ -0,0 +1,197 @@ +"""统一入口 ``svfs <子命令>``（sqlite_vfs.cli.main）。 + +约定： +* 每个子命令与旧的 ``vfs*`` 别名命令跑同一段 ``execute``，输出与退出码一致； +* 退出码：0 成功，1 失败，argparse 用法错误 2； +* 不带子命令 → 打印帮助并以 0 退出（而不是甩 usage 错误）。 + +端到端走子进程（与 test_cli.py 同一套 run_cli），保证测的就是用户敲的那条命令。 +""" + +import os +import unittest + +import helpers + +from sqlite_vfs.core import SQLiteVFS + + +class DispatcherTest(helpers.VFSTestCase): + def run_svfs(self, *args): + return helpers.run_cli("sqlite_vfs.cli.main", *args) + + def setUp(self): + super().setUp() + self.src = helpers.make_tree( + self.path("src"), {"a.txt": b"aaa", "sub/b.txt": b"bbb"} + ) + self.db = self.path("app.svfs") + + def pack(self, *extra): + result = self.run_svfs("pack", self.src, self.db, "--compress", *extra) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + return result + + # ------------------------------------------------------------------ + + def test_no_subcommand_prints_help_and_exits_zero(self): + result = self.run_svfs() + self.assertEqual(result.returncode, 0) + self.assertIn("svfs pack", result.stdout) + self.assertIn("usage", result.stdout.lower()) + + def test_help_lists_all_subcommands(self): + result = self.run_svfs("--help") + self.assertEqual(result.returncode, 0) + for name in ("pack", "unpack", "verify", "sign", "info", "migrate", "publish", "fetch"): + self.assertIn(name, result.stdout) + + def test_version_flag(self): + result = self.run_svfs("--version") + self.assertEqual(result.returncode, 0) + self.assertIn("svfs", result.stdout) + + def test_unknown_command_is_usage_error(self): + result = self.run_svfs("nope") + self.assertEqual(result.returncode, 2) + + +class SubcommandFlowTest(helpers.VFSTestCase): + """每个子命令端到端一遍：这条链就是 README 上的主流程。""" + + def setUp(self): + super().setUp() + self.src = helpers.make_tree( + self.path("src"), + {"a.txt": b"aaa", "sub/b.txt": b"bbb", "dup.txt": b"aaa"}, # dup 与 a 同内容 + ) + self.db = self.path("app.svfs") + + def run_svfs(self, *args, expect=0): + result = helpers.run_cli("sqlite_vfs.cli.main", *args) + self.assertEqual( + result.returncode, expect, f"svfs {args}\n{result.stdout}\n{result.stderr}" + ) + return result + + def test_full_lifecycle(self): + self.run_svfs("pack", self.src, self.db, "--compress", "--name", "demo") + + # info：统计与去重收益 + info = self.run_svfs("info", self.db) + self.assertIn("demo", info.stdout) + self.assertIn("schema v3", info.stdout) + self.assertIn("去重+压缩共省下", info.stdout) + + # sign / verify + self.run_svfs("sign", self.db, "--key", "k1", "--key-id", "ci") + verify = self.run_svfs("verify", self.db, "--key", "k1") + self.assertIn("签名: OK", verify.stdout) + wrong = self.run_svfs("verify", self.db, "--key", "别的密钥", expect=1) + self.assertIn("签名: 失败", wrong.stdout) + + # unpack + 内容比对 + out = self.path("out") + self.run_svfs("unpack", self.db, out) + helpers.assert_tree_equal( + self, helpers.read_tree(self.src), helpers.read_tree(out) + ) + + # publish → fetch 回环 + public = self.path("public") + self.run_svfs("publish", self.db, public) + self.assertTrue(os.path.exists(os.path.join(public, "manifest.json"))) + back = self.path("back") + self.run_svfs("fetch", os.path.join(public, "manifest.json"), back) + helpers.assert_tree_equal( + self, helpers.read_tree(self.src), helpers.read_tree(back) + ) + + def test_info_json(self): + self.run_svfs("pack", self.src, self.db) + import json + + result = self.run_svfs("info", self.db, "--json") + data = json.loads(result.stdout) + self.assertEqual(data["name"], "packed_filesystem") + self.assertEqual(data["files"], 3) + self.assertEqual(data["unique_contents"], 2) # a.txt 与 dup.txt 去重 + self.assertGreater(data["saved_bytes"], 0) + self.assertEqual(len(data["snapshots"]), 1) + + def test_sign_json_and_snapshot_selection(self): + from sqlite_vfs.folder_packer import FolderPacker + + self.run_svfs("pack", self.src, self.db) + # 打一版历史出来 + vfs = SQLiteVFS(self.db, compress=True) + try: + with helpers.quiet(): + FolderPacker(vfs, fs_name="demo").pack_folder( + self.src, incremental=True + ) + vfs.create_snapshot("v2", copy_entries=True) + ids = [row["id"] for row in vfs.list_snapshots()] + finally: + vfs.close() + + result = self.run_svfs("sign", self.db, "--key", "k", "--snapshot", str(ids[0]), "--json") + import json + + data = json.loads(result.stdout) + self.assertEqual(data["snapshot"], ids[0]) + self.assertTrue(data["signature"]) + + # 验签指定同一快照 → OK；最新快照（未签名）→ 失败 + self.run_svfs("verify", self.db, "--key", "k", "--snapshot", str(ids[0])) + self.run_svfs("verify", self.db, "--key", "k", expect=1) + + def test_pack_snapshot_flag_creates_version(self): + self.run_svfs("pack", self.src, self.db) + result = self.run_svfs("pack", self.src, self.db, "--incremental", "--snapshot", "v2") + self.assertIn("已创建快照 v2", result.stdout) + vfs = SQLiteVFS(self.db, readonly=True) + try: + names = [row["name"] for row in vfs.list_snapshots()] + finally: + vfs.close() + self.assertEqual(len(names), 2) + self.assertIn("v2", names) + + def test_error_paths_report_and_exit_1(self): + self.run_svfs("pack", self.path("不存在"), self.db, expect=1) + self.run_svfs("info", self.path("不存在.svfs"), expect=1) + self.run_svfs("sign", self.path("不存在.svfs"), "--key", "k", expect=1) + + self.pack_helper() + # 对不是 SVFS 的文件 info/verify/sign 都要干净地失败 + garbage = self.path("garbage.db") + with open(garbage, "wb") as handle: + handle.write(b"x" * 4096) + self.run_svfs("info", garbage, expect=1) + self.run_svfs("verify", garbage, expect=1) + self.run_svfs("sign", garbage, "--key", "k", expect=1) + + def pack_helper(self): + self.run_svfs("pack", self.src, self.db, "--compress") + + def test_legacy_aliases_share_the_same_execute(self): + """旧命令与新入口是同一段实现：退出码一致，产出的库摘要一致（内容寻址）。""" + fresh_db = self.path("alias.svfs") + new_run = helpers.run_cli("sqlite_vfs.cli.main", "pack", self.src, self.db, "--compress") + legacy_run = helpers.run_cli("sqlite_vfs.cli.packer", self.src, fresh_db, "--compress") + self.assertEqual(new_run.returncode, 0, new_run.stderr) + self.assertEqual(legacy_run.returncode, 0, legacy_run.stderr) + + def digest(path): + vfs = SQLiteVFS(path, readonly=True) + try: + return vfs.get_stats()["digest"] + finally: + vfs.close() + + self.assertEqual(digest(self.db), digest(fresh_db)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_pack_progress.py b/tests/test_pack_progress.py new file mode 100644 index 0000000..dad2852 --- /dev/null +++ b/tests/test_pack_progress.py @@ -0,0 +1,145 @@ +"""打包进度回调（pack_folder 的 progress 参数）。 + +约定： +* 每处理一个条目（目录也算）回调一次，``done`` 从 1 递增到 ``total``； +* ``total`` 是完整遍历一遍得到的条目数，为了拿到它多做一次"只 stat"的预扫描， + 因此**只有传了回调时**才算，不传回调的路径（CLI）开销一点不变； +* 回调抛出的异常会一路向上抛——GUI 的中断按钮就靠这个语义。 +""" + +import unittest + +import helpers + +from sqlite_vfs.core import SQLiteVFS +from sqlite_vfs.folder_packer import FolderPacker + + +class PackProgressTest(helpers.VFSTestCase): + def setUp(self): + super().setUp() + self.src = helpers.make_tree( + self.path("src"), + { + "a.txt": b"aaa", + "b.txt": b"bbb", + "sub/c.txt": b"ccc", + "sub/deep/d.txt": b"ddd", + "empty_dir/.keep": b"", + }, + ) + # 条目数 = 5 个文件（含 empty_dir/.keep）+ 3 个目录（sub / sub/deep / empty_dir） + self.expected_entries = 8 + + def collect(self, incremental=False): + calls = [] + + def progress(done, total, rel_path): + calls.append((done, total, rel_path)) + + vfs = SQLiteVFS(self.path("prog.svfs"), compress=True) + try: + packer = FolderPacker(vfs, fs_name="prog") + with helpers.quiet(): + ok = packer.pack_folder(self.src, incremental=incremental, progress=progress) + finally: + vfs.close() + return ok, calls + + def test_full_progress_is_one_call_per_entry(self): + ok, calls = self.collect() + self.assertTrue(ok) + self.assertEqual(len(calls), self.expected_entries) + self.assertEqual([done for done, _total, _path in calls], list(range(1, len(calls) + 1))) + + def test_total_matches_entry_count(self): + _ok, calls = self.collect() + totals = {total for _done, total, _path in calls} + self.assertEqual(totals, {self.expected_entries}) + + def test_progress_paths_are_relative_and_normalized(self): + _ok, calls = self.collect() + paths = [path.replace("\\", "/") for _done, _total, path in calls] + self.assertEqual(sorted(paths), sorted(["a.txt", "b.txt", "sub", "sub/c.txt", + "sub/deep", "sub/deep/d.txt", + "empty_dir", "empty_dir/.keep"])) + + def test_incremental_progress_counts_all_entries(self): + ok, calls = self.collect(incremental=True) + self.assertTrue(ok) + self.assertEqual(len(calls), self.expected_entries) + self.assertEqual({total for _done, total, _path in calls}, {self.expected_entries}) + + def test_result_is_same_with_and_without_progress(self): + """带回调不该改变打包结果：条目集合必须一模一样。""" + db_with = self.path("with.svfs") + db_without = self.path("without.svfs") + + vfs = SQLiteVFS(db_with, compress=True) + try: + with helpers.quiet(): + FolderPacker(vfs, fs_name="prog").pack_folder( + self.src, progress=lambda *_args: None + ) + finally: + vfs.close() + + helpers.pack(self.src, db_without, compress=True) + + def snapshot(db_path): + vfs = SQLiteVFS(db_path, readonly=True) + try: + return [ + (row["path"], row["is_directory"], row["file_size"], row["content_hash"]) + for row in vfs.all_entries() + ], sorted(row["hash"] for row in vfs.list_blobs()) + finally: + vfs.close() + + self.assertEqual(snapshot(db_with), snapshot(db_without)) + + def test_no_precount_when_progress_is_absent(self): + """不传回调就不该多做那次预扫描（CLI 的开销必须保持原样）。""" + vfs = SQLiteVFS(self.path("nopre.svfs"), compress=True) + try: + packer = FolderPacker(vfs, fs_name="prog") + original = packer._count_entries + called = [] + + def spy(source_folder): + called.append(source_folder) + return original(source_folder) + + packer._count_entries = spy + with helpers.quiet(): + self.assertTrue(packer.pack_folder(self.src)) + self.assertEqual(called, []) + finally: + vfs.close() + + def test_exception_from_callback_propagates(self): + """回调抛异常要能中断打包——GUI 的"中断"就是靠这个语义实现的。""" + + class Stop(Exception): + pass + + seen = [] + + def progress(done, total, rel_path): + seen.append(rel_path) + if done >= 2: + raise Stop("用户中断") + + vfs = SQLiteVFS(self.path("cancel.svfs"), compress=True) + try: + packer = FolderPacker(vfs, fs_name="prog") + with helpers.quiet(): + with self.assertRaises(Stop): + packer.pack_folder(self.src, progress=progress) + self.assertEqual(len(seen), 2) + finally: + vfs.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_publish.py b/tests/test_publish.py index 511ce67..67e7227 100644 --- a/tests/test_publish.py +++ b/tests/test_publish.py @@ -449,6 +449,26 @@ class FetchOverHttpTest(HttpServerFixture): with open(os.path.join(self.path("unverified"), "sub", "b.js"), "rb") as f: self.assertEqual(b"console.log(2)", f.read()) + def test_server_index_json_gets_actionable_error(self): + # 最常见的误操作：把服务器根地址（索引 JSON）当清单地址填进来， + # 报错要直接给出可用的插件 id，而不是一个看不懂的 format=None + index = self.path("index.json") + with open(index, "w", encoding="utf-8") as f: + json.dump({"service": "svfs-plugin-server", + "plugins": ["com.svfs.demo", "com.example.other"]}, f) + with self.assertRaises(ValueError) as ctx: + publish.load_manifest(index) + message = str(ctx.exception) + self.assertIn("服务器索引", message) + self.assertIn("com.svfs.demo", message) + + empty = self.path("empty-index.json") + with open(empty, "w", encoding="utf-8") as f: + json.dump({"service": "svfs-plugin-server", "plugins": []}, f) + with self.assertRaises(ValueError) as ctx: + publish.load_manifest(empty) + self.assertIn("还没有插件", str(ctx.exception)) + def test_manifest_from_url_mismatch_is_reported(self): _src, objects, base, _db = self._export_and_serve() with self.assertRaises(Exception):