svfs/tests/test_snapshots.py

"""快照与时间旅行。

一个库里可以存同一棵树的历史版本：``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()