from backend.public_enquiries import TABLES as INTAKE_TABLES """Exact customer review recovery, adversarial archives and observed migration.""" import hashlib,json,shutil,sqlite3,zipfile from pathlib import Path import pytest from fastapi.testclient import TestClient from backend import recovery from backend.app import create_app from backend.store import DEMO_BUSINESSES from backend.customer_review_models import TABLES from test_boundaries import app,client,write,job,ORIGIN from test_customer_reviews import complex_history,selection,preview,confirm,customer,ROOT from test_identity import new_business from test_recovery import table_rows,new_login,rewrite_archive from test_portal_access import PASSWORD def run_customer_drill(tmp_path): root=tmp_path/'fictional';application=create_app(root);owner=client(application);records={} for n,slug in enumerate(('oak-and-vale','harbour-homeworks','willow-customer','cedar-customer')): if n<2:c=client(application,n);bid=DEMO_BUSINESSES[n]['id'];membership=c.get('/api/session').json()['membership_id'] else: b=new_business(owner,slug).json();bid=b['business_id'];membership=b['membership_id'] switched=write(owner,'/api/business/switch',{'membership_id':membership});owner.headers['x-csrf-token']=switched.json()['csrf'];c=owner a=complex_history(application,c);j=a['job'];target=job(c,name='Retained Fictional Customer') twin=job(c);pairs=c.get(ROOT+'/duplicates').json()['pairs'];p=next(p for p in pairs if {v['id'] for v in p['customers']}=={j['customer_id'],twin['customer_id']}) assert write(c,ROOT+'/pairs',{'left_id':p['customers'][0]['id'],'right_id':p['customers'][1]['id'],'version':0,'context_hash':p['context_hash'],'state':'keep_separate','confirmed':True,'reason':'Distinct fictional people checked by exact record'}).status_code==200 plan=preview(c,selection(c,j,target));assert plan['can_confirm'];r=confirm(c,plan);assert r.status_code==200,r.text cust=customer(c,target['customer_id']);assert write(c,'/api/customers/'+cust['id']+'/private-notes',{'version':cust['record_version'],'notes':'Fictional owner-private preference'},method='put').status_code==200 from test_workflow import edit edit(c,j,completed_date='') row={'customer_id':target['customer_id'],'customer_version':customer(c,target['customer_id'])['version'],'job_id':j['id'],'job_version':c.get('/api/jobs/'+j['id']).json()['version'],'invoice_id':a['invoice']['id'],'completed_date':'2026-10-01','evidence_note':'Checked an invented original completion note'} plan=write(c,ROOT+'/completion-preview',{'rows':[row]}).json();assert plan['can_confirm'] saved=write(c,ROOT+'/completions',{'rows':[row],'preview_hash':plan['preview_hash'],'reason':'Owner reviewed fictional completion evidence','confirmed':True});assert saved.status_code==200,saved.text records[bid]={'slug':slug,'member':membership,'source':j['customer_id'],'target':target['customer_id'],'twin':twin['customer_id'],'job':j['id'],'portal':a['portal'],'portal_id':a['portal_id'],'code':a['code'],'invoice':a['invoice']['id'],'before':table_rows(root,bid),'photos':{p.name:p.read_bytes() for p in (root/bid/'photos').iterdir()}} archive=tmp_path/'customers.zip';saved=recovery.backup(root,archive);assert recovery.verify(archive,saved['sha256'])['valid'] with zipfile.ZipFile(archive) as z: assert not any('capture' in p for p in z.namelist()) data=b''.join(z.read(n) for n in z.namelist()) for r in records.values():assert r['code'].encode() not in data shutil.rmtree(root);target=tmp_path/'restored';recovery.restore(archive,saved['sha256'],target,confirm_fresh=True);restored=create_app(target) unchanged=TABLES+('customers','jobs','job_workflow','job_workflow_history','quotes','invoices','document_events','quote_responses','payments','payment_details','field_logs','field_sessions','field_intervals','field_time_links','time_entries','document_sources','source_history','job_photos','photo_history','feedback_responses') for bid,r in records.items(): after=table_rows(target,bid) for table in unchanged: if table=='customers': assert [{k:v for k,v in row.items() if k not in ('portal_version','portal_hash')} for row in after[table]]==[{k:v for k,v in row.items() if k not in ('portal_version','portal_hash')} for row in r['before'][table]] else:assert after[table]==r['before'][table],table assert {p.name:p.read_bytes() for p in (target/bid/'photos').iterdir()}==r['photos'] a=TestClient(restored,headers=ORIGIN);a.cookies.update(r['portal'].cookies);base='/api/portal/'+r['slug'] assert a.get(base+'/me').status_code==401 assert a.post(base+'/access/login',json={'portal_id':r['portal_id'],'password':PASSWORD}).status_code==401 assert write(a,'/api/feedback/'+r['slug']+'/access',{'code':r['code']}).status_code==401 assert a.get('/api/public/'+r['slug']+'/testimonials').json()['count']==0 fresh,_=new_login(restored,target,'harbour' if bid=='harbour' else 'oak') if bid not in ('oak','harbour'): switched=write(fresh,'/api/business/switch',{'membership_id':r['member']});assert switched.status_code==200;fresh.headers['x-csrf-token']=switched.json()['csrf'] assert customer(fresh,r['source'])['state']=='merged';history=fresh.get('/api/customers/'+r['target']+'/history').json() assert history['totals']['jobs']==2 and history['totals']['received_pence']==10 assert fresh.get('/api/documents/invoice/'+r['invoice']).json()['snapshot']['customer']['id']==r['source'] foreign=next(v for k,v in records.items() if k!=bid) assert fresh.get('/api/customers/'+foreign['target']+'/history').status_code==404 assert write(fresh,'/api/customers/'+r['source']+'/portal-access/invite',{'version':customer(fresh,r['source'])['version']}).status_code==409 second=recovery.backup(target,tmp_path/'second.zip');assert recovery.verify(second['archive'],second['sha256'])['valid'] return {'businesses':4,'source_deleted_before_restore':True,'snapshot_format':recovery.VERSION,'snapshot_sha256':saved['sha256'],'snapshot_files':saved['files'],'customer_records':12,'retained_aliases':4,'reviewed_transfers':4,'keep_separate_decisions':4,'evidenced_completion_reviews':4,'exact_jobs_documents_and_original_customer_identity_preserved':True,'receipts_and_private_notes_preserved':True,'field_sources_photos_quote_responses_and_feedback_histories_preserved':True,'old_credentials_and_feedback_codes_rejected':True,'cross_business_customer_history_rejected':True,'aliases_cannot_be_invited':True,'second_generation_backup_verified':True,'browser_verified':False} def test_four_business_customer_disappearance_and_restore(tmp_path): report=run_customer_drill(tmp_path) (tmp_path/'customer-recovery-drill.json').write_text(json.dumps(report,indent=2)+'\n') @pytest.mark.parametrize('tamper',['baseline','alias','ledger','plan','actor','document_identity','receipt','photo_identity','photo_permission','timer_identity','source_identity','portal_revocation','pair_history']) def test_customer_restore_rejects_inconsistent_identity_before_creating_target(app,tmp_path,tamper): c=client(app);a=complex_history(app,c);j=a['job'];target=job(c) p=c.get(ROOT+'/duplicates').json()['pairs'][0] assert write(c,ROOT+'/pairs',{'left_id':p['customers'][0]['id'],'right_id':p['customers'][1]['id'],'version':0,'context_hash':p['context_hash'],'state':'keep_separate','confirmed':True,'reason':'Reviewed fictional exact pair'}).status_code==200 assert confirm(c,preview(c,selection(c,j,target))).status_code==200 archive=tmp_path/'source.zip';recovery.backup(app.state.store.root,archive) def mutate(files,meta): name='businesses/oak/business.sqlite3';path=tmp_path/'altered.sqlite3';path.write_bytes(files[name]) with sqlite3.connect(path) as db: if tamper=='baseline':db.execute('DELETE FROM customer_record_history WHERE version=1') elif tamper=='alias':db.execute('UPDATE customer_records SET canonical_id=customer_id WHERE canonical_id IS NOT NULL') elif tamper=='ledger':db.execute('DELETE FROM customer_transfer_jobs') elif tamper=='plan':db.execute("UPDATE customer_transfers SET preview_hash=?",('a'*64,)) elif tamper=='actor':db.execute("UPDATE customer_transfers SET actor='foreign-owner'") elif tamper=='document_identity': value=json.loads(db.execute('SELECT snapshot FROM invoices').fetchone()[0]);value['customer']['id']=target['customer_id'];db.execute('UPDATE invoices SET snapshot=?',(json.dumps(value),)) elif tamper=='receipt':db.execute('UPDATE payments SET amount_pence=999') elif tamper in ('photo_identity','timer_identity','source_identity'): table,nested={'photo_identity':('photo_history',None),'timer_identity':('field_history','session'),'source_identity':('source_history',None)}[tamper] for rid,raw in db.execute('SELECT id,snapshot FROM '+table).fetchall(): v=json.loads(raw);(v[nested] if nested else v)['customer_id']=target['customer_id'];db.execute('UPDATE '+table+' SET snapshot=? WHERE id=?',(json.dumps(v),rid)) elif tamper=='photo_permission':db.execute("UPDATE photo_history SET action='photo.review' WHERE action='photo.permission_suspended'") elif tamper=='portal_revocation':db.execute("DELETE FROM portal_events WHERE action='revoked'") elif tamper=='pair_history':db.execute('DELETE FROM customer_pair_history') files[name]=path.read_bytes();meta['files'][name].update(bytes=len(files[name]),sha256=hashlib.sha256(files[name]).hexdigest()) bad=tmp_path/'altered.zip';sha=rewrite_archive(archive,bad,mutate);target=tmp_path/'must-not-exist' with pytest.raises((recovery.RecoveryError,ValueError,KeyError)):recovery.restore(bad,sha,target,confirm_fresh=True) assert not target.exists() def test_format_ten_in_place_migration_observes_contacts_without_changing_records(tmp_path): root=tmp_path/'old';application=create_app(root);c=client(application);a=complex_history(application,c);before=table_rows(root,'oak') for bid in ('oak','harbour'): with sqlite3.connect(root/bid/'business.sqlite3') as db: for table in INTAKE_TABLES+TABLES:db.execute('DROP TABLE '+table) migrated=create_app(root);after=table_rows(root,'oak') for table in before: if table not in TABLES:assert before[table]==after[table],table assert len(after['customer_record_history'])==1 and after['customer_record_history'][0]['action']=='initialisation' assert a['portal'].get('/api/portal/oak-and-vale/me').status_code==200 saved=recovery.backup(root,tmp_path/'new.zip');assert recovery.verify(saved['archive'],saved['sha256'])['valid'] def test_pinned_format_ten_archive_migrates_without_synthetic_history(tmp_path): root=tmp_path/'old';application=create_app(root);c=client(application);j=job(c);original=tmp_path/'original.zip';recovery.backup(root,original) def mutate(files,meta): meta.update(version=10,prototype_version='0.13.0') for name,data in list(files.items()): if not name.startswith('businesses/') or not name.endswith('.sqlite3'):continue p=tmp_path/(name.split('/')[1]+'.sqlite3');p.write_bytes(data) with sqlite3.connect(p) as db: for table in INTAKE_TABLES+TABLES:db.execute('DROP TABLE '+table) files[name]=p.read_bytes();meta['files'][name].update(bytes=len(files[name]),sha256=hashlib.sha256(files[name]).hexdigest()) old=tmp_path/'v10.zip';sha=rewrite_archive(original,old,mutate);assert recovery.verify(old,sha)['valid'] target=tmp_path/'restored';recovery.restore(old,sha,target,confirm_fresh=True);new=create_app(target);fresh,_=new_login(new,target,'oak') history=fresh.get('/api/customers/'+j['customer_id']+'/history').json();assert len(history['record_history'])==1 and history['transfers']==[] assert history['customer']['state']=='active' and history['customer']['notes']=='' saved=recovery.backup(target,tmp_path/'second.zip');assert recovery.verify(saved['archive'],saved['sha256'])['valid']