fix+feat: 修复打包/解包主链路，并把存储层重建为内容寻址 + 快照
起始状态：主链路 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。
This commit is contained in:
@@ -0,0 +1,993 @@
|
||||
"""SQLiteVFS 核心行为的契约测试。
|
||||
|
||||
用例按缺陷编号分组，命名对应 BUG_REPORT.md 里的条目，便于"哪条修好了"一目了然。
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import sqlite3
|
||||
import unittest
|
||||
import warnings
|
||||
|
||||
import helpers
|
||||
|
||||
from sqlite_vfs.core import SQLiteVFS, VFSIntegrityError
|
||||
|
||||
|
||||
class SchemaContractTest(helpers.VFSTestCase):
|
||||
"""P0-1：建表列名必须与 INSERT / read_file 使用的列名一致。"""
|
||||
|
||||
def test_expected_columns_exist(self):
|
||||
"""v3：目录项与内容分家，files 只留元数据 + content_hash。"""
|
||||
vfs = SQLiteVFS(":memory:")
|
||||
self.addCleanup(vfs.close)
|
||||
|
||||
file_columns = helpers.column_names(vfs.conn, "files")
|
||||
for name in (
|
||||
"snapshot_id",
|
||||
"path",
|
||||
"name",
|
||||
"parent_path",
|
||||
"is_directory",
|
||||
"file_size",
|
||||
"modified_time",
|
||||
"permissions",
|
||||
"content_hash",
|
||||
):
|
||||
self.assertIn(name, file_columns, f"files 表缺少列 {name}")
|
||||
for name in ("content", "freq_blob", "compress_algo"):
|
||||
self.assertNotIn(
|
||||
name, file_columns, f"{name} 应当已经搬到 blobs 表"
|
||||
)
|
||||
|
||||
blob_columns = helpers.column_names(vfs.conn, "blobs")
|
||||
for name in ("hash", "size", "stored_size", "content", "freq_blob", "compress_algo"):
|
||||
self.assertIn(name, blob_columns, f"blobs 表缺少列 {name}")
|
||||
|
||||
snapshot_columns = helpers.column_names(vfs.conn, "snapshots")
|
||||
for name in ("id", "name", "parent_id", "total_files", "total_size", "digest"):
|
||||
self.assertIn(name, snapshot_columns, f"snapshots 表缺少列 {name}")
|
||||
|
||||
def test_add_file_actually_stores_a_row(self):
|
||||
"""add_file 之后库里必须真的有这条记录。
|
||||
|
||||
缺陷版本里 INSERT 的列名（compress_algo）与建表列名（compresse_algo）
|
||||
不一致，异常被 except 吞掉，于是打包"成功"但库里一个文件都没有。
|
||||
"""
|
||||
src = helpers.write_file(self.path("input.bin"), b"payload-bytes")
|
||||
vfs = SQLiteVFS(":memory:")
|
||||
self.addCleanup(vfs.close)
|
||||
|
||||
with helpers.quiet():
|
||||
vfs.add_file(src, "input.bin")
|
||||
|
||||
self.assertTrue(vfs.exists("input.bin"), "add_file 没有把文件写进数据库")
|
||||
self.assertEqual(b"payload-bytes", vfs.read_file("input.bin"))
|
||||
|
||||
|
||||
class PersistenceTest(helpers.VFSTestCase):
|
||||
"""P0-4：add_file 必须自己提交事务，不能依赖后续调用代为 commit。"""
|
||||
|
||||
def test_add_file_survives_close_without_further_commit(self):
|
||||
db = self.path("commit.svfs")
|
||||
src = helpers.write_file(self.path("x.bin"), b"payload")
|
||||
|
||||
vfs = SQLiteVFS(db)
|
||||
with helpers.quiet():
|
||||
vfs.add_file(src, "x.bin")
|
||||
vfs.close() # 故意不再调用任何会 commit 的方法
|
||||
|
||||
reopened = SQLiteVFS(db)
|
||||
self.addCleanup(reopened.close)
|
||||
self.assertTrue(
|
||||
reopened.exists("x.bin"),
|
||||
"add_file 之后直接 close()，数据丢失——事务没有提交",
|
||||
)
|
||||
|
||||
|
||||
class FileContentTest(helpers.VFSTestCase):
|
||||
"""P0-1 的下游：写入与读出的字节必须完全一致。"""
|
||||
|
||||
def _write_and_read(self, data, rel_path="data.bin", compress=False):
|
||||
src = helpers.write_file(self.path("input.bin"), data)
|
||||
vfs = SQLiteVFS(":memory:", compress=compress)
|
||||
self.addCleanup(vfs.close)
|
||||
with helpers.quiet():
|
||||
vfs.add_file(src, rel_path)
|
||||
return vfs, vfs.read_file(rel_path)
|
||||
|
||||
def test_small_file_roundtrip(self):
|
||||
vfs, content = self._write_and_read(b"hello sqlite vfs")
|
||||
self.assertEqual(b"hello sqlite vfs", content)
|
||||
self.assertEqual("none", vfs.get_blob_info("data.bin")["compress_algo"])
|
||||
|
||||
def test_empty_file_roundtrip(self):
|
||||
_, content = self._write_and_read(b"")
|
||||
self.assertEqual(b"", content)
|
||||
|
||||
def test_compressible_file_uses_zlib(self):
|
||||
data = b"repeat me please. " * 500 # 9000 字节，高于阈值
|
||||
vfs, content = self._write_and_read(data, compress=True)
|
||||
self.assertEqual(data, content)
|
||||
self.assertEqual(
|
||||
"zlib",
|
||||
vfs.get_blob_info("data.bin")["compress_algo"],
|
||||
"默认算法应当是 zlib",
|
||||
)
|
||||
self.assertIsNone(
|
||||
vfs.get_blob_info("data.bin")["freq_blob"],
|
||||
"zlib 不需要频率表，freq_blob 应当为 NULL",
|
||||
)
|
||||
|
||||
def test_incompressible_file_falls_back_to_none(self):
|
||||
import random
|
||||
|
||||
data = random.Random(7).randbytes(4096)
|
||||
vfs, content = self._write_and_read(data, compress=True)
|
||||
self.assertEqual(data, content)
|
||||
self.assertEqual(
|
||||
"none",
|
||||
vfs.get_blob_info("data.bin")["compress_algo"],
|
||||
"随机数据压不动时应当保留原始字节",
|
||||
)
|
||||
|
||||
def test_compression_applies_below_the_old_1kb_bar(self):
|
||||
"""zlib 包头只有几字节，小文件也值得压（旧的哈夫曼阈值是 1KB）。"""
|
||||
data = b"body{margin:0}" * 30 # 450 字节
|
||||
vfs, content = self._write_and_read(data, compress=True)
|
||||
self.assertEqual(data, content)
|
||||
self.assertEqual("zlib", vfs.get_blob_info("data.bin")["compress_algo"])
|
||||
|
||||
def test_tiny_file_is_not_compressed(self):
|
||||
data = b"hi"
|
||||
vfs, content = self._write_and_read(data, compress=True)
|
||||
self.assertEqual(data, content)
|
||||
self.assertEqual(
|
||||
"none",
|
||||
vfs.get_blob_info("data.bin")["compress_algo"],
|
||||
"低于 zlib 的最小压缩体积时应当原样存储",
|
||||
)
|
||||
|
||||
def test_file_size_recorded_matches_content(self):
|
||||
data = b"1234567890"
|
||||
vfs, _ = self._write_and_read(data)
|
||||
self.assertEqual(10, vfs.get_file_info("data.bin")["file_size"])
|
||||
|
||||
def test_read_file_offset_and_size(self):
|
||||
vfs, _ = self._write_and_read(b"0123456789")
|
||||
self.assertEqual(b"234", vfs.read_file("data.bin", 2, 3))
|
||||
self.assertEqual(b"56789", vfs.read_file("data.bin", 5))
|
||||
self.assertEqual(b"", vfs.read_file("data.bin", 10))
|
||||
|
||||
def test_read_file_rejects_directory_and_missing_path(self):
|
||||
vfs = SQLiteVFS(":memory:")
|
||||
self.addCleanup(vfs.close)
|
||||
helpers.seed_rows(vfs, [{"path": "d", "is_directory": True}])
|
||||
with self.assertRaises(IsADirectoryError):
|
||||
vfs.read_file("d")
|
||||
with self.assertRaises(FileNotFoundError):
|
||||
vfs.read_file("nope.txt")
|
||||
|
||||
|
||||
class CompressionAlgorithmTest(helpers.VFSTestCase):
|
||||
"""zlib 是默认算法；huffman 保留只有一个理由——读得懂早期版本打的包。"""
|
||||
|
||||
DATA = b"function render(el){return el.innerHTML}\n" * 200
|
||||
|
||||
def _vfs_with(self, algorithm, compress=True):
|
||||
src = helpers.write_file(self.path("input.bin"), self.DATA)
|
||||
vfs = SQLiteVFS(":memory:", compress=compress, compress_algorithm=algorithm)
|
||||
self.addCleanup(vfs.close)
|
||||
with helpers.quiet():
|
||||
vfs.add_file(src, "data.bin")
|
||||
return vfs
|
||||
|
||||
def test_default_algorithm_is_zlib(self):
|
||||
from sqlite_vfs import compression
|
||||
|
||||
self.assertEqual("zlib", compression.DEFAULT_ALGORITHM)
|
||||
vfs = self._vfs_with(compression.DEFAULT_ALGORITHM)
|
||||
self.assertEqual("zlib", vfs.get_blob_info("data.bin")["compress_algo"])
|
||||
|
||||
def test_each_algorithm_roundtrips(self):
|
||||
for algorithm in ("zlib", "huffman"):
|
||||
with self.subTest(algorithm=algorithm):
|
||||
vfs = self._vfs_with(algorithm)
|
||||
self.assertEqual(
|
||||
algorithm, vfs.get_blob_info("data.bin")["compress_algo"]
|
||||
)
|
||||
self.assertEqual(self.DATA, vfs.read_file("data.bin"))
|
||||
|
||||
def test_early_huffman_packs_are_still_readable(self):
|
||||
"""早期版本用 huffman 打的包必须还能读——这是保留该算法唯一的理由。"""
|
||||
vfs = self._vfs_with("huffman")
|
||||
blob = vfs.get_blob_info("data.bin")
|
||||
self.assertIsNotNone(blob["freq_blob"], "huffman 必须带频率表")
|
||||
self.assertEqual(self.DATA, vfs.read_file("data.bin"))
|
||||
|
||||
def test_zlib_compresses_better_than_huffman(self):
|
||||
"""同一份文本，zlib 的结果必须明显小于 huffman（LZ77 能吃重复）。"""
|
||||
stored = {}
|
||||
for algorithm in ("zlib", "huffman"):
|
||||
vfs = self._vfs_with(algorithm)
|
||||
stored[algorithm] = vfs.get_blob_info("data.bin")["stored_size"]
|
||||
self.assertLess(
|
||||
stored["zlib"],
|
||||
stored["huffman"],
|
||||
f"zlib={stored['zlib']} huffman={stored['huffman']}",
|
||||
)
|
||||
|
||||
def test_compression_disabled_stores_raw(self):
|
||||
vfs = self._vfs_with("zlib", compress=False)
|
||||
self.assertEqual("none", vfs.get_blob_info("data.bin")["compress_algo"])
|
||||
self.assertEqual(self.DATA, vfs.read_file("data.bin"))
|
||||
|
||||
def test_unknown_algorithm_rejected_at_construction(self):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
SQLiteVFS(":memory:", compress_algorithm="lzma")
|
||||
self.assertIn("lzma", str(ctx.exception))
|
||||
|
||||
def test_unknown_algorithm_in_database_raises_on_read(self):
|
||||
"""库里的算法名本版本不认识时要明确报错，而不是返回坏数据。"""
|
||||
vfs = self._vfs_with("zlib", compress=False)
|
||||
vfs.conn.execute(
|
||||
"UPDATE blobs SET compress_algo = 'lzma', content = ?",
|
||||
(sqlite3.Binary(b"whatever"),),
|
||||
)
|
||||
vfs.conn.commit()
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
vfs.read_file("data.bin")
|
||||
self.assertIn("lzma", str(ctx.exception))
|
||||
self.assertIn("data.bin", str(ctx.exception))
|
||||
|
||||
|
||||
class DirectoryListingTest(helpers.VFSTestCase):
|
||||
"""P0-2：list_directory 必须按 parent_path 过滤。
|
||||
|
||||
用原始 SQL 造数据，避免被 P0-1 连带打挂，保证失败原因单一。
|
||||
"""
|
||||
|
||||
def _vfs(self):
|
||||
vfs = SQLiteVFS(":memory:")
|
||||
self.addCleanup(vfs.close)
|
||||
helpers.seed_rows(
|
||||
vfs,
|
||||
[
|
||||
{"path": "doc", "is_directory": True},
|
||||
{"path": "docs", "is_directory": True},
|
||||
{"path": "doc/a.txt", "content": b"a"},
|
||||
{"path": "doc/sub", "is_directory": True},
|
||||
{"path": "doc/sub/b.txt", "content": b"b"},
|
||||
{"path": "docs/c.txt", "content": b"c"},
|
||||
{"path": "empty_dir", "is_directory": True},
|
||||
{"path": "top.txt", "content": b"t"},
|
||||
],
|
||||
)
|
||||
return vfs
|
||||
|
||||
def test_root_lists_top_level_only(self):
|
||||
vfs = self._vfs()
|
||||
paths = sorted(row["path"] for row in vfs.list_directory("/"))
|
||||
self.assertEqual(["doc", "docs", "empty_dir", "top.txt"], paths)
|
||||
|
||||
def test_subdirectory_lists_its_children(self):
|
||||
vfs = self._vfs()
|
||||
paths = sorted(row["path"] for row in vfs.list_directory("doc"))
|
||||
self.assertEqual(["doc/a.txt", "doc/sub"], paths)
|
||||
|
||||
def test_nested_directory_lists_its_children(self):
|
||||
vfs = self._vfs()
|
||||
paths = sorted(row["path"] for row in vfs.list_directory("doc/sub"))
|
||||
self.assertEqual(["doc/sub/b.txt"], paths)
|
||||
|
||||
def test_directories_come_before_files(self):
|
||||
vfs = self._vfs()
|
||||
rows = vfs.list_directory("doc")
|
||||
self.assertEqual([1, 0], [row["is_directory"] for row in rows])
|
||||
|
||||
def test_leaf_directory_is_empty(self):
|
||||
vfs = self._vfs()
|
||||
self.assertEqual([], vfs.list_directory("empty_dir"))
|
||||
self.assertEqual([], vfs.list_directory("no/such/dir"))
|
||||
|
||||
def test_paths_are_returned_as_dicts_with_datetime(self):
|
||||
vfs = self._vfs()
|
||||
rows = vfs.list_directory("doc")
|
||||
self.assertIsInstance(rows[0], dict)
|
||||
self.assertIn("created_time", rows[0])
|
||||
|
||||
|
||||
class MoveAndRemoveTest(helpers.VFSTestCase):
|
||||
"""P1-1 / P1-2 / P1-3：前缀误伤、重复改名、根目录整体移动。"""
|
||||
|
||||
def _seed(self, rows=None):
|
||||
vfs = SQLiteVFS(":memory:")
|
||||
self.addCleanup(vfs.close)
|
||||
helpers.seed_rows(
|
||||
vfs,
|
||||
rows
|
||||
or [
|
||||
{"path": "doc", "is_directory": True},
|
||||
{"path": "doc/a.txt", "content": b"a"},
|
||||
{"path": "doc/sub", "is_directory": True},
|
||||
{"path": "doc/sub/b.txt", "content": b"b"},
|
||||
{"path": "docs", "is_directory": True},
|
||||
{"path": "docs/c.txt", "content": b"c"},
|
||||
{"path": "top.txt", "content": b"t"},
|
||||
],
|
||||
)
|
||||
return vfs
|
||||
|
||||
def test_remove_directory_does_not_touch_prefix_sibling(self):
|
||||
"""删除 doc 不能连带删掉 docs（LIKE 'doc%' 会把 docs 一起匹配上）。"""
|
||||
vfs = self._seed()
|
||||
vfs.remove("doc")
|
||||
|
||||
paths = helpers.all_paths(vfs)
|
||||
self.assertNotIn("doc", paths)
|
||||
self.assertEqual(
|
||||
{"docs", "docs/c.txt", "top.txt"},
|
||||
paths,
|
||||
"删除 doc 误伤了 docs",
|
||||
)
|
||||
|
||||
def test_remove_file_with_underscore_is_not_a_wildcard(self):
|
||||
"""文件名里的 _ 不能被当成 LIKE 的单字符通配符。"""
|
||||
vfs = self._seed(
|
||||
[
|
||||
{"path": "a_b.txt", "content": b"1"},
|
||||
{"path": "axb.txt", "content": b"2"},
|
||||
]
|
||||
)
|
||||
vfs.remove("a_b.txt")
|
||||
self.assertEqual({"axb.txt"}, helpers.all_paths(vfs))
|
||||
|
||||
def test_remove_file_with_percent_is_not_a_wildcard(self):
|
||||
vfs = self._seed(
|
||||
[
|
||||
{"path": "100%done.txt", "content": b"1"},
|
||||
{"path": "100Xdoney.txt", "content": b"2"},
|
||||
]
|
||||
)
|
||||
vfs.remove("100%done.txt")
|
||||
self.assertEqual({"100Xdoney.txt"}, helpers.all_paths(vfs))
|
||||
|
||||
def test_remove_missing_path_reports_error(self):
|
||||
vfs = self._seed()
|
||||
with self.assertRaises(FileNotFoundError):
|
||||
vfs.remove("no/such/thing.txt")
|
||||
|
||||
def test_mv_dir_rename_is_not_applied_twice(self):
|
||||
"""P1-2: mv_dir("doc","doc_new") 的第二条 UPDATE 会再次命中已改名的目录自身。"""
|
||||
vfs = self._seed()
|
||||
vfs.mv_dir("doc", "doc_new")
|
||||
|
||||
self.assertEqual(
|
||||
{
|
||||
"doc_new",
|
||||
"doc_new/a.txt",
|
||||
"doc_new/sub",
|
||||
"doc_new/sub/b.txt",
|
||||
"docs",
|
||||
"docs/c.txt",
|
||||
"top.txt",
|
||||
},
|
||||
helpers.all_paths(vfs),
|
||||
)
|
||||
|
||||
def test_mv_dir_does_not_touch_prefix_sibling(self):
|
||||
vfs = self._seed()
|
||||
vfs.mv_dir("doc", "doc_new")
|
||||
self.assertIn("docs", helpers.all_paths(vfs))
|
||||
self.assertIn("docs/c.txt", helpers.all_paths(vfs))
|
||||
|
||||
def test_mv_dir_updates_children_parent_paths(self):
|
||||
vfs = self._seed()
|
||||
vfs.mv_dir("doc", "moved")
|
||||
|
||||
parents = helpers.path_parents(vfs)
|
||||
self.assertEqual("moved", parents["moved/a.txt"])
|
||||
self.assertEqual("moved", parents["moved/sub"])
|
||||
self.assertEqual("moved/sub", parents["moved/sub/b.txt"])
|
||||
|
||||
def test_mv_dir_into_existing_parent_directory(self):
|
||||
vfs = self._seed()
|
||||
vfs.mk_dir("nested")
|
||||
vfs.mv_dir("doc", "nested/doc")
|
||||
self.assertIn("nested/doc", helpers.all_paths(vfs))
|
||||
self.assertIn("nested/doc/a.txt", helpers.all_paths(vfs))
|
||||
self.assertEqual("nested", helpers.path_parents(vfs)["nested/doc"])
|
||||
|
||||
def test_mv_dir_to_missing_parent_raises(self):
|
||||
"""目标的父目录不存在时要明确报错，而不是造出一个不可达的目录。"""
|
||||
vfs = self._seed()
|
||||
with self.assertRaises(NotADirectoryError):
|
||||
vfs.mv_dir("doc", "nested/doc")
|
||||
self.assertIn("doc", helpers.all_paths(vfs))
|
||||
|
||||
def test_mv_dir_into_itself_raises(self):
|
||||
vfs = self._seed()
|
||||
with self.assertRaises(ValueError):
|
||||
vfs.mv_dir("doc", "doc/inner")
|
||||
|
||||
def test_mv_dir_root_creates_destination_and_fixes_parent_paths(self):
|
||||
"""P1-3: 把根目录整体移进 archive 时，archive 目录行必须存在，
|
||||
且顶层条目的 parent_path 不能带尾斜杠（'archive/'）。"""
|
||||
vfs = self._seed()
|
||||
vfs.mv_dir("/", "archive")
|
||||
|
||||
parents = helpers.path_parents(vfs)
|
||||
self.assertIn("archive", parents, "目标目录 archive 没有对应的目录行")
|
||||
# archive 自己是顶层条目，parent_path 约定为空
|
||||
self.assertEqual("", parents["archive"])
|
||||
self.assertEqual("archive", parents["archive/doc"])
|
||||
self.assertEqual("archive", parents["archive/docs"])
|
||||
self.assertEqual("archive", parents["archive/top.txt"])
|
||||
self.assertEqual("archive/doc", parents["archive/doc/a.txt"])
|
||||
self.assertEqual("archive/doc/sub", parents["archive/doc/sub/b.txt"])
|
||||
|
||||
self.assertEqual(
|
||||
sorted(["archive"]),
|
||||
sorted(row["path"] for row in vfs.list_directory("/")),
|
||||
)
|
||||
|
||||
def test_mv_dir_to_existing_destination_raises_and_keeps_source(self):
|
||||
vfs = self._seed()
|
||||
with self.assertRaises(Exception):
|
||||
vfs.mv_dir("doc", "docs")
|
||||
paths = helpers.all_paths(vfs)
|
||||
self.assertIn("doc", paths)
|
||||
self.assertIn("doc/a.txt", paths)
|
||||
self.assertIn("docs", paths)
|
||||
|
||||
def test_mv_file_updates_name_and_parent(self):
|
||||
vfs = self._seed()
|
||||
vfs.mv_file("top.txt", "docs/moved.txt")
|
||||
parents = helpers.path_parents(vfs)
|
||||
self.assertNotIn("top.txt", parents)
|
||||
self.assertIn("docs/moved.txt", parents)
|
||||
self.assertEqual("docs", parents["docs/moved.txt"])
|
||||
|
||||
|
||||
class MkDirTest(helpers.VFSTestCase):
|
||||
"""P1-4：mk_dir 重复创建要有明确报错，不能留下悬挂事务。"""
|
||||
|
||||
def test_mk_dir_creates_directory(self):
|
||||
vfs = SQLiteVFS(":memory:")
|
||||
self.addCleanup(vfs.close)
|
||||
vfs.mk_dir("new_dir")
|
||||
self.assertTrue(vfs.is_dir("new_dir"))
|
||||
self.assertEqual(["new_dir"], [row["path"] for row in vfs.list_directory("/")])
|
||||
|
||||
def test_mk_dir_existing_raises_cleanly(self):
|
||||
vfs = SQLiteVFS(":memory:")
|
||||
self.addCleanup(vfs.close)
|
||||
vfs.mk_dir("new_dir")
|
||||
with self.assertRaises(Exception):
|
||||
vfs.mk_dir("new_dir")
|
||||
# 连接仍然可用，后续写入不受影响
|
||||
vfs.mk_dir("another")
|
||||
self.assertTrue(vfs.is_dir("another"))
|
||||
|
||||
|
||||
class AddFileConflictTest(helpers.VFSTestCase):
|
||||
"""P1-4：同一路径重复写入要能被调用方感知，而不是静默失败。"""
|
||||
|
||||
def test_duplicate_add_file_is_reported_and_keeps_first_content(self):
|
||||
src = helpers.write_file(self.path("a.bin"), b"first")
|
||||
src2 = helpers.write_file(self.path("b.bin"), b"second")
|
||||
|
||||
vfs = SQLiteVFS(":memory:")
|
||||
self.addCleanup(vfs.close)
|
||||
with helpers.quiet():
|
||||
first = vfs.add_file(src, "same.bin")
|
||||
second = vfs.add_file(src2, "same.bin")
|
||||
|
||||
self.assertTrue(first, "首次写入应当成功")
|
||||
self.assertFalse(second, "重复路径的写入应当失败并返回 False")
|
||||
self.assertEqual(1, helpers.count_files(vfs))
|
||||
self.assertEqual(b"first", vfs.read_file("same.bin"))
|
||||
# 失败后连接仍可用
|
||||
with helpers.quiet():
|
||||
self.assertTrue(vfs.add_file(src2, "other.bin"))
|
||||
self.assertEqual(2, helpers.count_files(vfs))
|
||||
|
||||
|
||||
class ExportTest(helpers.VFSTestCase):
|
||||
"""P0-2 / P1-6 / P1-10：导出的目录树必须与库内容严格一致。"""
|
||||
|
||||
def _vfs(self):
|
||||
vfs = SQLiteVFS(":memory:")
|
||||
self.addCleanup(vfs.close)
|
||||
helpers.seed_rows(
|
||||
vfs,
|
||||
[
|
||||
{"path": "doc", "is_directory": True},
|
||||
{"path": "doc/a.txt", "content": b"a"},
|
||||
{"path": "doc/sub", "is_directory": True},
|
||||
{"path": "doc/sub/b.txt", "content": b"b"},
|
||||
{"path": "empty_dir", "is_directory": True},
|
||||
{"path": "top.txt", "content": b"t"},
|
||||
],
|
||||
)
|
||||
return vfs
|
||||
|
||||
def test_export_file_writes_exact_bytes(self):
|
||||
vfs = self._vfs()
|
||||
target = self.path("out/top.txt")
|
||||
with helpers.quiet():
|
||||
self.assertTrue(vfs.export_file("top.txt", target))
|
||||
with open(target, "rb") as f:
|
||||
self.assertEqual(b"t", f.read())
|
||||
|
||||
def test_export_file_accepts_bare_filename(self):
|
||||
"""P1-6: local_path 没有目录部分时 dirname 为 ''，不能抛 FileNotFoundError。"""
|
||||
vfs = self._vfs()
|
||||
cwd = os.getcwd()
|
||||
os.chdir(self.tmp)
|
||||
try:
|
||||
with helpers.quiet():
|
||||
ok = vfs.export_file("top.txt", "bare.txt")
|
||||
self.assertTrue(ok, "导出到当前目录下的裸文件名应当成功")
|
||||
self.assertTrue(os.path.exists("bare.txt"))
|
||||
finally:
|
||||
os.chdir(cwd)
|
||||
|
||||
def test_export_directory_full_tree(self):
|
||||
vfs = self._vfs()
|
||||
out = self.path("out")
|
||||
with helpers.quiet():
|
||||
vfs.export_directory("/", out)
|
||||
|
||||
expected = {
|
||||
"doc": None,
|
||||
"doc/a.txt": b"a",
|
||||
"doc/sub": None,
|
||||
"doc/sub/b.txt": b"b",
|
||||
"empty_dir": None,
|
||||
"top.txt": b"t",
|
||||
}
|
||||
helpers.assert_tree_equal(self, expected, helpers.read_tree(out))
|
||||
|
||||
def test_export_subdirectory_does_not_recurse_forever(self):
|
||||
"""P0-2: list_directory 按 path 过滤时，导出 doc 会无限自我递归。"""
|
||||
vfs = self._vfs()
|
||||
out = self.path("out")
|
||||
with helpers.quiet():
|
||||
vfs.export_directory("doc", out)
|
||||
|
||||
expected = {"a.txt": b"a", "sub": None, "sub/b.txt": b"b"}
|
||||
helpers.assert_tree_equal(self, expected, helpers.read_tree(out))
|
||||
|
||||
def test_export_filesystem_matches_tree(self):
|
||||
vfs = self._vfs()
|
||||
out = self.path("out")
|
||||
with helpers.quiet():
|
||||
vfs.export_filesystem(out)
|
||||
expected = {
|
||||
"doc": None,
|
||||
"doc/a.txt": b"a",
|
||||
"doc/sub": None,
|
||||
"doc/sub/b.txt": b"b",
|
||||
"empty_dir": None,
|
||||
"top.txt": b"t",
|
||||
}
|
||||
helpers.assert_tree_equal(self, expected, helpers.read_tree(out))
|
||||
|
||||
|
||||
class SchemaCompatibilityTest(helpers.VFSTestCase):
|
||||
"""P1-7：旧格式数据库必须被明确拒绝，而不是连上之后静默失败。"""
|
||||
|
||||
def test_synthesized_old_schema_db_is_rejected(self):
|
||||
err = helpers.vfs_format_error()
|
||||
if err is None:
|
||||
self.fail(
|
||||
"core 没有定义 VFSFormatError 一类的异常，"
|
||||
"调用方无法区分「旧格式库」与「正常库」，只能在导出阶段看到一堆导出失败"
|
||||
)
|
||||
db = helpers.make_old_schema_db(self.path("legacy.svfs"))
|
||||
with self.assertRaises(err) as ctx:
|
||||
SQLiteVFS(db)
|
||||
# 报错必须直接给出下一步动作，而不是只说"格式不对"
|
||||
self.assertIn("vfsmigrate", str(ctx.exception))
|
||||
|
||||
def test_v2_schema_db_is_rejected_with_migration_hint(self):
|
||||
db = helpers.make_v2_schema_db(
|
||||
self.path("v2.svfs"), rows=[{"path": "a.txt", "content": b"a"}]
|
||||
)
|
||||
with self.assertRaises(helpers.vfs_format_error()) as ctx:
|
||||
SQLiteVFS(db)
|
||||
self.assertIn("vfsmigrate", str(ctx.exception))
|
||||
|
||||
def test_repository_sample_db_is_rejected(self):
|
||||
if not os.path.exists(helpers.SAMPLE_DB):
|
||||
self.skipTest("仓库内没有 appcation.svfs")
|
||||
err = helpers.vfs_format_error()
|
||||
if err is None:
|
||||
self.fail("core 没有定义 VFSFormatError 一类的异常")
|
||||
with self.assertRaises(err):
|
||||
SQLiteVFS(helpers.SAMPLE_DB)
|
||||
|
||||
def test_fresh_database_is_accepted(self):
|
||||
vfs = SQLiteVFS(self.path("fresh.svfs"))
|
||||
self.addCleanup(vfs.close)
|
||||
helpers.seed_rows(vfs, [{"path": "a.txt", "content": b"x"}])
|
||||
self.assertTrue(vfs.exists("a.txt"))
|
||||
|
||||
def test_non_sqlite_file_is_rejected(self):
|
||||
junk = helpers.write_file(self.path("junk.svfs"), b"this is not a database")
|
||||
with self.assertRaises(sqlite3.DatabaseError):
|
||||
SQLiteVFS(junk)
|
||||
|
||||
def test_legacy_pack_is_readable_after_migration(self):
|
||||
"""迁移工具必须真的能读懂最早那版格式（compressed 布尔 + zlib）。
|
||||
|
||||
详细覆盖见 test_migration.py；这里只保证"v1 库能被迁到当前版本且内容一致"。
|
||||
"""
|
||||
import zlib
|
||||
|
||||
from sqlite_vfs import migration
|
||||
|
||||
payload = b"legacy compressed content " * 50
|
||||
db = helpers.make_old_schema_db(
|
||||
self.path("legacy.svfs"),
|
||||
extra_rows=[
|
||||
{
|
||||
"path": "packed.txt",
|
||||
"content": zlib.compress(payload),
|
||||
"compressed": True,
|
||||
"file_size": len(payload),
|
||||
}
|
||||
],
|
||||
)
|
||||
result = migration.migrate(db)
|
||||
self.assertTrue(result["migrated"])
|
||||
|
||||
vfs = SQLiteVFS(db)
|
||||
self.addCleanup(vfs.close)
|
||||
self.assertEqual(
|
||||
payload, vfs.read_file("packed.txt"), "旧版 zlib 压缩的数据应当能读出来"
|
||||
)
|
||||
self.assertEqual(
|
||||
b"hello", vfs.read_file("legacy.txt"), "旧版未压缩的数据应当能读出来"
|
||||
)
|
||||
self.assertEqual("zlib", vfs.get_blob_info("packed.txt")["compress_algo"])
|
||||
|
||||
|
||||
class TimestampStorageTest(helpers.VFSTestCase):
|
||||
"""P2-4：时间戳必须显式转成字符串，不能依赖 sqlite3 已弃用的默认 adapter。
|
||||
|
||||
这条同时钉住两件事：**不再有弃用警告**，以及**盘上格式与历史保持一致**
|
||||
（格式一变，旧库的读取与快照摘要都会受影响）。
|
||||
"""
|
||||
|
||||
#: sqlite3 默认 adapter 当年的输出格式就是 val.isoformat(" ")
|
||||
STORED_TIME_RE = re.compile(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(\.\d+)?$")
|
||||
|
||||
def _exercise_every_write_path(self, vfs):
|
||||
src = helpers.write_file(self.path("in.txt"), b"payload")
|
||||
with helpers.quiet():
|
||||
vfs.add_file(src, "in.txt")
|
||||
vfs.mk_dir("made")
|
||||
vfs.add_directory("sub", self.path("."))
|
||||
vfs.create_snapshot("v2", copy_entries=True)
|
||||
vfs.sign("k")
|
||||
|
||||
def test_no_deprecation_warning_from_any_write_path(self):
|
||||
vfs = SQLiteVFS(self.path("t.svfs"))
|
||||
self.addCleanup(vfs.close)
|
||||
with warnings.catch_warnings():
|
||||
# 任何 DeprecationWarning 都会变成错误——包括 sqlite3 的 adapter 警告
|
||||
warnings.simplefilter("error", DeprecationWarning)
|
||||
self._exercise_every_write_path(vfs)
|
||||
|
||||
def test_stored_text_keeps_the_historical_format(self):
|
||||
vfs = SQLiteVFS(self.path("fmt.svfs"))
|
||||
self.addCleanup(vfs.close)
|
||||
src = helpers.write_file(self.path("in.txt"), b"payload")
|
||||
with helpers.quiet():
|
||||
vfs.add_file(src, "in.txt")
|
||||
vfs.create_snapshot("v2", copy_entries=True)
|
||||
vfs.sign("k")
|
||||
|
||||
for value in vfs.conn.execute(
|
||||
"SELECT created_time, modified_time FROM files WHERE path = 'in.txt'"
|
||||
).fetchone():
|
||||
self.assertRegex(value, self.STORED_TIME_RE, "files 时间戳格式变了")
|
||||
|
||||
self.assertRegex(
|
||||
vfs.conn.execute(
|
||||
"SELECT created_time FROM snapshots WHERE id = 2"
|
||||
).fetchone()[0],
|
||||
self.STORED_TIME_RE,
|
||||
"snapshots 时间戳格式变了",
|
||||
)
|
||||
self.assertRegex(
|
||||
vfs.conn.execute("SELECT signed_time FROM signatures LIMIT 1").fetchone()[0],
|
||||
self.STORED_TIME_RE,
|
||||
"signatures 时间戳格式变了",
|
||||
)
|
||||
|
||||
def test_stored_text_parses_back_to_the_file_mtime(self):
|
||||
vfs = SQLiteVFS(":memory:")
|
||||
self.addCleanup(vfs.close)
|
||||
src = helpers.write_file(self.path("in.txt"), b"payload")
|
||||
with helpers.quiet():
|
||||
vfs.add_file(src, "in.txt")
|
||||
|
||||
raw = vfs.conn.execute(
|
||||
"SELECT modified_time FROM files WHERE path = 'in.txt'"
|
||||
).fetchone()[0]
|
||||
parsed = SQLiteVFS.parse_time(raw)
|
||||
self.assertIsNotNone(parsed)
|
||||
self.assertAlmostEqual(os.path.getmtime(src), parsed.timestamp(), delta=2)
|
||||
# get_file_info 走的也是同一条解析路径
|
||||
self.assertAlmostEqual(
|
||||
os.path.getmtime(src),
|
||||
vfs.get_file_info("in.txt")["modified_time"].timestamp(),
|
||||
delta=2,
|
||||
)
|
||||
|
||||
|
||||
class ReadFileBoundaryTest(helpers.VFSTestCase):
|
||||
"""read_file 的 offset/size 边界语义（此前只有间接覆盖）。"""
|
||||
|
||||
DATA = b"0123456789"
|
||||
|
||||
def _vfs(self, compress=False):
|
||||
vfs = SQLiteVFS(":memory:", compress=compress)
|
||||
self.addCleanup(vfs.close)
|
||||
helpers.seed_rows(vfs, [{"path": "n.bin", "content": self.DATA}])
|
||||
return vfs
|
||||
|
||||
def test_full_read(self):
|
||||
self.assertEqual(self.DATA, self._vfs().read_file("n.bin"))
|
||||
|
||||
def test_zero_size(self):
|
||||
self.assertEqual(b"", self._vfs().read_file("n.bin", 0, 0))
|
||||
|
||||
def test_offset_in_the_middle(self):
|
||||
self.assertEqual(b"456789", self._vfs().read_file("n.bin", 4))
|
||||
|
||||
def test_size_larger_than_remaining_clamps(self):
|
||||
self.assertEqual(b"789", self._vfs().read_file("n.bin", 7, 999))
|
||||
|
||||
def test_offset_at_eof_returns_empty(self):
|
||||
self.assertEqual(b"", self._vfs().read_file("n.bin", len(self.DATA)))
|
||||
|
||||
def test_offset_beyond_eof_returns_empty(self):
|
||||
self.assertEqual(b"", self._vfs().read_file("n.bin", len(self.DATA) + 100))
|
||||
|
||||
def test_negative_offset_is_rejected(self):
|
||||
"""负数会走 Python 切片语义（从末尾数），对调用方是陷阱，直接拒绝。"""
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
self._vfs().read_file("n.bin", -3)
|
||||
self.assertIn("offset", str(ctx.exception))
|
||||
|
||||
def test_negative_size_is_rejected(self):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
self._vfs().read_file("n.bin", 0, -1)
|
||||
self.assertIn("size", str(ctx.exception))
|
||||
|
||||
def test_boundary_semantics_hold_for_compressed_content(self):
|
||||
"""压缩路径下 offset/size 作用的是**解压后**的字节。"""
|
||||
data = b"repeat me please. " * 500
|
||||
vfs = SQLiteVFS(":memory:", compress=True)
|
||||
self.addCleanup(vfs.close)
|
||||
src = helpers.write_file(self.path("big.bin"), data)
|
||||
with helpers.quiet():
|
||||
vfs.add_file(src, "big.bin")
|
||||
self.assertEqual("zlib", vfs.get_blob_info("big.bin")["compress_algo"])
|
||||
|
||||
self.assertEqual(data[:10], vfs.read_file("big.bin", 0, 10))
|
||||
self.assertEqual(data[100:130], vfs.read_file("big.bin", 100, 30))
|
||||
self.assertEqual(data[-5:], vfs.read_file("big.bin", len(data) - 5))
|
||||
with self.assertRaises(ValueError):
|
||||
vfs.read_file("big.bin", -1)
|
||||
|
||||
|
||||
class BlobPrimitiveTest(helpers.VFSTestCase):
|
||||
"""内容寻址读原语：list_blobs / read_blob / all_entries / get_blob_info。"""
|
||||
|
||||
def _vfs(self, compress=True):
|
||||
vfs = SQLiteVFS(":memory:", compress=compress)
|
||||
self.addCleanup(vfs.close)
|
||||
helpers.seed_rows(
|
||||
vfs,
|
||||
[
|
||||
{"path": "a.txt", "content": b"same"},
|
||||
{"path": "dup.txt", "content": b"same"},
|
||||
{"path": "sub/b.txt", "content": b"other"},
|
||||
{"path": "sub", "is_directory": True},
|
||||
],
|
||||
)
|
||||
return vfs
|
||||
|
||||
def test_list_blobs_returns_one_row_per_unique_content(self):
|
||||
vfs = self._vfs()
|
||||
hashes = [b["hash"] for b in vfs.list_blobs()]
|
||||
self.assertEqual(2, len(hashes), "相同内容只应有一行")
|
||||
self.assertEqual(sorted(hashes), hashes, "应当按 hash 排序")
|
||||
|
||||
def test_list_blobs_reports_sizes_without_content(self):
|
||||
vfs = self._vfs()
|
||||
for info in vfs.list_blobs():
|
||||
self.assertEqual(
|
||||
{"hash", "size", "stored_size", "compress_algo"}, set(info)
|
||||
)
|
||||
|
||||
def test_read_blob_returns_original_bytes(self):
|
||||
vfs = self._vfs()
|
||||
digest = vfs.get_blob_info("a.txt")["hash"]
|
||||
self.assertEqual(b"same", vfs.read_blob(digest))
|
||||
self.assertEqual(b"same", vfs.read_blob(digest, verify_hash=True))
|
||||
|
||||
def test_read_blob_unknown_hash_raises(self):
|
||||
with self.assertRaises(FileNotFoundError):
|
||||
self._vfs().read_blob("0" * 64)
|
||||
|
||||
def test_read_blob_detects_tampering_when_asked(self):
|
||||
vfs = self._vfs(compress=False)
|
||||
digest = vfs.get_blob_info("sub/b.txt")["hash"]
|
||||
# 换成同样长度的别的内容：大小检查看不出来，只有哈希校验能发现
|
||||
vfs.conn.execute(
|
||||
"UPDATE blobs SET content = X'0000000000' WHERE hash = ?", (digest,)
|
||||
)
|
||||
vfs.conn.commit()
|
||||
|
||||
self.assertEqual(
|
||||
b"\x00" * 5, vfs.read_blob(digest), "不校验时按原样返回（结构性检查放行）"
|
||||
)
|
||||
with self.assertRaises(VFSIntegrityError) as ctx:
|
||||
vfs.read_blob(digest, verify_hash=True)
|
||||
self.assertIn("哈希不匹配", str(ctx.exception))
|
||||
|
||||
def test_read_blob_detects_a_size_mismatch_without_asking(self):
|
||||
"""大小对不上是便宜的结构性检查，不需要调用方显式要求。"""
|
||||
vfs = self._vfs(compress=False)
|
||||
digest = vfs.get_blob_info("sub/b.txt")["hash"]
|
||||
vfs.conn.execute(
|
||||
"UPDATE blobs SET content = X'00' WHERE hash = ?", (digest,)
|
||||
)
|
||||
vfs.conn.commit()
|
||||
with self.assertRaises(VFSIntegrityError) as ctx:
|
||||
vfs.read_blob(digest)
|
||||
self.assertIn("大小", str(ctx.exception))
|
||||
|
||||
def test_all_entries_covers_the_whole_tree(self):
|
||||
vfs = self._vfs()
|
||||
self.assertEqual(
|
||||
["a.txt", "dup.txt", "sub", "sub/b.txt"],
|
||||
[e["path"] for e in vfs.all_entries()],
|
||||
)
|
||||
kinds = {e["path"]: e["is_directory"] for e in vfs.all_entries()}
|
||||
self.assertEqual(1, kinds["sub"])
|
||||
self.assertEqual(0, kinds["a.txt"])
|
||||
|
||||
def test_all_entries_is_snapshot_scoped(self):
|
||||
vfs = self._vfs()
|
||||
vfs.create_snapshot("v2")
|
||||
self.assertEqual([], vfs.all_entries())
|
||||
self.assertEqual(4, len(vfs.all_entries(snapshot_id=1)))
|
||||
|
||||
def test_get_blob_info_returns_none_for_a_directory(self):
|
||||
self.assertIsNone(self._vfs().get_blob_info("sub"))
|
||||
|
||||
def test_get_blob_info_returns_none_for_a_missing_path(self):
|
||||
self.assertIsNone(self._vfs().get_blob_info("nope"))
|
||||
|
||||
|
||||
class AddDirectoryContractTest(helpers.VFSTestCase):
|
||||
"""add_directory 的返回值与 replace 语义（此前只有间接覆盖）。"""
|
||||
|
||||
def _vfs(self):
|
||||
vfs = SQLiteVFS(":memory:")
|
||||
self.addCleanup(vfs.close)
|
||||
helpers.make_tree(self.path("src"), {"sub/deep": None})
|
||||
return vfs
|
||||
|
||||
def test_returns_true_and_creates_the_row(self):
|
||||
vfs = self._vfs()
|
||||
self.assertTrue(vfs.add_directory("sub", self.path("src")))
|
||||
self.assertTrue(vfs.is_dir("sub"))
|
||||
self.assertEqual("", helpers.path_parents(vfs)["sub"])
|
||||
|
||||
def test_returns_false_for_a_missing_source_directory(self):
|
||||
vfs = self._vfs()
|
||||
with helpers.quiet():
|
||||
self.assertFalse(vfs.add_directory("nope", self.path("src")))
|
||||
self.assertFalse(vfs.exists("nope"))
|
||||
self.assertEqual(set(), helpers.all_paths(vfs), "失败不该留下半条记录")
|
||||
|
||||
def test_duplicate_without_replace_fails_cleanly(self):
|
||||
vfs = self._vfs()
|
||||
self.assertTrue(vfs.add_directory("sub", self.path("src")))
|
||||
with helpers.quiet():
|
||||
self.assertFalse(
|
||||
vfs.add_directory("sub", self.path("src")), "重复路径应当失败"
|
||||
)
|
||||
self.assertEqual(1, len(helpers.all_paths(vfs)))
|
||||
self.assertTrue(vfs.is_dir("sub"), "原记录仍在")
|
||||
|
||||
def test_replace_updates_in_place(self):
|
||||
vfs = self._vfs()
|
||||
vfs.add_directory("sub", self.path("src"))
|
||||
vfs.conn.execute("UPDATE files SET permissions = 0 WHERE path = 'sub'")
|
||||
vfs.conn.commit()
|
||||
|
||||
self.assertTrue(vfs.add_directory("sub", self.path("src"), replace=True))
|
||||
self.assertEqual(1, len(helpers.all_paths(vfs)), "replace 不该留下两行")
|
||||
self.assertNotEqual(0, vfs.get_file_info("sub")["permissions"])
|
||||
|
||||
def test_connection_survives_a_failed_insert(self):
|
||||
vfs = self._vfs()
|
||||
vfs.add_directory("sub", self.path("src"))
|
||||
with helpers.quiet():
|
||||
vfs.add_directory("sub", self.path("src"))
|
||||
self.assertTrue(vfs.add_directory("deep", self.path("src", "sub")))
|
||||
|
||||
|
||||
class CleanTest(helpers.VFSTestCase):
|
||||
"""clean() 此前只被 packer 间接调用过。"""
|
||||
|
||||
def test_clean_wipes_every_table(self):
|
||||
vfs = SQLiteVFS(self.path("c.svfs"))
|
||||
self.addCleanup(vfs.close)
|
||||
helpers.seed_rows(vfs, [{"path": "a.txt", "content": b"a"}])
|
||||
vfs.add_vfs_metadata("pack")
|
||||
vfs.create_snapshot("v2", copy_entries=True)
|
||||
vfs.sign("k")
|
||||
|
||||
vfs.clean()
|
||||
|
||||
for table in ("files", "blobs", "snapshots", "signatures", "filesystem_metadata"):
|
||||
with self.subTest(table=table):
|
||||
count = vfs.conn.execute(
|
||||
f"SELECT COUNT(*) AS c FROM {table}"
|
||||
).fetchone()["c"]
|
||||
self.assertEqual(0, count, f"{table} 没被清空")
|
||||
self.assertIsNone(vfs.snapshot_id)
|
||||
self.assertEqual((0, 0), (vfs.total_files, vfs.total_size))
|
||||
|
||||
def test_pack_after_clean_starts_fresh(self):
|
||||
db = self.path("reuse.svfs")
|
||||
src = helpers.make_tree(self.path("src"), {"a.txt": b"a"})
|
||||
helpers.pack(src, db, name="first")
|
||||
|
||||
vfs = SQLiteVFS(db)
|
||||
vfs.clean()
|
||||
vfs.add_vfs_metadata("second")
|
||||
helpers.seed_rows(vfs, [{"path": "b.txt", "content": b"b"}])
|
||||
self.assertEqual({"b.txt"}, helpers.all_paths(vfs))
|
||||
self.assertEqual("second", vfs.get_stats()["name"])
|
||||
self.assertIsNone(vfs.get_file_info("a.txt"))
|
||||
vfs.close()
|
||||
|
||||
|
||||
class ExistsHelpersTest(helpers.VFSTestCase):
|
||||
"""exists / is_dir / is_file 三个判定（is_file 在 src 里没有任何调用方）。"""
|
||||
|
||||
def test_all_three_agree_with_the_entry_type(self):
|
||||
vfs = SQLiteVFS(":memory:")
|
||||
self.addCleanup(vfs.close)
|
||||
helpers.seed_rows(
|
||||
vfs, [{"path": "d", "is_directory": True}, {"path": "f", "content": b"x"}]
|
||||
)
|
||||
|
||||
self.assertTrue(vfs.exists("d"))
|
||||
self.assertTrue(vfs.is_dir("d"))
|
||||
self.assertFalse(vfs.is_file("d"))
|
||||
|
||||
self.assertTrue(vfs.exists("f"))
|
||||
self.assertFalse(vfs.is_dir("f"))
|
||||
self.assertTrue(vfs.is_file("f"))
|
||||
|
||||
for missing in ("nope", "d/nope"):
|
||||
with self.subTest(missing=missing):
|
||||
self.assertFalse(vfs.exists(missing))
|
||||
self.assertFalse(vfs.is_dir(missing))
|
||||
self.assertFalse(vfs.is_file(missing))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user