"""统一入口 ``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()