"""测试公共工具：路径引导、临时目录、目录树构造与比对、打包/解包/CLI 封装。 测试用标准库 unittest 编写，因此 `python -m unittest discover -s tests` 与 `pytest tests/` 都能运行，不需要额外安装依赖。 """ import contextlib import io import os import shutil import sqlite3 import subprocess import sys import tempfile import unittest REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) SRC_DIR = os.path.join(REPO_ROOT, "src") SAMPLE_DB = os.path.join(REPO_ROOT, "appcation.svfs") if SRC_DIR not in sys.path: sys.path.insert(0, SRC_DIR) # -------------------------------------------------------------------------- # 环境与清理 # -------------------------------------------------------------------------- def force_rmtree(path): """删除目录树，容忍超长路径。 P0-2 的缺陷会在导出时生成几百层 sub/sub/sub/... 的垃圾目录链， 普通 shutil.rmtree 在 Windows 上会因为超过 MAX_PATH 而删不掉。 """ if not path or not os.path.exists(path): return if os.name == "nt": shutil.rmtree("\\\\?\\" + os.path.abspath(path), ignore_errors=True) shutil.rmtree(path, ignore_errors=True) @contextlib.contextmanager def temp_dir(prefix="svfs_test_"): path = tempfile.mkdtemp(prefix=prefix) try: yield path finally: force_rmtree(path) @contextlib.contextmanager def quiet(): """屏蔽被测代码的进度输出，让测试失败信息保持可读。""" buf = io.StringIO() with contextlib.redirect_stdout(buf): yield buf class VFSTestCase(unittest.TestCase): """每个用例一份独立的临时目录。""" def setUp(self): self.tmp = tempfile.mkdtemp(prefix="svfs_test_") self.addCleanup(force_rmtree, self.tmp) def path(self, *parts): return os.path.join(self.tmp, *parts) # -------------------------------------------------------------------------- # 目录树构造与比对 # -------------------------------------------------------------------------- def make_tree(base_dir, spec): """按 spec 生成目录树。 spec 的键是相对路径（用 / 分隔），值为 None 表示目录，bytes 表示文件内容。 """ for rel_path, content in spec.items(): abs_path = os.path.join(base_dir, *rel_path.split("/")) if content is None: os.makedirs(abs_path, exist_ok=True) else: parent = os.path.dirname(abs_path) if parent: os.makedirs(parent, exist_ok=True) with open(abs_path, "wb") as f: f.write(content) return base_dir def write_file(path, data): parent = os.path.dirname(path) if parent: os.makedirs(parent, exist_ok=True) with open(path, "wb") as f: f.write(data) return path def read_bytes(path): with open(path, "rb") as f: return f.read() def md5_of(path): import hashlib return hashlib.md5(read_bytes(path)).hexdigest() def read_tree(base_dir): """读取目录树，返回 {相对路径(/ 分隔): bytes 或 None(目录)}。""" result = {} for root, dirs, files in os.walk(base_dir): for name in dirs: abs_path = os.path.join(root, name) result[os.path.relpath(abs_path, base_dir).replace("\\", "/")] = None for name in files: abs_path = os.path.join(root, name) with open(abs_path, "rb") as f: result[os.path.relpath(abs_path, base_dir).replace("\\", "/")] = f.read() return result def diff_trees(expected, actual, limit=15): """返回人类可读的目录树差异摘要，完全一致时返回 "无差异"。""" missing = sorted(set(expected) - set(actual)) extra = sorted(set(actual) - set(expected)) changed = sorted( rel for rel in set(expected) & set(actual) if expected[rel] != actual[rel] ) lines = [] if missing: lines.append(f"缺失 {len(missing)} 项，例如: {missing[:limit]}") if extra: lines.append(f"多出 {len(extra)} 项，例如: {extra[:limit]}") if changed: lines.append(f"内容不一致 {len(changed)} 项，例如: {changed[:limit]}") return "\n".join(lines) if lines else "无差异" def assert_tree_equal(case, expected, actual): case.assertEqual("无差异", diff_trees(expected, actual)) # -------------------------------------------------------------------------- # 数据库直接探查（绕过被测的公开 API，用于独立断言） # -------------------------------------------------------------------------- def all_paths(vfs): """返回当前快照里所有 path 的集合。""" return { row["path"] for row in vfs.conn.execute( "SELECT path FROM files WHERE snapshot_id = ?", (vfs._active_snapshot(),) ) } def path_parents(vfs): """返回 {path: parent_path}（当前快照）。""" return { row["path"]: row["parent_path"] for row in vfs.conn.execute( "SELECT path, parent_path FROM files WHERE snapshot_id = ?", (vfs._active_snapshot(),), ) } def stored_files(vfs): """返回 {path: content}，仅文件（内容来自 blobs 表）。""" return { row["path"]: row["content"] for row in vfs.conn.execute( """ SELECT f.path AS path, b.content AS content FROM files f JOIN blobs b ON b.hash = f.content_hash WHERE f.is_directory = 0 AND f.snapshot_id = ? """, (vfs._active_snapshot(),), ) } def blob_count(vfs): """库里 blob 的行数（去重后的内容份数）。""" return vfs.conn.execute("SELECT COUNT(*) AS c FROM blobs").fetchone()["c"] def stored_blobs(vfs): """返回 {hash: (size, stored_size, compress_algo)}。""" return { row["hash"]: (row["size"], row["stored_size"], row["compress_algo"]) for row in vfs.conn.execute( "SELECT hash, size, stored_size, compress_algo FROM blobs" ) } def column_names(conn, table="files"): return {row["name"] for row in conn.execute(f"PRAGMA table_info({table})")} def count_files(vfs): return vfs.conn.execute( "SELECT COUNT(*) AS c FROM files WHERE is_directory = 0 AND snapshot_id = ?", (vfs._active_snapshot(),), ).fetchone()["c"] def seed_rows(vfs, rows): """用原始 SQL 造数据（走 blobs + content_hash，等价于库内真实布局）。 只写新旧 schema 共有的列，这样"目录语义"类测试不会被 存储层的细节连带打挂，便于定位问题。 """ import hashlib cursor = vfs.conn.cursor() snapshot_id = vfs._ensure_snapshot() for row in rows: rel_path = row["path"] if "/" in rel_path: default_name = rel_path.rsplit("/", 1)[1] default_parent = rel_path.rsplit("/", 1)[0] else: default_name = rel_path default_parent = "" is_directory = bool(row.get("is_directory")) content = None if is_directory else (row.get("content") or b"") content_hash = None if content is not None: content_hash = hashlib.sha256(content).hexdigest() cursor.execute( """ INSERT OR IGNORE INTO blobs (hash, size, stored_size, content, freq_blob, compress_algo) VALUES (?, ?, ?, ?, NULL, 'none') """, (content_hash, len(content), len(content), sqlite3.Binary(content)), ) cursor.execute( """ INSERT INTO files (snapshot_id, path, name, parent_path, is_directory, file_size, content_hash) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( snapshot_id, rel_path, row.get("name", default_name), row.get("parent_path", default_parent), 1 if is_directory else 0, row.get("file_size", len(content) if content is not None else 0), content_hash, ), ) vfs.conn.commit() # 摘要与统计要一起写上：seed_rows 造出来的库应当等价于"正常打包出来的库"， # 否则 verify() 会因为"没有记录摘要"而判失败（这是有意的保守策略） vfs.recount() def make_old_schema_db(db_path, extra_rows=()): """生成 **v1** 旧格式数据库（content BLOB + compressed BOOLEAN）。 对应仓库里 appcation.svfs 的结构，也用于验证迁移工具能处理最早的格式。 """ conn = sqlite3.connect(db_path) try: conn.executescript( """ CREATE TABLE filesystem_metadata ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, created_time DATETIME DEFAULT CURRENT_TIMESTAMP, total_files INTEGER DEFAULT 0, total_size INTEGER DEFAULT 0 ); CREATE TABLE files ( id INTEGER PRIMARY KEY AUTOINCREMENT, path TEXT NOT NULL UNIQUE, name TEXT NOT NULL, parent_path TEXT, is_directory BOOLEAN NOT NULL DEFAULT 0, file_size INTEGER DEFAULT 0, created_time DATETIME, modified_time DATETIME, permissions INTEGER DEFAULT 0644, content BLOB, compressed BOOLEAN DEFAULT 0 ); INSERT INTO filesystem_metadata (name, total_files, total_size) VALUES ('legacy', 1, 5); INSERT INTO files (path, name, parent_path, is_directory, file_size, content) VALUES ('legacy.txt', 'legacy.txt', '', 0, 5, X'68656C6C6F'); """ ) for row in extra_rows: conn.execute( """ INSERT INTO files (path, name, parent_path, is_directory, file_size, content, compressed) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( row["path"], row["path"].rsplit("/", 1)[-1], row.get("parent_path", ""), 1 if row.get("is_directory") else 0, row.get("file_size", len(row.get("content") or b"")), sqlite3.Binary(row["content"]) if row.get("content") is not None else None, 1 if row.get("compressed") else 0, ), ) conn.commit() finally: conn.close() return db_path def make_v2_schema_db(db_path, rows=()): """生成 **v2** 格式数据库：内容与压缩算法列都在 files 表里（无 blobs/snapshots）。 rows 里可以给 content / compress_algo / freq_blob 来模拟"已经压缩过的内容"。 """ conn = sqlite3.connect(db_path) try: conn.executescript( """ CREATE TABLE filesystem_metadata ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, created_time DATETIME DEFAULT CURRENT_TIMESTAMP, total_files INTEGER DEFAULT 0, total_size INTEGER DEFAULT 0 ); CREATE TABLE files ( id INTEGER PRIMARY KEY AUTOINCREMENT, path TEXT NOT NULL UNIQUE, name TEXT NOT NULL, parent_path TEXT, is_directory BOOLEAN NOT NULL DEFAULT 0, file_size INTEGER DEFAULT 0, created_time DATETIME, modified_time DATETIME, permissions INTEGER DEFAULT 420, content BLOB, freq_blob BLOB, compress_algo TEXT DEFAULT 'none' ); INSERT INTO filesystem_metadata (name, total_files, total_size) VALUES ('v2pack', 0, 0); """ ) for row in rows: conn.execute( """ INSERT INTO files (path, name, parent_path, is_directory, file_size, content, freq_blob, compress_algo) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( row["path"], row["path"].rsplit("/", 1)[-1], row.get("parent_path", "" if "/" not in row["path"] else row["path"].rsplit("/", 1)[0]), 1 if row.get("is_directory") else 0, row.get("file_size", len(row.get("content") or b"")), sqlite3.Binary(row["content"]) if row.get("content") is not None else None, sqlite3.Binary(row["freq_blob"]) if row.get("freq_blob") else None, row.get("compress_algo", "none"), ), ) conn.commit() finally: conn.close() return db_path def vfs_format_error(): """返回（修复后应当存在的）数据库格式异常类型；未实现时返回 None。""" from sqlite_vfs import core return getattr(core, "VFSFormatError", None) # -------------------------------------------------------------------------- # 打包 / 解包 / CLI 封装 # -------------------------------------------------------------------------- def pack( source_dir, db_path, compress=False, exclude=None, name="test_fs", algorithm=None, quiet_output=True, ): from sqlite_vfs import compression from sqlite_vfs.core import SQLiteVFS from sqlite_vfs.folder_packer import FolderPacker kwargs = {} if algorithm is not None: kwargs["compress_algorithm"] = algorithm vfs = SQLiteVFS(db_path, compress=compress, **kwargs) try: packer = FolderPacker( sqlite_vfs=vfs, exclude_patterns=exclude or [], fs_name=name ) if quiet_output: with quiet(): packer.pack_folder(source_dir) else: packer.pack_folder(source_dir) finally: vfs.close() return db_path def unpack(db_path, out_dir, ignore_timestamps=False): from sqlite_vfs.core import SQLiteVFS from sqlite_vfs.folder_unpacker import FolderUnpacker vfs = SQLiteVFS(db_path) try: unpacker = FolderUnpacker(vfs) try: return unpacker.unpack_filesystem( out_dir, ignore_timestamps=ignore_timestamps ) except TypeError: # 兼容尚未支持该参数的老接口 return unpacker.unpack_filesystem(out_dir) finally: vfs.close() def run_cli(module, *args): """以子进程运行 CLI 模块，返回 CompletedProcess（text 模式）。""" env = dict(os.environ) env["PYTHONPATH"] = SRC_DIR + os.pathsep + env.get("PYTHONPATH", "") env["PYTHONIOENCODING"] = "utf-8" return subprocess.run( [sys.executable, "-m", module, *args], capture_output=True, text=True, encoding="utf-8", errors="replace", env=env, cwd=REPO_ROOT, ) def cli_output(result): """合并 stdout/stderr，便于断言错误信息。""" return (result.stdout or "") + (result.stderr or "")