書き込み試験に合格しても、既存の一覧や集計が移行前と同じ結果を返すとは限りません。
今回は、同じスナップショットを起点とする移行前後のSQLiteへ、既存業務を代表する3本のSELECTを実行します。列名、行順、値、SQLiteから返された型を突き合わせ、差分があれば切り替えを止めます。
稼働中のデータベースや、取得時点の異なるファイルは比較対象にしません。
3種類の試験を分ける
| 試験 | 確認できること | 確認できないこと |
|---|---|---|
| 構造検査 | 列、制約、インデックス、スキーマバージョン | 既存の検索が同じ結果を返すか |
| 書き込み試験 | 登録、更新、拒否、ロールバック | 一覧や集計の互換性 |
| 読み取り回帰テスト | 指定した検索の列名、行順、値、型 | 指定していない検索や、結果自体の業務上の正しさ |
新しく追加した列の機能確認など、移行後だけに存在する動作は、移行前との比較ではなく別の期待値テストで確認します。
今回の実行契約
| 項目 | 仕様 |
|---|---|
| 動作環境 | Python 3.10以上、外部ライブラリ不要 |
| 入力 | 同じスナップショットを起点とする、移行前後の独立したSQLiteファイル |
| 入力の状態 | 書き込みが終了し、隣に-wal、-shm、-journalがないこと |
| 接続 | URIのmode=roとPRAGMA query_only=ONを使用する |
| 比較 | 列名、行数、行順、値、値の型を完全一致で比較する |
| 生成物 | 新しいJSONレポート。既存ファイルは上書きしない |
| レポート内容 | 件数、列名、結果のSHA-256、最初の差分の種類と位置。セル値は記録しない |
| 対象外 | 性能測定、意図したデータ変換の承認、稼働中DBの比較、未指定クエリの保証 |
ファイルを読み取り専用で開く指定はmode=roです。query_onlyも設定しますが、これだけでデータベースファイルが完全な読み取り専用になるわけではありません。
複数行を返すクエリでは、一意になる並び順を指定します。同順位が残る場合は主キーまでORDER BYへ追加します。
1．再現用のSQLiteを作る
新しい空のディレクトリを用意します。コマンド例はPOSIX系シェル向けです。
mkdir sqlite-read-regression-demo
cd sqlite-read-regression-demo次のコードをmake_fixture.pyとして保存します。移行前のbaseline.sqliteを作り、そのファイルを複製してからpriority列を追加したcandidate.sqliteを作ります。データは動作確認用です。
import json
import shutil
import sqlite3
from pathlib import Path
BASELINE = Path('baseline.sqlite')
CANDIDATE = Path('candidate.sqlite')
OUTPUTS = (BASELINE, CANDIDATE)
SCHEMA_V1 = '''
CREATE TABLE review_items (
batch_id TEXT NOT NULL,
record_id TEXT NOT NULL,
proposed_category TEXT NOT NULL,
status TEXT NOT NULL CHECK (
status IN ('pending', 'accepted', 'corrected', 'rejected')
),
confirmed_category TEXT,
PRIMARY KEY (batch_id, record_id)
);
'''
ROWS = (
('B001', 'R001', '配送', 'pending', None),
('B001', 'R002', '請求', 'accepted', '請求'),
('B001', 'R003', '配送', 'corrected', 'その他'),
('B002', 'R004', '請求', 'rejected', None),
)
def create_baseline(path):
with path.open('xb'):
pass
connection = sqlite3.connect(path)
try:
connection.executescript(SCHEMA_V1)
connection.executemany(
'''
INSERT INTO review_items (
batch_id,
record_id,
proposed_category,
status,
confirmed_category
) VALUES (?, ?, ?, ?, ?)
''',
ROWS,
)
connection.execute('PRAGMA user_version = 1')
connection.commit()
except BaseException:
connection.rollback()
raise
finally:
connection.close()
def copy_database(source, destination):
with source.open('rb') as source_handle:
with destination.open('xb') as destination_handle:
shutil.copyfileobj(source_handle, destination_handle)
def migrate_candidate(path):
connection = sqlite3.connect(path)
try:
connection.execute(
'''
ALTER TABLE review_items
ADD COLUMN priority INTEGER NOT NULL DEFAULT 0
CHECK (priority BETWEEN 0 AND 3)
'''
)
connection.execute('PRAGMA user_version = 2')
connection.commit()
except BaseException:
connection.rollback()
raise
finally:
connection.close()
def main():
if any(path.exists() for path in OUTPUTS):
raise SystemExit('run this fixture in a new empty directory')
try:
create_baseline(BASELINE)
copy_database(BASELINE, CANDIDATE)
migrate_candidate(CANDIDATE)
except BaseException:
for path in OUTPUTS:
path.unlink(missing_ok=True)
raise
print(json.dumps(
{'files': [str(path) for path in OUTPUTS], 'status': 'created'},
ensure_ascii=False,
separators=(',', ':'),
sort_keys=True,
))
if __name__ == '__main__':
main()2．読み取り回帰テストを実装する
次のコードをsqlite_read_regression.pyとして保存します。実行するSQLはCASESへ固定し、コマンドラインから任意のSQLを渡せない形にします。
import argparse
import hashlib
import json
import os
import sqlite3
import sys
from pathlib import Path
MAX_ROWS = 10_000
MAX_COLUMNS = 100
MAX_RESULT_BYTES = 20 * 1024 * 1024
PROGRESS_INTERVAL = 1_000
MAX_PROGRESS_CALLS = 20_000
CASES = (
{
'name': 'pending_items',
'sql': '''SELECT batch_id, record_id, proposed_category, status
FROM review_items
WHERE status = ?
ORDER BY batch_id, record_id''',
'params': ('pending',),
},
{
'name': 'status_counts',
'sql': '''SELECT status, COUNT(*) AS item_count
FROM review_items
GROUP BY status
ORDER BY status''',
'params': (),
},
{
'name': 'final_categories',
'sql': '''SELECT
batch_id,
record_id,
CASE
WHEN status = 'corrected' THEN confirmed_category
ELSE proposed_category
END AS final_category
FROM review_items
WHERE status IN (?, ?)
ORDER BY batch_id, record_id''',
'params': ('accepted', 'corrected'),
},
)
class InputError(Exception):
pass
class CaseRunError(Exception):
pass
def compact_bytes(value):
return json.dumps(
value,
ensure_ascii=False,
separators=(',', ':'),
sort_keys=True,
allow_nan=False,
).encode('utf-8')
def emit(stream, value):
print(json.dumps(
value,
ensure_ascii=False,
separators=(',', ':'),
sort_keys=True,
allow_nan=False,
), file=stream, flush=True)
def sha256_file(path):
digest = hashlib.sha256()
with path.open('rb') as handle:
while True:
chunk = handle.read(1024 * 1024)
if not chunk:
break
digest.update(chunk)
return digest.hexdigest()
def find_sidecars(entries):
found = []
for label, path in entries:
for suffix in ('-wal', '-shm', '-journal'):
sidecar = path.with_name(path.name + suffix)
if sidecar.exists():
found.append((label, sidecar))
return found
def validate_inputs(baseline_path, candidate_path, report_path):
for label, path in (
('baseline', baseline_path),
('candidate', candidate_path),
):
if not path.is_file():
raise InputError(f'{label} is not a file: {path}')
if baseline_path.samefile(candidate_path):
raise InputError('baseline and candidate must be different files')
if report_path.exists():
raise InputError(f'report already exists: {report_path}')
if not report_path.parent.is_dir():
raise InputError(f'report directory does not exist: {report_path.parent}')
names = set()
for case in CASES:
name = case['name']
if name in names:
raise InputError(f'duplicate case name: {name}')
names.add(name)
words = case['sql'].split(None, 1)
if not words or words[0].upper() != 'SELECT':
raise InputError(f'case must start with SELECT: {name}')
def open_readonly(path):
uri = path.resolve().as_uri() + '?mode=ro'
connection = sqlite3.connect(
uri,
uri=True,
isolation_level=None,
timeout=5.0,
)
try:
connection.execute('PRAGMA query_only = ON')
connection.execute('BEGIN')
connection.execute('SELECT COUNT(*) FROM sqlite_master').fetchone()
return connection
except BaseException:
connection.close()
raise
def normalize_value(value):
if value is None:
return ['null', None]
if type(value) is int:
return ['integer', str(value)]
if type(value) is float:
return ['real', value.hex()]
if type(value) is str:
return ['text', value]
if isinstance(value, (bytes, bytearray, memoryview)):
return ['blob', bytes(value).hex()]
raise CaseRunError(
f'unsupported SQLite result type: {type(value).__name__}'
)
def add_digest_piece(digest, piece):
digest.update(len(piece).to_bytes(8, 'big'))
digest.update(piece)
def read_result(connection, case):
calls = 0
cursor = None
def progress():
nonlocal calls
calls += 1
return 1 if calls > MAX_PROGRESS_CALLS else 0
connection.set_progress_handler(progress, PROGRESS_INTERVAL)
try:
cursor = connection.execute(case['sql'], case['params'])
if cursor.description is None:
raise CaseRunError('query did not return a result set')
columns = [description[0] for description in cursor.description]
if len(columns) > MAX_COLUMNS:
raise CaseRunError('column limit exceeded')
digest = hashlib.sha256()
header_piece = compact_bytes(['columns', columns])
total_bytes = len(header_piece)
if total_bytes > MAX_RESULT_BYTES:
raise CaseRunError('result byte limit exceeded')
add_digest_piece(digest, header_piece)
rows = []
while True:
row = cursor.fetchone()
if row is None:
break
if len(rows) >= MAX_ROWS:
raise CaseRunError('row limit exceeded')
normalized = [normalize_value(value) for value in row]
row_piece = compact_bytes(normalized)
total_bytes += len(row_piece)
if total_bytes > MAX_RESULT_BYTES:
raise CaseRunError('result byte limit exceeded')
add_digest_piece(digest, row_piece)
rows.append(normalized)
return {
'columns': columns,
'rows': rows,
'row_count': len(rows),
'sha256': digest.hexdigest(),
}
except sqlite3.OperationalError as exc:
if calls > MAX_PROGRESS_CALLS:
raise CaseRunError('query operation budget exceeded') from exc
raise
finally:
connection.set_progress_handler(None, 0)
if cursor is not None:
cursor.close()
def definition_hash(case):
payload = {
'params': list(case['params']),
'sql': case['sql'],
}
return hashlib.sha256(compact_bytes(payload)).hexdigest()
def result_summary(result):
return {
'columns': result['columns'],
'row_count': result['row_count'],
'sha256': result['sha256'],
}
def compare_case(baseline_connection, candidate_connection, case):
payload = {
'definition_sha256': definition_hash(case),
'name': case['name'],
}
results = {}
errors = {}
for label, connection in (
('baseline', baseline_connection),
('candidate', candidate_connection),
):
try:
results[label] = read_result(connection, case)
except (sqlite3.Error, CaseRunError, UnicodeError) as exc:
errors[label] = {
'message': str(exc),
'type': type(exc).__name__,
}
if errors:
for label, result in results.items():
payload[label] = result_summary(result)
payload['errors'] = errors
payload['status'] = 'error'
return payload
baseline = results['baseline']
candidate = results['candidate']
payload['baseline'] = result_summary(baseline)
payload['candidate'] = result_summary(candidate)
first_difference = None
if baseline['columns'] != candidate['columns']:
common_columns = min(
len(baseline['columns']),
len(candidate['columns']),
)
column_index = common_columns + 1
for index in range(common_columns):
if baseline['columns'][index] != candidate['columns'][index]:
column_index = index + 1
break
first_difference = {
'column_index': column_index,
'kind': 'columns',
}
else:
common_rows = min(baseline['row_count'], candidate['row_count'])
for row_index in range(common_rows):
baseline_row = baseline['rows'][row_index]
candidate_row = candidate['rows'][row_index]
if baseline_row == candidate_row:
continue
for cell_index, (baseline_cell, candidate_cell) in enumerate(
zip(baseline_row, candidate_row)
):
if baseline_cell != candidate_cell:
difference_kind = (
'type'
if baseline_cell[0] != candidate_cell[0]
else 'value'
)
first_difference = {
'column': baseline['columns'][cell_index],
'column_index': cell_index + 1,
'kind': difference_kind,
'row_number': row_index + 1,
}
break
break
if first_difference is None and (
baseline['row_count'] != candidate['row_count']
):
first_difference = {
'kind': 'row_count',
'row_number': common_rows + 1,
}
if first_difference is None:
payload['status'] = 'passed'
else:
payload['first_difference'] = first_difference
payload['status'] = 'failed'
return payload
def write_report(report_path, document):
payload = json.dumps(
document,
ensure_ascii=False,
indent=2,
sort_keys=True,
allow_nan=False,
)
created = False
try:
with report_path.open('x', encoding='utf-8', newline='') as handle:
created = True
handle.write(payload)
handle.write(chr(10))
handle.flush()
os.fsync(handle.fileno())
except FileExistsError as exc:
raise InputError(f'report already exists: {report_path}') from exc
except BaseException:
if created:
try:
report_path.unlink()
except OSError:
pass
raise
def run(args):
baseline_path = args.baseline
candidate_path = args.candidate
report_path = args.report
validate_inputs(baseline_path, candidate_path, report_path)
entries = (
('baseline', baseline_path),
('candidate', candidate_path),
)
initial_sidecars = find_sidecars(entries)
if initial_sidecars:
details = ', '.join(
f'{label}:{path.name}' for label, path in initial_sidecars
)
raise InputError(f'SQLite sidecar found: {details}')
initial_hashes = {
'baseline': sha256_file(baseline_path),
'candidate': sha256_file(candidate_path),
}
baseline_connection = None
candidate_connection = None
try:
baseline_connection = open_readonly(baseline_path)
candidate_connection = open_readonly(candidate_path)
baseline_version = baseline_connection.execute(
'PRAGMA user_version'
).fetchone()[0]
candidate_version = candidate_connection.execute(
'PRAGMA user_version'
).fetchone()[0]
case_reports = [
compare_case(baseline_connection, candidate_connection, case)
for case in CASES
]
finally:
for connection in (candidate_connection, baseline_connection):
if connection is not None:
connection.close()
final_hashes = {
'baseline': sha256_file(baseline_path),
'candidate': sha256_file(candidate_path),
}
post_sidecars = find_sidecars(entries)
stability_problems = []
for label in ('baseline', 'candidate'):
if initial_hashes[label] != final_hashes[label]:
stability_problems.append(f'{label}_changed_during_test')
for label, path in post_sidecars:
stability_problems.append(
f'{label}_sidecar_appeared:{path.name}'
)
if stability_problems:
overall_status = 'error'
exit_code = 2
elif all(case['status'] == 'passed' for case in case_reports):
overall_status = 'passed'
exit_code = 0
else:
overall_status = 'failed'
exit_code = 1
document = {
'case_count': len(case_reports),
'cases': case_reports,
'input_stability': {
'problems': stability_problems,
'status': 'passed' if not stability_problems else 'failed',
},
'inputs': {
'baseline': {
'file': baseline_path.name,
'sha256': initial_hashes['baseline'],
'user_version': baseline_version,
},
'candidate': {
'file': candidate_path.name,
'sha256': initial_hashes['candidate'],
'user_version': candidate_version,
},
},
'status': overall_status,
'version': 1,
}
write_report(report_path, document)
emit(sys.stdout, {
'case_count': len(case_reports),
'report': str(report_path),
'status': overall_status,
})
return exit_code
def main(argv=None):
parser = argparse.ArgumentParser()
parser.add_argument('baseline', type=Path)
parser.add_argument('candidate', type=Path)
parser.add_argument('report', type=Path)
args = parser.parse_args(argv)
try:
return run(args)
except InputError as exc:
emit(sys.stderr, {'error': str(exc), 'status': 'input_error'})
return 2
except KeyboardInterrupt:
emit(sys.stderr, {'error': 'interrupted', 'status': 'interrupted'})
return 130
except (OSError, sqlite3.Error) as exc:
emit(sys.stderr, {
'error': str(exc),
'error_type': type(exc).__name__,
'status': 'execution_error',
})
return 3
except Exception as exc:
emit(sys.stderr, {
'error': str(exc),
'error_type': type(exc).__name__,
'status': 'unexpected_error',
})
return 3
if __name__ == '__main__':
raise SystemExit(main())3．正常系を実行する
python make_fixture.py
python sqlite_read_regression.py baseline.sqlite candidate.sqlite report.json
echo $?標準出力と終了コードは次のようになります。
{"files":["baseline.sqlite","candidate.sqlite"],"status":"created"}
{"case_count":3,"report":"report.json","status":"passed"}
0レポートは標準ライブラリで整形できます。
python -m json.tool report.json| 項目 | 意味 |
|---|---|
definition_sha256 |
SQLとバインド値をまとめたテスト定義の識別値 |
columns |
SQLiteが返した列名と順序 |
row_count |
結果行数 |
sha256 |
型付きで正規化した列名と全行の結果ハッシュ |
first_difference |
最初に見つかった差分の種類と、1から始まる列番号または行番号 |
input_stability |
試験中に入力ファイルやサイドカーファイルが変化しなかったか |
first_difference.kindは、列構成の差ならcolumns、行数の差ならrow_count、セルの型の差ならtype、同じ型の値の差ならvalueです。列構成の差には最初に異なるcolumn_index、セルの差にはrow_number、column_index、列名を記録します。行数の差では、片方だけに存在する最初の行番号をrow_numberへ記録します。
値は内部でinteger、real、text、blob、nullのタグ付き表現へ変換します。Pythonでは1 == 1.0が真になりますが、このテストでは整数と実数を別の型として検出します。
4．差分を検出する
次の操作は、ここで作ったcandidate.sqliteだけに実行してください。
make_difference.pyとして保存します。
import sqlite3
connection = sqlite3.connect('candidate.sqlite')
try:
cursor = connection.execute(
'''
UPDATE review_items
SET status = ?, confirmed_category = ?
WHERE batch_id = ? AND record_id = ?
''',
('accepted', '配送', 'B001', 'R001'),
)
if cursor.rowcount != 1:
raise RuntimeError('expected exactly one updated row')
connection.commit()
except BaseException:
connection.rollback()
raise
finally:
connection.close()report.jsonは上書きできないため、新しい出力名を指定します。
python make_difference.py
python sqlite_read_regression.py baseline.sqlite candidate.sqlite report-diff.json
echo $?検索結果が一致しないため、終了コードは1になります。
{"case_count":3,"report":"report-diff.json","status":"failed"}
1report-diff.jsonには、各失敗ケースで最初に見つかった差分の種類と位置が記録されます。セル値は複製しません。ただし、列名、件数、ハッシュも業務データから派生した情報です。レポートには元データに準じたアクセス制御を適用します。
終了コードを後続処理へ渡す
| 終了コード | 意味 | 判断 |
|---|---|---|
| 0 | 全クエリが一致し、入力も変化しなかった | ほかの切り替え条件の確認へ進む |
| 1 | 結果差分またはクエリ実行エラー | 切り替えを止め、SQLと移行内容を確認する |
| 2 | 入力契約違反、既存レポート、サイドカー、入力内容の変化 | 隔離コピーと実行条件を確認する |
| 3 | ファイル操作、SQLite接続、予期しない実行エラー | 実行環境を確認する |
| 130 | Ctrl+Cによる中断 | 未完了として扱う |
argparseが引数不足を検出した場合も終了コードは2です。この場合は通常の使用方法が表示され、1行JSONにはなりません。
実データへ合わせるときの注意点
- 稼働中DBを直接指定しない：SQLiteのバックアップAPIなどで閉じたスナップショットを作り、移行候補はそのスナップショットから作成します。
- 重要な検索を選ぶ：未処理一覧、確定結果、請求対象など、切り替え後に使う検索を少数から登録します。
- 並び順を固定する：複数行を比較する場合は、一意キーまで含めた
ORDER BYを指定します。ORDER BYがない結果や、同順位が残る結果の順序は保証されません。 - 意図した差分を混ぜない：移行で値を変える仕様なら、完全一致の条件を緩めず、変換後の期待値を確認する別テストに分けます。
- 同じ誤りは検出できない：移行前後が同じ誤った値を返せば合格します。業務上の正解を確認するテストの代わりにはなりません。
- 大規模な抽出には使わない：1ケースあたり1万行、100列、正規化したJSON断片の合計20MiBで停止します。進捗ハンドラーによる上限は、処理時間の上限ではありません。
- ハッシュを改ざん証明にしない：SHA-256は対象や結果の取り違え検出には使えますが、データベースとレポートの両方を変更できる利用者に対する改ざん証明にはなりません。
最初は、切り替え後に必ず使う検索を1本だけCASESへ移します。隔離コピーで合格条件を確認してから、一覧、集計、確定データの順に増やします。