"""SVFS Studio 的逻辑层测试（不需要 PySide6）。 界面靠眼睛验收，但"大小算错、差异分错类、只读判断漏了写语句"这类问题不能靠眼睛， 所以这些纯逻辑都被钉在测试里。测试只 import ``svfs_studio.format`` / ``svfs_studio.analysis``，两个模块都是纯标准库的。 """ import os import sqlite3 import unittest import helpers from sqlite_vfs.core import SQLiteVFS from sqlite_vfs.folder_packer import FolderPacker from svfs_studio import analysis, format as fmt class HumanFormatTest(unittest.TestCase): def test_human_size(self): self.assertEqual(fmt.human_size(None), "-") self.assertEqual(fmt.human_size(0), "0 B") self.assertEqual(fmt.human_size(512), "512 B") self.assertEqual(fmt.human_size(1024), "1.00 KB") self.assertEqual(fmt.human_size(1536), "1.50 KB") self.assertEqual(fmt.human_size(1024 ** 2), "1.00 MB") self.assertEqual(fmt.human_size(-2048), "-2.00 KB") def test_human_count(self): self.assertEqual(fmt.human_count(1234567), "1,234,567") self.assertEqual(fmt.human_count(None), "-") def test_perm_string(self): self.assertEqual(fmt.perm_string(420), "rw-r--r--") # 420 == 0o644 self.assertEqual(fmt.perm_string(493), "rwxr-xr-x") # 493 == 0o755 self.assertEqual(fmt.perm_string(0), "---------") self.assertEqual(fmt.perm_string(None), "-") self.assertEqual(fmt.perm_octal(420), "644") def test_human_time_accepts_db_string_and_datetime(self): self.assertEqual(fmt.human_time("2026-09-14 10:23:45"), "2026-09-14 10:23") self.assertEqual(fmt.human_time(None), "-") self.assertEqual(fmt.human_time(""), "-") self.assertEqual(fmt.human_time("不是时间"), "不是时间") def test_ratio_and_percent(self): self.assertEqual(fmt.ratio_text(50, 100), "50.0%") self.assertEqual(fmt.ratio_text(0, 0), "-") self.assertEqual(fmt.percent_text(25, 100), "25.0%") self.assertEqual(fmt.percent_text(1, 0), "-") def test_sniff_text_decodes_utf8_then_gbk(self): self.assertEqual(fmt.sniff_text("纯文本".encode("utf-8")), (False, "utf-8", "纯文本")) # GBK 的字节不是合法 UTF-8，必须回退到 gbk 而不是显示乱码 binary, encoding, text = fmt.sniff_text("中文内容测试".encode("gbk")) self.assertFalse(binary) self.assertEqual(encoding, "gbk") self.assertEqual(text, "中文内容测试") def test_sniff_text_flags_binary(self): binary, encoding, text = fmt.sniff_text(b"\x00\x01\x02\x03") self.assertTrue(binary) self.assertIsNone(text) self.assertIsNone(encoding) def test_sniff_text_honours_bom(self): binary, encoding, text = fmt.sniff_text(b"\xef\xbb\xbfhi") self.assertFalse(binary) self.assertEqual(encoding, "utf-8-sig") def test_hexdump_layout(self): lines, truncated = fmt.hexdump(b"Hello, SVFS!", limit=8) self.assertTrue(truncated) self.assertEqual(len(lines), 1) self.assertTrue(lines[0].startswith("00000000 48 65 6c 6c 6f 2c 20 53")) self.assertTrue(lines[0].endswith("|Hello, S|")) def test_hexdump_full_line_alignment(self): lines, truncated = fmt.hexdump(bytes(range(16))) self.assertFalse(truncated) self.assertEqual(len(lines), 1) self.assertIn("|................|", lines[0]) # 两行时 ASCII 列要对齐（同一列起） lines, _ = fmt.hexdump(bytes(range(20))) self.assertEqual(len(lines), 2) self.assertEqual(lines[0].index("|"), lines[1].index("|")) self.assertTrue(lines[1].startswith("00000010")) class SnapshotDiffTest(unittest.TestCase): @staticmethod def row(path, is_directory=False, size=0, content_hash=None, mtime="2026-01-01 00:00:00", permissions=420): return { "path": path, "is_directory": is_directory, "file_size": size, "content_hash": content_hash, "modified_time": mtime, "permissions": permissions, } def test_classifies_every_kind(self): old = [ self.row("keep.txt", size=3, content_hash="aaa"), self.row("gone.txt", size=3, content_hash="bbb"), self.row("edit.txt", size=3, content_hash="ccc"), self.row("meta.txt", size=3, content_hash="ddd"), self.row("swap", is_directory=False, size=1, content_hash="eee"), self.row("dir", is_directory=True), ] new = [ self.row("keep.txt", size=3, content_hash="aaa"), self.row("edit.txt", size=5, content_hash="zzz"), self.row("meta.txt", size=3, content_hash="ddd", mtime="2026-02-02 00:00:00"), self.row("swap", is_directory=True), self.row("dir", is_directory=True), self.row("fresh.txt", size=9, content_hash="fff"), ] diff = analysis.snapshot_diff(old, new) self.assertEqual(diff["counts"]["added"], 1) self.assertEqual(diff["added"][0]["path"], "fresh.txt") self.assertEqual(diff["counts"]["removed"], 1) self.assertEqual(diff["removed"][0]["path"], "gone.txt") self.assertEqual([item["path"] for item in diff["content"]], ["edit.txt"]) self.assertEqual(diff["content"][0]["size_delta"], 2) self.assertEqual([item["path"] for item in diff["type"]], ["swap"]) self.assertEqual([item["path"] for item in diff["meta"]], ["meta.txt"]) self.assertEqual(diff["unchanged"], 2) # keep.txt / dir self.assertEqual(diff["total_changed"], 5) def test_directory_mtime_is_ignored(self): """往目录里加文件会改目录 mtime，那不算"变更"（否则差异视图全是噪音）。""" old = [self.row("dir", is_directory=True, mtime="2026-01-01 00:00:00")] new = [self.row("dir", is_directory=True, mtime="2026-03-03 12:00:00")] diff = analysis.snapshot_diff(old, new) self.assertEqual(diff["total_changed"], 0) self.assertEqual(diff["unchanged"], 1) def test_same_size_different_hash_is_content_change(self): old = [self.row("a.bin", size=1024, content_hash="old")] new = [self.row("a.bin", size=1024, content_hash="new")] diff = analysis.snapshot_diff(old, new) self.assertEqual(diff["counts"]["content"], 1) self.assertEqual(diff["content"][0]["size_delta"], 0) def test_permission_only_change_is_meta(self): old = [self.row("x.sh", size=5, content_hash="h", permissions=420)] new = [self.row("x.sh", size=5, content_hash="h", permissions=493)] diff = analysis.snapshot_diff(old, new) self.assertEqual(diff["counts"]["meta"], 1) def test_empty_inputs(self): diff = analysis.snapshot_diff([], []) self.assertEqual(diff["total_changed"], 0) self.assertEqual(diff["unchanged"], 0) class StorageAnalysisTest(unittest.TestCase): def test_storage_summary_accounts_for_each_stage(self): entries = [ {"path": "a.txt", "is_directory": False, "file_size": 100, "content_hash": "h1"}, {"path": "b.txt", "is_directory": False, "file_size": 100, "content_hash": "h1"}, {"path": "dir", "is_directory": True, "file_size": 0, "content_hash": None}, ] blobs = [{"hash": "h1", "size": 100, "stored_size": 40, "compress_algo": "zlib"}] summary = analysis.storage_summary(entries, blobs, db_file_size=12288) self.assertEqual(summary["total_files"], 2) self.assertEqual(summary["total_dirs"], 1) self.assertEqual(summary["logical_size"], 200) self.assertEqual(summary["unique_bytes"], 100) self.assertEqual(summary["dedup_saved"], 100) self.assertEqual(summary["stored_bytes"], 40) self.assertEqual(summary["compression_saved"], 60) self.assertEqual(summary["db_overhead"], 12288 - 40) self.assertEqual(summary["blob_count"], 1) def test_algo_distribution_sorted_by_stored_size(self): blobs = [ {"hash": "a", "size": 10, "stored_size": 10, "compress_algo": "none"}, {"hash": "b", "size": 100, "stored_size": 30, "compress_algo": "zlib"}, {"hash": "c", "size": 100, "stored_size": 20, "compress_algo": "zlib"}, ] distribution = analysis.algo_distribution(blobs) self.assertEqual([item["algo"] for item in distribution], ["zlib", "none"]) self.assertEqual(distribution[0]["stored_size"], 50) self.assertEqual(distribution[0]["count"], 2) def test_duplicate_groups_waste(self): entries = [ {"path": "a/vendor.js", "is_directory": False, "content_hash": "h", "file_size": 60}, {"path": "b/vendor.js", "is_directory": False, "content_hash": "h", "file_size": 60}, {"path": "c/vendor.js", "is_directory": False, "content_hash": "h", "file_size": 60}, {"path": "only.js", "is_directory": False, "content_hash": "k", "file_size": 5}, {"path": "dir", "is_directory": True, "content_hash": None, "file_size": 0}, ] blobs = {"h": {"size": 60, "stored_size": 20, "compress_algo": "zlib"}} groups = analysis.duplicate_groups(entries, blobs) self.assertEqual(len(groups), 1) self.assertEqual(groups[0]["refcount"], 3) self.assertEqual(groups[0]["wasted_bytes"], 120) self.assertEqual(groups[0]["paths"], ["a/vendor.js", "b/vendor.js", "c/vendor.js"]) def test_top_files_carries_blob_info(self): entries = [ {"path": "big.bin", "is_directory": False, "file_size": 900, "content_hash": "h"}, {"path": "small.bin", "is_directory": False, "file_size": 9, "content_hash": "k"}, ] blobs = {"h": {"size": 900, "stored_size": 300, "compress_algo": "zlib"}} top = analysis.top_files(entries, blobs, limit=1) self.assertEqual(len(top), 1) self.assertEqual(top[0]["path"], "big.bin") self.assertEqual(top[0]["stored_size"], 300) self.assertEqual(top[0]["algo"], "zlib") class FakeCursor: def __init__(self, rows): self._rows = rows def fetchone(self): return self._rows[0] class NoDbstatConn: """模拟没编译 dbstat 的 SQLite：只答得出 COUNT(*)。""" def execute(self, sql, *args): if "dbstat" in sql.lower(): raise sqlite3.OperationalError("no such table: dbstat") return FakeCursor([(7,)]) class SqliteInternalsTest(helpers.VFSTestCase): def setUp(self): super().setUp() self.src = helpers.make_tree(self.path("src"), {"a.txt": b"aaa", "sub/b.txt": b"bbb"}) self.db = self.path("p.svfs") helpers.pack(self.src, self.db, compress=True) def test_read_header_parses_real_file(self): header = analysis.read_header(self.db) self.assertTrue(header["valid"]) self.assertEqual(header["page_size"], 4096) self.assertGreater(header["page_count"], 0) self.assertEqual(header["text_encoding_name"], "UTF-8") def test_read_header_rejects_non_sqlite(self): path = self.path("garbage.bin") with open(path, "wb") as handle: handle.write(b"x" * 200) header = analysis.read_header(path) self.assertFalse(header["valid"]) self.assertIn("SQLite", header["error"]) def test_read_header_rejects_tiny_file(self): path = self.path("tiny.bin") with open(path, "wb") as handle: handle.write(b"SQL") self.assertFalse(analysis.read_header(path)["valid"]) def test_internals_reports_occupancy_and_objects(self): vfs = SQLiteVFS(self.db, readonly=True) try: info = analysis.sqlite_internals(vfs.conn, self.db) finally: vfs.close() self.assertTrue(info["ok"], info.get("error")) self.assertIn(info["occupancy_mode"], ("dbstat", "rowcount")) names = {obj["name"] for obj in info["objects"]} self.assertTrue({"files", "blobs", "snapshots", "signatures"} <= names) self.assertEqual(info["total_bytes"], info["pragmas"]["page_size"] * info["pragmas"]["page_count"]) self.assertEqual(info["total_bytes"] - info["free_bytes"], info["used_bytes"]) rows = {row["name"]: row for row in info["occupancy"]} self.assertIsNotNone(rows["files"]["rows"]) if info["occupancy_mode"] == "dbstat": self.assertGreater(rows["files"]["bytes"], 0) def test_occupancy_falls_back_without_dbstat(self): objects = [ {"type": "table", "name": "files", "table": "files", "rootpage": 2, "sql": ""}, {"type": "index", "name": "idx", "table": "files", "rootpage": 3, "sql": ""}, ] occupancy, mode = analysis._occupancy(NoDbstatConn(), objects) self.assertEqual(mode, "rowcount") self.assertEqual([row["name"] for row in occupancy], ["files"]) self.assertEqual(occupancy[0]["rows"], 7) self.assertIsNone(occupancy[0]["bytes"]) class ReadonlySqlTest(unittest.TestCase): def test_accepts_readonly_statements(self): for sql in ( "SELECT * FROM files", "select count(*) from blobs", " WITH x AS (SELECT 1) SELECT * FROM x ", "PRAGMA page_count", "EXPLAIN SELECT 1", "SELECT 1;", ): ok, reason = analysis.check_readonly_sql(sql) self.assertTrue(ok, f"{sql!r} 应被接受，却被拒绝：{reason}") def test_rejects_writes_and_multi_statements(self): for sql in ( "", " ", "INSERT INTO files (path) VALUES ('x')", "DELETE FROM blobs", "DROP TABLE files", "UPDATE snapshots SET name = 'x'", "SELECT 1; DELETE FROM files", "VACUUM", ): ok, reason = analysis.check_readonly_sql(sql) self.assertFalse(ok, f"{sql!r} 应被拒绝") self.assertTrue(reason) class SqlConsoleTest(helpers.VFSTestCase): def setUp(self): super().setUp() self.src = helpers.make_tree( self.path("src"), {"a.txt": b"aaa", "b.txt": b"aaa", "sub/c.txt": b"ccc"} ) self.db = self.path("q.svfs") helpers.pack(self.src, self.db, compress=True) def test_query_and_truncation(self): vfs = SQLiteVFS(self.db, readonly=True) try: columns, rows, truncated = analysis.run_readonly_query( vfs.conn, "SELECT path, file_size FROM files ORDER BY path", limit=2 ) finally: vfs.close() self.assertEqual(columns, ["path", "file_size"]) self.assertEqual(len(rows), 2) self.assertTrue(truncated) def test_write_attempt_fails_at_sqlite_level(self): """就算白名单被绕过，mode=ro 连接也会拒绝写——两道闸都要在。""" vfs = SQLiteVFS(self.db, readonly=True) try: with self.assertRaises(sqlite3.OperationalError): vfs.conn.execute("DELETE FROM files") finally: vfs.close() def test_presets_all_run(self): vfs = SQLiteVFS(self.db, readonly=True) try: for title, sql in analysis.SQL_PRESETS: columns, rows, _truncated = analysis.run_readonly_query(vfs.conn, sql) self.assertTrue(columns, f"预设 {title} 没返回列") finally: vfs.close() def test_refcounts_and_orphans(self): vfs = SQLiteVFS(self.db, readonly=True) try: counts = analysis.refcounts(vfs.conn) hashes = {row["hash"] for row in vfs.list_blobs()} # a.txt 与 b.txt 内容相同，共用一个 blob self.assertTrue(any(value == 2 for value in counts.values())) self.assertEqual(analysis.orphan_blobs(vfs.conn)["count"], 0) self.assertEqual(set(counts), hashes) finally: vfs.close() def test_orphan_detection_on_stray_blob(self): """手工塞一条没人引用的 blob，孤立检测要能抓到（GC 前的预览就靠它）。""" vfs = SQLiteVFS(self.db) try: vfs.conn.execute( "INSERT INTO blobs (hash, size, stored_size, content, freq_blob, compress_algo)" " VALUES ('deadbeef', 123, 123, X'00', NULL, 'none')" ) vfs.conn.commit() finally: vfs.close() vfs = SQLiteVFS(self.db, readonly=True) try: orphans = analysis.orphan_blobs(vfs.conn) self.assertEqual(orphans["count"], 1) self.assertEqual(orphans["size"], 123) finally: vfs.close() def test_snapshot_stats_reports_shared_blobs(self): vfs = SQLiteVFS(self.db, compress=True) try: parent = vfs._active_snapshot() new_id = vfs.create_snapshot("v2", parent_id=parent, copy_entries=True) stats = analysis.snapshot_stats(vfs, new_id) finally: vfs.close() # 整份复制过去的新快照：所有内容都是"别处也在用"的 self.assertEqual(stats["files"], 3) self.assertEqual(stats["shared_blobs"], stats["blob_count"]) self.assertGreater(stats["shared_bytes"], 0) class LocalCompareTest(helpers.VFSTestCase): def setUp(self): super().setUp() self.src = helpers.make_tree( self.path("src"), {"a.txt": b"aaa", "b.txt": b"bbb", "sub/c.txt": b"ccc"} ) self.db = self.path("c.svfs") helpers.pack(self.src, self.db, compress=True) def entries(self): vfs = SQLiteVFS(self.db, readonly=True) try: return vfs.all_entries() finally: vfs.close() def read_pack(self, path): vfs = SQLiteVFS(self.db, readonly=True) try: return vfs.read_file(path) finally: vfs.close() def test_identical_tree_matches(self): result = analysis.local_vs_snapshot_diff(self.src, self.entries()) self.assertTrue(result["ok"], result["counts"]) self.assertEqual(result["counts"]["identical"], 3) self.assertEqual(result["total_local"], 3) def test_detects_each_difference(self): helpers.write_file(os.path.join(self.src, "a.txt"), b"AAA") # 内容不同 helpers.write_file(os.path.join(self.src, "new.txt"), b"new") # 仅本地 os.remove(os.path.join(self.src, "b.txt")) # 仅包内 helpers.write_file(os.path.join(self.src, "sub", "c.txt"), b"ccc") # 内容相同 os.utime(os.path.join(self.src, "sub", "c.txt"), (1, 1)) # 只改了 mtime result = analysis.local_vs_snapshot_diff( self.src, self.entries(), read_content=self.read_pack ) self.assertFalse(result["ok"]) self.assertEqual([item["path"] for item in result["content"]], ["a.txt"]) self.assertEqual([item["path"] for item in result["only_local"]], ["new.txt"]) self.assertEqual([item["path"] for item in result["only_pack"]], ["b.txt"]) self.assertEqual([item["path"] for item in result["meta"]], ["sub/c.txt"]) def test_size_only_mode_needs_no_read_content(self): helpers.write_file(os.path.join(self.src, "a.txt"), b"AAAA") result = analysis.local_vs_snapshot_diff(self.src, self.entries()) self.assertEqual([item["path"] for item in result["content"]], ["a.txt"]) def test_progress_reports_every_local_file(self): seen = [] analysis.local_vs_snapshot_diff( self.src, self.entries(), progress=lambda done, total, path: seen.append((done, total, path)), ) self.assertEqual(len(seen), 3) self.assertEqual([done for done, _total, _p in seen], [1, 2, 3]) self.assertEqual({total for _d, total, _p in seen}, {3}) def test_skips_vcs_directories(self): helpers.write_file(os.path.join(self.src, ".git", "config"), b"[core]") result = analysis.local_vs_snapshot_diff(self.src, self.entries()) self.assertEqual(result["counts"]["only_local"], 0) self.assertTrue(result["ok"]) if __name__ == "__main__": unittest.main()