"""内容寻址与去重。 核心承诺：**相同内容只存一份**。这里用 blob 行数与 stored_size 直接验证， 而不是只看读回来的字节——字节对不代表没多存。 """ import os import unittest import helpers from sqlite_vfs.core import SQLiteVFS class DedupTest(helpers.VFSTestCase): def _add(self, vfs, name, data): src = helpers.write_file(self.path(f"src_{name.replace('/', '_')}"), data) with helpers.quiet(): self.assertTrue(vfs.add_file(src, name), f"写入失败: {name}") def _vfs(self, **kwargs): vfs = SQLiteVFS(":memory:", **kwargs) self.addCleanup(vfs.close) return vfs def test_identical_files_share_one_blob(self): vfs = self._vfs() self._add(vfs, "a.txt", b"same content") self._add(vfs, "b.txt", b"same content") self._add(vfs, "c.txt", b"same content") self.assertEqual(3, helpers.count_files(vfs)) self.assertEqual(1, helpers.blob_count(vfs), "相同内容应当只占一行 blob") def test_identical_content_in_different_directories_shares_blob(self): vfs = self._vfs() self._add(vfs, "doc/logo.png", b"PNGDATA" * 100) self._add(vfs, "assets/logo.png", b"PNGDATA" * 100) self.assertEqual(1, helpers.blob_count(vfs)) def test_distinct_content_gets_distinct_blobs(self): vfs = self._vfs() self._add(vfs, "a.txt", b"aaa") self._add(vfs, "b.txt", b"bbb") self.assertEqual(2, helpers.blob_count(vfs)) def test_empty_files_share_one_blob(self): vfs = self._vfs() self._add(vfs, "e1.txt", b"") self._add(vfs, "e2.txt", b"") self.assertEqual(1, helpers.blob_count(vfs)) self.assertEqual(b"", vfs.read_file("e1.txt")) def test_replacing_with_identical_content_does_not_add_blob(self): vfs = self._vfs() self._add(vfs, "a.txt", b"payload") vfs.remove("a.txt") self._add(vfs, "a.txt", b"payload") self.assertEqual(1, helpers.blob_count(vfs), "重新写入相同内容不该新增 blob") def test_rewrite_different_content_keeps_only_new_blob(self): vfs = self._vfs() self._add(vfs, "a.txt", b"first") vfs.remove("a.txt") self._add(vfs, "a.txt", b"second") self.assertEqual(1, helpers.blob_count(vfs), "旧内容已经没人引用，应当被回收") def test_blob_records_original_and_stored_size(self): vfs = self._vfs(compress=True) data = b"compressible text " * 200 self._add(vfs, "big.txt", data) blob = vfs.get_blob_info("big.txt") self.assertEqual(len(data), blob["size"], "size 记的是原始大小") self.assertLess(blob["stored_size"], len(data), "stored_size 记的是实际占用") self.assertEqual("zlib", blob["compress_algo"]) def test_dedup_skips_recompression(self): """第二次写入相同内容时不该再压一遍——这是增量打包能变快的原因。""" vfs = self._vfs(compress=True) data = b"same bytes " * 500 src = helpers.write_file(self.path("in.bin"), data) with helpers.quiet(): vfs.add_file(src, "one.bin") first = vfs.get_blob_info("one.bin") vfs.add_file(src, "two.bin") second = vfs.get_blob_info("two.bin") self.assertEqual(first["hash"], second["hash"]) self.assertEqual(1, helpers.blob_count(vfs)) # 两个路径指向同一份压缩数据，占用没有翻倍 self.assertEqual(first["stored_size"], second["stored_size"]) def test_repacking_identical_tree_does_not_grow_storage(self): src = helpers.make_tree( self.path("tree"), {"a.txt": b"hello", "b.txt": b"hello", "sub/c.txt": b"world"}, ) db = self.path("p.svfs") helpers.pack(src, db, compress=True) first_size = os.path.getsize(db) vfs = SQLiteVFS(db) self.assertEqual(2, helpers.blob_count(vfs), "hello 与 world 各一份") vfs.close() helpers.pack(src, db, compress=True) vfs = SQLiteVFS(db) self.addCleanup(vfs.close) self.assertEqual(2, helpers.blob_count(vfs)) self.assertEqual(3, helpers.count_files(vfs)) self.assertLess( os.path.getsize(db), first_size * 1.5, "重打同一棵树不该让库明显变大", ) class BlobGarbageCollectionTest(helpers.VFSTestCase): def _vfs(self, **kwargs): vfs = SQLiteVFS(":memory:", **kwargs) self.addCleanup(vfs.close) helpers.seed_rows( vfs, [ {"path": "keep.txt", "content": b"keep"}, {"path": "shared.txt", "content": b"shared"}, {"path": "also_shared.txt", "content": b"shared"}, {"path": "temp", "is_directory": True}, {"path": "temp/gone.txt", "content": b"gone"}, ], ) return vfs def test_remove_file_collects_unreferenced_blob(self): vfs = self._vfs() self.assertEqual(3, helpers.blob_count(vfs)) vfs.remove("temp/gone.txt") self.assertEqual(2, helpers.blob_count(vfs), "'gone' 已无人引用，应被回收") def test_remove_directory_collects_blobs(self): vfs = self._vfs() vfs.remove("temp") self.assertEqual(2, helpers.blob_count(vfs)) def test_shared_blob_survives_partial_removal(self): vfs = self._vfs() vfs.remove("shared.txt") self.assertEqual( 3, helpers.blob_count(vfs), "'shared' 还有另一个路径在用，不能回收" ) self.assertEqual(b"shared", vfs.read_file("also_shared.txt")) def test_gc_keeps_blobs_referenced_by_another_snapshot(self): """多快照共用内容时，删掉一个版本不能影响另一个版本。""" vfs = self._vfs() second = vfs.create_snapshot("v2", copy_entries=True) # 在 v2 里删掉一部分，v1 仍然引用着这些内容 vfs.remove("temp/gone.txt") vfs.remove("shared.txt") self.assertEqual( 3, helpers.blob_count(vfs), "v1 还在引用，blob 不该被回收" ) vfs.use_snapshot(1) self.assertEqual(b"gone", vfs.read_file("temp/gone.txt")) self.assertEqual(b"shared", vfs.read_file("shared.txt")) self.assertEqual(second, 2) def test_deleting_snapshot_allows_gc(self): """只被某个快照引用的内容，在删掉那个快照后应当被回收。""" vfs = self._vfs() vfs.create_snapshot("v2") # 空快照，不与 v1 共享内容 src = helpers.write_file(self.path("v2only.bin"), b"v2-only-content") with helpers.quiet(): vfs.add_file(src, "only-in-v2.txt") self.assertEqual(4, helpers.blob_count(vfs)) vfs.delete_snapshot(2) self.assertEqual( 3, helpers.blob_count(vfs), "只被 v2 引用的内容应当被回收" ) self.assertFalse(vfs.exists("only-in-v2.txt")) # v1 完全不受影响 self.assertEqual(b"shared", vfs.read_file("shared.txt")) self.assertEqual(b"gone", vfs.read_file("temp/gone.txt")) def test_gc_returns_deleted_count(self): vfs = self._vfs() # 直接造一个没人引用的 blob vfs.conn.execute( "INSERT INTO blobs (hash, size, stored_size, content) " "VALUES ('orphan', 1, 1, X'00')" ) vfs.conn.commit() self.assertEqual(1, vfs.gc_blobs(), "应当回收到孤立的 blob") self.assertEqual(0, vfs.gc_blobs(), "再跑一次应当没有可回收的") def test_clear_snapshot_keeps_other_snapshot_content(self): vfs = self._vfs() vfs.create_snapshot("v2", copy_entries=True) vfs.clear_snapshot() self.assertEqual(0, helpers.count_files(vfs), "当前快照已清空") self.assertEqual(3, helpers.blob_count(vfs), "v1 还在引用") vfs.use_snapshot(1) self.assertEqual(5, len(helpers.all_paths(vfs))) self.assertEqual(b"keep", vfs.read_file("keep.txt")) if __name__ == "__main__": unittest.main()