chakcy
8489b4052d
三处写入侧的缺口，都是被下游"模块银行"（module_bank，把模块打包进 svfs 包 再按需导入）真实逼出来的： - add_bytes(rel_path, data, ...)：从内存写文件，不经过磁盘。受保护的字节码、 密文这类绝对不能落盘的内容因此可以直接入库——先写临时文件再 add_file 会把 明文留在磁盘上，而磁盘残留是这类工具最容易被忽视的泄漏点。 - make_parents=True（默认）：补齐缺失的父目录行。list_directory 是按 parent_path 过滤的，而"某个目录存在"的唯一标志就是它自己那一行——少一层 目录行，export_directory 从根就递归不下去，export_filesystem() 会静默导出 一个空目录。add_file 一并受益（内部改为读盘后调 add_bytes，行为不变， 含 created_time 取 st_ctime 这个细节）。 - commit=False + 公开的 commit()：一组相关文件可以写成一个事务，中途任何一步 失败整批回滚，不会在库里留下半套数据。 - rename_snapshot(id, name)：create/list/delete 都有，唯独缺改名；而"把发布名 落在快照上"是版本管理的基本动作。**定格一个版本靠的正是它**：先给当前快照 改名，再另建一个工作快照接手后续写入——直接 create_snapshot(name) 的话， 名字落在会继续被写入的那一版上，下一次改动就把这个"版本"改了。 - DEFAULT_DIR_PERMISSIONS：补出来的目录行用 0o755（表的默认值 0o644 是按文件 定的，套到目录上会让导出的目录不可进入）。 - 版本 0.3.0 → 0.4.0（add_file 开始建父目录行属于可见的行为变化）。 测试：tests/test_core_add_bytes.py（22 条：往返/空内容/bytearray/时间戳与权限/ 压缩/去重/覆盖/空路径/只读拒绝/父目录行与 export 目录树/批量提交与回滚/ add_file 委托行为）+ tests/test_snapshots.py 新增 5 条改名用例（只改名字不动 内容与摘要/空名拒绝/不存在的快照报错/允许重名/改名+派生工作快照之后 已定格的版本不被后续写入改动）。完整测试套件逐个文件跑通。
381 lines
15 KiB
Python
381 lines
15 KiB
Python
"""快照与时间旅行。
|
||
|
||
一个库里可以存同一棵树的历史版本：``files`` 表的唯一约束从 ``path``
|
||
变成 ``UNIQUE(snapshot_id, path)``，所以同一路径在不同快照里各有一行。
|
||
"""
|
||
|
||
import os
|
||
import unittest
|
||
|
||
import helpers
|
||
|
||
from sqlite_vfs.core import SQLiteVFS
|
||
|
||
|
||
class SnapshotLifecycleTest(helpers.VFSTestCase):
|
||
def _vfs(self, **kwargs):
|
||
vfs = SQLiteVFS(self.path("s.svfs"), **kwargs)
|
||
self.addCleanup(vfs.close)
|
||
return vfs
|
||
|
||
def test_first_write_creates_a_default_snapshot(self):
|
||
vfs = self._vfs()
|
||
self.assertIsNone(vfs.snapshot_id)
|
||
src = helpers.write_file(self.path("a.txt"), b"a")
|
||
with helpers.quiet():
|
||
vfs.add_file(src, "a.txt")
|
||
|
||
self.assertIsNotNone(vfs.snapshot_id)
|
||
snapshots = vfs.list_snapshots()
|
||
self.assertEqual(1, len(snapshots))
|
||
self.assertEqual(vfs.snapshot_id, snapshots[0]["id"])
|
||
|
||
def test_add_vfs_metadata_names_the_default_snapshot(self):
|
||
vfs = self._vfs()
|
||
vfs.add_vfs_metadata("my-pack")
|
||
self.assertEqual("my-pack", vfs.list_snapshots()[0]["name"])
|
||
|
||
def test_readonly_never_creates_a_snapshot(self):
|
||
"""只读打开不能因为"读一下"就写库——否则只读挂载会失败。"""
|
||
db = self.path("ro.svfs")
|
||
vfs = SQLiteVFS(db)
|
||
vfs.close()
|
||
before = helpers.md5_of(db)
|
||
|
||
readonly = SQLiteVFS(db, readonly=True)
|
||
self.addCleanup(readonly.close)
|
||
self.assertIsNone(readonly._active_snapshot())
|
||
self.assertEqual([], readonly.list_directory("/"))
|
||
self.assertIsNone(readonly.get_file_info("anything"))
|
||
self.assertEqual(0, readonly.get_stats()["total_files"])
|
||
self.assertEqual(before, helpers.md5_of(db), "只读会话改动了库文件")
|
||
|
||
def test_create_snapshot_becomes_active_and_is_listed(self):
|
||
vfs = self._vfs()
|
||
src = helpers.write_file(self.path("a.txt"), b"a")
|
||
with helpers.quiet():
|
||
vfs.add_file(src, "a.txt")
|
||
first = vfs.snapshot_id
|
||
|
||
second = vfs.create_snapshot("v2")
|
||
self.assertNotEqual(first, second)
|
||
self.assertEqual(second, vfs.snapshot_id, "新建的快照应当成为当前快照")
|
||
self.assertEqual(["v2"], [s["name"] for s in vfs.list_snapshots() if s["id"] == second])
|
||
self.assertEqual(0, helpers.count_files(vfs), "新快照本身是空的")
|
||
|
||
def test_use_snapshot_switches_the_view(self):
|
||
vfs = self._vfs()
|
||
src = helpers.write_file(self.path("a.txt"), b"a")
|
||
with helpers.quiet():
|
||
vfs.add_file(src, "a.txt")
|
||
first = vfs.snapshot_id
|
||
vfs.create_snapshot("v2")
|
||
|
||
vfs.use_snapshot(first)
|
||
self.assertEqual(first, vfs.snapshot_id)
|
||
self.assertTrue(vfs.exists("a.txt"))
|
||
|
||
def test_use_missing_snapshot_raises(self):
|
||
vfs = self._vfs()
|
||
with self.assertRaises(FileNotFoundError):
|
||
vfs.use_snapshot(999)
|
||
|
||
def test_constructor_releases_the_connection_when_snapshot_is_invalid(self):
|
||
"""构造期抛异常也要释放连接，否则 Windows 上库文件被锁住删不掉。
|
||
|
||
注意：POSIX 上删除已打开的文件总是成功，所以这条断言主要是在 Windows 上生效
|
||
（那正是它被发现的地方）。
|
||
"""
|
||
db = self.path("locked.svfs")
|
||
SQLiteVFS(db).close() # 先建出一个合法的库
|
||
with self.assertRaises(FileNotFoundError):
|
||
SQLiteVFS(db, snapshot_id=999)
|
||
|
||
os.remove(db) # 句柄没释放的话这里会 PermissionError
|
||
self.assertFalse(os.path.exists(db))
|
||
|
||
def test_reopen_defaults_to_the_latest_snapshot(self):
|
||
db = self.path("multi.svfs")
|
||
vfs = SQLiteVFS(db)
|
||
src = helpers.write_file(self.path("a.txt"), b"a")
|
||
with helpers.quiet():
|
||
vfs.add_file(src, "a.txt")
|
||
vfs.create_snapshot("v2")
|
||
vfs.close()
|
||
|
||
reopened = SQLiteVFS(db)
|
||
self.addCleanup(reopened.close)
|
||
self.assertEqual(2, reopened.get_stats()["snapshot_id"])
|
||
self.assertEqual("v2", reopened.get_stats()["snapshot_name"])
|
||
|
||
def test_delete_snapshot_removes_entries_and_signature_rows(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")
|
||
vfs.sign("k")
|
||
self.assertEqual(1, len(vfs.list_signatures(2)))
|
||
|
||
deleted = vfs.delete_snapshot(2)
|
||
self.assertEqual(0, deleted, "v2 是空快照，没有条目被删")
|
||
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):
|
||
"""构造两个版本的库：v1 有 a.txt=first，v2 里 a.txt=second 且多了 b.txt。"""
|
||
db = self.path("hist.svfs")
|
||
vfs = SQLiteVFS(db)
|
||
self.addCleanup(vfs.close)
|
||
vfs.add_vfs_metadata("history")
|
||
|
||
first = helpers.write_file(self.path("a1.txt"), b"first")
|
||
with helpers.quiet():
|
||
vfs.add_file(first, "a.txt")
|
||
vfs.recount()
|
||
|
||
vfs.create_snapshot("v2", copy_entries=True)
|
||
vfs.remove("a.txt")
|
||
second = helpers.write_file(self.path("a2.txt"), b"second")
|
||
extra = helpers.write_file(self.path("b.txt"), b"b")
|
||
with helpers.quiet():
|
||
vfs.add_file(second, "a.txt")
|
||
vfs.add_file(extra, "b.txt")
|
||
vfs.recount()
|
||
return vfs
|
||
|
||
def test_same_path_has_different_content_per_snapshot(self):
|
||
vfs = self._pack_v1_then_v2()
|
||
self.assertEqual(b"second", vfs.read_file("a.txt"))
|
||
|
||
vfs.use_snapshot(1)
|
||
self.assertEqual(b"first", vfs.read_file("a.txt"))
|
||
|
||
def test_new_files_are_invisible_in_the_older_snapshot(self):
|
||
vfs = self._pack_v1_then_v2()
|
||
self.assertTrue(vfs.exists("b.txt"))
|
||
vfs.use_snapshot(1)
|
||
self.assertFalse(vfs.exists("b.txt"))
|
||
|
||
def test_copy_entries_forks_from_the_parent(self):
|
||
vfs = self._pack_v1_then_v2()
|
||
vfs.use_snapshot(1)
|
||
self.assertEqual({b"first"}, {vfs.read_file("a.txt")})
|
||
self.assertEqual({"a.txt"}, helpers.all_paths(vfs))
|
||
|
||
def test_blobs_are_shared_between_snapshots(self):
|
||
vfs = self._pack_v1_then_v2()
|
||
# first / second / b 三份内容
|
||
self.assertEqual(3, helpers.blob_count(vfs))
|
||
|
||
def test_each_snapshot_has_its_own_digest(self):
|
||
vfs = self._pack_v1_then_v2()
|
||
digests = {s["id"]: s["digest"] for s in vfs.list_snapshots()}
|
||
self.assertEqual(2, len(digests))
|
||
self.assertTrue(all(digests.values()), "每个快照都应当有摘要")
|
||
self.assertNotEqual(digests[1], digests[2], "内容不同，摘要必须不同")
|
||
|
||
def test_counts_are_per_snapshot(self):
|
||
vfs = self._pack_v1_then_v2()
|
||
self.assertEqual(2, vfs.get_stats()["total_files"])
|
||
vfs.use_snapshot(1)
|
||
self.assertEqual(1, vfs.get_stats()["total_files"])
|
||
|
||
def test_export_respects_the_active_snapshot(self):
|
||
vfs = self._pack_v1_then_v2()
|
||
out_v2 = self.path("out_v2")
|
||
with helpers.quiet():
|
||
vfs.export_filesystem(out_v2)
|
||
|
||
vfs.use_snapshot(1)
|
||
out_v1 = self.path("out_v1")
|
||
with helpers.quiet():
|
||
vfs.export_filesystem(out_v1)
|
||
|
||
self.assertEqual(
|
||
{"a.txt": b"first"}, {k: v for k, v in helpers.read_tree(out_v1).items()}
|
||
)
|
||
self.assertEqual(
|
||
{"a.txt": b"second", "b.txt": b"b"},
|
||
{k: v for k, v in helpers.read_tree(out_v2).items()},
|
||
)
|
||
|
||
def test_list_directory_is_snapshot_scoped(self):
|
||
vfs = self._pack_v1_then_v2()
|
||
self.assertEqual(
|
||
["a.txt", "b.txt"], sorted(r["path"] for r in vfs.list_directory("/"))
|
||
)
|
||
vfs.use_snapshot(1)
|
||
self.assertEqual(
|
||
["a.txt"], sorted(r["path"] for r in vfs.list_directory("/"))
|
||
)
|
||
|
||
def test_remove_in_one_snapshot_does_not_affect_the_other(self):
|
||
vfs = self._pack_v1_then_v2()
|
||
vfs.remove("a.txt")
|
||
self.assertFalse(vfs.exists("a.txt"))
|
||
vfs.use_snapshot(1)
|
||
self.assertEqual(b"first", vfs.read_file("a.txt"))
|
||
|
||
def test_mkdir_in_one_snapshot_does_not_leak(self):
|
||
vfs = self._pack_v1_then_v2()
|
||
vfs.mk_dir("only-here")
|
||
self.assertTrue(vfs.is_dir("only-here"))
|
||
vfs.use_snapshot(1)
|
||
self.assertFalse(vfs.exists("only-here"))
|
||
|
||
def test_snapshot_id_can_be_chosen_at_construction(self):
|
||
vfs = self._pack_v1_then_v2()
|
||
db = vfs.db_path
|
||
first = SQLiteVFS(db, snapshot_id=1)
|
||
self.addCleanup(first.close)
|
||
self.assertEqual(b"first", first.read_file("a.txt"))
|
||
self.assertEqual(1, first.get_stats()["snapshot_id"])
|
||
|
||
|
||
class UnpackerSnapshotTest(helpers.VFSTestCase):
|
||
def _two_versions(self, db_name="u.svfs"):
|
||
"""v1 有 one.txt，v2（派生自 v1）删掉 one.txt 并加上 two.txt。"""
|
||
db = self.path(db_name)
|
||
vfs = SQLiteVFS(db)
|
||
vfs.add_vfs_metadata("u")
|
||
one = helpers.write_file(self.path("one.txt"), b"one")
|
||
with helpers.quiet():
|
||
vfs.add_file(one, "one.txt")
|
||
vfs.recount()
|
||
vfs.create_snapshot("v2", copy_entries=True)
|
||
vfs.remove("one.txt")
|
||
two = helpers.write_file(self.path("two.txt"), b"two")
|
||
with helpers.quiet():
|
||
vfs.add_file(two, "two.txt")
|
||
vfs.recount()
|
||
return vfs
|
||
|
||
def test_unpacker_list_root_is_snapshot_scoped(self):
|
||
from sqlite_vfs.folder_unpacker import FolderUnpacker
|
||
|
||
vfs = self._two_versions()
|
||
self.addCleanup(vfs.close)
|
||
|
||
self.assertEqual(
|
||
["two.txt"], [r["path"] for r in FolderUnpacker(vfs).list_root()]
|
||
)
|
||
vfs.use_snapshot(1)
|
||
self.assertEqual(
|
||
["one.txt"], [r["path"] for r in FolderUnpacker(vfs).list_root()]
|
||
)
|
||
|
||
def test_unpacker_export_follows_the_snapshot(self):
|
||
from sqlite_vfs.folder_unpacker import FolderUnpacker
|
||
|
||
vfs = self._two_versions("u2.svfs")
|
||
self.addCleanup(vfs.close)
|
||
|
||
out = self.path("out")
|
||
with helpers.quiet():
|
||
FolderUnpacker(vfs).unpack_filesystem(out)
|
||
self.assertEqual({"two.txt": b"two"}, helpers.read_tree(out))
|
||
|
||
vfs.use_snapshot(1)
|
||
out1 = self.path("out1")
|
||
with helpers.quiet():
|
||
FolderUnpacker(vfs).unpack_filesystem(out1)
|
||
self.assertEqual({"one.txt": b"one"}, helpers.read_tree(out1))
|
||
|
||
def test_unpacker_check_database_counts_only_the_active_snapshot(self):
|
||
from sqlite_vfs.folder_unpacker import FolderUnpacker
|
||
|
||
vfs = self._two_versions("u3.svfs")
|
||
self.addCleanup(vfs.close)
|
||
|
||
# v2 只有一个条目（two.txt），v1 也只有一个（one.txt）；
|
||
# 两个快照加起来是 2 条，所以"报 1"就证明统计是分快照的
|
||
ok, message = FolderUnpacker(vfs).check_database()
|
||
self.assertTrue(ok, message)
|
||
self.assertIn("快照 2", message)
|
||
self.assertIn("1 个文件/目录", message)
|
||
|
||
vfs.use_snapshot(1)
|
||
ok, message = FolderUnpacker(vfs).check_database()
|
||
self.assertTrue(ok, message)
|
||
self.assertIn("快照 1", message)
|
||
self.assertIn("1 个文件/目录", message)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|