浅念
601451f22b
Overlay ~150 upstream items onto our V8.12 base, preserving trade-pro
theme, registry tests, and design docs. Re-applied our patch set
(discovery/env_for/blocks.css/type-mapping/theme_color/frontend entries,
test_theme_pro_v3 mods) — all anchors survived upstream evolution.
New platform enhancement (fixes upgrade-introduced regression):
multilingual.site_locale now honors settings_json.locale, plus
PUT /api/i18n/sites/{id}/locale — standalone English sites get
<html lang='en'> under the V8.15 locale-rewrite pipeline.
compose: adopt V8.21.3 (image 8.21.3-dev.1, /api/ready probes, MySQL/S3
env) with project name pinned to siteforge-v8-10 (data volumes).
Tests: 356 passed / 11 failed — all 11 confirmed pre-existing in
pristine V8.21.3 on this host (10 upstream Windows env + our build test
now green after locale fix). Patch registry updated in plan doc.
Co-Authored-By: Claude Code <noreply@anthropic.com>
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)
|