"""Execute crash, restart, contention, snapshot and recovery checks on synthetic catalogs."""
import concurrent.futures, hashlib, http.client, json, os, shutil, sqlite3, subprocess, sys, tempfile, threading, time
from pathlib import Path
from app import Application, Catalog
from operations import backup, restore, inspect_database
from operations_report import Record
ROOT=Path(__file__).resolve().parent
HEADER='part_number,description,price,currency,unit\n'
def approve(c,bid):
 b=c.batch(bid);return c.approve(bid,b['base_version'],b['review_digest'])
def fingerprint(path):
 with sqlite3.connect(path) as db:rows={t:db.execute('SELECT * FROM '+t+' ORDER BY 1,2').fetchall() for t in ['suppliers','items','batches','audit']}
 return hashlib.sha256(json.dumps(rows,sort_keys=True).encode()).hexdigest()
def child(mode,path,bid):
 c=Catalog(path)
 if mode=='--crash-before-commit':
  original=c._audit
  def audit(db,batch,action,actor,detail):
   original(db,batch,action,actor,detail)
   if action=='applied':os._exit(86)
  c._audit=audit;approve(c,bid)
 elif mode=='--crash-after-commit':approve(c,bid);os._exit(87)
def request(s,path,body=None,headers=None):
 c=http.client.HTTPConnection('127.0.0.1',s.server_port,timeout=4);c.request('POST' if body is not None else 'GET',path,json.dumps(body) if body is not None else None,{'Content-Type':'application/json',**(headers or {})});r=c.getresponse();out=(r.status,json.loads(r.read()),dict(r.getheaders()));c.close();return out
def launch(c):
 s=Application(('127.0.0.1',0),c);t=threading.Thread(target=s.serve_forever,daemon=True);t.start();return s,t
