svfs/tests/test_roundtrip.py

"""端到端回归测试：打包 → 解包，必须逐字节无损。

这是本项目最核心的验收测试：只要它绿了，说明"把文件夹装进一个 .svfs 再拿出来"
这条主链路是通的。
"""

import os
import random
import unittest

import helpers

from sqlite_vfs.core import SQLiteVFS


def build_spec():
    """构造一个覆盖面较广的目录树。

    刻意包含：嵌套目录、空目录、空文件、压缩阈值两侧的样本、
    可压缩与不可压缩内容、中文路径、带空格的路径、
    以及含 % 和 _ 的文件名（这两个字符在 SQL LIKE 里是通配符）。
    """
    rnd = random.Random(20240913)
    return {
        "top.txt": b"top level\n",
        "empty.txt": b"",
        "one_byte.bin": b"\x00",
        "below_threshold.bin": rnd.randbytes(999),
        "compressible.txt": b"hello world, this is a highly compressible payload. " * 200,
        "random.bin": rnd.randbytes(4096),
        "empty_dir": None,
        "sub/deep/deeper/nested.txt": b"nested\n",
        "sub/space name.txt": b"spaces\n",
        "sub/中文名.txt": "中文内容，用来验证非 ASCII 路径与内容。\n".encode("utf-8"),
        "sub/percent%name_.txt": b"wildcards\n",
        "doc/a.txt": b"a\n",
        "docs/b.txt": b"b\n",
    }


class RoundTripTest(helpers.VFSTestCase):
    def _pack(self, compress):
        src = helpers.make_tree(self.path("src"), build_spec())
        db = helpers.pack(src, self.path("out.svfs"), compress=compress, name="roundtrip")
        return src, db

    def _assert_lossless(self, src, db, ignore_timestamps=False):
        out = self.path("out")
        with helpers.quiet():
            helpers.unpack(db, out, ignore_timestamps=ignore_timestamps)
        # 以磁盘上的源目录为基准：build_spec 里没显式写出的中间目录
        # （sub、sub/deep 等）也必须在导出结果里出现
        helpers.assert_tree_equal(self, helpers.read_tree(src), helpers.read_tree(out))

    def test_roundtrip_without_compression(self):
        src, db = self._pack(compress=False)
        self._assert_lossless(src, db)

    def test_roundtrip_with_compression(self):
        src, db = self._pack(compress=True)
        self._assert_lossless(src, db)

    def test_roundtrip_with_ignore_timestamps(self):
        src, db = self._pack(compress=True)
        self._assert_lossless(src, db, ignore_timestamps=True)

    def test_every_source_file_is_stored(self):
        """P0-1：打包后库里必须有和源目录一一对应的文件记录。"""
        src, db = self._pack(compress=True)
        vfs = SQLiteVFS(db)
        self.addCleanup(vfs.close)

        expected = {rel for rel, content in build_spec().items() if content is not None}
        stored = set(helpers.stored_files(vfs))
        self.assertEqual(
            "无差异",
            helpers.diff_trees({p: None for p in expected}, {p: None for p in stored}),
        )

    def test_metadata_counts_match_database(self):
        """P1-5：上报的统计数字必须来自真实入库的数据，而不是遍历计数。"""
        src, db = self._pack(compress=True)
        vfs = SQLiteVFS(db)
        self.addCleanup(vfs.close)

        stats = vfs.get_stats()
        stored_count = helpers.count_files(vfs)
        source_count = len(
            [rel for rel, content in build_spec().items() if content is not None]
        )

        self.assertEqual(source_count, stored_count, "入库文件数 ≠ 源文件数")
        self.assertEqual(
            source_count,
            stats["total_files"],
            "元数据里的文件数与实际入库数不一致",
        )
        # total_size 记录的是原始文件大小，压缩后 content 里存的是压缩数据，
        # 所以这里要和源文件大小比，而不是库里 blob 的长度
        source_size = sum(
            len(content) for content in build_spec().values() if content is not None
        )
        self.assertEqual(source_size, stats["total_size"])

    def test_compression_reduces_database_size(self):
        """用一个足够大的重复文本，让压缩效果超过 SQLite 页粒度（4KB）的噪声。

        build_spec 里的小文件即使压得动，省下的字节也可能被页对齐吃掉，
        所以这里单独造一份 92KB 的样本。
        """
        payload = b"hello world, this is a highly compressible payload. " * 2000
        src = helpers.make_tree(self.path("big_src"), {"big.txt": payload})
        plain_db = helpers.pack(src, self.path("plain.svfs"), compress=False)
        compressed_db = helpers.pack(src, self.path("compressed.svfs"), compress=True)

        plain_size = os.path.getsize(plain_db)
        compressed_size = os.path.getsize(compressed_db)
        self.assertLess(
            compressed_size,
            plain_size,
            f"压缩后数据库应当更小: plain={plain_size}, compressed={compressed_size}",
        )

    def test_stored_blob_of_compressible_file_is_smaller(self):
        """压缩不只是"标记了一下"：库里真正存的字节必须变少。"""
        src, db = self._pack(compress=True)
        vfs = SQLiteVFS(db)
        self.addCleanup(vfs.close)

        blob = vfs.get_blob_info("compressible.txt")
        self.assertEqual("zlib", blob["compress_algo"])
        self.assertIsNone(blob["freq_blob"], "zlib 不需要频率表")
        with open(os.path.join(src, "compressible.txt"), "rb") as f:
            original = f.read()
        self.assertEqual(len(original), blob["size"], "blob 记录的应当是原始大小")
        # 高度重复的文本，zlib 至少能压掉一半（huffman 只能压到 ~56%）
        self.assertLess(blob["stored_size"], len(original) // 2)
        self.assertEqual(original, vfs.read_file("compressible.txt"))

    def test_packed_database_contains_no_duplicate_paths(self):
        """唯一性现在是"每个快照内"的：UNIQUE(snapshot_id, path)。"""
        src, db = self._pack(compress=False)
        vfs = SQLiteVFS(db)
        self.addCleanup(vfs.close)
        snapshot_id = vfs._active_snapshot()
        total = vfs.conn.execute(
            "SELECT COUNT(*) AS c FROM files WHERE snapshot_id = ?", (snapshot_id,)
        ).fetchone()["c"]
        self.assertEqual(total, len(helpers.all_paths(vfs)))


if __name__ == "__main__":
    unittest.main()