备份、恢复与事件响应

New-Item -ItemType Directory -Force l8-incident
cd l8-incident

mkdir -p l8-incident
cd l8-incident

"""执行SQLite快照恢复和日志重放；时间线使用固定演练分钟。"""
import sqlite3

REPLAY_JOURNAL = True
JOURNAL = [(4, "午餐报销", 16), (5, "出差申请", 19)]

def seed():
    db = sqlite3.connect(":memory:")
    db.execute("CREATE TABLE notes (id INTEGER PRIMARY KEY, text TEXT NOT NULL)")
    db.executemany("INSERT INTO notes VALUES (?, ?)",
                   [(1, "入职材料"), (2, "交通报销"), (3, "休假流程")])
    db.commit()
    return db

def recover(replay=True, journal=JOURNAL):
    live = seed()
    snapshot = sqlite3.connect(":memory:")
    # 12分做快照，后续写入不会出现在该副本中。
    live.backup(snapshot)
    live.executemany("INSERT INTO notes VALUES (?, ?)", [(i, text) for i, text, _ in JOURNAL])
    live.commit()
    expected = live.execute("SELECT id, text FROM notes ORDER BY id").fetchall()
    live.close()  # 演练中放弃原连接，恢复到独立数据库。
    restored = sqlite3.connect(":memory:")
    snapshot.backup(restored)
    watermark = 12
    if replay:
        for note_id, text, minute in journal:
            # 相同id、相同内容可重放；冲突内容必须停下来调查。
            old = restored.execute("SELECT text FROM notes WHERE id=?", (note_id,)).fetchone()
            if old and old[0] != text:
                restored.close()
                snapshot.close()
                raise ValueError("重放冲突")
            restored.execute("INSERT OR IGNORE INTO notes VALUES (?, ?)", (note_id, text))
            watermark = max(watermark, minute)
        restored.commit()
    actual = restored.execute("SELECT id, text FROM notes ORDER BY id").fetchall()
    restored.close()
    snapshot.close()
    # 故障20分，恢复26分；19分为最后一条已确认写入。
    return actual, actual == expected, 19 - watermark, 26 - 20

def main():
    notes, complete, loss_minutes, restore_minutes = recover(REPLAY_JOURNAL)
    print("12分: 快照3条 / 16分: 写入4 / 19分: 写入5")
    print("20分: 故障，暂停写入 / 26分: 恢复校验")
    print("恢复ID:", [note_id for note_id, _ in notes])
    print("内容核对:", "一致" if complete else "缺少已确认写入")
    print(f"已确认写入覆盖缺口: {loss_minutes}分钟 / 演练恢复时间: {restore_minutes}分钟")

if __name__ == "__main__":
    main()

"""核对恢复内容、快照缺口、重复重放和冲突，不依赖原库。"""
import unittest
from restore_notes import JOURNAL, recover

class RestoreTests(unittest.TestCase):
    def test_complete_restore(self):
        notes, complete, loss, elapsed = recover()
        self.assertEqual([i for i, _ in notes], [1, 2, 3, 4, 5])
        self.assertTrue(complete)
        self.assertEqual((loss, elapsed), (0, 6))

    def test_snapshot_alone(self):
        notes, complete, loss, _ = recover(False)
        self.assertEqual(len(notes), 3)
        self.assertFalse(complete)
        self.assertEqual(loss, 7)

    def test_duplicate_journal(self):
        self.assertTrue(recover(journal=JOURNAL + JOURNAL)[1])

    def test_conflicting_write(self):
        with self.assertRaisesRegex(ValueError, "重放冲突"):
            recover(journal=JOURNAL + [(4, "冲突内容", 20)])

    def test_incomplete_journal_detected(self):
        self.assertFalse(recover(journal=JOURNAL[:1])[1])

if __name__ == "__main__":
    unittest.main()

ls
python -X utf8 restore_notes.py
python -X utf8 -m unittest -v test_restore_notes.py

12分: 快照3条 / 16分: 写入4 / 19分: 写入5
20分: 故障，暂停写入 / 26分: 恢复校验
恢复ID: [1, 2, 3, 4, 5]
内容核对: 一致
已确认写入覆盖缺口: 0分钟 / 演练恢复时间: 6分钟

python -X utf8 restore_notes.py