def stop(s,t):s.shutdown();s.server_close();t.join()
def main():
 r=Record('Catalog import operations rehearsal')
 with tempfile.TemporaryDirectory() as td:
  home=Path(td);path=home/'catalog.sqlite';c=Catalog(path);c.seed();text=HEADER+'BRG-10,Bearing,13.25,USD,EA\nSHIM-2,Shim,2.10,USD,EA\n';bid=c.stage('ACME',text)['id'];before=fingerprint(path)
  run=subprocess.run([sys.executable,__file__,'--crash-before-commit',str(path),bid],capture_output=True,timeout=15)
  r.check('crash.rollback','Abrupt process exit before commit rolls back every application write',{'exit':86,'unchanged':True,'status':'staged'},{'exit':run.returncode,'unchanged':fingerprint(path)==before,'status':Catalog(path).batch(bid)['status']},'Child calls os._exit after item, version, batch and audit writes, before commit. Parent reopens SQLite.')
  run=subprocess.run([sys.executable,__file__,'--crash-after-commit',str(path),bid],capture_output=True,timeout=15);restarted=Catalog(path);result=approve(restarted,bid);applied=sum(x['action']=='applied' for x in restarted.batch(bid)['audit'])
  r.check('crash.response','Retry after committed child exits returns original approval',{'exit':87,'replayed':True,'version':1,'applications':1},{'exit':run.returncode,'replayed':result['replayed'],'version':result['version'],'applications':applied},'Child commits then exits without returning a response. Parent replays the same batch identity.')
  count=8;rows=250
  def import_supplier(i):
   local=Catalog(path);supplier=f'SUP{i}';source=HEADER+''.join(f'PART-{j},Synthetic part {j},{j%40+1}.25,USD,EA\n' for j in range(rows));b=local.stage(supplier,source)['id'];approve(local,b);return local.items(supplier)['total']
  started=time.perf_counter()
  with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:totals=list(pool.map(import_supplier,range(count)))
  elapsed=time.perf_counter()-started
  r.check('load.concurrent','Eight suppliers preserve all rows under four concurrent writers',[rows]*count,totals,'Separate SQLite connections stage and commit eight 250-row catalogs. SQLite serializes writes.')
  versions={s['id']:s['version'] for s in c.summary()['suppliers'] if s['id'].startswith('SUP')}
  r.check('load.versions','Each concurrent supplier advances exactly once',{f'SUP{i}':1 for i in range(count)},versions,'Read persisted supplier versions after every worker completes.')
  r.metrics['concurrent_imports']={'suppliers':count,'rows_per_supplier':rows,'workers':4,'elapsed_seconds':round(elapsed,4),'committed_rows_per_second':round(count*rows/elapsed,1),'qualification':'One synthetic shared-host run; not sustained production capacity.'}
  s,t=launch(c)
  try:
   status,login,headers=request(s,'/api/login',{'username':'reviewer','password':'review-demo'});cookie=headers['Set-Cookie'].split(';')[0]
   r.check('http.local','Reviewer login and authenticated read succeed',[200,200],[status,request(s,'/api/summary',headers={'Cookie':cookie})[0]],'Real loopback HTTP requests.')
   r.check('http.boundary','Foreign Host and Origin requests are rejected',[403,403],[request(s,'/api/summary',headers={'Cookie':cookie,'Host':'evil.invalid'})[0],request(s,'/api/summary',headers={'Cookie':cookie,'Origin':'https://evil.invalid'})[0]],'Otherwise authenticated requests with foreign Host or Origin.')
  finally:stop(s,t)
  s,t=launch(Catalog(path))
  try:r.check('restart.session','Process-local sessions do not survive restart',401,request(s,'/api/summary',headers={'Cookie':cookie})[0],'Create a new application with the same database and replay the old cookie.')
  finally:stop(s,t)
  with c.connect() as active:
   active.execute('BEGIN IMMEDIATE');active.execute("INSERT INTO suppliers VALUES('UNCOMMITTED',0)");snapshot=backup(path,home/'snapshot');active.rollback()
  with sqlite3.connect(home/'snapshot/store.sqlite') as db:transient=db.execute("SELECT COUNT(*) FROM suppliers WHERE id='UNCOMMITTED'").fetchone()[0]
  r.check('backup.snapshot','Online snapshot excludes uncommitted data and passes integrity',{'transient_rows':0,'integrity':'ok'},{'transient_rows':transient,'integrity':snapshot['integrity']},'Hold a WAL writer open while SQLite online backup reads a committed snapshot.')
  started=time.perf_counter();restored=restore(home/'snapshot',home/'restored');restore_time=time.perf_counter()-started
  r.check('restore.business','Restored supplier, item, batch and audit content is identical',True,fingerprint(path)==fingerprint(restored['database']),'Verify checksum and integrity, restore to a new directory, hash all business rows.')
  r.check('restore.idempotency','Restored batch still prevents duplicate application',True,approve(Catalog(restored['database']),bid)['replayed'],'Execute approval again against the restored database.')
  shutil.copytree(home/'snapshot',home/'damaged')
  with (home/'damaged/store.sqlite').open('r+b') as f:f.seek(100);f.write(b'BROKEN')
  try:restore(home/'damaged',home/'must-not-exist');rejected=False
  except ValueError:rejected=not (home/'must-not-exist').exists()
  r.check('restore.corruption','Corrupt snapshot rejected before destination creation',True,rejected,'Modify six snapshot bytes; SHA-256 no longer matches.')
  (home/'occupied').mkdir();(home/'occupied/keep.txt').write_text('keep')
  try:restore(home/'snapshot',home/'occupied');preserved=False
  except FileExistsError:preserved=(home/'occupied/keep.txt').read_text()=='keep'
  r.check('restore.no-overwrite','Existing restore destination is preserved',True,preserved,'Attempt restore into an occupied directory.')
  r.metrics['recovery']={'snapshot_bytes':snapshot['bytes'],'snapshot_seconds':snapshot['elapsed_seconds'],'restore_seconds':round(restore_time,6),'rows':snapshot['table_counts'],'nominal_backup_interval_seconds':900,'theoretical_rpo_bound_seconds':round(900+snapshot['elapsed_seconds'],6),'rpo_assumption':'A successful snapshot at least every 15 minutes; source and backup not lost together. No unattended schedule was run.'}
  facts=inspect_database(path);r.check('integrity.final','Final source passes integrity and foreign-key checks',{'integrity':'ok','foreign_key_errors':0},{k:facts[k] for k in ['integrity','foreign_key_errors']},'PRAGMA integrity_check and foreign_key_check after crashes, imports and backup.')
  schedule=subprocess.run([sys.executable,str(ROOT/'backup_schedule.py'),'--db',str(path),'--out',str(home/'scheduled'),'--interval','0.01','--count','2'],capture_output=True,text=True,timeout=15)
  snapshots=list((home/'scheduled').glob('snapshot-*'))
  r.check('backup.schedule','Finite scheduler creates two checked snapshots and stops',{'exit':0,'snapshots':2,'valid':2},{'exit':schedule.returncode,'snapshots':len(snapshots),'valid':sum(inspect_database(p/'store.sqlite')['integrity']=='ok' for p in snapshots)},'Run the supplied CLI in a subprocess with two snapshots and a 10 ms interval. This exercises scheduling without leaving a background process.')
 return 0 if r.write(['app.py','schema.sql','operations.py','operations_rehearsal.py','operations_report.py','backup_schedule.py'],['Abrupt process termination exercises SQLite recovery, not physical power loss or loss of source and backup together.','Synthetic CSVs and public demonstration accounts were used. The service is loopback-only; no internet deployment or external identity provider is claimed.','Checksums detect accidental modification. Snapshots are not encrypted or digitally signed and need restricted storage permissions.','Durations describe one local run. Detection, provisioning and restart are outside the measured restore copy time.','The 15-minute recovery-point bound is a theoretical schedule requirement, not an enabled backup service.']) else 1
if __name__=='__main__':
 if len(sys.argv)>1 and sys.argv[1].startswith('--crash-'):child(*sys.argv[1:])
 else:raise SystemExit(main())