"""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()