svfs/tests/test_dedup.py

"""内容寻址与去重。

核心承诺：**相同内容只存一份**。这里用 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()