"""Real SQLite/ZIP checks on bounded synthetic data in an isolated temp directory. Follows restore_lab.py's capture/archive/restore/verify sequence. The original reference is retained separately; this adapter adds failures and cleanup. """ from contextlib import closing import hashlib import math from pathlib import Path import sqlite3 import tempfile import time import zipfile SCENARIOS=('intact','missing_note','modified_note','truncated_database','wrong_record') DEFAULT=dict(restore_case='intact',record_count=1) NOTE=b'Synthetic restore fixture; not production data.\n' EXPECTED='restore-check' def validate(p): if not isinstance(p,dict) or set(p)!=set(DEFAULT) or not isinstance(p['restore_case'],str) or p['restore_case'] not in SCENARIOS: raise ValueError('Unknown or missing parameters') n=p['record_count'] if type(n) not in (int,float) or not math.isfinite(n) or not 1<=n<=50 or n!=round(n): raise ValueError('Record count must be an integer from 1 to 50') def digest(path):return hashlib.sha256(path.read_bytes()).hexdigest() def prepare(root,p): """Only fixed filenames and generated data; no user archives or host inputs.""" source,snapshot,restored=(root/name for name in ('source','snapshot','restored')) for directory in (source,snapshot,restored):directory.mkdir() (source/'note.txt').write_bytes(NOTE) with closing(sqlite3.connect(source/'app.db')) as db: with db: db.execute('CREATE TABLE notes(id INTEGER PRIMARY KEY, text TEXT)') db.executemany('INSERT INTO notes VALUES(?, ?)',[(i,EXPECTED if i==1 else f'synthetic-note-{i}') for i in range(1,int(p['record_count'])+1)]) if p['restore_case']=='wrong_record':db.execute('UPDATE notes SET text=? WHERE id=1',('wrong-before-backup',)) captured_rows=db.execute('SELECT COUNT(*) FROM notes').fetchone()[0] with closing(sqlite3.connect(source/'app.db')) as db,closing(sqlite3.connect(snapshot/'app.db')) as dest:db.backup(dest) (snapshot/'note.txt').write_bytes((source/'note.txt').read_bytes()) manifest={name:digest(snapshot/name) for name in ('app.db','note.txt')} with zipfile.ZipFile(root/'backup.zip','w',zipfile.ZIP_DEFLATED) as archive: for name in manifest:archive.write(snapshot/name,name) with zipfile.ZipFile(root/'backup.zip') as archive: for name in manifest:(restored/name).write_bytes(archive.read(name)) if p['restore_case']=='missing_note':(restored/'note.txt').unlink() elif p['restore_case']=='modified_note':(restored/'note.txt').write_bytes(NOTE+b'Changed after extraction.\n') elif p['restore_case']=='truncated_database':(restored/'app.db').write_bytes((restored/'app.db').read_bytes()[:32]) return restored,manifest,captured_rows def inspect(restored,manifest): files=[] for name,expected in manifest.items(): target=restored/name;actual=None try: actual=digest(target);state='pass' if actual==expected else 'fail';reason='hash_match' if state=='pass' else 'hash_mismatch' except FileNotFoundError:state,reason='fail','missing_file' except OSError:state,reason='error','file_read_error' files.append(dict(name=name,state=state,reason=reason,expected_sha256=expected,actual_sha256=actual)) hash_state='error' if any(f['state']=='error' for f in files) else 'pass' if all(f['state']=='pass' for f in files) else 'fail' checks=dict(file_hashes=dict(state=hash_state,reason='captured_manifest'),sqlite_integrity=dict(state='not_run',reason='database_unavailable'),application_read=dict(state='not_run',reason='database_unavailable')) observed=None;rows=None try: # Read-only prevents creating an empty database if restoration lost it. with closing(sqlite3.connect((restored/'app.db').as_uri()+'?mode=ro',uri=True)) as db: integrity=[r[0] for r in db.execute('PRAGMA integrity_check').fetchall()] checks['sqlite_integrity']=dict(state='pass' if integrity==['ok'] else 'fail',reason='integrity_check',details=integrity) if integrity==['ok']: try: record=db.execute('SELECT text FROM notes WHERE id=1').fetchone() observed=record[0] if record else None rows=db.execute('SELECT COUNT(*) FROM notes').fetchone()[0] checks['application_read']=dict(state='pass' if observed==EXPECTED else 'fail',reason='known_record') except sqlite3.Error:checks['application_read']=dict(state='error',reason='query_error') else:checks['application_read']=dict(state='not_run',reason='integrity_not_passed') except sqlite3.Error: checks['sqlite_integrity']=dict(state='error',reason='database_error') checks['application_read']=dict(state='not_run',reason='integrity_not_passed') return files,checks,observed,rows def run(name,p): if name!='restore':raise ValueError('Unknown experiment') validate(p);start=time.perf_counter() with tempfile.TemporaryDirectory(prefix='duskcoil-browser-restore-') as directory: root=Path(directory);restored,manifest,captured=prepare(root,p) files,checks,observed,rows=inspect(restored,manifest) metrics=[dict(id='captured_records',value=captured,unit='records'),dict(id='restored_file_count',value=sum((restored/n).is_file() for n in manifest),unit='files'),dict(id='matching_file_count',value=sum(f['state']=='pass' for f in files),unit='files'),dict(id='readable_records',value=rows,unit='records')] if rows is None:metrics[-1]['reason']='Database query could not run' result=dict(metrics=metrics,series=[],stdout='',provenance=dict(data_type='synthetic',checks=checks,files=files,expected_record=EXPECTED,observed_record=observed, stages=['capture','archive','extract','verify'],scenario=p['restore_case'], limitations='Temporary synthetic fixture only. Application check is one SQL record, not a service restart. Hashes compare to this run\'s captured manifest; not proof of capture consistency.')) result['provenance']['temporary_directory_removed']=not root.exists() result['provenance']['compute_ms']=(time.perf_counter()-start)*1000 return result if __name__=='__main__': import argparse,json parser=argparse.ArgumentParser(description=__doc__);parser.add_argument('--parameters',default=json.dumps(DEFAULT)) print(json.dumps(run('restore',json.loads(parser.parse_args().parameters)),allow_nan=False))