让 Codex 先演练 SQLite 迁移，检查成功与失败留下的状态
先在原创临时副本里跑出可核对的前后状态，再评估迁移代码。
1. 先约定迁移完成后应该留下什么
让 Codex 给 SQLite 表加一个字段，通常很快就能得到几条 SQL。真正需要验收的是这几条语句执行之后的数据库：旧行有没有按预期补值，约束是否生效，版本号是否跟着变化；如果最后一步失败，前面的列改名和索引又留下了什么。
这次演练把一个很小的成员表从版本 1 迁到版本 2。原表允许 handle 为空，迁移将它改名为 email，再加上非空与唯一性要求，同时新增默认值为 active 的 status。数据完全由脚本生成，全部数据库放在新建的临时目录。每次运行都会建立自己的源库，复制后只修改副本。
本文的实验实际运行于 Python 3.12.14 / SQLite 3.53.1。文末完整脚本连续运行两次，JSON 输出一致，2026-10-20 修订后 20 个 unittest 测试通过。实际 Codex 会话未运行；下文给出可以交给 Codex 的完整任务模板，实验结果来自这里列出的 Python 程序。
2. 先检查运行时绑定的 SQLite 版本
本例直接使用 ALTER COLUMN email SET NOT NULL。SQLite 在 3.53.0 才加入 SET NOT NULL 和 DROP NOT NULL，旧版本不能照抄这条语句。Python 的版本号不能替代 SQLite 的版本号；两台都装 Python 3.12 的机器，sqlite3 模块绑定的 SQLite 可能不同。
python3 -c "import platform, sqlite3; print(platform.python_version()); print(sqlite3.sqlite_version)"本文要求 Python ≥ 3.12，因为代码明确设置 sqlite3 的 autocommit 参数；同时要求 SQLite ≥ 3.53.0。脚本在创建 fixture 前检查版本，不满足时直接停止。本篇不提供旧版本的重建表分支；涉及旧版本或复杂依赖对象时，应另行设计并测试迁移。版本依据可参阅 SQLite 3.53.0 发布说明。
3. 从两张表开始，写下固定的前后状态
源库有 teams 与 members 两张表。members.team_id 引用 teams.id；源库 user_version 为 1。正常数据只有团队 1 和两名成员，邮箱使用保留的 .invalid 域名。这里没有读取用户库，也不接受业务数据库路径参数。
teams: (1, 'synthetic-team')
members: (1, 'a@example.invalid', 1)
(2, 'b@example.invalid', 1)
user_version: 1成功后，members 的列顺序应为 id、email、team_id、status，两行都应获得 active。schema 中应该出现 email TEXT NOT NULL 和 members_email_unique 唯一索引，user_version 应为 2。测试里的 AFTER_SCHEMA 和 AFTER_ROWS 是独立手写的预期，未从程序输出反向生成。
members: (1, 'a@example.invalid', 1, 'active')
(2, 'b@example.invalid', 1, 'active')
user_version: 2只比较行数会漏掉错误默认值；只比较 user_version 会漏掉索引；只看提交前的连接又会漏掉持久状态。脚本因此记录完整 schema、按 id 排序的行以及 user_version，再关闭副本并重开，重新读取一次。
4. 让前检报告具体的脏数据
脏数据 fixture 在同一个表里追加两行：id=3 的 handle 为 NULL；id=4 重复使用 a@example.invalid，并引用不存在的团队 99。只有构造这份合成 fixture 时，脚本才在事务外暂时关闭外键检查，随后重新开启。这样的历史坏数据需要主动检查，开启 foreign_keys 并不会自动修复它。
前检分别执行 foreign_key_check、NULL 查询和重复值分组查询。实际结果如下，记录同时指出了三类问题：
foreign_key_violations: [('members', 4, 'teams', 0)]
null_handles: [(3,)]
duplicate_handles: [('a@example.invalid', 2)]foreign_key_check 的四列依次是子表名、违例行的 rowid、父表名、外键编号。这份 fixture 的 id 是 INTEGER PRIMARY KEY，因此这里 rowid=4 对应成员 id=4。不要把末尾的 0 读成出错团队的 id。
重复值检查沿用此表默认的文本比较规则。它没有覆盖自定义 collation、表达式索引或带条件的唯一索引。实际业务若采用其他唯一性规则，前检也要用相同规则，否则“没有查到重复”仍可能在建索引时失败。
脚本先把前检结果放进报告；BEGIN IMMEDIATE 后又执行一次前检，之后才改 schema。合成实验没有并发写者，重复前检只是把事务边界写清楚，并不构成对生产竞争条件的完整验证。
5. 把迁移与显式回滚放进同一段控制流
外键开关在开始事务之前设置，并立刻读取确认值为 1。单独的测试证明：先 BEGIN，再执行 PRAGMA foreign_keys=ON，开关仍然为 0。把这行语句放进迁移事务的最前面并不能补救。
主程序选用 LEGACY_TRANSACTION_CONTROL + isolation_level=None，自己用 SQL BEGIN IMMEDIATE、COMMIT 和 ROLLBACK 管理边界。这是一种便于观察边界的教学配置。Python 文档推荐的 PEP 249 风格是 autocommit=False；后文会分别观察它们的行为。
迁移由五条预先定义的语句组成，使用 execute 逐条执行。新增非空列时给出非 NULL 默认值 active；已有两行会按该默认值读取。提交前还检查 foreign_key_check 与 integrity_check。这里的 integrity_check=ok 也不能代替逐条业务预期测试。
故障情形在五条迁移语句执行完后，故意插入重复 email。实际异常为 UNIQUE constraint failed: members.email。默认 ABORT 冲突处理通常只撤销当前语句，因此程序捕获异常后检查 in_transaction，并执行 SQL ROLLBACK。代码重新抛出异常，让外层报告保留失败原因。
回滚后的 schema 重新出现 handle 列，email 唯一索引消失，status 列消失，原来两行与 user_version=1 恢复。关闭并重开副本后仍然一致。这一结论限定于同一尚未提交事务内的失败；已经 COMMIT 的迁移不能靠这次 ROLLBACK 撤销。
6. 同样的 executescript，为什么回滚结果不同
迁移代码常被压缩成一段 SQL 字符串交给 executescript。下面的反例专门观察它与事务配置的关系：先建 probe 表并确认提交，再写入尚未提交的 id=1；脚本先 ADD COLUMN marker，接着重复插入 id=1。第二条语句失败后，先调用 Connection.rollback()，再按需执行 SQL ROLLBACK。
在 legacy + DEFERRED 配置下，executescript 在运行脚本之前提交了已有事务。此前插入的行被提交，新增 marker 列随后也落地。后面的重复主键错误不会把它们收回。legacy + isolation_level=None 即使事先显式 BEGIN，本次结果也一样。因此，主迁移使用 isolation_level=None 并不足以允许中途改用 executescript。
autocommit=False 没有这个 legacy 前置提交。调用 rollback() 后，新增列与行都撤回。此时 in_transaction 仍然为 True，因为 Python 立即开启了新事务，不能据此判断“回滚失败”。探针随后结束这个空事务，再检查最终状态。
autocommit=True 配合手写 BEGIN 时，executescript 同样没有 legacy 前置提交，但 Connection.rollback() 在这种配置下不执行事务回滚。原事务仍然开着，直到执行 SQL ROLLBACK，列与行才撤回。判断这类代码时，需要同时看 Python autocommit 配置与底层事务状态。
这些结果对应本文固定的短脚本；脚本内部若自己包含事务语句，行为还取决于那些语句。要核对 API 约定，可以直接看 Python 3.12 executescript 条目。主迁移刻意不使用 executescript，也不靠分号 split 来解析 SQL。
7. 源库不变需要区分逻辑状态与文件字节
fixture 创建完成后，源库先关闭，记录文件 SHA-256；然后以 mode=ro 重开，通过 Connection.backup 写入新的临时副本。源连接关闭后才开始迁移副本。最后又只读重开源库，比较其 schema、rows、user_version，并在关闭后重新计算文件哈希。
本次三个情形中，源库的逻辑状态与 SHA-256 都保持一致。文件哈希检查针对这里没有并发写者、采用默认日志模式的合成源库。不要把这种单文件比较直接用于活跃的 WAL 数据库，也不要把它当成并发备份完整性的证明。
对于失败的副本，本文只要求完整逻辑状态等于 before。数据库文件中的底层布局或其他字节细节不属于这个断言。将两种检查分开，报告里的“保持不变”才不会同时混指数据库内容与二进制文件。
8. 完整演练程序
在一个空目录中，将下面代码保存为 rehearse.py。fixture 生成、复制、前检、成功与失败演练、重开检查以及四种 executescript 探针都在同一个文件中。代码没有额外依赖，也不会读取外部数据库路径。数据库位于 TemporaryDirectory，程序退出后自动清理；报告只写到标准输出。
请按下文命令运行，不加 Python 的 -O 选项；这份教学脚本使用 assert 执行部分验收，-O 会关闭这些断言。
2026-10-20 修订：原版在三个预期情形下能正常关闭连接，但意外 SQL、备份或快照错误会跳过后续 close。实际复现是在迁移第一条语句后执行不存在表的 ALTER TABLE：migrate 已回滚并抛出 OperationalError，外层副本连接却仍能执行 SELECT 1。本版将 run 中的连接放进 closing_connection 关闭作用域，并在 connect、fixture 初始化失败及 script_probe 退出时关闭各自持有的连接。源连接仍在迁移副本前关闭；未知异常继续向外抛出，不伪装成成功报告。
事务回滚和资源关闭要分别验收。直接写 with db 并不会关闭 sqlite3 连接，见 Python 3.12 连接上下文管理器说明；contextlib.closing 可用于简单关闭；本例的 closing_connection 进一步保留先发生的异常：若 close 又报错，只给原异常附加说明，不覆盖它；若之前没有异常，关闭错误仍向外抛出。每个连接只尝试关闭一次。迁移的 BEGIN、COMMIT、ROLLBACK 仍由原来的控制流管理，不改用连接上下文管理器隐式提交。
"""Original synthetic fixtures only. Never accepts a user database path."""
import hashlib, json, platform, sqlite3, sys, tempfile
from pathlib import Path
from contextlib import contextmanager
def close_after_error(db, primary):
"""Attempt close once; a cleanup error must not replace the original failure."""
try:
db.close()
except BaseException as cleanup:
primary.add_note(f'Connection close also failed: {cleanup!r}')
@contextmanager
def closing_connection(db):
try:
yield db
except BaseException as primary:
close_after_error(db, primary)
raise
else:
# With no earlier failure, a close failure remains the reported error.
db.close()
def require_versions():
if sys.version_info < (3, 12):
raise RuntimeError('This demo requires Python >=3.12')
if sqlite3.sqlite_version_info < (3, 53, 0):
raise RuntimeError('This demo requires SQLite >=3.53.0 for SET NOT NULL')
def connect(path, readonly=False):
require_versions()
target = path.resolve().as_uri() + '?mode=ro' if readonly else path
db = sqlite3.connect(target, uri=readonly,
autocommit=sqlite3.LEGACY_TRANSACTION_CONTROL,
isolation_level=None)
try:
db.execute('PRAGMA foreign_keys=ON')
assert db.execute('PRAGMA foreign_keys').fetchone() == (1,)
return db
except BaseException as primary:
close_after_error(db, primary)
raise
def snapshot(db):
schema = db.execute("SELECT type,name,tbl_name,sql FROM sqlite_schema WHERE name NOT LIKE 'sqlite_%' ORDER BY type,name").fetchall()
return {'schema': schema,
'rows': {name: db.execute(f'SELECT * FROM "{name}" ORDER BY id').fetchall()
for name in ('teams', 'members')},
'user_version': db.execute('PRAGMA user_version').fetchone()[0]}
def fixture(path, dirty=False):
assert not path.exists()
db = connect(path)
try:
db.execute('CREATE TABLE teams(id INTEGER PRIMARY KEY,name TEXT NOT NULL)')
db.execute('CREATE TABLE members(id INTEGER PRIMARY KEY,handle TEXT,team_id INTEGER REFERENCES teams(id))')
db.execute("INSERT INTO teams VALUES(1,'synthetic-team')")
db.executemany('INSERT INTO members VALUES(?,?,?)', [(1,'a@example.invalid',1),(2,'b@example.invalid',1)])
if dirty:
db.execute('PRAGMA foreign_keys=OFF')
db.executemany('INSERT INTO members VALUES(?,?,?)', [(3,None,1),(4,'a@example.invalid',99)])
db.execute('PRAGMA foreign_keys=ON')
db.execute('PRAGMA user_version=1')
return db
except BaseException as primary:
close_after_error(db, primary)
raise
def preflight(db):
return {
'foreign_key_violations': db.execute('PRAGMA foreign_key_check').fetchall(),
'null_handles': db.execute('SELECT id FROM members WHERE handle IS NULL ORDER BY id').fetchall(),
'duplicate_handles': db.execute('SELECT handle,count(*) FROM members WHERE handle IS NOT NULL GROUP BY handle HAVING count(*)>1 ORDER BY handle').fetchall(),
}
MIGRATION = [
'ALTER TABLE members RENAME COLUMN handle TO email',
'ALTER TABLE members ALTER COLUMN email SET NOT NULL',
"ALTER TABLE members ADD COLUMN status TEXT NOT NULL DEFAULT 'active'",
'CREATE UNIQUE INDEX members_email_unique ON members(email)',
'PRAGMA user_version=2',
]
def migrate(db, inject_failure=False):
if sqlite3.sqlite_version_info < (3,53,0):
raise RuntimeError('This specific demo requires SQLite >=3.53.0 for SET NOT NULL')
db.execute('BEGIN IMMEDIATE')
try:
checks = preflight(db)
if any(checks.values()):
raise ValueError('preflight rejected: '+json.dumps(checks))
for sql in MIGRATION:
db.execute(sql)
if inject_failure:
db.execute("INSERT INTO members(id,email,team_id) VALUES(3,'a@example.invalid',1)")
assert db.execute('PRAGMA foreign_key_check').fetchall() == []
assert db.execute('PRAGMA integrity_check').fetchall() == [('ok',)]
db.execute('COMMIT')
return {'status':'committed','checks':checks}
except BaseException:
if db.in_transaction:
db.execute('ROLLBACK')
raise
def script_probe(mode, isolation_level, *, explicit_begin):
db = sqlite3.connect(':memory:',autocommit=mode,isolation_level=isolation_level)
with closing_connection(db):
db.execute('CREATE TABLE probe(id INTEGER PRIMARY KEY)')
db.commit()
if explicit_begin:
db.execute('BEGIN')
db.execute('INSERT INTO probe VALUES(1)')
before_script = db.in_transaction
error = None
try:
db.executescript('ALTER TABLE probe ADD COLUMN marker TEXT; INSERT INTO probe(id) VALUES(1);')
except sqlite3.IntegrityError as exc:
error = str(exc)
after_error = db.in_transaction
db.rollback()
after_method = db.in_transaction
after_method_rows = db.execute('SELECT * FROM probe').fetchall()
if db.in_transaction:
db.execute('ROLLBACK')
after_sql_rows = db.execute('SELECT * FROM probe').fetchall()
columns = [x[1] for x in db.execute('PRAGMA table_info(probe)')]
return {'autocommit':str(mode),'isolation_level':isolation_level,
'before_script_in_transaction':before_script,'error':error,
'after_error_in_transaction':after_error,'after_rollback_method_in_transaction':after_method,
'rows_after_rollback_method':after_method_rows,'rows_after_sql_rollback':after_sql_rows,
'final_columns':columns}
def run():
require_versions()
output={'versions':{'python':platform.python_version(),'sqlite':sqlite3.sqlite_version},'scenarios':[]}
with tempfile.TemporaryDirectory(prefix='schema-rehearsal-') as tmp:
root=Path(tmp)
for name, dirty, injected in [('success',False,False),('preflight_reject',True,False),('mid_migration_failure',False,True)]:
source_path = root/(name+'-source.db')
# A Connection's own context manager does not close it.
# Close scopes also run on snapshot, backup or unexpected SQL failures.
with closing_connection(fixture(source_path,dirty)) as source:
initial=snapshot(source)
source_hash = hashlib.sha256(source_path.read_bytes()).hexdigest()
copy_path=root/(name+'-copy.db')
with closing_connection(connect(copy_path)) as copy:
with closing_connection(connect(source_path, readonly=True)) as source:
source.backup(copy)
before=snapshot(copy)
checks=preflight(copy)
try:
outcome=migrate(copy,injected)
except (ValueError,sqlite3.IntegrityError) as exc:
outcome={'status':'rejected' if dirty else 'rolled_back','error':str(exc)}
after=snapshot(copy)
with closing_connection(connect(copy_path)) as reopened:
persisted=snapshot(reopened)
assert after==persisted
with closing_connection(connect(source_path, readonly=True)) as source:
assert initial==snapshot(source)
assert hashlib.sha256(source_path.read_bytes()).hexdigest()==source_hash
if name!='success': assert before==after
else:
assert after['user_version']==2
assert after['rows']['members']==[(1,'a@example.invalid',1,'active'),(2,'b@example.invalid',1,'active')]
output['scenarios'].append({'name':name,'preflight':checks,'outcome':outcome,'before':before,'after':after,'reopened_equals_after':True,'source_unchanged':True,'source_file_sha256_unchanged':True,'failed_copy_unchanged':before==after})
probes=[script_probe(sqlite3.LEGACY_TRANSACTION_CONTROL,'DEFERRED',explicit_begin=False),
script_probe(sqlite3.LEGACY_TRANSACTION_CONTROL,None,explicit_begin=True),
script_probe(False,None,explicit_begin=False),
script_probe(True,None,explicit_begin=True)]
assert probes[0]['final_columns']==probes[1]['final_columns']==['id','marker']
assert probes[0]['rows_after_sql_rollback']==probes[1]['rows_after_sql_rollback']==[(1,None)]
assert probes[2]['final_columns']==probes[3]['final_columns']==['id']
assert probes[2]['rows_after_sql_rollback']==probes[3]['rows_after_sql_rollback']==[]
assert probes[3]['after_rollback_method_in_transaction'] is True
output['executescript_probes']=probes
output['all_assertions_passed']=True
return output
def main():
print(json.dumps(run(),ensure_ascii=False,indent=2))
if __name__=='__main__': main()
9. 独立固定预期与约束测试
将下面代码保存为同目录的 test_rehearse.py。前三个情形核对完整的 schema 和行值；三个约束测试分别插入 NULL email、重复 email 和孤儿 team_id，确认都被数据库拒绝。后面的测试核对事务开关与 executescript 的结果。
schema SQL 文本预期按本文实测 SQLite 3.53.1 的输出逐字写下。若未来版本改变 sqlite_schema 中 SQL 的格式，测试会暴露差异；应先核对实际 schema 语义，再决定是否调整预期，不要直接用新输出覆盖测试。
2026-10-20 新增 9 项 CleanupTests：正常完成时检查全部 19 个连接均已关闭；分别注入连接初始化、fixture 建表、初始快照、backup、迁移 SQL 与 executescript 的失败，检查异常仍抛出且已取得的连接全部关闭、关闭计数恰好为 1。另两项测试检查两个清理错误不会替换原来的 backup 错误，以及单独发生的关闭错误不会被隐藏。故障只作用于程序新建的合成连接，不访问外部数据库。原版运行这组新增测试时，六个失败路径均暴露未关闭连接；修订后原有 11 项与新增 9 项全部通过。
"""Handwritten expectations: do not generate them from the migration output."""
import sqlite3
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
import rehearse
from rehearse import fixture, migrate, run, snapshot, script_probe
BEFORE_SCHEMA = [
('table', 'members', 'members', 'CREATE TABLE members(id INTEGER PRIMARY KEY,handle TEXT,team_id INTEGER REFERENCES teams(id))'),
('table', 'teams', 'teams', 'CREATE TABLE teams(id INTEGER PRIMARY KEY,name TEXT NOT NULL)'),
]
AFTER_SCHEMA = [
('index', 'members_email_unique', 'members', 'CREATE UNIQUE INDEX members_email_unique ON members(email)'),
('table', 'members', 'members', "CREATE TABLE members(id INTEGER PRIMARY KEY,email TEXT NOT NULL,team_id INTEGER REFERENCES teams(id), status TEXT NOT NULL DEFAULT 'active')"),
('table', 'teams', 'teams', 'CREATE TABLE teams(id INTEGER PRIMARY KEY,name TEXT NOT NULL)'),
]
CLEAN_ROWS = [(1, 'a@example.invalid', 1), (2, 'b@example.invalid', 1)]
DIRTY_ROWS = CLEAN_ROWS + [(3, None, 1), (4, 'a@example.invalid', 99)]
AFTER_ROWS = [(1, 'a@example.invalid', 1, 'active'), (2, 'b@example.invalid', 1, 'active')]
def expected(schema, members, version):
return {'schema': schema, 'rows': {'teams': [(1, 'synthetic-team')], 'members': members}, 'user_version': version}
class RehearsalTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.report = run()
cls.scenarios = {s['name']: s for s in cls.report['scenarios']}
def test_success_full_state(self):
s = self.scenarios['success']
self.assertEqual(s['before'], expected(BEFORE_SCHEMA, CLEAN_ROWS, 1))
self.assertEqual(s['after'], expected(AFTER_SCHEMA, AFTER_ROWS, 2))
self.assertEqual(s['outcome']['status'], 'committed')
self.assertEqual(s['preflight'], {'foreign_key_violations': [], 'null_handles': [], 'duplicate_handles': []})
def test_dirty_preflight_exact_findings(self):
s = self.scenarios['preflight_reject']
self.assertEqual(s['preflight'], {
'foreign_key_violations': [('members', 4, 'teams', 0)],
'null_handles': [(3,)], 'duplicate_handles': [('a@example.invalid', 2)]})
self.assertEqual(s['outcome']['status'], 'rejected')
self.assertEqual(s['before'], expected(BEFORE_SCHEMA, DIRTY_ROWS, 1))
self.assertEqual(s['after'], s['before'])
def test_injected_failure_restores_schema_rows_version(self):
s = self.scenarios['mid_migration_failure']
self.assertEqual(s['outcome'], {'status': 'rolled_back', 'error': 'UNIQUE constraint failed: members.email'})
self.assertEqual(s['after'], expected(BEFORE_SCHEMA, CLEAN_ROWS, 1))
self.assertEqual(s['before'], s['after'])
def test_reopened_copies_and_sources(self):
for s in self.report['scenarios']:
with self.subTest(scenario=s['name']):
self.assertTrue(s['reopened_equals_after'])
self.assertTrue(s['source_unchanged'])
self.assertTrue(s['source_file_sha256_unchanged'])
def assert_constraint(self, row, error):
with tempfile.TemporaryDirectory(prefix='schema-constraint-') as tmp:
db = fixture(Path(tmp)/'fixture.db')
try:
migrate(db)
before = snapshot(db)
with self.assertRaisesRegex(sqlite3.IntegrityError, error):
db.execute('INSERT INTO members(id,email,team_id) VALUES(?,?,?)', row)
self.assertEqual(snapshot(db), before)
finally:
db.close()
def test_not_null_is_enforced(self):
self.assert_constraint((3, None, 1), 'NOT NULL constraint failed')
def test_unique_is_enforced(self):
self.assert_constraint((3, 'a@example.invalid', 1), 'UNIQUE constraint failed')
def test_foreign_key_is_enforced(self):
self.assert_constraint((3, 'c@example.invalid', 99), 'FOREIGN KEY constraint failed')
def test_foreign_keys_inside_transaction_is_noop(self):
db = sqlite3.connect(':memory:', autocommit=sqlite3.LEGACY_TRANSACTION_CONTROL, isolation_level=None)
try:
db.execute('PRAGMA foreign_keys=OFF')
db.execute('BEGIN')
db.execute('PRAGMA foreign_keys=ON')
self.assertEqual(db.execute('PRAGMA foreign_keys').fetchone(), (0,))
db.execute('ROLLBACK')
db.execute('PRAGMA foreign_keys=ON')
self.assertEqual(db.execute('PRAGMA foreign_keys').fetchone(), (1,))
finally:
db.close()
def test_legacy_executescript_commits_both_configurations(self):
for isolation, explicit in [('DEFERRED', False), (None, True)]:
with self.subTest(isolation=isolation):
p = script_probe(sqlite3.LEGACY_TRANSACTION_CONTROL, isolation, explicit_begin=explicit)
self.assertTrue(p['before_script_in_transaction'])
self.assertFalse(p['after_error_in_transaction'])
self.assertEqual(p['final_columns'], ['id', 'marker'])
self.assertEqual(p['rows_after_sql_rollback'], [(1, None)])
def test_autocommit_false_rollback_starts_new_transaction(self):
p = script_probe(False, None, explicit_begin=False)
self.assertTrue(p['after_error_in_transaction'])
self.assertTrue(p['after_rollback_method_in_transaction'])
self.assertEqual(p['rows_after_rollback_method'], [])
self.assertEqual(p['rows_after_sql_rollback'], [])
self.assertEqual(p['final_columns'], ['id'])
def test_autocommit_true_needs_sql_rollback(self):
p = script_probe(True, None, explicit_begin=True)
self.assertTrue(p['after_error_in_transaction'])
self.assertTrue(p['after_rollback_method_in_transaction'])
self.assertEqual(p['rows_after_rollback_method'], [(1, None)])
self.assertEqual(p['rows_after_sql_rollback'], [])
self.assertEqual(p['final_columns'], ['id'])
class CleanupTests(unittest.TestCase):
"""Faults are injected only into newly created synthetic connections."""
def setUp(self):
self.opened = []
self.fault = None
self.fail_close = False
self.close_calls = {}
owner = self
real_connect = sqlite3.connect
class TrackedConnection(sqlite3.Connection):
def close(self):
owner.close_calls[id(self)] = owner.close_calls.get(id(self), 0) + 1
super().close()
if owner.fail_close:
raise sqlite3.OperationalError('injected close failure after release')
def execute(self, sql, *args, **kwargs):
if sql == owner.fault:
raise sqlite3.OperationalError('injected execute failure')
return super().execute(sql, *args, **kwargs)
def backup(self, *args, **kwargs):
if owner.fault == 'backup':
raise sqlite3.OperationalError('injected backup failure')
return super().backup(*args, **kwargs)
def executescript(self, *args, **kwargs):
if owner.fault == 'executescript':
raise sqlite3.OperationalError('injected script failure')
return super().executescript(*args, **kwargs)
def tracked(*args, **kwargs):
conn = real_connect(*args, factory=TrackedConnection, **kwargs)
self.opened.append(conn)
return conn
self.patcher = patch.object(sqlite3, 'connect', side_effect=tracked)
self.patcher.start()
self.addCleanup(self.patcher.stop)
self.addCleanup(self.close_remaining)
def close_remaining(self):
# Test cleanup also closes leaked handles when checking an older version.
for conn in self.opened:
try:
conn.execute('SELECT 1')
except sqlite3.ProgrammingError:
continue
# The original source can leak handles; cleanup is not the assertion.
sqlite3.Connection.close(conn)
def assert_all_closed(self):
self.assertTrue(self.opened)
for number, conn in enumerate(self.opened):
self.assertEqual(self.close_calls.get(id(conn)), 1)
with self.subTest(connection=number), self.assertRaises(sqlite3.ProgrammingError):
conn.execute('SELECT 1')
def test_success_closes_all_connections_including_probes(self):
run()
self.assertEqual(len(self.opened), 19)
self.assert_all_closed()
def test_unexpected_migration_sql_error_propagates_and_closes(self):
statements = rehearse.MIGRATION[:1] + ['ALTER TABLE missing_table ADD COLUMN x']
with patch.object(rehearse, 'MIGRATION', statements):
with self.assertRaisesRegex(sqlite3.OperationalError, 'missing_table'):
run()
self.assert_all_closed()
def test_initial_snapshot_error_closes_fixture(self):
with patch.object(rehearse, 'snapshot', side_effect=RuntimeError('injected snapshot failure')):
with self.assertRaisesRegex(RuntimeError, 'snapshot failure'):
run()
self.assert_all_closed()
def test_backup_error_closes_source_and_copy(self):
self.fault = 'backup'
with self.assertRaisesRegex(sqlite3.OperationalError, 'backup failure'):
run()
self.assertEqual(len(self.opened), 3)
self.assert_all_closed()
def test_fixture_creation_error_closes_connection(self):
self.fault = 'CREATE TABLE members(id INTEGER PRIMARY KEY,handle TEXT,team_id INTEGER REFERENCES teams(id))'
with self.assertRaisesRegex(sqlite3.OperationalError, 'execute failure'):
run()
self.assert_all_closed()
def test_connection_setup_error_closes_connection(self):
self.fault = 'PRAGMA foreign_keys=ON'
with self.assertRaisesRegex(sqlite3.OperationalError, 'execute failure'):
run()
self.assert_all_closed()
def test_unexpected_script_error_closes_probe(self):
self.fault = 'executescript'
with self.assertRaisesRegex(sqlite3.OperationalError, 'script failure'):
script_probe(True, None, explicit_begin=True)
self.assert_all_closed()
def test_primary_backup_error_survives_two_close_errors(self):
# Use only owned memory connections: both close methods release and then fail.
source = sqlite3.connect(':memory:', autocommit=True)
target = sqlite3.connect(':memory:', autocommit=True)
self.fault = 'backup'
self.fail_close = True
with self.assertRaisesRegex(sqlite3.OperationalError, 'backup failure') as caught:
with rehearse.closing_connection(target):
with rehearse.closing_connection(source):
source.backup(target)
self.assertEqual(len(caught.exception.__notes__), 2)
self.assertTrue(all('close failure' in note for note in caught.exception.__notes__))
self.assert_all_closed()
def test_close_error_without_primary_is_not_hidden(self):
conn = sqlite3.connect(':memory:', autocommit=True)
self.fail_close = True
with self.assertRaisesRegex(sqlite3.OperationalError, 'close failure'):
with rehearse.closing_connection(conn):
conn.execute('SELECT 1')
self.assert_all_closed()
if __name__ == '__main__':
unittest.main(verbosity=2)
10. 在空目录复跑并阅读结果
保存两个文件后，在同一目录执行：
python3 -B rehearse.py > report-1.json
python3 -B rehearse.py > report-2.json
python3 -B -c "from pathlib import Path; a=Path('report-1.json').read_bytes(); b=Path('report-2.json').read_bytes(); assert a==b; print('reports identical')"
python3 -B -m unittest -v test_rehearse本次输出为 reports identical，unittest 汇总为 Ran 20 tests 与 OK。报告中的三个 outcome 分别是 committed、rejected 和 rolled_back。每个情形的 reopened_equals_after、source_unchanged 与 source_file_sha256_unchanged 都为 true。success 的 failed_copy_unchanged 为 false 是预期结果，因为成功迁移本来就会改变副本。
成功副本最后是两行 active 成员与 user_version=2；脏数据副本仍保留两条坏记录，schema 和版本均未改变；故障注入副本恢复原来的两行和版本 1。程序没有替脏数据作清洗决定，拒绝后应该由迁移设计者明确修复规则。
如果只有报告 JSON 看起来正确，但 unittest 失败，先检查失败的固定预期。若运行时版本不满足门槛，到此停止。本教程没有通过安装新运行时或连接其他数据库来绕过门槛。
2026-10-20 回归记录：Python 3.12.14 / SQLite 3.53.1，修订后的两次完整 JSON 逐字节一致，也与修订前的正常输出一致。20 项测试通过，说明新增清理没有改变原有三种迁移结果和四组 executescript 探针。错误路径验证的是可控异常下的连接关闭、恰好一次关闭尝试与异常传播。close 错误由测试子类在真实连接已经关闭后再抛出，用于验证异常保留，不代表验证了真实关闭失败后的资源状态。未测试进程强杀、断电、磁盘故障或 Windows 文件删除语义。
11. 把验收条件一并交给 Codex
如果希望让 Codex 为类似改动准备迁移，可以在空工作目录中使用下面的任务。它把允许操作的文件范围、fixture、迁移顺序与验收要求一起写清楚。本文没有运行这段 Codex 会话，因此不会把前面的测试结果归因于 Codex 的生成代码。
请在当前空目录实现一个可复跑的 SQLite schema 迁移演练。只允许标准库与当前已经安装的 Python；不安装依赖，不联网，不调用模型，不访问现有数据库、用户目录、服务或机密。不启动服务或端口。若版本不足或访问被拒绝，立即停止并报告，不寻找替代访问路径。
先输出 Python 与 sqlite3.sqlite_version。要求 Python >=3.12，SQLite >=3.53.0；低版本停止。本次使用 SET NOT NULL，不扩展到重建表实现。
所有数据库必须由程序在 TemporaryDirectory 中原创生成，不能接收外部数据库路径。原始 schema：teams(id INTEGER PRIMARY KEY,name TEXT NOT NULL)；members(id INTEGER PRIMARY KEY,handle TEXT,team_id INTEGER REFERENCES teams(id))；user_version=1。有效数据：teams=(1,'synthetic-team')；members=(1,'a@example.invalid',1),(2,'b@example.invalid',1)。另外生成脏数据源，追加 members=(3,NULL,1),(4,'a@example.invalid',99)，这一步仅在合成 fixture 的事务外暂时关闭外键，随后恢复开启。
源库创建完成后关闭并记录 SHA-256，再以 mode=ro 重开，用 sqlite3.Connection.backup 复制到新临时库。只在副本迁移；源库逻辑快照与关闭后的文件哈希都要前后核验。不要把失败副本的逻辑相等写成字节相等，也不要推广到 WAL 或并发备份。
迁移前报告 FK 违例、NULL handle 和重复非 NULL handle。事务外启用 foreign_keys 并断言为 1。使用明确的 LEGACY_TRANSACTION_CONTROL + isolation_level=None，以逐条 execute 执行 SQL BEGIN IMMEDIATE、迁移语句及 SQL COMMIT/ROLLBACK；事务开始后重复前检。不得用 executescript 执行主迁移，不通过 split(';') 解析任意脚本。
迁移依次为：handle 重命名 email；email SET NOT NULL；新增 status TEXT NOT NULL DEFAULT 'active'；建立 members_email_unique UNIQUE INDEX(email)；user_version=2。提交前检查 foreign_key_check 与 integrity_check。
覆盖三种情形：有效数据成功；脏数据前检拒绝；全部迁移语句执行后、COMMIT 前故意插入重复 email，捕获 UNIQUE 错误并显式回滚。每种情形输出 before/after 的完整 schema、排序后的行、user_version，关闭副本再重开核验持久状态。失败时应恢复迁移前的完整逻辑状态。
独立编写固定预期测试，不能从运行结果生成预期。成功后另外尝试 NULL email、重复 email、孤儿 team_id，三种插入都应失败。验证事务内开启 foreign_keys 无效。用独立内存库比较 executescript 在 legacy+DEFERRED、legacy+None+BEGIN、autocommit=False、autocommit=True+BEGIN 下的失败与回滚，输出 rollback() 后和 SQL ROLLBACK 后的行与列，并解释 in_transaction。
交付完整源码、测试和命令。先人工列出预期，再跑两次并比较结果，报告实际版本、失败信息及测试数。任何未执行步骤标为未执行。不要声称本演练覆盖生产断电恢复、并发写者、任意触发器/视图、应用兼容性或已提交迁移的逆向恢复。不要发布内容。
审核 Codex 的交付时，先看它是否真的执行了版本检查和两次复跑，再核对测试预期是不是独立写下的。不要接受一张“全部通过”的摘要来替代可读的测试文件。尤其留意异常处理是否只调用了 rollback() 却没说明当前 autocommit 配置，或迁移中途换成了 executescript。
12. 这次演练结束后，还有哪些上线问题
这份程序证明了固定 fixture 与固定五步迁移在已记录版本上的结果，也演示了一个提交前失败路径。它没有测试任意触发器、视图、自定义排序规则、活跃 WAL 写者、锁等待、磁盘故障或断电恢复，也没有验证新旧应用同时访问时的兼容性。
实际项目还需要按自己的 schema 依赖与数据规模设计检查，确定上线前备份的可恢复性，并单独安排已提交迁移的逆向方案。本文可直接复用的是“先写固定预期，再在合成副本里记录成功和失败留下的状态”这一做法。是否适用于生产库，需要新的测试证据。
13. 补充演练：活跃 WAL 下，备份到底看见了哪一批行
前面的迁移演练从已关闭的默认日志模式源库开始，末节“没有覆盖 WAL”的说明描述的是那次实验。这里于 2026-10-13 追加一个独立小案例，只补验备份入口：源库还开着、已提交数据仍在 WAL 时，怎样确认拿到的副本包含了预期行。前面的迁移与失败回滚脚本没有在 WAL 下重新运行，这段补充不能扩大它们的验证范围。
本例最值得留意的结果是：仅复制 source.db 得到的文件，重开后 integrity_check 返回 ok，却遗漏了已经提交的行 2。换成 Connection.backup 后，还要问一句：来源连接是否停留在旧读事务里？同一数据库的两个读连接，本次就分别备份出了 [1] 和 [1, 2]。
实测范围：Linux、Python 3.12.14、SQLite 3.53.1，完全原创的临时 items 表，20 项 unittest 测试通过；正常父环境与刻意污染的父环境结果相同。代码没有读取真实数据库、调用模型或运行 Codex 会话，也没有在 backup 调用期间继续提交写入。两个新增图是按本次结果绘制的原创示意图。
14. 把 checkpoint、读事务和提交排成可复验的顺序
只建一张 items(id, label) 表。行 1 的 label 是 baseline，行 2 是 committed-in-wal，行 3 是 uncommitted。连接明确使用 autocommit=True，所以普通 INSERT 执行完就提交；需要保留事务的地方用 SQL BEGIN、COMMIT、ROLLBACK。
- 写连接启用 WAL，暂时设置 wal_autocheckpoint=0；创建表并插入行 1，然后执行 wal_checkpoint(TRUNCATE)，确认返回 [0, 0, 0]。
- 旧读连接执行 BEGIN，再执行 SELECT 并读完结果，固定只看见行 1 的读快照。只写 BEGIN 不足以替代这里的实际读取。
- 写连接提交行 2，并保持连接打开。程序检查 -wal 大于 32 bytes，确认它不只是一个空文件或文件头。
- 依次做主文件单独复制、新读连接 backup、旧读事务 backup。每份输出关闭目标连接后，再用新连接重开核对。
- 旧读连接执行 SQL COMMIT 后再 backup。接着写连接显式 BEGIN IMMEDIATE，插入行 3 但不提交，再通过另一只读连接 backup，最后回滚行 3。
关闭自动 checkpoint、保留写连接，是为了让这个小实验稳定保留 WAL 帧，并非生产配置建议。SQLite 官方 WAL 文档说明，提交可以发生在 WAL 中，WAL 也是数据库持久状态的一部分。最后一个连接关闭常会触发清理，所以“先关闭全部连接，再观察文件”会改变本例想验证的前提。
15. 五份输出，分别对照固定行集合
main-only.db rows = [[1, "baseline"]] integrity = ["ok"]
fresh.db rows = [[1, "baseline"], [2, "committed-in-wal"]] integrity = ["ok"]
pinned.db rows = [[1, "baseline"]] integrity = ["ok"]
released.db rows = [[1, "baseline"], [2, "committed-in-wal"]] integrity = ["ok"]
during-uncommitted.db rows = [[1, "baseline"], [2, "committed-in-wal"]] integrity = ["ok"]主文件单独复制的反例已经说明：可以正常打开、schema 正确、integrity_check=ok，都不能代替“已提交业务行是否完整”的断言。本例的主文件 SHA-256 在行 2 提交前后完全相同，而新读连接已经看到行 2；因此不能沿用前文针对已关闭源库的主文件哈希检查，来证明活跃 WAL 数据库逻辑状态没有变化。
旧读连接的备份只有行 1，是本例明确固定的旧快照，并不表示 backup 漏复制了该连接可见的行。SQLite 隔离文档解释了读事务期间持续看到旧快照，以及另一连接未提交写入的不可见性。这里在旧读事务结束后，用同一个连接再备份，结果才包含行 2。
本例每次 backup 期间没有新提交，不能由此给出繁忙数据库“绝对最新”的时点保证。SQLite 在线备份说明另行讨论了备份中途其他连接写入导致重启，以及频繁重启可能无法完成的情况。也不要把本例改成按顺序复制 .db、-wal、-shm 就当作已验证的活跃库备份方案。
16. 先在空目录运行，再检查拒绝覆盖与中止边界
把下面三个完整源码块分别保存成 wal_backup_demo.py、test_wal_backup.py、verify_repro.py，放到同一个空目录。只用 Python 标准库。本补充沿用原文 Python ≥3.12、SQLite ≥3.53.0 的实验基线；这不是说 WAL 或 backup 到这些版本才可用。先查看 sqlite3 模块实际绑定的 SQLite 版本，不要只看 Python 版本。
python3 -I -B -c "import platform,sqlite3; print(platform.python_version(), sqlite3.sqlite_version)"
python3 -I -B wal_backup_demo.py
python3 -I -B -m unittest discover -s . -p test_wal_backup.py -v
python3 -I -B verify_repro.py实验脚本不接受任何数据库路径参数，所有数据库都新建在 TemporaryDirectory 中，退出时清理；JSON 报告打印到标准输出。测试入口使用新建目录，拒绝重用已有目录。单份输出创建前检查同名文件、目录、符号链接与 -wal/-shm/-journal，并用排他创建拒绝覆盖；这些保护只针对自有临时目录，不是恶意文件系统中的竞态安全方案。
backup 每步复制 1 页，进度回调检查 2 秒协作式预算和最多 1000 次回调，连接锁等待设为 0.05 秒。它们是演练保护，不能强制中断一次卡住的底层 I/O，也不是严格 2 秒墙钟承诺。复跑器另外给每个子进程 20 秒上限。测试故意从仍在写事务的同一个源连接调用 backup，确认回调次数上限会中止并删除本次新建的输出；正常案例始终使用独立读连接。
源码显式关闭所有连接，并在 finally 中结束尚未完成的事务。autocommit=True 时，Python 的 .commit() 不会替代本例的 SQL COMMIT，依据见 Python 3.12 commit 文档。backup 的 pages/progress 参数见 Python 3.12 backup 文档；来源与目标连接限制见 SQLite Backup API。
17. 完整源码：构造活跃 WAL 并重开五份输出
wal_backup_demo.py
"""Synthetic WAL experiment. No user database or filesystem path is accepted by CLI."""
from __future__ import annotations
import hashlib
import json
import math
import os
from pathlib import Path
import platform
import shutil
import sqlite3
import sys
import tempfile
import time
class BackupBudgetExceeded(TimeoutError):
pass
class BackupBudget:
"""Cooperative callback budget, not a hard wall-clock deadline."""
def __init__(self, seconds=2.0, max_callbacks=1000, clock=time.monotonic):
if not isinstance(seconds, (int, float)) or isinstance(seconds, bool):
raise ValueError("seconds must be a positive finite number")
if not math.isfinite(seconds) or seconds <= 0:
raise ValueError("seconds must be a positive finite number")
if type(max_callbacks) is not int or max_callbacks < 1:
raise ValueError("max_callbacks must be a positive integer")
self.clock = clock
self.deadline = clock() + seconds
self.max_callbacks = max_callbacks
self.calls = 0
self.statuses = []
def __call__(self, status, remaining, total):
self.calls += 1
self.statuses.append(status)
if self.clock() >= self.deadline:
raise BackupBudgetExceeded("backup callback deadline reached")
if self.calls >= self.max_callbacks and status != sqlite3.SQLITE_DONE:
raise BackupBudgetExceeded("backup callback limit reached")
def check_runtime():
# Match the original article's baseline, rather than claiming the API is new.
if sys.version_info < (3, 12) or sqlite3.sqlite_version_info < (3, 53, 0):
raise RuntimeError("this rehearsal requires Python >=3.12 and SQLite >=3.53.0")
def connect(path, *, readonly=False):
path = Path(path).resolve()
if readonly:
return sqlite3.connect(path.as_uri() + "?mode=ro&cache=private", uri=True,
autocommit=True, timeout=0.05)
return sqlite3.connect(path, autocommit=True, timeout=0.05)
def rows(conn):
return [list(row) for row in conn.execute("SELECT id,label FROM items ORDER BY id")]
def snapshot(path):
"""Open a NEW read-only connection after the output writer was closed."""
conn = connect(path, readonly=True)
try:
return {
"rows": rows(conn),
"schema": [list(row) for row in conn.execute(
"SELECT type,name,sql FROM sqlite_schema "
"WHERE name NOT LIKE 'sqlite_%' ORDER BY type,name")],
"integrity": [row[0] for row in conn.execute("PRAGMA integrity_check")],
}
finally:
conn.close()
def sidecars(path):
return [Path(str(path) + suffix) for suffix in ("-wal", "-shm", "-journal")]
def reserve_new(path):
"""Only for this program's private temporary directory, not hostile paths."""
path = Path(path)
for candidate in [path, *sidecars(path)]:
if os.path.lexists(candidate):
raise FileExistsError(f"refusing existing output or sidecar: {candidate.name}")
# Exclusive creation also refuses an existing/dangling symlink at this instant.
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
os.close(fd)
return path
def remove_owned_output(path):
for candidate in [path, *sidecars(path)]:
candidate.unlink(missing_ok=True)
def backup_new(source, path, *, seconds=2.0, max_callbacks=1000):
budget = BackupBudget(seconds, max_callbacks) # Validate before file creation.
path = reserve_new(path)
target = None
completed = False
try:
target = connect(path)
source.backup(target, pages=1, progress=budget, sleep=0.01)
completed = True
finally:
if target is not None:
target.close()
if not completed:
remove_owned_output(path)
return {"snapshot": snapshot(path), "progress": {
"callbacks": budget.calls, "statuses": budget.statuses,
"ended_with_done": bool(budget.statuses and budget.statuses[-1] == sqlite3.SQLITE_DONE),
}}
def raw_main_copy(source_path, output_path):
output_path = reserve_new(output_path)
completed = False
try:
shutil.copyfile(source_path, output_path)
completed = True
finally:
if not completed:
remove_owned_output(output_path)
return snapshot(output_path)
def sha256(path):
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
def run_case(directory):
"""Internal fixture entry point. Refuses a pre-existing case directory."""
check_runtime()
directory = Path(directory).resolve()
directory.mkdir() # No exist_ok: never reuse another experiment's files.
source_path = directory / "source.db"
handles = []
def opened(*, readonly=False):
conn = connect(source_path, readonly=readonly)
handles.append(conn)
return conn
try:
writer = opened()
mode = writer.execute("PRAGMA journal_mode=WAL").fetchone()[0]
if mode != "wal":
raise RuntimeError("WAL was not enabled")
writer.execute("PRAGMA wal_autocheckpoint=0")
writer.execute("CREATE TABLE items(id INTEGER PRIMARY KEY,label TEXT NOT NULL)")
writer.execute("INSERT INTO items VALUES(1,'baseline')")
checkpoint = list(writer.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone())
if checkpoint != [0, 0, 0]:
raise RuntimeError(f"baseline checkpoint did not finish: {checkpoint}")
baseline_hash = sha256(source_path)
old_reader = opened(readonly=True)
old_reader.execute("BEGIN")
old_rows = rows(old_reader) # BEGIN alone does not establish the read snapshot.
writer.execute("INSERT INTO items VALUES(2,'committed-in-wal')")
wal_bytes = Path(str(source_path) + "-wal").stat().st_size
if wal_bytes <= 32:
raise RuntimeError("expected WAL frames after the committed insert")
committed_hash = sha256(source_path)
fresh_reader = opened(readonly=True)
report = {
"runtime": {"python": platform.python_version(), "sqlite": sqlite3.sqlite_version,
"system": platform.system()},
"fixture": {"journal_mode": mode, "baseline_checkpoint": checkpoint,
"wal_bytes_after_commit": wal_bytes,
"main_sha256_before_row2": baseline_hash,
"main_sha256_after_row2": committed_hash,
"old_reader_first_rows": old_rows,
"fresh_reader_committed_rows": rows(fresh_reader)},
"main_file_only": raw_main_copy(source_path, directory / "main-only.db"),
"fresh_backup": backup_new(fresh_reader, directory / "fresh.db"),
"pinned_backup": backup_new(old_reader, directory / "pinned.db"),
}
report["old_reader_still_in_transaction"] = old_reader.in_transaction
old_reader.execute("COMMIT") # SQL COMMIT: autocommit=True makes .commit() a no-op.
report["released_backup"] = backup_new(old_reader, directory / "released.db")
writer.execute("BEGIN IMMEDIATE")
writer.execute("INSERT INTO items VALUES(3,'uncommitted')")
report["writer_uncommitted_rows"] = rows(writer)
uncommitted_reader = opened(readonly=True)
report["during_uncommitted_backup"] = backup_new(
uncommitted_reader, directory / "during-uncommitted.db")
writer.execute("ROLLBACK")
report["source_after_rollback"] = snapshot(source_path)
report["wal_still_nonempty_before_close"] = Path(str(source_path) + "-wal").stat().st_size > 32
return report
finally:
# Roll back unfinished transactions on both success and exception paths.
for conn in reversed(handles):
try:
if conn.in_transaction:
conn.execute("ROLLBACK")
finally:
conn.close()
def main():
if len(sys.argv) != 1:
raise SystemExit("No arguments accepted: this script only creates synthetic temporary databases")
with tempfile.TemporaryDirectory(prefix="wal-rehearsal-") as td:
report = run_case(Path(td) / "case")
print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
if __name__ == "__main__":
main()
18. 独立手写预期：行、schema 与失败边界
ONE、TWO、THREE、SCHEMA 和 EXPECTED_FILES 都在测试文件中独立写出，没有从程序报告反向生成。除五种可见性结果，测试还重新打开所有输出、检查非空 WAL 与主文件哈希反例、验证拒绝覆盖、无效预算、时钟与回调上限、源连接关闭错误、故障路径连接清理，以及拒绝真实数据库路径参数。
test_wal_backup.py
"""Hand-written row/schema expectations, deliberately independent of program output."""
import json
import os
from pathlib import Path
import sqlite3
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
import wal_backup_demo as demo
ONE = [[1, "baseline"]]
TWO = [[1, "baseline"], [2, "committed-in-wal"]]
THREE = [[1, "baseline"], [2, "committed-in-wal"], [3, "uncommitted"]]
SCHEMA = [["table", "items", "CREATE TABLE items(id INTEGER PRIMARY KEY,label TEXT NOT NULL)"]]
EXPECTED_FILES = {
"main-only.db": ONE,
"fresh.db": TWO,
"pinned.db": ONE,
"released.db": TWO,
"during-uncommitted.db": TWO,
}
class WalCaseTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.temp = tempfile.TemporaryDirectory(prefix="wal-test-")
cls.addClassCleanup(cls.temp.cleanup)
cls.case = Path(cls.temp.name) / "case"
cls.report = demo.run_case(cls.case)
def test_main_only_is_readable_but_misses_committed_row(self):
actual = self.report["main_file_only"]
self.assertEqual(actual, {"rows": ONE, "schema": SCHEMA, "integrity": ["ok"]})
self.assertNotEqual(actual["rows"], TWO)
def test_new_reader_backup_contains_wal_commit(self):
self.assertEqual(self.report["fresh_backup"]["snapshot"]["rows"], TWO)
def test_existing_read_transaction_keeps_old_snapshot(self):
self.assertEqual(self.report["fixture"]["old_reader_first_rows"], ONE)
self.assertTrue(self.report["old_reader_still_in_transaction"])
self.assertEqual(self.report["pinned_backup"]["snapshot"]["rows"], ONE)
def test_sql_commit_releases_the_old_snapshot(self):
self.assertEqual(self.report["released_backup"]["snapshot"]["rows"], TWO)
def test_other_connections_uncommitted_row_is_excluded(self):
self.assertEqual(self.report["writer_uncommitted_rows"], THREE)
self.assertEqual(self.report["during_uncommitted_backup"]["snapshot"]["rows"], TWO)
self.assertEqual(self.report["source_after_rollback"]["rows"], TWO)
def test_nonempty_wal_and_main_hash_do_not_mean_same_logical_state(self):
f = self.report["fixture"]
self.assertEqual(f["journal_mode"], "wal")
self.assertEqual(f["baseline_checkpoint"], [0, 0, 0])
self.assertGreater(f["wal_bytes_after_commit"], 32)
self.assertTrue(self.report["wal_still_nonempty_before_close"])
self.assertEqual(f["main_sha256_before_row2"], f["main_sha256_after_row2"])
self.assertEqual(f["fresh_reader_committed_rows"], TWO)
self.assertNotEqual(f["old_reader_first_rows"], f["fresh_reader_committed_rows"])
def test_every_file_reopened_after_all_original_connections_close(self):
for name, expected_rows in EXPECTED_FILES.items():
with self.subTest(name=name):
self.assertEqual(demo.snapshot(self.case / name), {
"rows": expected_rows, "schema": SCHEMA, "integrity": ["ok"]})
def test_successful_backups_finish_with_sqlite_done(self):
for key in ("fresh_backup", "pinned_backup", "released_backup", "during_uncommitted_backup"):
with self.subTest(key=key):
result = self.report[key]
self.assertEqual(result["snapshot"]["schema"], SCHEMA)
self.assertEqual(result["snapshot"]["integrity"], ["ok"])
self.assertTrue(result["progress"]["ended_with_done"])
self.assertEqual(result["progress"]["statuses"][-1], sqlite3.SQLITE_DONE)
self.assertLessEqual(result["progress"]["callbacks"], 1000)
def test_existing_case_directory_is_rejected(self):
before = (self.case / "source.db").read_bytes()
with self.assertRaises(FileExistsError):
demo.run_case(self.case)
self.assertEqual((self.case / "source.db").read_bytes(), before)
class GuardTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory(prefix="wal-guard-")
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
self.source = sqlite3.connect(":memory:", autocommit=True)
self.addCleanup(self.source.close)
self.source.execute("CREATE TABLE items(id INTEGER PRIMARY KEY,label TEXT NOT NULL)")
self.source.execute("INSERT INTO items VALUES(1,'baseline')")
def test_existing_output_file_preserved(self):
out = self.root / "already.db"
out.write_bytes(b"do not replace me")
with self.assertRaises(FileExistsError):
demo.backup_new(self.source, out)
self.assertEqual(out.read_bytes(), b"do not replace me")
def test_existing_directory_and_dangling_symlink_rejected(self):
directory = self.root / "directory.db"
directory.mkdir()
link = self.root / "link.db"
link.symlink_to(self.root / "missing.db")
for out in (directory, link):
with self.subTest(out=out.name), self.assertRaises(FileExistsError):
demo.backup_new(self.source, out)
self.assertTrue(directory.is_dir())
self.assertTrue(link.is_symlink())
self.assertFalse((self.root / "missing.db").exists())
def test_existing_sidecars_preserved(self):
for suffix in ("-wal", "-shm", "-journal"):
out = self.root / (suffix[1:] + ".db")
sidecar = Path(str(out) + suffix)
sidecar.write_bytes(b"sidecar sentinel")
with self.subTest(suffix=suffix), self.assertRaises(FileExistsError):
demo.backup_new(self.source, out)
self.assertEqual(sidecar.read_bytes(), b"sidecar sentinel")
self.assertFalse(out.exists())
def test_invalid_budget_rejected_before_creating_output(self):
for kwargs in ({"seconds": 0}, {"seconds": -1}, {"seconds": float("inf")},
{"seconds": float("nan")}, {"seconds": True},
{"max_callbacks": 0}, {"max_callbacks": 1.5}):
out = self.root / "invalid.db"
with self.subTest(kwargs=kwargs), self.assertRaises(ValueError):
demo.backup_new(self.source, out, **kwargs)
self.assertFalse(out.exists())
def test_callback_deadline_is_checked(self):
times = iter([10.0, 12.0])
budget = demo.BackupBudget(seconds=2, clock=lambda: next(times))
with self.assertRaises(demo.BackupBudgetExceeded):
budget(sqlite3.SQLITE_OK, 1, 2)
def test_callback_count_is_bounded_even_if_clock_does_not_advance(self):
budget = demo.BackupBudget(seconds=2, max_callbacks=2, clock=lambda: 0)
budget(sqlite3.SQLITE_BUSY, 0, 0)
with self.assertRaises(demo.BackupBudgetExceeded):
budget(sqlite3.SQLITE_BUSY, 0, 0)
self.assertEqual(budget.calls, 2)
def test_same_connection_write_transaction_aborts_and_removes_partial_output(self):
# Deliberate misuse only in this negative fixture, never in the main case.
self.source.execute("BEGIN IMMEDIATE")
self.source.execute("INSERT INTO items VALUES(2,'not-committed')")
out = self.root / "aborted.db"
try:
with self.assertRaises(demo.BackupBudgetExceeded):
demo.backup_new(self.source, out, max_callbacks=3)
self.assertFalse(out.exists())
self.assertFalse(any(p.exists() for p in demo.sidecars(out)))
self.assertTrue(self.source.in_transaction)
finally:
self.source.execute("ROLLBACK")
self.assertEqual(demo.rows(self.source), ONE)
def test_source_closed_error_cleans_owned_output(self):
self.source.close()
out = self.root / "closed.db"
with self.assertRaises(sqlite3.ProgrammingError):
demo.backup_new(self.source, out)
self.assertFalse(out.exists())
def test_raw_copy_refuses_existing_target(self):
src, out = self.root / "source", self.root / "target"
src.write_bytes(b"source bytes")
out.write_bytes(b"target sentinel")
with self.assertRaises(FileExistsError):
demo.raw_main_copy(src, out)
self.assertEqual(out.read_bytes(), b"target sentinel")
def test_exception_path_closes_all_case_connections(self):
opened = []
original_connect = demo.connect
def tracking(*args, **kwargs):
conn = original_connect(*args, **kwargs)
opened.append(conn)
return conn
with patch.object(demo, "connect", side_effect=tracking), \
patch.object(demo, "raw_main_copy", side_effect=RuntimeError("injected failure")):
with self.assertRaisesRegex(RuntimeError, "injected failure"):
demo.run_case(self.root / "fault-case")
self.assertGreaterEqual(len(opened), 3)
for conn in opened:
with self.assertRaises(sqlite3.ProgrammingError):
conn.execute("SELECT 1")
def test_cli_refuses_real_database_path_argument(self):
sentinel = self.root / "do-not-open.db"
sentinel.write_bytes(b"not a SQLite database")
result = subprocess.run([sys.executable, "-I", "-B", str(Path(demo.__file__)), str(sentinel)],
text=True, capture_output=True, timeout=10)
self.assertNotEqual(result.returncode, 0)
self.assertIn("No arguments accepted", result.stderr)
self.assertEqual(sentinel.read_bytes(), b"not a SQLite database")
if __name__ == "__main__":
unittest.main(verbosity=2)
19. 完整复跑器：污染父环境，仍只运行这份临时实验
复跑器把三个文件复制到新临时目录，分别在正常环境和污染环境运行。污染项包括 PYTHONPATH、PYTHONHOME、PYTHONSTARTUP、HOME、SQLITE_HISTORY；污染目录还放置会立即报错的 sqlite3.py、sitecustomize.py 与 usercustomize.py，以及不可改动的 keep.db 哨兵。它使用 -I 隔离 Python 启动环境，unittest discover 只引入复制后的测试目录。两份完整 JSON 必须相等，20 项测试必须实际运行，哨兵文件名与字节哈希必须不变。
这只排除了列出的父环境干扰，不是对恶意 Python 解释器、任意操作系统配置或所有环境变量的隔离承诺。源代码块与离线 ZIP 中的副本还分别在新目录复跑过，不能用工作目录里已存在的输出文件替代实际执行。
verify_repro.py
"""Re-run copied sources in clean and polluted parents; no project files required."""
import hashlib
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
FILES = ("wal_backup_demo.py", "test_wal_backup.py", "verify_repro.py")
def file_hash(path):
return hashlib.sha256(path.read_bytes()).hexdigest()
def run_variant(root, source, *, polluted):
destination = root / ("polluted" if polluted else "clean")
destination.mkdir()
for name in FILES:
shutil.copyfile(source / name, destination / name)
sentinels = root / (destination.name + "-sentinels")
sentinels.mkdir()
for name in ("sqlite3.py", "sitecustomize.py", "usercustomize.py"):
(sentinels / name).write_text("raise RuntimeError('parent import pollution reached')\n")
(sentinels / "keep.db").write_bytes(b"unrelated synthetic sentinel: do not open")
before = {p.name: file_hash(p) for p in sentinels.iterdir()}
env = os.environ.copy()
cwd = destination
if polluted:
env.update({"PYTHONPATH": str(sentinels), "PYTHONHOME": str(sentinels),
"PYTHONSTARTUP": str(sentinels / "sitecustomize.py"),
"HOME": str(sentinels), "SQLITE_HISTORY": str(sentinels / "keep.db")})
cwd = sentinels
program = subprocess.run([sys.executable, "-I", "-B", str(destination / FILES[0])],
cwd=cwd, env=env, capture_output=True, text=True, timeout=20)
if program.returncode != 0:
raise RuntimeError(program.stderr)
report = json.loads(program.stdout)
tests = subprocess.run([sys.executable, "-I", "-B", "-m", "unittest", "discover",
"-s", str(destination), "-p", "test_wal_backup.py", "-v"],
cwd=cwd, env=env, capture_output=True, text=True, timeout=20)
if tests.returncode != 0:
raise RuntimeError(tests.stderr)
if "Ran 20 tests" not in tests.stderr:
raise AssertionError("expected 20 tests to run")
after = {p.name: file_hash(p) for p in sentinels.iterdir()}
if before != after:
raise AssertionError("unrelated sentinel files changed")
return {"report": report, "tests_exit": tests.returncode,
"test_count_20": "Ran 20 tests" in tests.stderr,
"sentinels_unchanged": before == after}
def main():
source = Path(__file__).resolve().parent
with tempfile.TemporaryDirectory(prefix="wal-repro-") as td:
root = Path(td)
clean = run_variant(root, source, polluted=False)
polluted = run_variant(root, source, polluted=True)
if clean["report"] != polluted["report"]:
raise AssertionError("clean and polluted runs differ")
print(json.dumps({"clean": clean, "polluted": polluted,
"reports_identical": True, "per_child_timeout_seconds": 20},
ensure_ascii=False, indent=2, sort_keys=True))
if __name__ == "__main__":
main()
20. 本次完整输出与测试回执
下面是 2026-10-13 的完整实验 JSON 与 20 项测试日志。WAL 大小 4152 bytes、文件哈希、两次进度状态 [0, 101] 是本次环境的观察值，不要求别的 SQLite 构建得到同样字节；固定业务预期是上一节手写的行与 schema。0 为 SQLITE_OK，101 为 SQLITE_DONE。两次直接运行的 JSON 逐字节一致。
完整实验 JSON
{
"during_uncommitted_backup": {
"progress": {
"callbacks": 2,
"ended_with_done": true,
"statuses": [
0,
101
]
},
"snapshot": {
"integrity": [
"ok"
],
"rows": [
[
1,
"baseline"
],
[
2,
"committed-in-wal"
]
],
"schema": [
[
"table",
"items",
"CREATE TABLE items(id INTEGER PRIMARY KEY,label TEXT NOT NULL)"
]
]
}
},
"fixture": {
"baseline_checkpoint": [
0,
0,
0
],
"fresh_reader_committed_rows": [
[
1,
"baseline"
],
[
2,
"committed-in-wal"
]
],
"journal_mode": "wal",
"main_sha256_after_row2": "c8223748d2ca20734062c2ab53f86c0f37b50eb36e8374a3c67057685136a27e",
"main_sha256_before_row2": "c8223748d2ca20734062c2ab53f86c0f37b50eb36e8374a3c67057685136a27e",
"old_reader_first_rows": [
[
1,
"baseline"
]
],
"wal_bytes_after_commit": 4152
},
"fresh_backup": {
"progress": {
"callbacks": 2,
"ended_with_done": true,
"statuses": [
0,
101
]
},
"snapshot": {
"integrity": [
"ok"
],
"rows": [
[
1,
"baseline"
],
[
2,
"committed-in-wal"
]
],
"schema": [
[
"table",
"items",
"CREATE TABLE items(id INTEGER PRIMARY KEY,label TEXT NOT NULL)"
]
]
}
},
"main_file_only": {
"integrity": [
"ok"
],
"rows": [
[
1,
"baseline"
]
],
"schema": [
[
"table",
"items",
"CREATE TABLE items(id INTEGER PRIMARY KEY,label TEXT NOT NULL)"
]
]
},
"old_reader_still_in_transaction": true,
"pinned_backup": {
"progress": {
"callbacks": 2,
"ended_with_done": true,
"statuses": [
0,
101
]
},
"snapshot": {
"integrity": [
"ok"
],
"rows": [
[
1,
"baseline"
]
],
"schema": [
[
"table",
"items",
"CREATE TABLE items(id INTEGER PRIMARY KEY,label TEXT NOT NULL)"
]
]
}
},
"released_backup": {
"progress": {
"callbacks": 2,
"ended_with_done": true,
"statuses": [
0,
101
]
},
"snapshot": {
"integrity": [
"ok"
],
"rows": [
[
1,
"baseline"
],
[
2,
"committed-in-wal"
]
],
"schema": [
[
"table",
"items",
"CREATE TABLE items(id INTEGER PRIMARY KEY,label TEXT NOT NULL)"
]
]
}
},
"runtime": {
"python": "3.12.14",
"sqlite": "3.53.1",
"system": "Linux"
},
"source_after_rollback": {
"integrity": [
"ok"
],
"rows": [
[
1,
"baseline"
],
[
2,
"committed-in-wal"
]
],
"schema": [
[
"table",
"items",
"CREATE TABLE items(id INTEGER PRIMARY KEY,label TEXT NOT NULL)"
]
]
},
"wal_still_nonempty_before_close": true,
"writer_uncommitted_rows": [
[
1,
"baseline"
],
[
2,
"committed-in-wal"
],
[
3,
"uncommitted"
]
]
}
完整测试回执
test_callback_count_is_bounded_even_if_clock_does_not_advance (test_wal_backup.GuardTests.test_callback_count_is_bounded_even_if_clock_does_not_advance) ... ok
test_callback_deadline_is_checked (test_wal_backup.GuardTests.test_callback_deadline_is_checked) ... ok
test_cli_refuses_real_database_path_argument (test_wal_backup.GuardTests.test_cli_refuses_real_database_path_argument) ... ok
test_exception_path_closes_all_case_connections (test_wal_backup.GuardTests.test_exception_path_closes_all_case_connections) ... ok
test_existing_directory_and_dangling_symlink_rejected (test_wal_backup.GuardTests.test_existing_directory_and_dangling_symlink_rejected) ... ok
test_existing_output_file_preserved (test_wal_backup.GuardTests.test_existing_output_file_preserved) ... ok
test_existing_sidecars_preserved (test_wal_backup.GuardTests.test_existing_sidecars_preserved) ... ok
test_invalid_budget_rejected_before_creating_output (test_wal_backup.GuardTests.test_invalid_budget_rejected_before_creating_output) ... ok
test_raw_copy_refuses_existing_target (test_wal_backup.GuardTests.test_raw_copy_refuses_existing_target) ... ok
test_same_connection_write_transaction_aborts_and_removes_partial_output (test_wal_backup.GuardTests.test_same_connection_write_transaction_aborts_and_removes_partial_output) ... ok
test_source_closed_error_cleans_owned_output (test_wal_backup.GuardTests.test_source_closed_error_cleans_owned_output) ... ok
test_every_file_reopened_after_all_original_connections_close (test_wal_backup.WalCaseTests.test_every_file_reopened_after_all_original_connections_close) ... ok
test_existing_case_directory_is_rejected (test_wal_backup.WalCaseTests.test_existing_case_directory_is_rejected) ... ok
test_existing_read_transaction_keeps_old_snapshot (test_wal_backup.WalCaseTests.test_existing_read_transaction_keeps_old_snapshot) ... ok
test_main_only_is_readable_but_misses_committed_row (test_wal_backup.WalCaseTests.test_main_only_is_readable_but_misses_committed_row) ... ok
test_new_reader_backup_contains_wal_commit (test_wal_backup.WalCaseTests.test_new_reader_backup_contains_wal_commit) ... ok
test_nonempty_wal_and_main_hash_do_not_mean_same_logical_state (test_wal_backup.WalCaseTests.test_nonempty_wal_and_main_hash_do_not_mean_same_logical_state) ... ok
test_other_connections_uncommitted_row_is_excluded (test_wal_backup.WalCaseTests.test_other_connections_uncommitted_row_is_excluded) ... ok
test_sql_commit_releases_the_old_snapshot (test_wal_backup.WalCaseTests.test_sql_commit_releases_the_old_snapshot) ... ok
test_successful_backups_finish_with_sqlite_done (test_wal_backup.WalCaseTests.test_successful_backups_finish_with_sqlite_done) ... ok
----------------------------------------------------------------------
Ran 20 tests in 0.057s
OK
父环境复跑摘要
{
"clean_tests": 20,
"polluted_tests": 20,
"reports_identical": true,
"clean_sentinels_unchanged": true,
"polluted_sentinels_unchanged": true,
"per_child_timeout_seconds": 20
}21. 把备份的验收时点也写进 Codex 任务
如果要让 Codex 扩展前面的迁移演练，先让它保留这组源状态、备份时点和重开验收条件。下面是一份可复制任务模板；本文没有把它发送给模型运行。
请只在新建临时目录扩展这份 SQLite 教学演练，不读取真实业务库。
保留已有迁移脚本和测试；另建 WAL 备份 fixture。
先把行 1 checkpoint 到主文件，用 BEGIN + SELECT 固定旧读快照，
再提交行 2 并证明 -wal 非空。分别比较主文件单拷、新读连接 backup、
旧读事务 backup，以及该事务 SQL COMMIT 后的 backup。
再从另一个连接插入但不提交行 3，验证只读连接的备份没有行 3。
每份输出都要关闭目标连接后重开，对照独立手写 rows、schema、integrity 预期。
保留主文件复制 integrity_check=ok 但缺少已提交行的反例。
拒绝覆盖已有输出；给 backup 回调与子进程设置中止边界；finally 关闭连接。
不要把主文件 SHA-256 不变当作活跃 WAL 逻辑状态不变的证明。
报告实际版本、完整命令、失败与成功输出及未测范围。这次补充不覆盖生产并发、备份期间连续提交、崩溃或断电恢复、网络文件系统、磁盘空间不足、加密扩展、多数据库原子性或提交后迁移逆向恢复。当前 SQLite WAL 文档还说明了 WAL-reset 问题及已修复版本；本次绑定 SQLite 3.53.1，没有测试旧版本并发写入/checkpoint 的安全性。真实部署应单独核对所用构建与修复状态。
本次核对了两个 SVG 的自然比例渲染、HTML 静态结构、完整内嵌源码和离线复跑。没有运行完整浏览器页面验收，也没有验证 CodexApp 页面或移动端交互。本补充保留前 12 节和原文章日期、版本等元数据，追加日期在补充开头单独说明。
旧版 SQLite 可以直接运行吗？
不可以。本例使用 SQLite 3.53.0 新增的 SET NOT NULL，脚本要求 Python >=3.12 与 SQLite >=3.53.0，并在版本不足时停止。
失败回滚代表可以撤销已经提交的迁移吗？
不代表。故障注入发生在 COMMIT 前，回滚只恢复当前尚未提交事务。已提交迁移的逆向恢复需要另外设计和测试。