From 26f31a09ecb84f1eb4dbc3866f4b936be299205c Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Wed, 22 Jul 2026 15:48:11 +0800 Subject: [PATCH] =?UTF-8?q?fix(versioning):=20=E4=BF=AE=E5=A4=8D=20shallow?= =?UTF-8?q?=20=E5=BF=AB=E7=85=A7=E6=97=A5=E5=BF=97=E9=87=8D=E5=A4=8D?= =?UTF-8?q?=E8=A1=8C=E5=AF=BC=E8=87=B4=E7=9A=84=20diff=20=E9=85=8D?= =?UTF-8?q?=E5=AF=B9=E9=94=99=E8=AF=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因：state.jsonl 采用只追加日志，track_edit 的中间半成品快照与 make_snapshot 的完整快照各自成行；加载时每行都被回放为独立快照， _find_snapshot_pair 简单取下标-1 作为上一快照，导致 diff 基准落在 同一条消息自己的半成品行上——半成品中缺失的已跟踪文件被全部判为 added 并按全文件计新增（如 +1928 -0）。 彻底修复（非补丁）： - 持久化改为原子全量重写（temp + os.replace），结构性保证磁盘上 一条 message_id 永远只有一行 - _load_state 幂等重载并在加载时折叠历史重复行（保留每消息最后一行）， 旧脏数据无需迁移即可透明愈合 - track_edit / make_snapshot 在按 state 文件加锁的临界区内执行 重载->变更->保存，避免短生命周期实例间丢行 - _find_snapshot_pair 防御性跳过同 message_id 行 - 新增 6 个回归测试（test/test_shallow_versioning.py） 验证：新测试 6/6 通过；test_server_refactor_smoke 6/6 通过； 用 conv_20260721_175016_796 真实数据只读复算，seq8 由 +1295/-0 修正为 +796/-7（3 文件），seq10 由 +1928/-0 修正为 +592/-3（4 文件）， seq31 与历史正确记录 +482/-1 完全一致。 --- modules/shallow_versioning.py | 256 +++++++++++++++++++++----------- test/test_shallow_versioning.py | 205 +++++++++++++++++++++++++ 2 files changed, 376 insertions(+), 85 deletions(-) create mode 100644 test/test_shallow_versioning.py diff --git a/modules/shallow_versioning.py b/modules/shallow_versioning.py index 2d23f196..e8ee84c0 100644 --- a/modules/shallow_versioning.py +++ b/modules/shallow_versioning.py @@ -6,13 +6,34 @@ Mirrors Claude Code's file-history approach: - A snapshot is taken per user message, referencing the latest backup version of every tracked file at that point in time. - Rewind restores the tracked files to the state recorded for a target message. + +Persistence model (state.jsonl) +-------------------------------- +INVARIANT: the on-disk state file holds **exactly one row per message_id** — +the most complete snapshot known for that message. While a message is being +processed, its single row grows as new files get tracked (track_edit rewrites +it); when the message finishes, make_snapshot replaces it with the final full +snapshot. Every mutation rewrites the whole file atomically (temp + replace), +so the invariant holds at rest, not only in memory. + +Legacy append-only logs may contain multiple rows per message_id (intermediate +track_edit journals plus the final snapshot). _load_state collapses those +duplicates keeping the last row, which heals old state files transparently; +the next mutation rewrites the file clean permanently. + +Concurrency: mutations (track_edit / make_snapshot) run inside a per-state-file +process lock as reload -> mutate -> save critical sections, so short-lived +manager instances (one per edit / per task end) can never lose each other's +rows. Readers only see whole files because writes go through os.replace. """ from __future__ import annotations import hashlib import json +import os import shutil +import threading from dataclasses import dataclass, field from datetime import datetime from pathlib import Path @@ -37,6 +58,22 @@ class ShallowSnapshot: timestamp: str +# Process-level locks keyed by state file path, so that separate manager +# instances targeting the same conversation serialize their mutations. +_STATE_LOCKS: Dict[str, threading.Lock] = {} +_STATE_LOCKS_GUARD = threading.Lock() + + +def _state_lock_for(state_file: Path) -> threading.Lock: + key = str(state_file) + with _STATE_LOCKS_GUARD: + lock = _STATE_LOCKS.get(key) + if lock is None: + lock = threading.Lock() + _STATE_LOCKS[key] = lock + return lock + + class ShallowVersioningManager: """Lightweight per-conversation backup for files edited by AI tools.""" @@ -50,6 +87,7 @@ class ShallowVersioningManager: self.save_root = (Path(data_dir).expanduser().resolve() / "save" / self.conversation_id).resolve() self.backup_dir = self.save_root / "shallow_backups" self.state_file = self.backup_dir / "state.jsonl" + self._state_lock = _state_lock_for(self.state_file) self._tracked_files: Set[str] = set() self._snapshots: List[ShallowSnapshot] = [] self._snapshot_sequence = 0 @@ -59,32 +97,84 @@ class ShallowVersioningManager: # persistence # ---------------------------- def _load_state(self) -> None: + """(Re)load snapshots from disk, collapsing duplicate rows per message_id. + + Idempotent: resets in-memory state first, so it can be called inside + mutation critical sections to pick up rows written by other instances. + Legacy logs with multiple rows for the same message_id keep only the + last (most complete) row, ordered by last occurrence. + """ + self._tracked_files = set() + self._snapshots = [] + self._snapshot_sequence = 0 if not self.state_file.exists(): return try: - for raw in self.state_file.read_text(encoding="utf-8", errors="ignore").splitlines(): - line = raw.strip() - if not line: - continue - try: - row = json.loads(line) - except Exception: - continue - if not isinstance(row, dict): - continue - snapshot = self._deserialize_snapshot(row) - if snapshot: - self._snapshots.append(snapshot) - self._snapshot_sequence += 1 - for path in snapshot.tracked_file_backups: - self._tracked_files.add(path) + raw_lines = self.state_file.read_text(encoding="utf-8", errors="ignore").splitlines() except OSError: - pass + return + parsed: List[ShallowSnapshot] = [] + for raw in raw_lines: + line = raw.strip() + if not line: + continue + try: + row = json.loads(line) + except Exception: + continue + if not isinstance(row, dict): + continue + snapshot = self._deserialize_snapshot(row) + if snapshot: + parsed.append(snapshot) + # Collapse duplicates: keep the last row per message_id. + last_index: Dict[str, int] = {} + for idx, snapshot in enumerate(parsed): + last_index[str(snapshot.message_id or "")] = idx + self._snapshots = [ + snapshot for idx, snapshot in enumerate(parsed) if last_index[str(snapshot.message_id or "")] == idx + ] + self._snapshot_sequence = len(self._snapshots) + for snapshot in self._snapshots: + self._tracked_files.update(snapshot.tracked_file_backups.keys()) - def _persist_snapshot(self, snapshot: ShallowSnapshot) -> None: + def _save_state(self) -> None: + """Persist the full snapshot list atomically (temp file + os.replace). + + Must be called with self._state_lock held. Rewriting the whole file is + what enforces the one-row-per-message invariant on disk. + """ self.backup_dir.mkdir(parents=True, exist_ok=True) - with self.state_file.open("a", encoding="utf-8") as fh: - fh.write(json.dumps(self._serialize_snapshot(snapshot), ensure_ascii=False) + "\n") + tmp_path = self.state_file.parent / (self.state_file.name + ".tmp") + payload = "".join( + json.dumps(self._serialize_snapshot(snapshot), ensure_ascii=False) + "\n" + for snapshot in self._snapshots + ) + tmp_path.write_text(payload, encoding="utf-8") + os.replace(tmp_path, self.state_file) + + def _upsert_snapshot(self, snapshot: ShallowSnapshot) -> None: + """Insert or replace the snapshot for its message_id in memory. + + Removes every existing entry with the same message_id and places the + new snapshot at the position of the first removed entry (appends when + no entry existed), preserving chronological order. + """ + target_id = str(snapshot.message_id or "") + first_match = -1 + for idx, existing in enumerate(self._snapshots): + if str(existing.message_id or "") == target_id: + first_match = idx + break + self._snapshots = [ + existing for existing in self._snapshots if str(existing.message_id or "") != target_id + ] + if first_match < 0: + self._snapshots.append(snapshot) + self._snapshot_sequence += 1 + else: + insert_at = min(first_match, len(self._snapshots)) + self._snapshots.insert(insert_at, snapshot) def _serialize_snapshot(self, snapshot: ShallowSnapshot) -> Dict[str, Any]: return { @@ -204,83 +294,70 @@ class ShallowVersioningManager: perf_log("[ShallowVersioning] track_edit called", extra={ "tracking_path": tracking_path, "message_id": message_id, - "already_tracked": tracking_path in self._tracked_files, - "snapshots_count": len(self._snapshots), }) - if tracking_path in self._tracked_files: - return + # Critical section: reload so rows written by other (short-lived) + # instances are visible, then mutate and save atomically. + with self._state_lock: + self._load_state() + if tracking_path in self._tracked_files: + return - resolved = self._resolve_file_path(tracking_path) - backup = self._create_backup(resolved, 1) - self._tracked_files.add(tracking_path) + resolved = self._resolve_file_path(tracking_path) + backup = self._create_backup(resolved, 1) + self._tracked_files.add(tracking_path) - # 归属到当前 message_id 对应的最新 snapshot；不存在则新建。 - # 避免把本次输入的修改追加到上一个输入的 snapshot。 - now = datetime.now().isoformat() - target_snapshot = self._find_latest_snapshot_by_message_id(message_id) - if target_snapshot is not None: - target_snapshot.tracked_file_backups[tracking_path] = backup - self._persist_snapshot(target_snapshot) - else: - snapshot = ShallowSnapshot( - message_id=message_id, - tracked_file_backups={tracking_path: backup}, - timestamp=now, - ) - self._snapshots.append(snapshot) - self._snapshot_sequence += 1 - self._persist_snapshot(snapshot) + # 归属到当前 message_id 对应的 snapshot；不存在则新建。 + # 避免把本次输入的修改追加到上一个输入的 snapshot。 + now = datetime.now().isoformat() + target_snapshot = self._find_latest_snapshot_by_message_id(message_id) + if target_snapshot is not None: + target_snapshot.tracked_file_backups[tracking_path] = backup + else: + target_snapshot = ShallowSnapshot( + message_id=message_id, + tracked_file_backups={tracking_path: backup}, + timestamp=now, + ) + self._upsert_snapshot(target_snapshot) + self._save_state() def make_snapshot(self, message_id: str) -> Dict[str, Any]: """Create a snapshot for the given message, backing up changed tracked files.""" from utils.perf_log import perf_log perf_log("[ShallowVersioning] make_snapshot called", extra={ "message_id": message_id, - "tracked_files_count": len(self._tracked_files), - "snapshots_count": len(self._snapshots), }) - tracked_file_backups: Dict[str, FileBackup] = {} - now = datetime.now().isoformat() + # Critical section: reload so track_edit rows from other instances are + # visible before deciding what changed and which version comes next. + with self._state_lock: + self._load_state() + tracked_file_backups: Dict[str, FileBackup] = {} + now = datetime.now().isoformat() - for tracking_path in self._tracked_files: - resolved = self._resolve_file_path(tracking_path) - latest_backup = self._get_latest_backup_version(tracking_path) - next_version = (latest_backup.version + 1) if latest_backup else 1 + for tracking_path in self._tracked_files: + resolved = self._resolve_file_path(tracking_path) + latest_backup = self._get_latest_backup_version(tracking_path) + next_version = (latest_backup.version + 1) if latest_backup else 1 - if latest_backup and not self._file_changed_since_backup(resolved, latest_backup): - tracked_file_backups[tracking_path] = latest_backup - continue + if latest_backup and not self._file_changed_since_backup(resolved, latest_backup): + tracked_file_backups[tracking_path] = latest_backup + continue - tracked_file_backups[tracking_path] = self._create_backup(resolved, next_version) + tracked_file_backups[tracking_path] = self._create_backup(resolved, next_version) - # Merge in backups from the latest snapshot for any files that may have been added - # by track_edit during the async window. - if self._snapshots: - for tracking_path, backup in self._snapshots[-1].tracked_file_backups.items(): - if tracking_path not in tracked_file_backups: - tracked_file_backups[tracking_path] = backup + snapshot = ShallowSnapshot( + message_id=message_id, + tracked_file_backups=tracked_file_backups, + timestamp=now, + ) + # 一条消息只保留一个快照：替换 track_edit 阶段留下的进行中的行。 + self._upsert_snapshot(snapshot) - snapshot = ShallowSnapshot( - message_id=message_id, - tracked_file_backups=tracked_file_backups, - timestamp=now, - ) - # 如果当前 message_id 已存在 snapshot（由 track_edit 预创建），则替换它， - # 避免同一个输入节点留下多个 snapshot 导致 diff 只计算最后两次之间的变化。 - existing_index = -1 - for idx, s in enumerate(self._snapshots): - if str(s.message_id or "") == str(message_id or ""): - existing_index = idx - if existing_index >= 0: - self._snapshots[existing_index] = snapshot - else: - self._snapshots.append(snapshot) - self._snapshot_sequence += 1 - self._persist_snapshot(snapshot) + # Evict old snapshots while keeping the most recent MAX_SNAPSHOTS. + if len(self._snapshots) > self.MAX_SNAPSHOTS: + self._snapshots = self._snapshots[-self.MAX_SNAPSHOTS :] - # Evict old snapshots while keeping the most recent MAX_SNAPSHOTS. - if len(self._snapshots) > self.MAX_SNAPSHOTS: - self._snapshots = self._snapshots[-self.MAX_SNAPSHOTS :] + self._save_state() perf_log("[ShallowVersioning] make_snapshot done", extra={ "message_id": message_id, @@ -372,16 +449,26 @@ class ShallowVersioningManager: return insertions, deletions def _find_snapshot_pair(self, message_id: str): - """Return (target_snapshot, previous_snapshot) for the given message_id.""" + """Return (target_snapshot, previous_snapshot) for the given message_id. + + The previous snapshot is the nearest preceding snapshot belonging to a + *different* message_id, so duplicate rows of the same message can never + be mistaken for the diff base. + """ target_snapshot: Optional[ShallowSnapshot] = None target_index = -1 + target_id = str(message_id or "") for idx, snapshot in enumerate(self._snapshots): - if snapshot.message_id == message_id: + if str(snapshot.message_id or "") == target_id: target_snapshot = snapshot target_index = idx if not target_snapshot or target_index < 0: return None, None - previous_snapshot = self._snapshots[target_index - 1] if target_index > 0 else None + previous_snapshot: Optional[ShallowSnapshot] = None + for idx in range(target_index - 1, -1, -1): + if str(self._snapshots[idx].message_id or "") != target_id: + previous_snapshot = self._snapshots[idx] + break return target_snapshot, previous_snapshot def get_diff_stats(self, message_id: str) -> Optional[Dict[str, Any]]: @@ -444,7 +531,6 @@ class ShallowVersioningManager: insertions, deletions = self._count_line_changes(previous_lines, target_lines) - resolved = self._resolve_file_path(tracking_path) status = "deleted" if target_backup.backup_file_name is None else ("added" if previous_backup is None or previous_backup.backup_file_name is None else "modified") result.append({ "path": tracking_path, diff --git a/test/test_shallow_versioning.py b/test/test_shallow_versioning.py new file mode 100644 index 00000000..ad99def1 --- /dev/null +++ b/test/test_shallow_versioning.py @@ -0,0 +1,205 @@ +"""Regression tests for shallow versioning snapshot pairing. + +Covers the bug where intermediate track_edit journal rows persisted to +state.jsonl were replayed as independent snapshots, causing diffs to be +computed against a partial snapshot of the *same* message (every tracked +file showed as "added" with full insertions, e.g. +1928 -0). +""" + +from __future__ import annotations + +import json +import shutil +import tempfile +import unittest +from pathlib import Path + +from modules.shallow_versioning import ShallowVersioningManager + + +def _new_manager(project: Path, data: Path, conv: str = "conv_test") -> ShallowVersioningManager: + """Mirror production usage: a fresh short-lived instance per operation.""" + return ShallowVersioningManager(project_path=project, data_dir=data, conversation_id=conv) + + +def _read_state_rows(data: Path, conv: str = "conv_test"): + state_file = data / "save" / conv / "shallow_backups" / "state.jsonl" + if not state_file.exists(): + return [] + return [json.loads(line) for line in state_file.read_text(encoding="utf-8").splitlines() if line.strip()] + + +class ShallowVersioningPairingTest(unittest.TestCase): + def setUp(self) -> None: + self.tmp = Path(tempfile.mkdtemp(prefix="shallow_ver_test_")) + self.project = self.tmp / "project" + self.data = self.tmp / "data" + self.project.mkdir(parents=True) + self.data.mkdir(parents=True) + + def tearDown(self) -> None: + shutil.rmtree(self.tmp, ignore_errors=True) + + def _write(self, rel: str, content: str) -> None: + path = self.project / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + + # ------------------------------------------------------------------ + # realistic multi-instance flow (the original bug scenario) + # ------------------------------------------------------------------ + def test_diff_pairs_with_previous_message_across_instances(self) -> None: + # message 1: AI creates two new files (each track_edit runs in its own instance) + self._write("a.txt", "a1\na2\n") + _new_manager(self.project, self.data).track_edit("a.txt", "msg_1") + self._write("b.txt", "b1\nb2\n") + _new_manager(self.project, self.data).track_edit("b.txt", "msg_1") + _new_manager(self.project, self.data).make_snapshot("msg_1") + + # message 2: modify a.txt, create c.txt + self._write("a.txt", "a1\na2 changed\na3\n") + _new_manager(self.project, self.data).track_edit("a.txt", "msg_2") + self._write("c.txt", "c1\n") + _new_manager(self.project, self.data).track_edit("c.txt", "msg_2") + _new_manager(self.project, self.data).make_snapshot("msg_2") + + # state file must hold exactly one row per message + rows = _read_state_rows(self.data) + msg_ids = [r["message_id"] for r in rows] + self.assertEqual(msg_ids.count("msg_1"), 1, f"duplicate msg_1 rows: {msg_ids}") + self.assertEqual(msg_ids.count("msg_2"), 1, f"duplicate msg_2 rows: {msg_ids}") + + # lazy reader (fresh instance, like the HTTP endpoints) + reader = _new_manager(self.project, self.data) + stats = reader.get_diff_stats("msg_2") or {} + files = {f["path"]: f for f in reader.get_file_diff_stats("msg_2")} + + # a.txt modified (+2 -1), c.txt added (+1 -0), b.txt untouched + self.assertEqual(set(files), {"a.txt", "c.txt"}, f"unexpected diff set: {files}") + self.assertEqual(files["a.txt"]["status"], "modified") + self.assertEqual((files["a.txt"]["insertions"], files["a.txt"]["deletions"]), (2, 1)) + self.assertEqual(files["c.txt"]["status"], "added") + self.assertEqual((files["c.txt"]["insertions"], files["c.txt"]["deletions"]), (1, 0)) + self.assertEqual(stats.get("insertions"), 3) + self.assertEqual(stats.get("deletions"), 1) + + # ------------------------------------------------------------------ + # legacy dirty logs: duplicates collapsed at load, healed on next save + # ------------------------------------------------------------------ + def test_legacy_duplicate_rows_healed_on_load(self) -> None: + conv = "conv_legacy" + backup_dir = self.data / "save" / conv / "shallow_backups" + backup_dir.mkdir(parents=True) + + # real backup files for a.txt v1/v2 and b.txt v1 + (backup_dir / "a@v1").write_text("a1\na2\n", encoding="utf-8") + (backup_dir / "a@v2").write_text("a1\na2 changed\na3\n", encoding="utf-8") + (backup_dir / "b@v1").write_text("b1\n", encoding="utf-8") + + def backup(name, version): + return {"backup_file_name": name, "version": version, "backup_time": "t"} + + # hand-craft the legacy append-only pattern: partials + full per message + legacy_rows = [ + {"message_id": "msg_1", "timestamp": "t", "tracked_file_backups": {"a.txt": backup("a@v1", 1)}}, + {"message_id": "msg_1", "timestamp": "t", "tracked_file_backups": {"a.txt": backup("a@v1", 1), "b.txt": backup("b@v1", 1)}}, + {"message_id": "msg_2", "timestamp": "t", "tracked_file_backups": {"a.txt": backup("a@v2", 2), "b.txt": backup("b@v1", 1)}}, + ] + state_file = backup_dir / "state.jsonl" + state_file.write_text("".join(json.dumps(r) + "\n" for r in legacy_rows), encoding="utf-8") + + manager = _new_manager(self.project, self.data, conv) + # load collapses duplicates: one snapshot per message + self.assertEqual(len(manager.list_snapshots()), 2) + # diff for msg_2 pairs against msg_1 full: only a.txt modified, NOT both added + files = {f["path"]: f for f in manager.get_file_diff_stats("msg_2")} + self.assertEqual(set(files), {"a.txt"}, f"unexpected diff set: {files}") + self.assertEqual(files["a.txt"]["status"], "modified") + self.assertEqual((files["a.txt"]["insertions"], files["a.txt"]["deletions"]), (2, 1)) + + # next mutation rewrites the file clean (one row per message) + self._write("a.txt", "a1\na2 changed\na3\na4\n") + _new_manager(self.project, self.data, conv).track_edit("a.txt", "msg_3") + _new_manager(self.project, self.data, conv).make_snapshot("msg_3") + rows = _read_state_rows(self.data, conv) + msg_ids = [r["message_id"] for r in rows] + self.assertEqual(len(msg_ids), len(set(msg_ids)), f"duplicates remain: {msg_ids}") + + # ------------------------------------------------------------------ + # message with exactly one newly tracked file (recorded stats were + # accidentally correct pre-fix; lazy view was not) + # ------------------------------------------------------------------ + def test_single_new_file_message(self) -> None: + self._write("a.txt", "a1\n") + _new_manager(self.project, self.data).track_edit("a.txt", "msg_1") + _new_manager(self.project, self.data).make_snapshot("msg_1") + + # msg_2 edits a.txt and tracks one new file b.txt + self._write("a.txt", "a1\na2\n") + _new_manager(self.project, self.data).track_edit("a.txt", "msg_2") + self._write("b.txt", "b1\nb2\nb3\n") + _new_manager(self.project, self.data).track_edit("b.txt", "msg_2") + _new_manager(self.project, self.data).make_snapshot("msg_2") + + reader = _new_manager(self.project, self.data) + files = {f["path"]: f for f in reader.get_file_diff_stats("msg_2")} + self.assertEqual(set(files), {"a.txt", "b.txt"}) + self.assertEqual(files["a.txt"]["status"], "modified") + self.assertEqual((files["a.txt"]["insertions"], files["a.txt"]["deletions"]), (1, 0)) + self.assertEqual(files["b.txt"]["status"], "added") + self.assertEqual((files["b.txt"]["insertions"], files["b.txt"]["deletions"]), (3, 0)) + + # ------------------------------------------------------------------ + # cross-instance visibility: mutation reloads state inside the lock + # ------------------------------------------------------------------ + def test_mutation_reloads_state_from_other_instances(self) -> None: + self._write("a.txt", "a1\n") + stale = _new_manager(self.project, self.data) # loads empty state + _new_manager(self.project, self.data).track_edit("a.txt", "msg_1") + _new_manager(self.project, self.data).make_snapshot("msg_1") + + # stale instance mutates: must not wipe msg_1's row + self._write("b.txt", "b1\n") + stale.track_edit("b.txt", "msg_2") + _new_manager(self.project, self.data).make_snapshot("msg_2") + + rows = _read_state_rows(self.data) + msg_ids = [r["message_id"] for r in rows] + self.assertIn("msg_1", msg_ids) + self.assertIn("msg_2", msg_ids) + self.assertEqual(len(msg_ids), 2) + + # ------------------------------------------------------------------ + # rewind still works + # ------------------------------------------------------------------ + def test_rewind_restores_tracked_files(self) -> None: + self._write("a.txt", "original\n") + _new_manager(self.project, self.data).track_edit("a.txt", "msg_1") + _new_manager(self.project, self.data).make_snapshot("msg_1") + + self._write("a.txt", "modified\n") + _new_manager(self.project, self.data).track_edit("a.txt", "msg_2") + _new_manager(self.project, self.data).make_snapshot("msg_2") + + result = _new_manager(self.project, self.data).rewind("msg_1") + self.assertTrue(result.get("success")) + self.assertEqual((self.project / "a.txt").read_text(encoding="utf-8"), "original\n") + + # ------------------------------------------------------------------ + # get_snapshot_by_seq maps to user messages correctly + # ------------------------------------------------------------------ + def test_get_snapshot_by_seq(self) -> None: + for i in (1, 2, 3): + self._write(f"f{i}.txt", f"content {i}\n") + _new_manager(self.project, self.data).track_edit(f"f{i}.txt", f"msg_{i}") + _new_manager(self.project, self.data).make_snapshot(f"msg_{i}") + + reader = _new_manager(self.project, self.data) + self.assertEqual(reader.get_snapshot_by_seq(1).message_id, "msg_1") + self.assertEqual(reader.get_snapshot_by_seq(2).message_id, "msg_2") + self.assertEqual(reader.get_snapshot_by_seq(3).message_id, "msg_3") + self.assertIsNone(reader.get_snapshot_by_seq(4)) + + +if __name__ == "__main__": + unittest.main()