"""SVFS Studio 的界面测试（offscreen 渲染，不需要显示器）。 界面好不好看没法断言，但这些都是可以钉死的： * 七个标签页都能渲染出来，没有 0 尺寸/被裁掉的文本； * 主题确实是暗色（采样渲染结果的像素算亮度）； * 预览显示的内容与 ``SQLiteVFS.read_file()`` 读出来的字节一致； * 旧格式包会给出横幅，而 SQL 控制台在旧格式包上照样能用； * SQL 控制台拒绝写语句；搜索能过滤树；校验/签名能跑通。 没装 PySide6 时整个模块跳过（GUI 是可选 extra，这不影响库本身）。 """ import os import sys import time import unittest import helpers from svfs_studio.format import short_hash # 必须在创建 QApplication 之前设定，否则会去找真实显示设备 os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") SRC_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src") if SRC_DIR not in sys.path: sys.path.insert(0, SRC_DIR) try: from PySide6.QtCore import QSettings from PySide6.QtWidgets import QApplication, QLabel, QWidget from svfs_studio.app import create_app from svfs_studio.sample import build_demo_folder from svfs_studio.widgets.cards import ElidedLabel from svfs_studio.widgets.shell import MainWindow HAVE_QT = True except ImportError as exc: # pragma: no cover - 取决于环境 HAVE_QT = False IMPORT_ERROR = str(exc) def pump(app, predicate=None, timeout=15.0, interval=0.02, settle=0.08): """转事件循环。 * 给了 ``predicate``：一直转到它成立（或超时），返回它是否成立； * 没给：只把事件循环转 ``settle`` 秒，让布局/定时器/排队信号落地。 （早期版本这里会白等满 timeout，七个标签页就是七个 15 秒。） """ if predicate is None: deadline = time.monotonic() + settle while time.monotonic() < deadline: app.processEvents() time.sleep(interval / 4) app.processEvents() return True deadline = time.monotonic() + timeout while time.monotonic() < deadline: app.processEvents() if predicate(): return True time.sleep(interval) app.processEvents() return bool(predicate()) def walk(widget): yield widget for child in widget.findChildren(QWidget): yield child def luminance(color): return 0.2126 * color.red() + 0.7152 * color.green() + 0.0722 * color.blue() @unittest.skipUnless(HAVE_QT, f"需要 PySide6：{'未安装' if not HAVE_QT else ''}") class StudioUITestCase(helpers.VFSTestCase): """所有界面测试的公共脚手架：一个共享的 QApplication + 每个用例一个窗口。""" @classmethod def setUpClass(cls): # 把 QSettings 关进临时目录：测试不该读用户真实的窗口几何/最近文件， # 否则"上次把窗口拖得很窄"会让某个用例莫名其妙地失败，也不该往注册表里写东西。 import tempfile settings_dir = tempfile.mkdtemp(prefix="svfs_studio_settings_") QSettings.setDefaultFormat(QSettings.Format.IniFormat) QSettings.setPath( QSettings.Format.IniFormat, QSettings.Scope.UserScope, settings_dir ) cls.app = QApplication.instance() or create_app([sys.argv[0]]) cls.app.setQuitOnLastWindowClosed(False) def setUp(self): super().setUp() self.window = MainWindow() self.window.resize(1400, 900) self.window.show() pump(self.app, interval=0.005) self.progress_events = [] self.window.client.progress.connect( lambda job, done, total, label: self.progress_events.append((job, done, total)) ) def tearDown(self): try: self.window.close_package(quiet=True) self.window._thread.quit() self.window._thread.wait(5000) self.window.close() finally: pump(self.app, interval=0.005) super().tearDown() # ------------------------------------------------------------------ def demo_package(self, target=None): """打一个演示包（与界面里「打个示例包看看」同一份数据）。""" target = target or self.path("demo.svfs") source = self.path("src_tree") os.makedirs(source, exist_ok=True) build_demo_folder(source) helpers.pack(source, target, compress=True) return target, source def load(self, path): """通过真实路径（后台线程 + 信号）打开一个包，并等到界面拿到数据。""" self.window.open_package(path) ok = pump(self.app, lambda: self.window.payload is not None, timeout=20) self.assertTrue(ok, "打开包超时") return self.window.payload def run_op(self, op, payload=None, timeout=30): """在真实的事件循环里跑一个后台操作，返回结果。 先等界面空闲再提交——独占车道一次只跑一个任务，界面上刚触发的"刷新"之类 还没结束时提交会被明确拒绝（这是刻意设计的，不是 bug）。 """ pump(self.app, lambda: not self.window.client.busy, timeout=20) box = {} def done(result): box["result"] = result def fail(title, detail): box["error"] = f"{title}：{detail}" self.window.client.submit(op, payload or {}, on_done=done, on_fail=fail, label=op) pump(self.app, lambda: "result" in box or "error" in box, timeout=timeout) self.assertNotIn("error", box, box.get("error")) return box["result"] def assert_rendered_cleanly(self, label): """渲染自查：没有 0 尺寸部件、没有该省略却没省略的文本、整体是暗色。""" image = self.window.grab().toImage() self.assertFalse(image.isNull(), f"{label}: 渲染结果为空") # 背景必须是暗的（左上角是侧栏，右下角是内容区边缘） for name, (x, y) in ( ("左上", (4, 4)), ("右下", (image.width() - 4, image.height() - 4)), ): value = luminance(image.pixelColor(x, y)) self.assertLess(value, 110, f"{label}: {name}像素亮度 {value:.0f}，暗色主题没生效") broken = [ widget for widget in walk(self.window) if widget.isVisible() and (widget.width() <= 0 or widget.height() <= 0) # 空的表头/滚动视口在 Qt 里本来就是 0 尺寸 and not widget.inherits("QHeaderView") and widget.objectName() != "qt_scrollarea_viewport" ] self.assertEqual( [(type(w).__name__, w.objectName()) for w in broken], [], f"{label}: 有部件被压成 0 尺寸" ) clipped = [] for widget in walk(self.window): if not isinstance(widget, QLabel) or isinstance(widget, ElidedLabel): continue if not widget.isVisible() or not widget.text() or widget.wordWrap(): continue if widget.sizeHint().width() > widget.width() + 2: clipped.append(widget.text()[:30]) self.assertEqual(clipped, [], f"{label}: 这些文本被裁掉了") class RenderTest(StudioUITestCase): def test_recent_list_remembers_opened_packages(self): target, _source = self.demo_package() self.load(target) self.assertEqual(self.window._recent_paths()[:1], [os.path.abspath(target)]) # 重复打开不该在列表里出现两次 self.load(target) self.assertEqual(self.window._recent_paths().count(os.path.abspath(target)), 1) self.window._fill_recent_menu() labels = [ action.text() for action in self.window.recent_menu.actions() if action.text() ] self.assertTrue(any(os.path.basename(target) in text for text in labels), labels) def test_all_tabs_render_cleanly(self): target, _source = self.demo_package() self.load(target) tabs = self.window.tabs self.assertEqual(tabs.count(), 7) for index in range(tabs.count()): tabs.setCurrentIndex(index) pump(self.app, interval=0.02) self.assert_rendered_cleanly(tabs.tabText(index)) def test_empty_state_renders_cleanly(self): self.assert_rendered_cleanly("空状态页") def test_overview_numbers_match_library(self): target, _source = self.demo_package() payload = self.load(target) overview = self.window._views["overview"] summary = payload["summary"] self.assertIn(str(summary["total_files"]), overview.card_files._value.text()) self.assertIn("份", overview.card_unique._value.text()) # 演示目录里 assets/vendor.js 与 admin/vendor.js 内容相同 → 必然有去重收益 self.assertGreater(summary["dedup_saved"], 0) self.assertGreater(summary["compression_saved"], 0) segments = overview.savings._segments self.assertEqual(len(segments), 3) total = sum(value for _label, value, _color in segments) self.assertEqual(total, summary["logical_size"], "堆叠条三段之和必须等于逻辑大小") def test_storage_tab_lists_every_blob(self): target, _source = self.demo_package() payload = self.load(target) self.window.goto_tab("storage") pump(self.app, interval=0.02) storage = self.window._views["storage"] self.assertEqual(storage.table.table.rowCount(), len(payload["blobs"])) self.assertEqual(len(payload["duplicates"]), 1) self.assertEqual(payload["duplicates"][0]["refcount"], 2) class BrowserTest(StudioUITestCase): def test_preview_content_matches_library(self): target, _source = self.demo_package() payload = self.load(target) browser = self.window._views["browser"] self.window.goto_tab("browser") path = "assets/vendor.js" browser.select_path(path) pump(self.app, lambda: bool(browser.text_view.toPlainText()), timeout=15) self.assertTrue(pump(self.app, lambda: "解码" in browser.preview_note.text(), timeout=15)) from sqlite_vfs.core import SQLiteVFS vfs = SQLiteVFS(target, readonly=True) try: expected = vfs.read_file(path) finally: vfs.close() shown = browser.text_view.toPlainText() self.assertTrue(expected.startswith(shown.encode("utf-8")) or shown.startswith("export const")) self.assertEqual(shown, expected.decode("utf-8")[: len(shown)]) self.assertIn(path, browser.path_label.text()) def test_detail_card_lists_duplicate_paths(self): target, _source = self.demo_package() self.load(target) browser = self.window._views["browser"] # 必须切到这一页：标签页没被选中的话整棵子树 isVisible() 都是 False self.window.goto_tab("browser") browser.select_path("assets/vendor.js") self.assertTrue( pump(self.app, lambda: browser.other_card.isVisible(), timeout=15), "同内容的其它路径没有显示出来", ) texts = [ browser.other_list.itemAt(index).widget().text() for index in range(browser.other_list.count()) if browser.other_list.itemAt(index).widget() is not None ] self.assertIn("admin/vendor.js", texts) def test_hex_view_renders_bytes(self): target, _source = self.demo_package() self.load(target) browser = self.window._views["browser"] browser.select_path("index.html") pump(self.app, lambda: bool(browser.text_view.toPlainText()), timeout=15) browser.mode_box.setCurrentIndex(1) # 十六进制 pump(self.app, lambda: bool(browser.hex_view.toPlainText()), timeout=15) first = browser.hex_view.toPlainText().splitlines()[0] self.assertTrue(first.startswith("00000000"), first) self.assertIn("|", first) def test_search_filters_the_tree(self): target, _source = self.demo_package() self.load(target) browser = self.window._views["browser"] before = browser.proxy.rowCount() browser.search.setText("vendor") pump(self.app, interval=0.02) self.assertLess(browser.proxy.rowCount(), before) self.assertGreater(browser.proxy.rowCount(), 0) browser.search.setText("") pump(self.app, interval=0.02) self.assertEqual(browser.proxy.rowCount(), before) class SnapshotViewTest(StudioUITestCase): def two_snapshot_package(self): """造一个有两版历史的包：v2 里改一个文件、删一个文件、加一个文件。""" from sqlite_vfs.core import SQLiteVFS source = self.path("tree") helpers.make_tree( source, {"keep.txt": b"keep", "edit.txt": b"old", "gone.txt": b"gone"} ) db = self.path("snap.svfs") helpers.pack(source, db, compress=True) vfs = SQLiteVFS(db, compress=True) try: parent = vfs._active_snapshot() vfs.create_snapshot("v2", parent_id=parent, copy_entries=True) helpers.write_file(os.path.join(source, "edit.txt"), b"new-content") helpers.write_file(os.path.join(source, "fresh.txt"), b"fresh") os.remove(os.path.join(source, "gone.txt")) with helpers.quiet(): from sqlite_vfs.folder_packer import FolderPacker FolderPacker(vfs, fs_name="snap").pack_folder( source, incremental=True, rehash=True ) ids = [row["id"] for row in vfs.list_snapshots()] finally: vfs.close() return db, ids def test_snapshot_list_and_diff(self): db, ids = self.two_snapshot_package() self.load(db) view = self.window._views["snapshots"] self.window.goto_tab("snapshots") pump(self.app, interval=0.02) self.assertEqual(view.tree.topLevelItemCount(), len(ids)) # 默认比较"最新两版" index_a = view.combo_a.findData(ids[0]) index_b = view.combo_b.findData(ids[1]) self.assertGreaterEqual(index_a, 0) self.assertGreaterEqual(index_b, 0) view.combo_a.setCurrentIndex(index_a) view.combo_b.setCurrentIndex(index_b) self.assertTrue( pump(self.app, lambda: bool(view.diff_tree.topLevelItemCount()), timeout=20), "差异树没有渲染出来", ) counts = view._diff["counts"] self.assertEqual(counts["added"], 1) # fresh.txt self.assertEqual(counts["removed"], 1) # gone.txt self.assertEqual(counts["content"], 1) # edit.txt self.assertEqual(view._diff["unchanged"], 1) # keep.txt self.assertIn("3 处变化", view.diff_badge.text()) def test_selecting_snapshot_switches_the_tree(self): db, ids = self.two_snapshot_package() self.load(db) self.assertEqual(self.window.payload["snapshot_id"], ids[1]) self.window.select_snapshot(ids[0]) self.assertTrue( pump(self.app, lambda: self.window.payload["snapshot_id"] == ids[0], timeout=20) ) browser = self.window._views["browser"] # v1 里 gone.txt 还在，v2 里没有 self.assertTrue(browser.model.find("gone.txt").isValid()) self.assertFalse(browser.model.find("fresh.txt").isValid()) class IntegrityViewTest(StudioUITestCase): def test_verify_and_sign_roundtrip(self): target, _source = self.demo_package() self.load(target) view = self.window._views["integrity"] self.window.goto_tab("integrity") result = self.run_op("verify", {"check_content": True}) self.assertTrue(result["report"]["ok"], result["report"]["errors"]) view._on_verify(result) pump(self.app, interval=0.02) self.assertEqual(view.result_badge.text(), "通过") self.assertIn("逐文件重算哈希 6", view.result_detail.text()) # 签名 → 验签 view.key_input.setText("测试密钥") view.key_id_input.setText("ci") view._sign() self.assertTrue( pump(self.app, lambda: bool(view.sign_note.text()), timeout=20) ) self.assertIn("已签名", view.sign_note.text()) verify = self.run_op( "verify_signature", {"key": "测试密钥", "key_id": "ci"}, ) self.assertTrue(verify["report"]["ok"], verify["report"]["errors"]) wrong = self.run_op("verify_signature", {"key": "别的密钥"}) self.assertFalse(wrong["report"]["ok"]) def test_signature_shows_up_in_table_after_refresh(self): target, _source = self.demo_package() self.load(target) self.run_op("sign", {"key": "k", "key_id": "id-1"}) self.window.refresh() self.assertTrue( pump(self.app, lambda: bool(self.window.payload.get("signatures")), timeout=20) ) view = self.window._views["integrity"] view._fill_signatures(self.window.payload) self.assertEqual(view.signature_table.table.rowCount(), 1) self.assertEqual(view.signature_table.table.item(0, 1).text(), "id-1") class InternalsViewTest(StudioUITestCase): def test_internals_tables_populated(self): target, _source = self.demo_package() self.load(target) self.window.goto_tab("internals") info = self.run_op("internals", {}) view = self.window._views["internals"] view._on_internals(info) pump(self.app, interval=0.02) self.assertTrue(info["header"]["valid"]) self.assertEqual(info["header"]["page_size"], 4096) self.assertEqual(info["occupancy_mode"], "dbstat") self.assertIn("SQLite format 3", view.header_badge.text()) self.assertGreater(view.table.table.rowCount(), 3) names = { view.table.table.item(row, 1).text() for row in range(view.table.table.rowCount()) } self.assertTrue({"files", "blobs", "snapshots"} <= names) def test_sql_console_runs_select_and_rejects_write(self): target, _source = self.demo_package() self.load(target) view = self.window._views["internals"] result = self.run_op("run_sql", {"sql": "SELECT count(*) AS n FROM files"}) view._on_result(result) pump(self.app, interval=0.02) self.assertEqual(result["columns"], ["n"]) self.assertEqual(view.result_table.rowCount(), 1) self.assertIn("返回 1 行", view.sql_note.text()) view.sql_edit.setPlainText("DELETE FROM files") view._run_sql() pump(self.app, lambda: "只读" in view.sql_note.text(), timeout=10) self.assertIn("只读控制台只允许", view.sql_note.text()) # 白名单挡不住的话，mode=ro 连接也会挡——两层都要在 view.sql_edit.setPlainText("SELECT * FROM files") view._run_sql() pump(self.app, lambda: "返回" in view.sql_note.text(), timeout=10) class LegacyPackageTest(StudioUITestCase): def legacy_path(self): candidate = os.path.join( os.path.dirname(SRC_DIR), "appcation.svfs" ) if not os.path.exists(candidate): self.skipTest("仓库里没有 appcation.svfs") return candidate def test_legacy_package_shows_banner_but_sql_still_works(self): path = self.legacy_path() payload = self.load(path) self.assertEqual(payload.get("legacy"), 1) self.assertTrue(self.window._banner_wrap.isVisible()) self.assertIn("schema v1", self.window.banner._title.text()) self.assertIn("去迁移", self.window.banner._button.text()) # 旧格式下 SQL 与内部结构仍然可用 result = self.run_op("run_sql", {"sql": "SELECT count(*) AS n FROM files"}) self.assertEqual(result["rows"][0][0], 39) info = self.run_op("internals", {}) self.assertTrue(info["header"]["valid"]) # 而需要 v3 结构的操作会明确报错，而不是静默返回空 with self.assertRaises(Exception) as caught: self.window.client.call_blocking("entries", {}) self.assertIn("旧格式", str(caught.exception)) def test_migrate_from_banner_entry(self): path = self.legacy_path() self.load(path) self.window._open_migrate_for_current() pump(self.app, interval=0.02) ops = self.window._views["ops"] self.assertEqual(ops.stack.currentIndex(), 2) self.assertEqual(ops.migrate_source.text(), path) self.assertIn("v1", ops.migrate_detect.text()) output = self.path("migrated.svfs") result = self.run_op( "migrate", {"path": path, "backup": False, "output": output}, timeout=60 ) self.assertEqual(result["from_version"], 1) self.assertEqual(result["to_version"], 3) self.assertTrue(os.path.exists(output)) ops._on_migrate_done(result) # 迁移完会自动打开新包；要等到界面真的切换到新包为止 # （原来的旧格式 payload 还在，所以不能用 "payload 不是 None" 当判据） self.assertTrue( pump( self.app, lambda: self.window.payload.get("legacy") is None and self.window.payload.get("path") == output, timeout=20, ), f"没等到界面切到迁移后的包：{self.window.payload.get('path')}", ) self.assertEqual(self.window.payload["stats"]["schema_version"], 3) class OpsViewTest(StudioUITestCase): def test_pack_through_the_ui_reports_progress(self): source = self.path("tree") helpers.make_tree(source, {"a.txt": b"a" * 5000, "sub/b.txt": b"b" * 5000}) target = self.path("packed.svfs") ops = self.window._views["ops"] ops.pack_source.setText(source) ops.pack_target.setText(target) ops.pack_compress.setChecked(True) ops._run_pack() self.assertTrue(pump(self.app, lambda: not self.window.client.busy, timeout=30)) self.assertTrue(os.path.exists(target)) jobs = {job for job, _done, _total in self.progress_events} self.assertIn("打包", jobs) self.assertTrue( any(total and done == total for _job, done, total in self.progress_events), f"进度没有走完：{self.progress_events[-3:]}", ) # 打完自动打开，界面上应该看到这两个文件 self.assertTrue(pump(self.app, lambda: self.window.payload is not None, timeout=20)) browser = self.window._views["browser"] self.assertTrue(browser.model.find("a.txt").isValid()) self.assertTrue(browser.model.find("sub/b.txt").isValid()) def test_unpack_then_compare_reports_identical(self): target, source = self.demo_package() self.load(target) output = self.path("exported") result = self.run_op("unpack", {"output": output}, timeout=30) self.assertEqual(result["kind"], "filesystem") self.assertGreater(result["count"], 0) payload = self.run_op("compare_local", {"local_root": source}, timeout=30) self.assertTrue(payload["result"]["ok"], payload["result"]["counts"]) def test_compare_detects_a_changed_file(self): target, source = self.demo_package() self.load(target) helpers.write_file(os.path.join(source, "app.js"), b"console.log('changed');\n") payload = self.run_op("compare_local", {"local_root": source}, timeout=30) counts = payload["result"]["counts"] self.assertEqual(counts["content"], 1) self.assertTrue(counts["only_pack"] + counts["identical"] >= 4) def test_gc_and_recount_via_ops(self): target, _source = self.demo_package() self.load(target) # 造一份没人引用的内容 from sqlite_vfs.core import SQLiteVFS vfs = SQLiteVFS(target) try: vfs.conn.execute( "INSERT INTO blobs (hash, size, stored_size, content, freq_blob, compress_algo)" " VALUES ('deadbeef', 100, 100, X'00', NULL, 'none')" ) vfs.conn.commit() finally: vfs.close() self.window.refresh() self.assertTrue(pump(self.app, lambda: self.window.payload["orphans"]["count"] == 1, timeout=20)) result = self.run_op("gc", {}) self.assertEqual(result["removed"], 1) self.assertEqual(result["after"]["count"], 0) recount = self.run_op("recount", {}) self.assertEqual(recount["recount"]["files"], 6) def test_publish_and_fetch_roundtrip(self): target, source = self.demo_package() self.load(target) out_dir = self.path("public") published = self.run_op( "publish", {"out_dir": out_dir, "shard": 2}, timeout=30 ) self.assertGreater(published["objects"], 0) self.assertTrue(os.path.exists(published["manifest"])) # 清单必须覆盖当前快照里的每一个条目（文件 + 目录） self.assertEqual(len(published["entries"]), len(self.window.payload["entries"])) fetched = self.run_op( "fetch", {"manifest": published["manifest"], "out_dir": self.path("back")}, timeout=60 ) self.assertEqual(fetched["files"], 6) self.assertEqual(fetched["verified"], published["objects"]) def test_fetch_result_matches_source_bytes(self): target, source = self.demo_package() self.load(target) out_dir = self.path("public") published = self.run_op("publish", {"out_dir": out_dir}, timeout=30) back = self.path("back") self.run_op("fetch", {"manifest": published["manifest"], "out_dir": back}, timeout=60) helpers.assert_tree_equal(self, helpers.read_tree(source), helpers.read_tree(back)) class OddFileTest(StudioUITestCase): """打开"不是 SVFS 包"的文件：既不能崩，也不能假装打开成功。""" def test_plain_sqlite_db_degrades_to_internals(self): import sqlite3 path = self.path("plain.db") connection = sqlite3.connect(path) connection.execute("CREATE TABLE t (a)") connection.execute("INSERT INTO t VALUES (1)") connection.commit() connection.close() payload = self.load(path) self.assertIsNone(payload.get("legacy"), "普通 SQLite 库不该被当成旧格式 SVFS") self.assertTrue(self.window._banner_wrap.isVisible()) self.assertIn("不是", self.window.banner._title.text()) # 内部结构与 SQL 依然可用——正好用来看清"这到底是什么文件" result = self.run_op("run_sql", {"sql": "SELECT name FROM sqlite_master"}) self.assertEqual(result["rows"], [("t",)]) info = self.run_op("internals", {}) self.assertTrue(info["header"]["valid"]) def test_non_database_file_reports_clearly(self): path = self.path("notes.txt") helpers.write_file(path, b"hello, definitely not a database\n" * 10) self.window.open_package(path) self.assertTrue( pump(self.app, lambda: self.window._banner_wrap.isVisible(), timeout=20) ) self.assertIn("无法打开", self.window.banner._title.text()) self.assertIn("也不是能读的 SQLite 数据库", self.window.banner._detail.text()) # 失败之后不能留下上一个包的影子数据 self.assertIsNone(self.window.payload) self.assertEqual(self.window.pages.currentIndex(), 0) def test_failed_open_clears_previous_package(self): target, _source = self.demo_package() self.load(target) self.assertIsNotNone(self.window.payload) bad = self.path("bad.bin") helpers.write_file(bad, b"\x00\x01\x02 not a db") self.window.open_package(bad) self.assertTrue(pump(self.app, lambda: self.window.payload is None, timeout=20)) overview = self.window._views["overview"] self.assertEqual(overview.info._row, 0, "概览页还留着上一个包的信息") class NewPackageTest(StudioUITestCase): """新建空包，再往里增量打文件——这是"从零开始用这个格式"的最短路径。""" def test_new_package_is_a_valid_empty_v3_package(self): target = self.path("fresh.svfs") self.assertFalse(os.path.exists(target)) payload = self.run_op("new_package", {"path": target, "name": "从零开始"}) self.assertTrue(payload.get("created")) self.assertEqual(payload["stats"]["name"], "从零开始") self.assertEqual(payload["stats"]["schema_version"], 3) self.assertEqual(payload["summary"]["total_files"], 0) self.assertEqual(payload["entries"], []) self.assertEqual(payload["blobs"], []) self.assertEqual(len(payload["snapshots"]), 1) # 空快照也要有摘要，而且校验要通过（不是"未计算摘要"那种半成品） self.assertTrue(payload["stats"]["digest"]) report = self.run_op("verify", {}) self.assertTrue(report["report"]["ok"], report["report"]["errors"]) def test_ui_flow_create_then_pack_into_it(self): target = self.path("collect.svfs") self.window.new_package(target) self.assertTrue( pump(self.app, lambda: self.window.payload is not None, timeout=20) ) self.assertEqual(self.window.payload["summary"]["total_files"], 0) self.assertIn("空包已建好", self.window.banner._title.text()) self.assertTrue(self.window._banner_wrap.isVisible()) # 从横幅的入口跳到工具箱：应该预填目标文件并勾好增量 self.window._focus_pack_into(target) pump(self.app, interval=0.02) ops = self.window._views["ops"] self.assertEqual(self.window.tabs.currentIndex(), self.window._tab_index["ops"]) self.assertEqual(ops.stack.currentIndex(), 0) self.assertEqual(ops.pack_target.text(), target) self.assertTrue(ops.pack_incremental.isChecked()) # 真打一轮进去：空包 → 有内容的包 source = self.path("drop") helpers.make_tree(source, {"hello.txt": b"hi", "sub/two.txt": b"two"}) ops.pack_source.setText(source) ops.pack_name.setText("从零开始") ops._run_pack() self.assertTrue( pump( self.app, lambda: self.window.payload is not None and self.window.payload["summary"]["total_files"] == 2, timeout=30, ), f"文件没打进空包：{(self.window.payload or {}).get('summary')}", ) browser = self.window._views["browser"] self.window.goto_tab("browser") self.assertTrue(browser.model.find("hello.txt").isValid()) self.assertTrue(browser.model.find("sub/two.txt").isValid()) self.assertEqual(self.window.payload["stats"]["name"], "从零开始") # 再打一轮，这次填上快照名：增量 + 快照名 = 新开一版历史（只写 delta） helpers.write_file(os.path.join(source, "third.txt"), b"three") ops.pack_snapshot.setText("v2") ops._run_pack() self.assertTrue( pump( self.app, lambda: len((self.window.payload or {}).get("snapshots") or []) == 2 and (self.window.payload or {}).get("summary", {}).get("total_files") == 3, timeout=30, ), "带快照名的增量打包没有产生第二版历史", ) # 第一版的快照名沿用建包时的默认名（取自文件名 collect.svfs → "collect"）； # 第二版用了显式的快照名 v2。断言"多了一版、且名字对得上"。 names = {row["name"] for row in self.window.payload["snapshots"]} self.assertEqual(names, {"collect", "v2"}) # 新版是基于旧版复制出来的：旧版内容还在，切换回去能看到 default_id = next( row["id"] for row in self.window.payload["snapshots"] if row["name"] == "collect" ) self.window.select_snapshot(default_id) self.assertTrue( pump( self.app, lambda: (self.window.payload or {}).get("summary", {}).get("total_files") == 2, timeout=20, ), "切回第一版应该还是 2 个文件（历史没有被动过）", ) # 回到最新版，第三个文件在那里 self.window.select_snapshot( max(row["id"] for row in self.window.payload["snapshots"]) ) self.assertTrue( pump( self.app, lambda: (self.window.payload or {}).get("summary", {}).get("total_files") == 3, timeout=20, ), "最新版应该有 3 个文件", ) def test_new_package_refuses_to_overwrite_content(self): target, _source = self.demo_package() before = os.path.getsize(target) with self.assertRaises(Exception) as caught: self.window.client.call_blocking("new_package", {"path": target}) self.assertIn("已经存在", str(caught.exception)) self.assertEqual(os.path.getsize(target), before, "已有包被改动了") def test_rename_package(self): target, _source = self.demo_package() self.load(target) payload = self.run_op("rename_package", {"name": "改名后的包"}) self.assertEqual(payload["stats"]["name"], "改名后的包") # 只动名字，条目与内容不能受影响 self.assertEqual(payload["summary"]["total_files"], 6) self.assertEqual(len(payload["snapshots"]), 1) self.assertEqual(payload["summary"]["stored_bytes"], self.window.payload["summary"]["stored_bytes"]) ops = self.window._views["ops"] ops._on_rename_done(payload) self.assertIn("改名后的包", self.window.payload["stats"]["name"]) with self.assertRaises(Exception) as caught: self.window.client.call_blocking("rename_package", {"name": " "}) self.assertIn("不能为空", str(caught.exception)) def test_pack_full_into_existing_package_refuses_new_snapshot(self): """全量打包会清空整库，所以"在新快照里累加"这种组合必须被拒绝而不是静默清库。""" target, source = self.demo_package() self.load(target) with self.assertRaises(Exception) as caught: self.window.client.call_blocking( "pack", { "source": source, "target": target, "incremental": False, "snapshot_name": "v2", }, ) self.assertIn("增量打包", str(caught.exception)) class EditTest(StudioUITestCase): """包内编辑：建目录 / 加文件 / 改名 / 删除。 这些是写路径，断言两条：动作生效了，而且没有偷偷改到别的东西（引用计数、快照数）。 """ def browser(self, target): self.load(target) self.window.goto_tab("browser") pump(self.app, interval=0.02) return self.window._views["browser"] def wait_entries(self, count): """等到当前快照的条目数变成 count。""" return pump( self.app, lambda: (self.window.payload or {}).get("summary", {}).get("total_entries") == count, timeout=20, ) def test_edit_buttons_follow_package_state(self): browser = self.window._views["browser"] self.assertFalse(browser.add_button.isEnabled(), "没打开包时不该能编辑") target, _source = self.demo_package() browser = self.browser(target) self.assertTrue(browser.add_button.isEnabled()) self.assertTrue(browser.mkdir_button.isEnabled()) # 没选中任何条目时，改名/删除不可用 self.assertFalse(browser.rename_button.isEnabled()) self.assertFalse(browser.delete_button.isEnabled()) def test_edit_keeps_package_verifiable(self): """改完库要能立刻校验通过：否则用户会以为库坏了。 靠的是编辑动作结束时调一次 recount()（刷新快照统计 + 摘要）， 与打包结束时做的是同一件事。 """ target, _source = self.demo_package() browser = self.browser(target) self.assertTrue(self.run_op("verify", {"check_content": True})["report"]["ok"]) local = self.path("extra.txt") helpers.write_file(local, "新增内容\n".encode("utf-8")) before = self.window.payload["summary"]["total_entries"] browser.add_local([local], dest="", replace=False) self.assertTrue(self.wait_entries(before + 1)) report = self.run_op("verify", {"check_content": True})["report"] self.assertTrue(report["ok"], report["errors"]) self.assertEqual(report["files_checked"], 7) # 快照行上的统计也同步了（不是只更新了界面） from sqlite_vfs.core import SQLiteVFS vfs = SQLiteVFS(target, readonly=True) try: self.assertEqual(vfs.get_stats()["total_files"], 7) finally: vfs.close() def test_edit_invalidates_existing_signature_and_says_so(self): target, _source = self.demo_package() browser = self.browser(target) self.run_op("sign", {"key": "k1", "key_id": "before-edit"}) self.assertTrue(self.run_op("verify_signature", {"key": "k1"})["report"]["ok"]) local = self.path("after.txt") helpers.write_file(local, b"after") before = self.window.payload["summary"]["total_entries"] browser.add_local([local], dest="", replace=False) self.assertTrue(self.wait_entries(before + 1)) # 摘要变了，旧签名必须验不过——这是设计如此，界面上也会提示 self.assertFalse( self.run_op("verify_signature", {"key": "k1"})["report"]["ok"], "改动之后旧签名不该还能验过", ) browser._after_edit(self.window.payload, "已加入 after.txt") self.assertIn("签名已失效", self.window.status_label.full_text()) def test_create_add_rename_delete_roundtrip(self): target, _source = self.demo_package() browser = self.browser(target) before = self.window.payload["summary"]["total_entries"] # 10 browser.create_directory("notes", parent="") self.assertTrue(self.wait_entries(before + 1), "建目录没生效") self.assertTrue(browser.model.find("notes").isValid()) local = self.path("note.txt") helpers.write_file(local, "编辑器加进来的内容\n".encode("utf-8")) browser.add_local([local], dest="notes", replace=False) self.assertTrue(self.wait_entries(before + 2), "加文件没生效") self.assertTrue(browser.model.find("notes/note.txt").isValid()) browser.rename_to("notes/note.txt", "renamed.txt") # 改名不改变条目数，所以要等"新路径出现"而不是等计数 self.assertTrue( pump( self.app, lambda: browser.model.find("notes/renamed.txt").isValid() and not browser.model.find("notes/note.txt").isValid(), timeout=20, ), "改名没生效", ) browser.delete_path("notes/renamed.txt") self.assertTrue(self.wait_entries(before + 1), "删除没生效") self.assertFalse(browser.model.find("notes/renamed.txt").isValid()) # 建目录再接上：删除目录会把子项一起带走 browser.delete_path("notes") self.assertTrue(self.wait_entries(before), "删目录没生效") self.assertFalse(browser.model.find("notes").isValid()) def test_added_file_content_matches_disk(self): target, _source = self.demo_package() browser = self.browser(target) local = self.path("payload.bin") data = bytes(range(256)) * 8 helpers.write_file(local, data) browser.add_local([local], dest="", replace=False) self.assertTrue(pump(self.app, lambda: browser.model.find("payload.bin").isValid(), timeout=20)) from sqlite_vfs.core import SQLiteVFS vfs = SQLiteVFS(target, readonly=True) try: self.assertEqual(vfs.read_file("payload.bin"), data) finally: vfs.close() def test_adding_a_folder_brings_the_whole_tree(self): target, _source = self.demo_package() browser = self.browser(target) folder = self.path("drop") helpers.make_tree( folder, {"a.txt": b"a", "sub/b.txt": b"b", "sub/deep/c.txt": b"c"} ) browser.add_local([folder], dest="incoming", replace=False) self.assertTrue( pump( self.app, lambda: browser.model.find("incoming/drop/sub/deep/c.txt").isValid(), timeout=20, ), "整棵目录没加进去", ) paths = {row["path"] for row in self.window.payload["entries"]} self.assertTrue( { "incoming/drop", "incoming/drop/a.txt", "incoming/drop/sub/b.txt", "incoming/drop/sub/deep/c.txt", } <= paths ) def test_name_conflict_is_reported_and_can_be_replaced(self): target, _source = self.demo_package() browser = self.browser(target) local = self.path("app.js") helpers.write_file(local, "console.log('替换后的内容');\n".encode("utf-8")) # 不替换：路径撞上包里的 app.js，结果应该是"跳过了" payload = self.run_op( "add_paths", {"local_paths": [local], "dest_dir": "", "replace_existing": False}, ) self.assertEqual(payload["conflicts"], ["app.js"]) self.assertEqual(payload["added_files"], []) # 冲突检测在 GUI 侧也要能提前算出来（按钮路径用它来决定要不要问） self.assertEqual(browser._conflicting_paths([local], ""), ["app.js"]) # 替换：内容真的换了，而且条目数不变（还是覆盖那个路径） before = len(self.window.payload["entries"]) payload = self.run_op( "add_paths", {"local_paths": [local], "dest_dir": "", "replace_existing": True}, ) self.assertEqual(payload["added_files"], ["app.js"]) self.assertEqual(len(payload["entries"]), before) from sqlite_vfs.core import SQLiteVFS vfs = SQLiteVFS(target, readonly=True) try: self.assertIn(b"\xe6\x9b\xbf\xe6\x8d\xa2", vfs.read_file("app.js")) finally: vfs.close() def test_delete_directory_takes_children_and_collects_orphans(self): target, _source = self.demo_package() browser = self.browser(target) before = self.window.payload["summary"]["total_entries"] browser.delete_path("assets") self.assertTrue(pump(self.app, lambda: not browser.model.find("assets").isValid(), timeout=20)) # assets 下有 logo.svg 与 vendor.js，连带目录行一共少 3 个条目 self.assertEqual(self.window.payload["summary"]["total_entries"], before - 3) # vendor.js 的内容还被 admin/vendor.js 引用着，所以那份 blob 必须留着 from sqlite_vfs.core import SQLiteVFS vfs = SQLiteVFS(target, readonly=True) try: hashes = {row["hash"] for row in vfs.list_blobs()} self.assertIn( vfs.get_file_info("admin/vendor.js")["content_hash"], hashes, "还有路径引用这份内容，不该被回收", ) finally: vfs.close() self.assertEqual(self.window.payload["orphans"]["count"], 0) def test_rename_rejects_collision_and_slash(self): target, _source = self.demo_package() browser = self.browser(target) self.assertIsNone(browser.rename_to("app.js", "index.html"), "重名不该放行") self.assertIsNone(browser.rename_to("app.js", "sub/app.js"), "带斜杠的名字不该放行") with self.assertRaises(Exception) as caught: self.run_op("rename_path", {"source": "app.js", "dest": "index.html"}) self.assertIn("已存在", str(caught.exception)) def test_edit_ops_refuse_legacy_package(self): legacy = os.path.join(os.path.dirname(SRC_DIR), "appcation.svfs") if not os.path.exists(legacy): self.skipTest("仓库里没有 appcation.svfs") self.load(legacy) for op, payload in ( ("mkdir", {"path": "x"}), ("rename_path", {"source": "a", "dest": "b"}), ("remove_paths", {"paths": ["a"]}), ("add_paths", {"local_paths": [__file__], "dest_dir": ""}), ): with self.assertRaises(Exception) as caught: self.window.client.call_blocking(op, payload) self.assertIn("旧格式", str(caught.exception), f"{op} 应该拒绝旧格式包") def test_mkdir_requires_existing_parent(self): target, _source = self.demo_package() self.load(target) with self.assertRaises(Exception) as caught: self.window.client.call_blocking("mkdir", {"path": "nope/deep"}) self.assertIn("父目录", str(caught.exception)) class TextEditTest(StudioUITestCase): """就地编辑文件内容：进入 → 改 → 保存 → 包内字节真的变了，且库仍可校验。""" def browser(self, target): self.load(target) self.window.goto_tab("browser") pump(self.app, interval=0.02) return self.window._views["browser"] def wait_badge(self, browser, content_hash): return pump( self.app, lambda: browser.hash_badge.text().endswith(short_hash(content_hash, 10)), timeout=15, ) def test_edit_roundtrip_updates_package(self): from sqlite_vfs.core import SQLiteVFS target, _source = self.demo_package() browser = self.browser(target) browser.select_path("app.js") self.assertTrue(pump(self.app, lambda: browser.edit_button.isEnabled(), timeout=15)) old_badge = browser.hash_badge.text() browser.edit_button.setChecked(True) self.assertTrue( pump( self.app, lambda: browser._editing and not browser.text_view.isReadOnly(), timeout=15, ), "没进入编辑模式", ) new_text = browser.text_view.toPlainText() + "\n// 由界面编辑器追加" browser.text_view.setPlainText(new_text) self.assertTrue(browser._edit_dirty) self.assertTrue(browser.save_button.isVisible() and browser.save_button.isEnabled()) browser.save_edit() self.assertTrue( pump(self.app, lambda: (self.window.payload or {}).get("saved"), timeout=20), "保存没有完成", ) saved = self.window.payload["saved"] self.assertEqual(saved["path"], "app.js") # 包内的字节与编辑器里的一致 vfs = SQLiteVFS(target, readonly=True) try: self.assertEqual(vfs.read_file("app.js").decode("utf-8"), new_text) finally: vfs.close() # 详情面板的哈希徽章刷新成了新内容哈希 self.assertTrue(self.wait_badge(browser, saved["hash"]), browser.hash_badge.text()) self.assertNotEqual(old_badge, browser.hash_badge.text()) # 库仍然可校验（编辑结束的 recount 刷新了摘要） report = self.run_op("verify", {"check_content": True})["report"] self.assertTrue(report["ok"], report["errors"]) # 退出编辑回到只读 browser.edit_button.setChecked(False) self.assertTrue(pump(self.app, lambda: browser.text_view.isReadOnly(), timeout=10)) def test_edit_gbk_file_keeps_encoding(self): from sqlite_vfs.core import SQLiteVFS target, _source = self.demo_package() browser = self.browser(target) local = self.path("gbk.txt") original = "这是 GBK 编码的内容" helpers.write_file(local, original.encode("gbk")) payload = self.run_op( "add_paths", {"local_paths": [local], "dest_dir": "", "replace_existing": False} ) self.window.apply_payload(payload) browser.select_path("gbk.txt") self.assertTrue(pump(self.app, lambda: browser.edit_button.isEnabled(), timeout=15)) browser.edit_button.setChecked(True) # _editing 在内容装载前就置位了；要等编辑器可写、编码嗅探完成 self.assertTrue( pump(self.app, lambda: browser._editing and not browser.text_view.isReadOnly(), timeout=15) ) self.assertEqual(browser._edit_encoding, "gbk", "GBK 文件应按 GBK 编辑") edited = browser.text_view.toPlainText() + "\n追加一行中文" browser.text_view.setPlainText(edited) browser.save_edit() self.assertTrue(pump(self.app, lambda: (self.window.payload or {}).get("saved"), timeout=20)) vfs = SQLiteVFS(target, readonly=True) try: raw = vfs.read_file("gbk.txt") finally: vfs.close() self.assertEqual(raw, edited.encode("gbk"), "保存必须保留 GBK 编码") self.assertEqual(self.window.payload["saved"]["encoding"], "gbk") def test_edit_guards(self): target, _source = self.demo_package() browser = self.browser(target) # 目录不可编辑 browser.select_path("docs") self.assertTrue(pump(self.app, lambda: b"_x" or browser._current == "docs", timeout=10)) self.assertTrue(pump(self.app, lambda: not browser.edit_button.isEnabled(), timeout=10)) # 二进制不可编辑（PNG 魔数 + NUL） binary = self.path("logo.bin") helpers.write_file(binary, b"\x89PNG\r\n\x1a\n" + b"\x00\x01" * 40) payload = self.run_op( "add_paths", {"local_paths": [binary], "dest_dir": "", "replace_existing": False} ) self.window.apply_payload(payload) browser.select_path("logo.bin") self.assertTrue(pump(self.app, lambda: browser._current == "logo.bin", timeout=10)) self.assertTrue(pump(self.app, lambda: not browser.edit_button.isEnabled(), timeout=10)) # 超过上限不可编辑 big = self.path("big.txt") helpers.write_file(big, b"x" * (1024 * 1024 + 100)) payload = self.run_op( "add_paths", {"local_paths": [big], "dest_dir": "", "replace_existing": False} ) self.window.apply_payload(payload) browser.select_path("big.txt") self.assertTrue(pump(self.app, lambda: browser._current == "big.txt", timeout=10)) self.assertTrue(pump(self.app, lambda: not browser.edit_button.isEnabled(), timeout=10)) self.assertIn("上限", browser.edit_button.toolTip()) # 空文件可以编辑（往里写内容是正常需求） empty = self.path("empty.txt") helpers.write_file(empty, b"") payload = self.run_op( "add_paths", {"local_paths": [empty], "dest_dir": "", "replace_existing": False} ) self.window.apply_payload(payload) browser.select_path("empty.txt") self.assertTrue(pump(self.app, lambda: browser._current == "empty.txt", timeout=10)) self.assertTrue(pump(self.app, lambda: browser.edit_button.isEnabled(), timeout=10)) def test_save_rejects_unencodable_text(self): """latin-1 文件里输入中文：编码失败要报人话，而不是半个坏文件进库。""" target, _source = self.demo_package() browser = self.browser(target) local = self.path("latin.txt") helpers.write_file(local, "caf\xe9".encode("latin-1")) payload = self.run_op( "add_paths", {"local_paths": [local], "dest_dir": "", "replace_existing": False} ) self.window.apply_payload(payload) browser.select_path("latin.txt") self.assertTrue(pump(self.app, lambda: browser.edit_button.isEnabled(), timeout=15)) browser.edit_button.setChecked(True) self.assertTrue( pump(self.app, lambda: browser._editing and not browser.text_view.isReadOnly(), timeout=15) ) self.assertEqual(browser._edit_encoding, "latin-1") box = {} self.window.client.submit( "save_text", {"path": "latin.txt", "text": "这里换成中文", "encoding": "latin-1"}, on_done=lambda r: box.setdefault("ok", r), on_fail=lambda t, d: box.setdefault("err", (t, d)), label="保存", ) self.assertTrue( pump(self.app, lambda: "ok" in box or "err" in box, timeout=20), "保存既没成功也没失败", ) self.assertIn("err", box) self.assertIn("编码", box["err"][1]) # 包里的内容保持原样（没有半写状态） from sqlite_vfs.core import SQLiteVFS vfs = SQLiteVFS(target, readonly=True) try: self.assertEqual(vfs.read_file("latin.txt"), "caf\xe9".encode("latin-1")) finally: vfs.close() def test_oversize_save_refused_by_service(self): """服务层兜底：对超过上限的文件做文本保存必须拒绝——编辑器装不下它， 任何"保存"都意味着把没显示的部分截掉。（往小文件里粘贴一大段是合法增长，不拦）""" target, _source = self.demo_package() self.load(target) big = self.path("big.txt") helpers.write_file(big, b"x" * (1024 * 1024 + 100)) payload = self.run_op( "add_paths", {"local_paths": [big], "dest_dir": "", "replace_existing": False} ) self.window.apply_payload(payload) box = {} self.window.client.submit( "save_text", {"path": "big.txt", "text": "替换", "encoding": "utf-8"}, on_done=lambda r: box.setdefault("ok", r), on_fail=lambda t, d: box.setdefault("err", (t, d)), label="保存", ) self.assertTrue( pump(self.app, lambda: "ok" in box or "err" in box, timeout=20), "超限保存既没成功也没失败", ) self.assertIn("err", box) self.assertIn("上限", box["err"][1]) # 包里的内容原封不动 from sqlite_vfs.core import SQLiteVFS vfs = SQLiteVFS(target, readonly=True) try: self.assertEqual(len(vfs.read_file("big.txt")), 1024 * 1024 + 100) finally: vfs.close() class CancelTest(StudioUITestCase): """中断语义：库里包在进度回调里抛异常，异常会一路抛回服务层。 这是"中断按钮不是装饰"的证据——顺带说明为什么只有带进度回调的任务才给中断按钮。 """ def test_cancel_interrupts_packing(self): source = self.path("big") helpers.make_tree( source, {f"f{index}.bin": bytes(40000 + index) for index in range(60)} ) target = self.path("big.svfs") ticks = [] def on_progress(job, done, total, label): if job != "打包": return ticks.append(done) if len(ticks) >= 3: # 在第三个进度点请求中断 → 下一个进度点抛 TaskCancelled self.window.client.cancel() self.window.client.progress.connect(on_progress) box = {} self.window.client.submit( "pack", {"source": source, "target": target, "compress": True}, on_done=lambda result: box.setdefault("result", result), on_fail=lambda title, detail: box.setdefault("error", (title, detail)), label="打包", ) self.assertTrue( pump(self.app, lambda: "result" in box or "error" in box, timeout=60), "任务既没成功也没失败", ) self.assertIn("error", box, f"中断没有生效：{box.get('result')}") self.assertEqual(box["error"][0], "已中断") self.assertLess(len(ticks), 60, "进度点比条目数还多，说明循环没被打断") # 被打断的包是个部分结果：库能打开，但条目数明显不全 from sqlite_vfs.core import SQLiteVFS vfs = SQLiteVFS(target, readonly=True) try: self.assertLess(len(vfs.all_entries()), 60) finally: vfs.close() if __name__ == "__main__": unittest.main()