"""v1 / v2 → v3 的格式迁移。 迁移是"只读旧库、重建新库、原子替换"：中间态写在旁边的临时文件里， 所以任何一步失败都不会让原库变成半成品；默认还会先留一份 .bak。 """ import os import sqlite3 import unittest import zlib import helpers from sqlite_vfs import migration from sqlite_vfs.core import SQLiteVFS class DetectVersionTest(helpers.VFSTestCase): def test_detects_each_format(self): v1 = helpers.make_old_schema_db(self.path("v1.svfs")) v2 = helpers.make_v2_schema_db( self.path("v2.svfs"), rows=[{"path": "a.txt", "content": b"a"}] ) v3 = self.path("v3.svfs") SQLiteVFS(v3).close() self.assertEqual(1, migration.detect_schema_version(v1)) self.assertEqual(2, migration.detect_schema_version(v2)) self.assertEqual(3, migration.detect_schema_version(v3)) def test_detects_non_vfs_database(self): path = self.path("plain.sqlite") conn = sqlite3.connect(path) conn.execute("CREATE TABLE unrelated (id INTEGER)") conn.commit() conn.close() self.assertIsNone(migration.detect_schema_version(path)) def test_missing_file_raises(self): with self.assertRaises(FileNotFoundError): migration.detect_schema_version(self.path("nope.svfs")) class MigrateFromV2Test(helpers.VFSTestCase): PAYLOAD = b"repeat me please. " * 100 def _make_v2(self, name="v2.svfs"): rows = [ {"path": "plain.txt", "content": b"plain v2 content"}, {"path": "dup1.txt", "content": self.PAYLOAD}, {"path": "dup2.txt", "content": self.PAYLOAD}, { "path": "compressed.txt", "content": zlib.compress(self.PAYLOAD), "compress_algo": "zlib", "file_size": len(self.PAYLOAD), }, {"path": "empty.txt", "content": b""}, {"path": "sub", "is_directory": True}, {"path": "sub/nested.txt", "content": b"nested"}, ] return helpers.make_v2_schema_db(self.path(name), rows=rows) def test_content_survives_byte_for_byte(self): db = self._make_v2() result = migration.migrate(db) self.assertTrue(result["migrated"]) self.assertEqual(2, result["from_version"]) self.assertEqual(3, migration.detect_schema_version(db)) vfs = SQLiteVFS(db) self.addCleanup(vfs.close) self.assertEqual(b"plain v2 content", vfs.read_file("plain.txt")) self.assertEqual(self.PAYLOAD, vfs.read_file("dup1.txt")) self.assertEqual(self.PAYLOAD, vfs.read_file("dup2.txt")) self.assertEqual( self.PAYLOAD, vfs.read_file("compressed.txt"), "旧库里 zlib 压缩过的内容必须解压后原样返回", ) self.assertEqual(b"", vfs.read_file("empty.txt")) self.assertEqual(b"nested", vfs.read_file("sub/nested.txt")) def test_structure_is_preserved(self): db = self._make_v2() migration.migrate(db) vfs = SQLiteVFS(db) self.addCleanup(vfs.close) self.assertTrue(vfs.is_dir("sub")) self.assertEqual( ["compressed.txt", "dup1.txt", "dup2.txt", "empty.txt", "plain.txt", "sub"], sorted(r["path"] for r in vfs.list_directory("/")), ) self.assertEqual(6, helpers.count_files(vfs)) def test_identical_content_is_deduplicated_during_migration(self): db = self._make_v2() result = migration.migrate(db) # dup1 / dup2 / compressed.txt 三份内容相同 → 只留一份 blob self.assertEqual(4, result["blobs"], "plain/empty/nested/payload 共四份") self.assertGreater(result["dedup_saved"], 0, "应当报告去重省下的字节数") vfs = SQLiteVFS(db) self.addCleanup(vfs.close) self.assertEqual(4, helpers.blob_count(vfs)) def test_backup_is_created_and_original_untouched(self): db = self._make_v2() before = helpers.md5_of(db) result = migration.migrate(db) self.assertTrue(result["backup"]) self.assertTrue(os.path.exists(result["backup"])) self.assertEqual( before, helpers.md5_of(result["backup"]), "备份必须与迁移前的库完全一致" ) self.assertEqual(2, migration.detect_schema_version(result["backup"])) self.assertNotEqual(before, helpers.md5_of(db)) def test_backup_can_be_disabled(self): db = self._make_v2() result = migration.migrate(db, backup=False) self.assertIsNone(result["backup"]) self.assertFalse(os.path.exists(db + ".bak")) def test_second_migration_is_a_no_op(self): db = self._make_v2() migration.migrate(db) digest_before = helpers.md5_of(db) again = migration.migrate(db) self.assertFalse(again["migrated"]) self.assertIn("已经是", again["reason"]) self.assertEqual(digest_before, helpers.md5_of(db), "重复迁移不该改动库") def test_migrated_pack_verifies_and_signs(self): db = self._make_v2() migration.migrate(db) vfs = SQLiteVFS(db) self.addCleanup(vfs.close) report = vfs.verify(check_content=True) self.assertTrue(report["ok"], report["errors"]) self.assertEqual(6, report["files_checked"]) self.assertEqual(6, report["content_checked"]) self.assertTrue(vfs.sign("k") and vfs.verify_signature("k")["ok"]) def test_metadata_is_carried_over(self): db = self.path("meta.svfs") conn = sqlite3.connect( helpers.make_v2_schema_db( db, rows=[{"path": "a.txt", "content": b"aaa"}] ) ) conn.execute( "UPDATE files SET modified_time = '2020-01-02 03:04:05', permissions = 493," " created_time = '2020-01-01 00:00:00' WHERE path = 'a.txt'" ) conn.commit() conn.close() migration.migrate(db) vfs = SQLiteVFS(db) self.addCleanup(vfs.close) info = vfs.get_file_info("a.txt") self.assertEqual(0o755, info["permissions"]) self.assertEqual(2020, info["modified_time"].year) self.assertEqual(3, info["modified_time"].hour) def test_snapshot_name_defaults_to_the_pack_name(self): db = self._make_v2() migration.migrate(db) vfs = SQLiteVFS(db) self.addCleanup(vfs.close) self.assertEqual("v2pack", vfs.list_snapshots()[0]["name"]) def test_custom_snapshot_name(self): db = self._make_v2() migration.migrate(db, snapshot_name="imported") vfs = SQLiteVFS(db) self.addCleanup(vfs.close) self.assertEqual("imported", vfs.list_snapshots()[0]["name"]) def test_migrate_to_a_new_file_leaves_the_original_alone(self): """原库在只读介质上时不能原地替换，所以支持输出到新路径。""" db = self._make_v2() before = helpers.md5_of(db) target = self.path("migrated.svfs") result = migration.migrate(db, output=target) self.assertTrue(result["migrated"]) self.assertFalse(result["in_place"]) self.assertEqual(os.path.abspath(target), result["output"]) self.assertIsNone(result["backup"], "有独立输出时不必再备份") self.assertEqual(before, helpers.md5_of(db), "原库必须一个字节都没变") self.assertEqual(2, migration.detect_schema_version(db)) self.assertEqual(3, migration.detect_schema_version(target)) vfs = SQLiteVFS(target) self.addCleanup(vfs.close) self.assertEqual(self.PAYLOAD, vfs.read_file("dup1.txt")) self.assertTrue(vfs.verify(check_content=True)["ok"]) def test_migration_of_a_readonly_file_into_a_new_file(self): db = self._make_v2() os.chmod(db, 0o444) try: result = migration.migrate(db, output=self.path("out.svfs")) self.assertTrue(result["migrated"]) finally: os.chmod(db, 0o644) def test_refuses_to_overwrite_an_existing_output(self): db = self._make_v2() target = helpers.write_file(self.path("taken.svfs"), b"occupied") with self.assertRaises(FileExistsError): migration.migrate(db, output=target) self.assertEqual(b"occupied", helpers.read_bytes(target)) def test_backup_can_be_forced_with_output(self): db = self._make_v2() result = migration.migrate( db, output=self.path("m2.svfs"), backup=True ) self.assertTrue(os.path.exists(result["backup"])) class MigrateFromV1Test(helpers.VFSTestCase): def test_v1_bool_flag_is_mapped_to_zlib(self): payload = b"legacy payload " * 50 db = helpers.make_old_schema_db( self.path("v1.svfs"), extra_rows=[ { "path": "packed.txt", "content": zlib.compress(payload), "compressed": True, "file_size": len(payload), }, {"path": "stored.txt", "content": b"stored raw"}, ], ) result = migration.migrate(db) self.assertEqual(1, result["from_version"]) vfs = SQLiteVFS(db) self.addCleanup(vfs.close) self.assertEqual(payload, vfs.read_file("packed.txt")) self.assertEqual(b"stored raw", vfs.read_file("stored.txt")) self.assertEqual(b"hello", vfs.read_file("legacy.txt")) self.assertEqual("zlib", vfs.get_blob_info("packed.txt")["compress_algo"]) self.assertEqual("none", vfs.get_blob_info("stored.txt")["compress_algo"]) self.assertTrue(vfs.verify(check_content=True)["ok"]) class MigrateCliTest(helpers.VFSTestCase): """迁移后的库要能被 CLI 正常解包。""" def test_migrated_database_unpacks_via_cli(self): payload = b"migrated content " * 20 src = helpers.make_tree(self.path("src"), {"a.txt": payload, "sub/b.txt": b"b"}) db = self.path("v2.svfs") helpers.pack(src, db, compress=True) # 把库降级成 v2 形态：内容搬回 files 表，去掉 blobs/snapshots conn = sqlite3.connect(db) conn.executescript( """ ALTER TABLE files ADD COLUMN content BLOB; ALTER TABLE files ADD COLUMN freq_blob BLOB; ALTER TABLE files ADD COLUMN compress_algo TEXT DEFAULT 'none'; UPDATE files SET content = (SELECT b.content FROM blobs b WHERE b.hash = files.content_hash), freq_blob = (SELECT b.freq_blob FROM blobs b WHERE b.hash = files.content_hash), compress_algo = (SELECT b.compress_algo FROM blobs b WHERE b.hash = files.content_hash); DROP TABLE blobs; DROP TABLE snapshots; DROP TABLE signatures; """ ) conn.commit() conn.close() self.assertEqual(2, migration.detect_schema_version(db)) out = self.path("out") failed = helpers.run_cli("sqlite_vfs.cli.unpacker", db, out) self.assertNotEqual(0, failed.returncode, "v2 库应当被拒绝并提示迁移") self.assertTrue(migration.migrate(db)["migrated"]) result = helpers.run_cli("sqlite_vfs.cli.unpacker", db, out) self.assertEqual(0, result.returncode, helpers.cli_output(result)) helpers.assert_tree_equal(self, helpers.read_tree(src), helpers.read_tree(out)) class MigrateInvalidInputTest(helpers.VFSTestCase): def test_non_vfs_database_is_rejected(self): path = self.path("plain.sqlite") conn = sqlite3.connect(path) conn.execute("CREATE TABLE unrelated (id INTEGER)") conn.commit() conn.close() with self.assertRaises(ValueError): migration.migrate(path) def test_missing_file_raises(self): with self.assertRaises(FileNotFoundError): migration.migrate(self.path("nope.svfs")) def test_original_survives_a_failure(self): """迁移失败时原库必须原样保留（中间态写在临时文件里）。""" path = self.path("plain.sqlite") conn = sqlite3.connect(path) conn.execute("CREATE TABLE unrelated (id INTEGER)") conn.commit() conn.close() before = helpers.md5_of(path) with self.assertRaises(ValueError): migration.migrate(path) self.assertEqual(before, helpers.md5_of(path)) self.assertFalse(os.path.exists(path + migration.TEMP_SUFFIX)) if __name__ == "__main__": unittest.main()