diff --git a/pyproject.toml b/pyproject.toml index 28c2a1d..553e064 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,7 +10,10 @@ readme = "README.md" # 仓库里没有 3.10+ 的语法（无 match、无 PEP 604 注解、无 zip(strict=)）。 # 说明：这里只在本机的 3.13 上跑过，"3.9 可用"是代码审查的结论，没有在 3.9 上实测。 requires-python = ">=3.9" -version = "0.3.0" +# 0.4.0: 新增 SQLiteVFS.add_bytes() / commit()（从内存写文件，不经过磁盘）， +# 且 add_file/add_bytes 默认会补齐缺失的父目录行——后者是可见的行为变化 +# （会让 export_directory/export_filesystem 从根递归得下去），所以抬次版本号。 +version = "0.4.0" # 可选集成：核心功能只用标准库，第三方框架一律走 extras。 # 例如 pip install "sqlite-vfs[fastapi]" 才会装上 FastAPI/Starlette。 diff --git a/src/sqlite_vfs/core.py b/src/sqlite_vfs/core.py index b403dc1..e17408b 100644 --- a/src/sqlite_vfs/core.py +++ b/src/sqlite_vfs/core.py @@ -34,6 +34,12 @@ from . import compression #: 2 = 内容与算法列在 files 表里，3 = blobs + snapshots（本版本）。 SCHEMA_VERSION = 3 +#: 由 add_bytes 补齐的父目录行所用的权限位。 +#: files 表的 schema 默认值 420(0o644) 是按**文件**定的，套到目录上会让 +#: 导出的目录不可进入（缺 x 位）——真实打包时目录权限来自 `st_mode & 0o777` +#: （通常是 0o755），这里跟它保持一致。 +DEFAULT_DIR_PERMISSIONS = 0o755 + class VFSFormatError(Exception): """数据库不是当前版本能够读取的格式（例如由旧版本创建）。""" @@ -512,6 +518,34 @@ class SQLiteVFS: ).fetchall() return [self._parse_row_datetime(row) for row in rows] + @_writable + def rename_snapshot(self, snapshot_id, name): + """给快照改名，返回改后的名字。 + + 版本管理里"把发布名/版本号落在快照上"是基本动作（create / list / + delete 都有，缺一个改名），而**定格一个版本靠的正是改名**：先给当前 + 快照改名，再另建一个工作快照接手后续写入，那样这个名字才真正代表 + "一份不再变动的历史内容"。 + + 只动 ``snapshots.name``，不影响内容、统计与摘要；重名是允许的 + （表上没有唯一约束），按名字找的时候取最新的那一个。 + """ + if self.conn is None: + raise ConnectionError("文件系统尚未连接") + name = str(name or "").strip() + if not name: + raise ValueError("快照名不能为空") + row = self.conn.execute( + "SELECT 1 FROM snapshots WHERE id = ?", (snapshot_id,) + ).fetchone() + if row is None: + raise FileNotFoundError(f"快照不存在: {snapshot_id}") + self.conn.execute( + "UPDATE snapshots SET name = ? WHERE id = ?", (name, snapshot_id) + ) + self.conn.commit() + return name + @_writable def delete_snapshot(self, snapshot_id, gc=True): """删除一个快照（连同它的目录项与签名），返回被删掉的条目数。""" @@ -731,25 +765,87 @@ class SQLiteVFS: ) return digest, False + def _ensure_dir_rows(self, cursor, normalized, snapshot_id, mtime_text): + """自上而下补齐缺失的父目录行，已存在的层不动。 + + 为什么必须补：``list_directory`` 是按 ``parent_path`` 过滤的，而"某个 + 目录存在"的唯一标志就是它自己那一行——少一层目录行，``export_directory`` + 从根就递归不下去，``export_filesystem()`` 会静默导出一个空目录。 + + 只算文件的 ``total_files`` / ``total_size`` 不在这里动（口径见 recount）。 + """ + missing = [] + parent = self._parent_of(normalized) + while parent: + if self.conn.execute( + "SELECT 1 FROM files WHERE snapshot_id = ? AND path = ?", + (snapshot_id, parent), + ).fetchone(): + break # 这一层已存在，它上面必然也都在（插入顺序保证），不必再往上查 + missing.append(parent) + parent = self._parent_of(parent) + + for dir_path in reversed(missing): # 从最上层往下逐层建 + cursor.execute( + """ + INSERT INTO files + (snapshot_id, path, name, parent_path, is_directory, + created_time, modified_time, permissions) + VALUES (?, ?, ?, ?, 1, ?, ?, ?) + """, + ( + snapshot_id, + dir_path, + self._basename_of(dir_path), + self._parent_of(dir_path), + mtime_text, + mtime_text, + DEFAULT_DIR_PERMISSIONS, + ), + ) + @_writable - def add_file(self, abs_path, rel_path, replace=False): - """添加文件到数据库。 + def add_bytes(self, rel_path, data, replace=False, make_parents=True, + mtime=None, created_time=None, permissions=0o644, + commit=True): + """把内存里的一段字节写成一个文件（不经过磁盘）。 - 返回是否成功；成功时数据已经提交（不再依赖调用方后续的 commit）。 - 内容按 sha256 去重：库里已有同样内容时不会再存一份，也不会重复压缩。 + 与 :meth:`add_file` 的唯一区别是"内容从哪来"：add_file 从磁盘读一个 + 真实文件，这里直接吃调用方手里的 bytes。**受保护的字节码、密文之类 + 绝对不能落盘的数据只能走这条路径**——先写临时文件再 add_file 会把 + 明文留在磁盘上，而磁盘残留往往是这类工具最容易被忽视的泄漏点。 - replace=True 时先删掉同路径的旧条目再写入（增量打包时覆盖用）。 + 内容按 sha256 去重（库里已有同样内容时不再存一份，也不重复压缩）。 + + 参数: + replace: True 时先删掉同路径旧条目再写入（覆盖写用） + make_parents: True 时补齐缺失的父目录行（见 :meth:`_ensure_dir_rows`） + mtime / created_time: 时间戳（datetime；None = 取当前时间） + permissions: 权限位（8 进制整数） + commit: False 时只写不提交，留给调用方批量写完再 :meth:`commit`—— + 一组互相关联的文件（例如一个模块的源码 + 元数据 + 字节码）因此 + 落在同一个事务里，崩在中间也不会出现"新元数据配旧内容"这种 + 要等到加载时才炸的错配。 + + 返回是否成功（不抛异常，与 add_file 一致）。失败时整批回滚：包括本次 + 之前那些 ``commit=False`` 的调用，所以批量写入是"要么全成要么全不成"。 """ if self.conn is None: raise ConnectionError("文件系统尚未连接") + raw = bytes(data) try: - stat = os.stat(abs_path) - with open(abs_path, "rb") as f: - raw = f.read() snapshot_id = self._ensure_snapshot() normalized = self._normalize(rel_path) + if not normalized: + raise ValueError("路径不能为空") + + now = datetime.now() + modified = _db_time(mtime or now) + created = _db_time(created_time or now) cursor = self.conn.cursor() + if make_parents: + self._ensure_dir_rows(cursor, normalized, snapshot_id, modified) digest, dedup_hit = self._store_blob(cursor, raw) if replace: cursor.execute( @@ -766,20 +862,21 @@ class SQLiteVFS: ( snapshot_id, normalized, - os.path.basename(rel_path), - self._normalize(os.path.dirname(rel_path)), - stat.st_size, - _db_time(datetime.fromtimestamp(stat.st_ctime)), - _db_time(datetime.fromtimestamp(stat.st_mtime)), - stat.st_mode & 0o777, + self._basename_of(normalized), + self._parent_of(normalized), + len(raw), + created, + modified, + permissions, digest, ), ) - self.conn.commit() + if commit: + self.conn.commit() self.total_files += 1 - self.total_size += stat.st_size + self.total_size += len(raw) mark = " 去重" if dedup_hit else "" - print(f"添加文件: {rel_path} ({stat.st_size} bytes{mark})") + print(f"添加文件: {normalized} ({len(raw)} bytes{mark})") return True except Exception as e: @@ -787,6 +884,44 @@ class SQLiteVFS: print(f"添加文件失败 {rel_path}: {e}") return False + def commit(self): + """提交当前事务。 + + 给 ``add_bytes(commit=False)`` 的批量写入收尾：一组相关文件全写完再一次 + 提交，中途失败时前面的写入也一起回滚，不会在库里留下半套数据。 + """ + if self.conn is None: + raise ConnectionError("文件系统尚未连接") + self.conn.commit() + + @_writable + def add_file(self, abs_path, rel_path, replace=False, make_parents=True): + """添加磁盘上的文件到数据库。 + + 返回是否成功；成功时数据已经提交（不再依赖调用方后续的 commit）。 + 内容按 sha256 去重：库里已有同样内容时不会再存一份，也不会重复压缩。 + + replace=True 时先删掉同路径的旧条目再写入（增量打包时覆盖用）； + make_parents=True 时顺带补齐缺失的父目录行（见 :meth:`_ensure_dir_rows`）。 + """ + try: + stat = os.stat(abs_path) + with open(abs_path, "rb") as f: + raw = f.read() + except OSError as e: + # 读不到就没写过任何东西，不需要回滚 + print(f"添加文件失败 {rel_path}: {e}") + return False + return self.add_bytes( + rel_path, + raw, + replace=replace, + make_parents=make_parents, + mtime=datetime.fromtimestamp(stat.st_mtime), + created_time=datetime.fromtimestamp(stat.st_ctime), + permissions=stat.st_mode & 0o777, + ) + def recount(self): """按库内实际内容重算当前快照的 total_files / total_size 并落库。 diff --git a/tests/test_core_add_bytes.py b/tests/test_core_add_bytes.py new file mode 100644 index 0000000..e2b477e --- /dev/null +++ b/tests/test_core_add_bytes.py @@ -0,0 +1,296 @@ +"""add_bytes 的契约测试：把内存里的字节直接写进库。 + +这个 API 存在的理由：受保护的字节码、密文之类**绝不能落盘**的数据只能这样 +进库——先写临时文件再 add_file 会把明文留在磁盘上。除了"不经过磁盘"本身， +这里还钉住它相对 add_file 的两条额外保证： + +* 补齐缺失的父目录行（少了目录行，export_directory 从根就递归不下去， + export_filesystem() 会静默导出一个空目录）； +* ``commit=False`` 的批量写入是"要么全成要么全不成"（写入过程中任何一步失败 + 都会把同一批里先前的写入一起回滚）。 +""" + +import os +from datetime import datetime + +import helpers + +from sqlite_vfs.core import SQLiteVFS + + +class AddBytesBasicsTest(helpers.VFSTestCase): + """写进去的字节要能一模一样读回来。""" + + def _vfs(self, **kwargs): + vfs = SQLiteVFS(":memory:", **kwargs) + self.addCleanup(vfs.close) + return vfs + + def test_writes_bytes_without_a_source_file_on_disk(self): + """核心契约：内容来自内存，磁盘上不需要（也不该有）任何源文件。""" + vfs = self._vfs() + with helpers.quiet(): + ok = vfs.add_bytes("input.bin", b"payload-bytes") + + self.assertTrue(ok) + self.assertTrue(vfs.exists("input.bin")) + self.assertEqual(b"payload-bytes", vfs.read_file("input.bin")) + + def test_empty_content_roundtrips(self): + vfs = self._vfs() + with helpers.quiet(): + self.assertTrue(vfs.add_bytes("empty.bin", b"")) + self.assertEqual(b"", vfs.read_file("empty.bin")) + self.assertEqual(0, vfs.get_file_info("empty.bin")["file_size"]) + + def test_file_size_matches_content_length(self): + vfs = self._vfs() + data = b"x" * 12345 + with helpers.quiet(): + vfs.add_bytes("big.bin", data) + self.assertEqual(len(data), vfs.get_file_info("big.bin")["file_size"]) + + def test_accepts_bytearray_and_memoryview(self): + """调用方手里的常常不是 bytes（bytearray / memoryview 很常见）。""" + vfs = self._vfs() + with helpers.quiet(): + self.assertTrue(vfs.add_bytes("ba.bin", bytearray(b"bytearray"))) + self.assertTrue(vfs.add_bytes("mv.bin", memoryview(b"memoryview"))) + self.assertEqual(b"bytearray", vfs.read_file("ba.bin")) + self.assertEqual(b"memoryview", vfs.read_file("mv.bin")) + + def test_mtime_and_permissions_are_recorded(self): + """时间戳与权限是调用方给的（内存内容没有 stat 可查）。""" + vfs = self._vfs() + stamp = datetime(2020, 1, 2, 3, 4, 5) + with helpers.quiet(): + vfs.add_bytes("stamped.bin", b"d", mtime=stamp, permissions=0o600) + + info = vfs.get_file_info("stamped.bin") + self.assertEqual(stamp, info["modified_time"]) + self.assertEqual(0o600, info["permissions"]) + + def test_compressible_content_uses_zlib_when_enabled(self): + vfs = self._vfs(compress=True) + data = b"repeat me please. " * 500 + with helpers.quiet(): + vfs.add_bytes("packed.txt", data) + + self.assertEqual(b"repeat me please. " * 500, vfs.read_file("packed.txt")) + self.assertEqual("zlib", vfs.get_blob_info("packed.txt")["compress_algo"]) + + def test_readonly_handle_refuses_to_write(self): + db = self.path("ro.svfs") + writer = SQLiteVFS(db) + with helpers.quiet(): + writer.add_bytes("keep.bin", b"keep") + writer.close() + + vfs = SQLiteVFS(db, readonly=True) + self.addCleanup(vfs.close) + with self.assertRaises(PermissionError): + vfs.add_bytes("nope.bin", b"nope") + + +class AddBytesDedupTest(helpers.VFSTestCase): + """内容寻址的既有语义对内存写入同样成立。""" + + def _vfs(self): + vfs = SQLiteVFS(":memory:") + self.addCleanup(vfs.close) + return vfs + + def test_identical_content_is_stored_once(self): + vfs = self._vfs() + with helpers.quiet(): + self.assertTrue(vfs.add_bytes("first.bin", b"same-bytes")) + self.assertTrue(vfs.add_bytes("second.bin", b"same-bytes")) + + self.assertEqual(1, len(vfs.list_blobs()), "相同内容应当只占一个 blob") + self.assertEqual(b"same-bytes", vfs.read_file("first.bin")) + self.assertEqual(b"same-bytes", vfs.read_file("second.bin")) + + def test_replace_overwrites_and_leaves_a_single_entry(self): + vfs = self._vfs() + with helpers.quiet(): + vfs.add_bytes("data.bin", b"old") + self.assertTrue(vfs.add_bytes("data.bin", b"new", replace=True)) + + self.assertEqual(b"new", vfs.read_file("data.bin")) + self.assertEqual({"data.bin"}, helpers.all_paths(vfs)) + + def test_duplicate_without_replace_is_reported_and_keeps_first_content(self): + """与 add_file 同样的冲突语义：不覆盖，返回 False。""" + vfs = self._vfs() + with helpers.quiet(): + self.assertTrue(vfs.add_bytes("same.bin", b"first")) + self.assertFalse(vfs.add_bytes("same.bin", b"second")) + + self.assertEqual(b"first", vfs.read_file("same.bin")) + + def test_path_that_normalizes_to_the_root_is_rejected(self): + """空路径与 "/" 都会规范化成根，写下去就是一行 path='' 的伪条目。 + + 注意只挡"规范化后为空"这一种：含空格的路径是**合法**文件名（与 + add_file 保持一致），不该在这里被拦掉。 + """ + vfs = self._vfs() + with helpers.quiet(): + self.assertFalse(vfs.add_bytes("", b"x")) + self.assertFalse(vfs.add_bytes("/", b"x")) + self.assertFalse(vfs.add_bytes("///", b"x")) + self.assertEqual(set(), helpers.all_paths(vfs)) + + +class AddBytesParentDirectoryTest(helpers.VFSTestCase): + """父目录行：export 能不能从根递归下去，全看它在不在。""" + + def _vfs(self): + vfs = SQLiteVFS(":memory:") + self.addCleanup(vfs.close) + return vfs + + def test_missing_parents_are_created_top_down(self): + vfs = self._vfs() + with helpers.quiet(): + self.assertTrue(vfs.add_bytes("a/b/c/deep.txt", b"x")) + + for directory in ("a", "a/b", "a/b/c"): + self.assertTrue(vfs.is_dir(directory), f"{directory} 应当被补成目录") + self.assertEqual("a/b/c", helpers.path_parents(vfs)["a/b/c/deep.txt"]) + + def test_make_parents_false_keeps_the_old_behaviour(self): + vfs = self._vfs() + with helpers.quiet(): + self.assertTrue( + vfs.add_bytes("a/b/c.txt", b"x", make_parents=False) + ) + + self.assertFalse(vfs.is_dir("a/b"), "显式关掉时不应该补目录行") + self.assertTrue(vfs.is_file("a/b/c.txt"), "文件本身仍然写进去了") + + def test_existing_directory_rows_are_left_alone(self): + """已经有目录行（正常打包出来的）时不能重复插入。""" + vfs = self._vfs() + with helpers.quiet(): + vfs.mk_dir("a") + vfs.add_bytes("a/x.txt", b"x") + + count = vfs.conn.execute( + "SELECT COUNT(*) AS n FROM files WHERE snapshot_id = ? AND path = 'a'", + (vfs._active_snapshot(),), + ).fetchone()["n"] + self.assertEqual(1, count, f"路径 a 应当只有一行，实际 {count} 行") + + def test_export_filesystem_writes_the_whole_nested_tree(self): + """这条才是 make_parents 的真正目的，也是它必须默认为 True 的原因。""" + vfs = self._vfs() + with helpers.quiet(): + vfs.add_bytes("modules/pkg/__init__.pyc", b"bytecode") + vfs.add_bytes("modules/pkg/util.py", b"# source") + vfs.add_bytes("top.bin", b"top") + + out = self.path("out") + with helpers.quiet(): + vfs.export_filesystem(out) + + expected = { + "top.bin": b"top", + "modules": None, + "modules/pkg": None, + "modules/pkg/__init__.pyc": b"bytecode", + "modules/pkg/util.py": b"# source", + } + helpers.assert_tree_equal(self, expected, helpers.read_tree(out)) + + +class AddBytesBatchCommitTest(helpers.VFSTestCase): + """commit=False：一组文件写成一个事务。""" + + def test_batch_persists_after_a_single_commit(self): + db = self.path("batch.svfs") + vfs = SQLiteVFS(db) + with helpers.quiet(): + vfs.add_bytes("source.py", b"# source", commit=False) + vfs.add_bytes("module.pyc", b"bytecode", commit=False) + vfs.commit() + vfs.close() + + reopened = SQLiteVFS(db) + self.addCleanup(reopened.close) + self.assertTrue(reopened.exists("source.py")) + self.assertEqual(b"bytecode", reopened.read_file("module.pyc")) + + def test_uncommitted_batch_is_lost_when_the_caller_never_commits(self): + """没有 commit 就不该落库——否则"一模块一事务"根本没有意义。""" + db = self.path("nocommit.svfs") + vfs = SQLiteVFS(db) + with helpers.quiet(): + vfs.add_bytes("source.py", b"# source", commit=False) + vfs.close() + + reopened = SQLiteVFS(db) + self.addCleanup(reopened.close) + self.assertFalse(reopened.exists("source.py")) + + def test_a_failed_write_rolls_back_the_whole_batch(self): + """失败要回滚整批（含之前那些 commit=False 的调用），不能留半套数据。""" + vfs = SQLiteVFS(":memory:") + self.addCleanup(vfs.close) + with helpers.quiet(): + self.assertTrue(vfs.add_bytes("good.bin", b"good", commit=False)) + self.assertFalse(vfs.add_bytes("", b"bad", commit=False)) + + self.assertFalse( + vfs.exists("good.bin"), "整批回滚后，先前写入的文件不该留下" + ) + + def test_commit_is_public_and_usable_on_a_readonly_handle(self): + """commit 只转发给连接；只读连接上没有待提交的事务，不该炸。""" + db = self.path("ro_commit.svfs") + writer = SQLiteVFS(db) + with helpers.quiet(): + writer.add_bytes("keep.bin", b"keep") + writer.close() + + vfs = SQLiteVFS(db, readonly=True) + self.addCleanup(vfs.close) + vfs.commit() + + +class AddFileDelegationTest(helpers.VFSTestCase): + """add_file 现在是 add_bytes 的"从磁盘读"外壳，行为不能跑偏。""" + + def _vfs(self): + vfs = SQLiteVFS(":memory:") + self.addCleanup(vfs.close) + return vfs + + def test_add_file_also_creates_parent_rows(self): + """修的是同一个静默 bug：只写文件行的话，导出的目录树是空的。""" + vfs = self._vfs() + src = helpers.write_file(self.path("deep.txt"), b"deep") + with helpers.quiet(): + self.assertTrue(vfs.add_file(src, "doc/sub/deep.txt")) + + self.assertTrue(vfs.is_dir("doc")) + self.assertTrue(vfs.is_dir("doc/sub")) + + def test_add_file_keeps_recording_the_source_stat_times(self): + vfs = self._vfs() + src = helpers.write_file(self.path("t.txt"), b"t") + os.utime(src, (1600000000, 1600000000)) + with helpers.quiet(): + vfs.add_file(src, "t.txt") + + info = vfs.get_file_info("t.txt") + self.assertEqual( + 1600000000, int(info["modified_time"].timestamp()), + "modified_time 应当来自源文件的 st_mtime", + ) + + def test_add_file_returns_false_for_a_missing_source(self): + vfs = self._vfs() + with helpers.quiet(): + self.assertFalse(vfs.add_file(self.path("nope.txt"), "nope.txt")) + self.assertFalse(vfs.exists("nope.txt")) diff --git a/tests/test_snapshots.py b/tests/test_snapshots.py index 1009d6f..98ca77f 100644 --- a/tests/test_snapshots.py +++ b/tests/test_snapshots.py @@ -122,6 +122,78 @@ class SnapshotLifecycleTest(helpers.VFSTestCase): self.assertEqual([1], [s["id"] for s in vfs.list_snapshots()]) self.assertIsNone(vfs.snapshot_id, "当前快照被删后应当需要重新解析") + def test_rename_snapshot_changes_the_name_only(self): + vfs = self._vfs() + src = helpers.write_file(self.path("a.txt"), b"a") + with helpers.quiet(): + vfs.add_file(src, "a.txt") + before = vfs.compute_digest(store=False) + + self.assertEqual("release-1.0", vfs.rename_snapshot(1, "release-1.0")) + self.assertEqual("release-1.0", vfs.get_stats()["snapshot_name"]) + self.assertEqual(["release-1.0"], + [s["name"] for s in vfs.list_snapshots()]) + self.assertEqual(b"a", vfs.read_file("a.txt"), "改名不该动内容") + self.assertEqual(before, vfs.compute_digest(store=False), + "改名不该动摘要") + + def test_rename_snapshot_rejects_blank_name(self): + vfs = self._vfs() + src = helpers.write_file(self.path("a.txt"), b"a") + with helpers.quiet(): + vfs.add_file(src, "a.txt") + for blank in ("", " ", None): + with self.assertRaises(ValueError): + vfs.rename_snapshot(1, blank) + + def test_rename_missing_snapshot_raises(self): + """与 use_snapshot 一致：快照不存在就报 FileNotFoundError。""" + vfs = self._vfs() + src = helpers.write_file(self.path("a.txt"), b"a") + with helpers.quiet(): + vfs.add_file(src, "a.txt") + with self.assertRaises(FileNotFoundError): + vfs.rename_snapshot(987654, "nope") + + def test_rename_allows_duplicate_names(self): + """表上没有唯一约束：重名允许（按名字找时由调用方定"取哪个"）。""" + vfs = self._vfs() + src = helpers.write_file(self.path("a.txt"), b"a") + with helpers.quiet(): + vfs.add_file(src, "a.txt") + vfs.create_snapshot("v2", copy_entries=True) + vfs.rename_snapshot(1, "dup") + vfs.rename_snapshot(2, "dup") + self.assertEqual(["dup", "dup"], + [s["name"] for s in vfs.list_snapshots()]) + + def test_rename_then_fork_freezes_the_named_version(self): + """定格一个版本的组合动作：把当前快照改名，再另建一个工作快照接手写入。 + + 这是"版本号该落在不再变动的那份内容上"的最小保证——不这样做的话， + 名字会落在**继续被写入**的快照上，下一次改动就把这个版本改了。 + """ + vfs = self._vfs() + src = helpers.write_file(self.path("a.txt"), b"v1") + with helpers.quiet(): + vfs.add_file(src, "a.txt") + + vfs.rename_snapshot(vfs._active_snapshot(), "release-1.0") + work = vfs.create_snapshot("work", copy_entries=True) + + # 在工作快照里改成 v2，已定格的 release-1.0 必须原封不动 + src2 = helpers.write_file(self.path("b.txt"), b"v2") + with helpers.quiet(): + vfs.add_file(src2, "a.txt", replace=True) + + self.assertEqual(b"v2", vfs.read_file("a.txt")) + release = SQLiteVFS(self.path("s.svfs"), snapshot_id=1) + self.addCleanup(release.close) + self.assertEqual("release-1.0", release.get_stats()["snapshot_name"]) + self.assertEqual(b"v1", release.read_file("a.txt"), + "定格之后的版本不该被后续写入改动") + self.assertEqual(work, 2) + class TimeTravelTest(helpers.VFSTestCase): def _pack_v1_then_v2(self):