79 lines
3.7 KiB
Python
79 lines
3.7 KiB
Python
|
"""Offline cross-engine transfer planning and all-or-nothing checks using a
|
||
|
|
SQLite-backed PG contract fixture. Actual PostgreSQL is tested only in CI."""
|
||
|
|
import sqlite3
|
||
|
|
import pytest
|
||
|
|
from pathlib import Path
|
||
|
|
from app.sqlite_pg_transfer import migrate, plan_tables, quote
|
||
|
|
from app.pg_disaster_recovery import RecoveryError
|
||
|
|
|
||
|
|
|
||
|
|
class PgCompatibleSQLite:
|
||
|
|
def __init__(self, path):
|
||
|
|
self.path=path
|
||
|
|
self.conn=None
|
||
|
|
def __enter__(self):
|
||
|
|
self.conn=sqlite3.connect(self.path)
|
||
|
|
self.conn.row_factory=sqlite3.Row
|
||
|
|
self.conn.execute('PRAGMA foreign_keys=ON')
|
||
|
|
return self
|
||
|
|
def __exit__(self,typ,val,tb):
|
||
|
|
if typ:self.conn.rollback()
|
||
|
|
else:self.conn.commit()
|
||
|
|
self.conn.close()
|
||
|
|
def execute(self,sql,params=()):
|
||
|
|
if 'information_schema.columns' in sql:
|
||
|
|
return self.conn.execute('SELECT name AS column_name FROM pragma_table_info(?)', params)
|
||
|
|
return self.conn.execute(sql, params)
|
||
|
|
|
||
|
|
|
||
|
|
def fixture_data(src,dest):
|
||
|
|
schema='''CREATE TABLE sites(id TEXT PRIMARY KEY,name TEXT);
|
||
|
|
CREATE TABLE entries(id TEXT PRIMARY KEY,site_id TEXT REFERENCES sites(id),title TEXT);
|
||
|
|
CREATE TABLE schema_migrations(version TEXT PRIMARY KEY,checksum TEXT);'''
|
||
|
|
for path in (src,dest):
|
||
|
|
with sqlite3.connect(path) as conn:conn.executescript(schema)
|
||
|
|
with sqlite3.connect(src) as conn:
|
||
|
|
conn.execute('INSERT INTO sites VALUES (?,?)',('a','中文站'))
|
||
|
|
conn.executemany('INSERT INTO entries VALUES (?,?,?)', [('1','a','Rox 01'),('2','a','BAO 5')])
|
||
|
|
conn.execute('INSERT INTO schema_migrations VALUES (?,?)',('v1','abc'))
|
||
|
|
with sqlite3.connect(dest) as conn:conn.execute('INSERT INTO schema_migrations VALUES (?,?)',('v1','abc'))
|
||
|
|
|
||
|
|
|
||
|
|
def test_dependent_tables_transferred_with_hashes(tmp_path):
|
||
|
|
src=tmp_path/'src.db';dest=tmp_path/'dest.db';fixture_data(src,dest)
|
||
|
|
with sqlite3.connect(src) as conn:
|
||
|
|
assert plan_tables(conn).index('sites')<plan_tables(conn).index('entries')
|
||
|
|
report=migrate(src,confirmed_offline=True,connect=lambda:PgCompatibleSQLite(dest),bootstrap=False,batch=1)
|
||
|
|
assert report['verified'] and report['tables']==2 and report['records']==3
|
||
|
|
with sqlite3.connect(dest) as conn:
|
||
|
|
assert conn.execute('SELECT title FROM entries WHERE id="2"').fetchone()[0]=='BAO 5'
|
||
|
|
assert 'sha256' in report['details']['entries']
|
||
|
|
|
||
|
|
|
||
|
|
def test_target_nonempty_refuses_all_changes(tmp_path):
|
||
|
|
src=tmp_path/'src.db';dest=tmp_path/'dest.db';fixture_data(src,dest)
|
||
|
|
with sqlite3.connect(dest) as conn:conn.execute('INSERT INTO sites VALUES (?,?)',('existing','keep'))
|
||
|
|
with pytest.raises(RecoveryError,match='not empty'):
|
||
|
|
migrate(src,confirmed_offline=True,connect=lambda:PgCompatibleSQLite(dest),bootstrap=False)
|
||
|
|
with sqlite3.connect(dest) as conn:
|
||
|
|
assert conn.execute('SELECT COUNT(*) FROM entries').fetchone()[0]==0
|
||
|
|
assert conn.execute('SELECT name FROM sites').fetchone()[0]=='keep'
|
||
|
|
|
||
|
|
|
||
|
|
def test_migration_guard_and_schema_mismatch(tmp_path):
|
||
|
|
src=tmp_path/'src.db';dest=tmp_path/'dest.db';fixture_data(src,dest)
|
||
|
|
with pytest.raises(RecoveryError,match='confirmed-offline'):
|
||
|
|
migrate(src,connect=lambda:PgCompatibleSQLite(dest),bootstrap=False)
|
||
|
|
with sqlite3.connect(dest) as conn:
|
||
|
|
conn.execute('DELETE FROM schema_migrations')
|
||
|
|
conn.execute('INSERT INTO schema_migrations VALUES (?,?)',('v1','wrong-checksum'))
|
||
|
|
with pytest.raises(RecoveryError,match='ledger'):
|
||
|
|
migrate(src,confirmed_offline=True,connect=lambda:PgCompatibleSQLite(dest),bootstrap=False)
|
||
|
|
|
||
|
|
|
||
|
|
def test_identifiers_are_validated(tmp_path):
|
||
|
|
with pytest.raises(RecoveryError):quote('entries; DROP TABLE sites')
|
||
|
|
conn=sqlite3.connect(':memory:')
|
||
|
|
with pytest.raises(RecoveryError,match='no sites'):
|
||
|
|
plan_tables(conn)
|