467 lines
15 KiB
Python
467 lines
15 KiB
Python
|
"""测试公共工具：路径引导、临时目录、目录树构造与比对、打包/解包/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 "")
|