chakcy
ecdfacc620
起始状态：主链路 100% 不可用，而且失败被静默吞掉——打包后库里 0 个文件、 界面仍打印"文件数量: N"；解包导出 0 个文件并在磁盘上生成几百层 sub/sub/sub 垃圾目录。 缺陷修复（4 个 P0 + 12 个 P1，逐条见 BUG_REPORT.md）： - 建表列名 compresse_algo 与 INSERT/read_file 用的 compress_algo 不一致 - list_directory 按 path 而非 parent_path 过滤，导致导出无限自递归 - CLI --ignore-timestamps 把 export_filesystem 赋成引用自身的 lambda - add_file 缺少 commit（直接调用 API 时静默丢数据） - LIKE 前缀误伤兄弟目录（删 doc 连带删 docs）；mv_dir 二次改名 - 失败路径一律返回 exit code 0；--exclude 是子串匹配而非 glob - FolderPacker.add_file 反向调用 export_file_safe，会用库内内容覆盖源文件 存储层（schema v3，格式变更，提供 vfsmigrate 迁移）： - 内容移入 blobs(hash PRIMARY KEY)：相同内容只存一份，跨路径/跨快照共享 - files 的唯一约束改为 UNIQUE(snapshot_id, path)：一个库存多个版本 - 完整性：每文件 sha256 + 快照级摘要 + HMAC-SHA256 签名，只读连接即可验证 - 增量打包：按 mtime+size 或 sha256 只写变化项，未变文件不重写不重压 - get_file_info 不再返回 content（元数据查询不该搬 BLOB） 新增能力： - 可选 FastAPI 静态文件插件（extras，核心仍零第三方依赖） - 内容寻址发布到对象存储 + 逐对象校验的取回客户端 - 压缩默认改回 zlib：实测比自研哈夫曼体积小 1/3~1/170、快数百倍 其它： - P2-4：时间戳显式转字符串，清掉 1240 条 DeprecationWarning（盘上格式不变） - read_file 负数 offset/size 改为报错（原先会走切片语义从末尾取） - requires-python 从 >=3.7 改为按实际 API 推导的 >=3.9 测试：0 → 343 个用例，纯标准库 unittest，有/无 FastAPI 两种环境都跑通。 证据见 docs/evidence-*.txt。
328 lines
12 KiB
Python
328 lines
12 KiB
Python
"""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()
|