"""Four-business disappearance, old schema migration and tampered intake archives.""" import copy,hashlib,json,shutil,sqlite3,uuid,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.public_enquiries import TABLES from test_boundaries import app,client,write,job,ORIGIN from test_public_enquiries import payload,photo,submit,saved from test_photos import listing,state,share,project,pub,reviewed from test_customer_reviews import complex_history from test_identity import new_business from test_recovery import table_rows,new_login,rewrite_archive from test_portal_access import PASSWORD def run_enquiry_drill(tmp_path): root=tmp_path/'fictional';application=create_app(root);owner=client(application);records={} for index,slug in enumerate(('oak-and-vale','harbour-homeworks','willow-enquiry','cedar-enquiry')): if index<2:c=client(application,index);bid=DEMO_BUSINESSES[index]['id'];membership=c.get('/api/session').json()['membership_id'] else: b=new_business(owner,slug).json();bid=b['business_id'];membership=b['membership_id'] switch=write(owner,'/api/business/switch',{'membership_id':membership});owner.headers['x-csrf-token']=switch.json()['csrf'];c=owner retained=complex_history(application,c) body=payload(message='Synthetic original intake '+slug,photos=[photo(colour) for colour in ['red','green','blue','yellow','purple']]);key=str(uuid.uuid4());path='/api/public/'+slug+'/enquiries' enquiry=saved(c,submit(application,body,key,path));zero=saved(c,submit(application,payload(preferred_date='',message='Separate matching-contact request '+slug),path=path)) assert enquiry['customer_id']!=zero['customer_id'];photos=listing(c,enquiry).json() assert state(c,enquiry,photos[0],'review').status_code==200 and share(c,enquiry,photos[0]).status_code==200 gallery=project(c,enquiry,photos[0],slug='intake-project');assert pub(c,gallery).status_code==200 records[bid]={'slug':slug,'membership':membership,'enquiry':enquiry,'zero':zero,'body':body,'key':key,'retained':retained,'gallery':gallery,'before':table_rows(root,bid),'bytes':{p.name:p.read_bytes() for p in (root/bid/'photos').iterdir()},'owner_cookies':dict(c.cookies)} archive=tmp_path/'enquiries.zip';snapshot=recovery.backup(root,archive);assert recovery.verify(archive,snapshot['sha256'])['valid'] shutil.rmtree(root) target=tmp_path/'restored';recovery.restore(archive,snapshot['sha256'],target,confirm_fresh=True);restored=create_app(target) unchanged=('enquiry_intakes','enquiry_photo_sources','photo_blobs','jobs','quotes','invoices','payments','payment_details','document_events','quote_responses','field_logs','field_sessions','field_intervals','field_time_links','time_entries','document_sources','source_history','feedback_responses') for bid,r in records.items(): after=table_rows(target,bid) for table in unchanged:assert after[table]==r['before'][table],(bid,table) assert {p.name:p.read_bytes() for p in (target/bid/'photos').iterdir()}==r['bytes'] old=TestClient(restored,headers=ORIGIN);old.cookies.update(r['owner_cookies']);assert old.get('/api/workspace').status_code==401 old=TestClient(restored,headers=ORIGIN);old.cookies.update(r['retained']['portal'].cookies) assert old.get('/api/portal/'+r['slug']+'/me').status_code==401 assert old.post('/api/portal/'+r['slug']+'/access/login',json={'portal_id':r['retained']['portal_id'],'password':PASSWORD}).status_code==401 assert write(old,'/api/feedback/'+r['slug']+'/access',{'code':r['retained']['code']}).status_code==401 fresh,_=new_login(restored,target,'harbour' if bid=='harbour' else 'oak') if bid not in ('oak','harbour'): switch=write(fresh,'/api/business/switch',{'membership_id':r['membership']});fresh.headers['x-csrf-token']=switch.json()['csrf'] current=fresh.get('/api/jobs/'+r['enquiry']['id']).json();assert current['enquiry_submission']==r['enquiry']['enquiry_submission'] photo_rows=listing(fresh,current).json();assert len(photo_rows)==5 and all(not p['reviewed'] and p['customer_share'] is None for p in photo_rows) originals={p['id']:p for p in r['before']['job_photos']} for p in after['job_photos']: if p['uploaded_by']!='public-enquiry':assert p==originals[p['id']] assert after['photo_history'][:len(r['before']['photo_history'])]==r['before']['photo_history'] assert after['gallery_history'][:len(r['before']['gallery_history'])]==r['before']['gallery_history'] assert len(after['photo_history'])==len(r['before']['photo_history'])+1 assert fresh.get('/api/public/'+r['slug']+'/projects/intake-project').status_code==404 assert pub(fresh,r['gallery']).status_code==409 assert fresh.get('/api/public/'+r['slug']+'/testimonials').json()['count']==0 # Exact public retries are non-authorizing and keep the original job. replay=submit(restored,r['body'],r['key'],'/api/public/'+r['slug']+'/enquiries');assert replay.json()['receipt']==current['id'] foreign=next(v for k,v in records.items() if k!=bid) assert fresh.get('/api/jobs/'+foreign['enquiry']['id']).status_code==404 assert fresh.get('/api/documents/invoice/'+r['retained']['invoice']['id']).status_code==200 second=recovery.backup(target,tmp_path/'second.zip');assert recovery.verify(second['archive'],second['sha256'])['valid'] return {'version':'0.25.0','recovery_format':recovery.VERSION,'businesses':4,'source_deleted_before_restore':True,'website_enquiries':8,'distinct_website_customers':8,'visitor_photos':20,'exact_normalized_bytes_and_original_facts_preserved':True,'original_intake_and_photo_provenance_tables_identical':True,'prior_finance_field_document_and_feedback_evidence_preserved':True,'prior_photo_and_gallery_history_retained':True,'visitor_photo_review_and_sharing_suspended':True,'visitor_photo_public_projects_withdrawn':True,'owner_republication_without_new_review_denied':True,'old_sessions_passwords_portal_and_feedback_access_denied':True,'exact_public_retries_return_original_receipt_without_private_access':True,'foreign_business_reads_denied':True,'second_generation_backup_verified':True,'same_mac_only':True,'browser_verified':False,'snapshot_sha256':snapshot['sha256']} def test_four_business_enquiry_disappearance_and_restore(tmp_path): result=run_enquiry_drill(tmp_path);(tmp_path/'enquiry-recovery-drill.json').write_text(json.dumps(result,indent=2)+'\n') @pytest.mark.parametrize('tamper',['original_customer','photo_job','date','source_hash','content_hash','operation_key','actor','invented_permission','recovery_actor','missing_source','photo_bytes']) def test_tampered_intake_archive_is_rejected_before_target_creation(app,tmp_path,tamper): owner=client(app);a=saved(owner,submit(app,payload(photos=[photo()])));b=saved(owner,submit(app,payload())) original=tmp_path/'original.zip';recovery.backup(app.state.store.root,original) def mutate(files,meta): name='businesses/oak/business.sqlite3';p=tmp_path/'altered.sqlite3';p.write_bytes(files[name]) with sqlite3.connect(p) as db: if tamper=='original_customer':db.execute('UPDATE enquiry_intakes SET customer_id=? WHERE job_id=?',(b['customer_id'],a['id'])) elif tamper=='photo_job':db.execute('UPDATE enquiry_photo_sources SET job_id=?',(b['id'],)) elif tamper=='date':db.execute("UPDATE enquiry_intakes SET submitted=json_set(submitted,'$.preferred_date','2026-02-30') WHERE job_id=?",(a['id'],)) elif tamper=='source_hash':db.execute('UPDATE enquiry_photo_sources SET source_sha256=?',('0'*64,)) elif tamper=='content_hash':db.execute('UPDATE enquiry_intakes SET content_sha256=?',('0'*64,)) elif tamper=='operation_key':db.execute('UPDATE enquiry_intakes SET operation_key=? WHERE job_id=?',('not-a-submission-key',a['id'])) elif tamper=='actor':db.execute("UPDATE photo_history SET actor_id='public-enquiry',action='photo.review'") elif tamper=='invented_permission':db.execute("UPDATE photo_history SET snapshot=json_set(snapshot,'$.reviewed',1)") elif tamper=='recovery_actor': db.row_factory=sqlite3.Row original=dict(db.execute('SELECT * FROM job_photos').fetchone());changed={**original,'reviewed':1,'version':2} db.execute('UPDATE job_photos SET reviewed=1,version=2') db.execute("INSERT INTO photo_history(photo_id,action,actor_id,membership_id,snapshot,created_at) VALUES(?,?,?,?,?,?)",(original['id'],'photo.enquiry_recovery_suspended','recovery','',json.dumps(changed),original['created_at'])) elif tamper=='missing_source':db.execute('DELETE FROM enquiry_photo_sources') elif tamper=='photo_bytes': image=next(n for n in files if n.startswith('businesses/oak/photos/')) files[image]=files[image][:-20];meta['files'][image].update(bytes=len(files[image]),sha256=hashlib.sha256(files[image]).hexdigest()) files[name]=p.read_bytes();meta['files'][name].update(bytes=len(files[name]),sha256=hashlib.sha256(files[name]).hexdigest()) bad=tmp_path/'tampered.zip';sha=rewrite_archive(original,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_eleven_migrates_without_fabricated_intake_or_photo_permission(tmp_path): root=tmp_path/'old';application=create_app(root);owner=client(application);j=job(owner);p=reviewed(owner,j) before=table_rows(root,'oak');original=tmp_path/'original.zip';recovery.backup(root,original) def old_ledger(db): assert db.execute('SELECT COUNT(*) FROM customer_merge_corrections').fetchone()[0]==0 db.execute('DROP TABLE customer_merge_corrections') assert db.execute('SELECT COUNT(*) FROM customer_transfer_jobs').fetchone()[0]==0 pinned=json.loads((Path(__file__).resolve().parents[1]/'docs/recovery-schema-v11.json').read_text()) sql=next(r[3] for r in pinned['tenant'] if r[1]=='customer_transfer_jobs') db.execute('DROP TABLE customer_transfer_jobs');db.execute(sql) def make_old(files,meta): meta.update(version=11,prototype_version='0.14.0') for name,data in list(files.items()): if not name.startswith('businesses/') or not name.endswith('.sqlite3'):continue path=tmp_path/(name.split('/')[1]+'.sqlite3');path.write_bytes(data) with sqlite3.connect(path) as db: for table in TABLES:db.execute('DROP TABLE '+table) old_ledger(db) files[name]=path.read_bytes();meta['files'][name].update(bytes=len(files[name]),sha256=hashlib.sha256(files[name]).hexdigest()) old=tmp_path/'format11.zip';sha=rewrite_archive(original,old,make_old);assert recovery.verify(old,sha)['valid'] target=tmp_path/'restored';recovery.restore(old,sha,target,confirm_fresh=True);restored=create_app(target);fresh,_=new_login(restored,target,'oak') assert fresh.get('/api/jobs/'+j['id']).json()['enquiry_submission'] is None after=table_rows(target,'oak');assert after['enquiry_intakes']==after['enquiry_photo_sources']==[] for table in ('jobs','job_photos','photo_history','photo_blobs'):assert after[table]==before[table] for bid in ('oak','harbour'): with sqlite3.connect(root/bid/'business.sqlite3') as db: for table in TABLES:db.execute('DROP TABLE '+table) old_ledger(db) migrated=create_app(root);assert owner.get('/api/session').status_code==200 current=client(migrated).get('/api/jobs/'+j['id']).json();assert current['enquiry_submission'] is None for table,rows in before.items():assert table_rows(root,'oak')[table]==rows,table def test_intake_retry_survives_pruning_general_operation_cache(app): key=str(uuid.uuid4());data=payload(photos=[photo()]);first=submit(app,data,key) with app.state.store.tenant('oak') as db:db.execute('DELETE FROM operations') second=submit(app,data,key);assert second.json()==first.json() assert submit(app,{**data,'message':'A different enquiry body'},key).status_code==409 with app.state.store.tenant('oak') as db:assert db.execute('SELECT COUNT(*) FROM jobs').fetchone()[0]==1