with sqlite3.connect(':memory:') as db:
    db.executescript('CREATE TABLE results(id INT PRIMARY KEY,summary TEXT); CREATE TABLE cursor(value INT);')
    db.execute('INSERT INTO cursor VALUES(?)', (data['initial_watermark'],))
    db.commit()
    try:
        with db:
            db.execute('INSERT INTO results VALUES(?,?)', (2, '模拟未提交结果'))
            db.execute('UPDATE cursor SET value=2')
            raise RuntimeError('Explicit local crash fixture')
    except RuntimeError:
        pass
    check('事务中断后水位与结果一起回滚', db.execute('SELECT value FROM cursor').fetchone()[0] == 1 and db.execute('SELECT COUNT(*) FROM results').fetchone()[0] == 0)
    with db:
        for row in output['rows']:
            db.execute('INSERT INTO results VALUES(?,?)', (row['id'], row['summary']))
            db.execute('UPDATE cursor SET value=?', (row['id'],))
    check('恢复后只处理2、3并提交水位3', db.execute('SELECT id FROM results ORDER BY id').fetchall() == [(2,), (3,)] and db.execute('SELECT value FROM cursor').fetchone()[0] == 3)
