from backend.public_enquiries import TABLES as INTAKE_TABLES
from backend.customer_review_models import TABLES as CUSTOMER_TABLES
"""Format 9 loss, migration and consistency failures using invented fixtures."""
from pathlib import Path
import hashlib,json,shutil,sqlite3,zipfile
from backend.workflow_recovery import TABLES as WORKFLOW_TABLES
import pytest
from fastapi.testclient import TestClient
from backend import recovery
from backend.app import create_app
from backend.store import DEMO_BUSINESSES,digest
from backend.feedback_models import TABLES
from test_boundaries import app,client,write,job,quote,ORIGIN
from test_feedback import prepare,respond,publish,completed,action,BASE,ACCESS,PUBLIC,payload,consent
from test_portal_access import enabled,response,PASSWORD
from test_documents import doc,act,schedule
from test_identity import new_business
from test_fieldwork import captured,draft
from test_photos import reviewed
from test_recurrence import definition,generate
from test_recovery import table_rows,new_login,rewrite_archive


def run_feedback_drill(tmp_path,populate_workflow=None):
    root=tmp_path/'fictional';application=create_app(root);owner=client(application);records={}
    for n,slug in enumerate(('oak-and-vale','harbour-homeworks','willow-feedback','cedar-feedback')):
        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'];r=write(owner,'/api/business/switch',{'membership_id':membership});owner.headers['x-csrf-token']=r.json()['csrf'];c=owner
        j,people,clock,tid=captured(c,application);photo=reviewed(c,j);q=draft(c,j,kind='quote').json();act(c,'quote',q,'share')
        p,grant,cap=enabled(application,c,j);base='/api/portal/'+slug
        assert write(p,base+'/quotes/'+q['id']+'/accept',response(c,q)).status_code==200
        inv=act(c,'quote',q,'invoice').json();act(c,'invoice',inv,'issue');total=doc(c,'invoice',inv)['total_pence'];stage=schedule(c,inv,(total//2,total-total//2))[0]
        assert write(c,'/api/invoices/'+inv['id']+'/schedule/'+stage['id']+'/request',{'version':doc(c,'invoice',inv)['version']}).status_code==200
        contract,_,_=definition(c,application,j,people[0]);generate(c,contract)
        completed(c,j);_,rid,code=prepare(c,j)
        a=TestClient(application,headers=ORIGIN);assert write(a,'/api/feedback/'+slug+'/submit',{'code':code,**payload(allow=True)}).status_code==200;publish(c,rid)
        twin=completed(c);_,pending,pending_code=prepare(c,twin);assert twin['customer_id']!=j['customer_id']
        assert [v['id'] for v in p.get(base+'/feedback').json()]==[rid]
        if populate_workflow:populate_workflow(application,c,j,twin)
        records[bid]={'slug':slug,'member':membership,'customer':j['customer_id'],'portal':p,'portal_id':cap['portal_id'],'rid':rid,'code':code,'pending':pending_code,'pending_id':pending,'invoice':inv,'before':table_rows(root,bid),'photos':{p.name:p.read_bytes() for p in (root/bid/'photos').iterdir()}}
    archive=tmp_path/'feedback.zip';saved=recovery.backup(root,archive);assert recovery.verify(archive,saved['sha256'])['valid']
    with zipfile.ZipFile(archive) as z:
        assert not any('capture' in name for name in z.namelist());all_bytes=b''.join(z.read(name) for name in z.namelist())
        for r in records.values():
            for code in (r['code'],r['pending']):assert code.encode() not in all_bytes and digest(code).encode() not in all_bytes
            for a in r['before']['portal_accounts']:
                if a['password']:assert a['password'].encode() not in all_bytes
    shutil.rmtree(root);target=tmp_path/'restored';recovery.restore(archive,saved['sha256'],target,confirm_fresh=True);restored=create_app(target)
    unchanged=('quotes','quote_links','quote_responses','invoices','payments','payment_stages','payment_request_snapshots','document_events','document_counters','field_logs','field_sessions','field_intervals','field_time_links','time_entries','document_sources','source_history','job_photos','recurring_contracts','recurrence_rules','recurrence_occurrences','recurrence_history','recurrence_runs','feedback_responses')
    for bid,r in records.items():
        after=table_rows(target,bid)
        for table in unchanged+(WORKFLOW_TABLES if populate_workflow else ()):assert after[table]==r['before'][table],table
        assert not after['feedback_tokens'] and not after['portal_tokens']
        assert all(v['state']=='revoked' for v in after['feedback_requests']);assert all(v['state']=='hidden' for v in after['feedback_publications'])
        assert after['feedback_history'][:len(r['before']['feedback_history'])]==r['before']['feedback_history']
        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+'/feedback').status_code==401
        assert a.post(base+'/access/login',json={'portal_id':r['portal_id'],'password':PASSWORD}).status_code==401
        for code in (r['code'],r['pending']):assert write(a,'/api/feedback/'+r['slug']+'/access',{'code':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 fresh.get('/api/documents/invoice/'+r['invoice']['id']+'/pdf').status_code==200
        foreign=next(v for k,v in records.items() if k!=bid);assert fresh.get(BASE+foreign['rid']).status_code==404
        assert not fresh.get(BASE+r['rid']+'/preview').json()['can_publish']
        assert action(fresh,r['rid'],'reissue').status_code==200
        new_code=fresh.get(BASE+r['rid']+'/capture').json()['code'];assert new_code!=r['code']
        view=write(a,'/api/feedback/'+r['slug']+'/access',{'code':new_code}).json()
        assert view['response']['answers']['overall_rating']==4 and not view['published']
        assert write(a,'/api/feedback/'+r['slug']+'/consent',{'code':new_code,'version':view['version'],'consent':consent(True),'confirmed':True}).status_code==200
        assert a.get('/api/public/'+r['slug']+'/testimonials').json()['count']==0;publish(fresh,r['rid']);assert a.get('/api/public/'+r['slug']+'/testimonials').json()['count']==1
    second=recovery.backup(target,tmp_path/'second.zip');assert recovery.verify(second['archive'],second['sha256'])['valid']
    return {'businesses':4,'exact_customers':8,'feedback_requests':8,'private_responses':4,'pending_requests':4,'original_responses_and_histories_preserved':True,'published_testimonials_paused_on_restore':4,'explicit_new_codes_customer_reapproval_and_owner_publication_verified':4,'old_credentials_sessions_and_captures_unusable':True,'tokens_and_hashes_and_captures_excluded':True,'same_contact_customers_remain_separate':True,'quote_responses':4,'issued_invoices':4,'sealed_payment_requests':4,'private_photos':4,'field_sources_and_recurrence_preserved':True,'source_deleted_before_restore':True,'second_generation_backup_verified':True,'snapshot_format':recovery.VERSION,'snapshot_sha256':saved['sha256'],'snapshot_files':saved['files'],'browser_verified':False}


def test_four_business_feedback_disappearance_and_restore(tmp_path):
    report=run_feedback_drill(tmp_path);(tmp_path/'feedback-recovery-drill.json').write_text(json.dumps(report,indent=2)+'\n')


@pytest.mark.parametrize('tamper',['customer_link','job_link','private_projection','projection_text','consent_refusal','consent_version','history_missing','history_original','history_foreign_actor','history_owner_consent','public_without_event','credential_history','token_export'])
def test_feedback_restore_fails_closed_for_inconsistent_links_or_consent(app,tmp_path,tamper):
    c=client(app);j,rid,code=prepare(c);respond(c,app,rid,code);publish(c,rid);other=completed(c)
    original=tmp_path/'original.zip';recovery.backup(app.state.store.root,original)
    def mutate(files,meta):
        name='businesses/oak/business.sqlite3';path=tmp_path/'changed.sqlite3';path.write_bytes(files[name])
        with sqlite3.connect(path) as db:
            if tamper=='customer_link':db.execute('UPDATE feedback_requests SET customer_id=?',(other['customer_id'],))
            elif tamper=='job_link':db.execute('UPDATE feedback_requests SET job_id=?',(other['id'],))
            elif tamper in ('private_projection','projection_text'):
                raw=db.execute('SELECT projection FROM feedback_publications').fetchone()[0];v=json.loads(raw)
                if tamper=='private_projection':v['comments']='PRIVATE leaked'
                else:v['statement']='Changed by owner'
                db.execute('UPDATE feedback_publications SET projection=?',(json.dumps(v),))
            elif tamper=='consent_refusal':db.execute("UPDATE feedback_consents SET state='withdrawn',display_name='',statement='',share_rating=0")
            elif tamper=='consent_version':db.execute('UPDATE feedback_publications SET consent_version=999')
            elif tamper=='history_missing':db.execute('DELETE FROM feedback_history WHERE version=2')
            elif tamper=='history_foreign_actor':db.execute("UPDATE feedback_history SET actor='harbour' WHERE action='published'")
            elif tamper=='history_owner_consent':db.execute("UPDATE feedback_history SET actor='oak' WHERE action='submitted'")
            elif tamper=='public_without_event':db.execute("UPDATE feedback_history SET action='hidden' WHERE action='published'")
            elif tamper=='token_export':db.execute('INSERT INTO feedback_tokens VALUES(?,?,?,?,?,?)',('a'*32,rid,digest(code),1,99999999999,'2026-10-28T12:00:00+00:00'))
            else:
                raw=db.execute('SELECT snapshot FROM feedback_history WHERE version=2').fetchone()[0];v=json.loads(raw)
                if tamper=='history_original':v['response']['answers']['comments']='Rewritten private answer'
                else:v['request']['code']=code
                db.execute('UPDATE feedback_history SET snapshot=? WHERE version=2',(json.dumps(v),))
        files[name]=path.read_bytes();meta['files'][name].update(bytes=len(files[name]),sha256=hashlib.sha256(files[name]).hexdigest())
    bad=tmp_path/'bad.zip';sha=rewrite_archive(original,bad,mutate);target=tmp_path/'target'
    with pytest.raises((recovery.RecoveryError,ValueError,KeyError)):recovery.restore(bad,sha,target,confirm_fresh=True)
    assert not target.exists()


def test_populated_format_eight_migrates_without_rewriting_issued_history(tmp_path):
    root=tmp_path/'v11';application=create_app(root);c=client(application);j,people,clock,tid=captured(c,application);photo=reviewed(c,j);q=draft(c,j,kind='quote').json();act(c,'quote',q,'share');p,grant,cap=enabled(application,c,j)
    assert write(p,'/api/portal/oak-and-vale/quotes/'+q['id']+'/accept',response(c,q)).status_code==200
    inv=act(c,'quote',q,'invoice').json();act(c,'invoice',inv,'issue');stage=schedule(c,inv,(doc(c,'invoice',inv)['total_pence'],))[0];assert write(c,'/api/invoices/'+inv['id']+'/schedule/'+stage['id']+'/request',{'version':doc(c,'invoice',inv)['version']}).status_code==200
    d,_,_=definition(c,application,j,people[0]);generate(c,d);before=table_rows(root,'oak');archive=tmp_path/'new.zip';recovery.backup(root,archive)
    def mutate(files,meta):
        meta['version']=8;meta['prototype_version']='0.11.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 INTAKE_TABLES+CUSTOMER_TABLES+WORKFLOW_TABLES+TABLES:db.execute('DROP TABLE '+table)
            files[name]=path.read_bytes();meta['files'][name].update(bytes=len(files[name]),sha256=hashlib.sha256(files[name]).hexdigest())
    old=tmp_path/'format8.zip';sha=rewrite_archive(archive,old,mutate);assert recovery.verify(old,sha)['valid'];target=tmp_path/'restored';recovery.restore(old,sha,target,confirm_fresh=True);after=table_rows(target,'oak')
    for table in ('quotes','quote_responses','invoices','payment_request_snapshots','field_logs','field_sessions','document_sources','job_photos','recurring_contracts','recurrence_rules','recurrence_occurrences','recurrence_runs'):assert before[table]==after[table],table
    assert all(not after[t] for t in TABLES)
    recovered=create_app(target);fresh,_=new_login(recovered,target,'oak');assert fresh.get('/api/feedback').json()['stats']['requests']==0
    completed(fresh,j);prepare(fresh,j)


def test_each_permission_and_moderation_transition_is_recoverable(app,tmp_path):
    c=client(app);_,rid,code=prepare(c);a,_=respond(c,app,rid,code);publish(c,rid)
    def check(n):
        saved=recovery.backup(app.state.store.root,tmp_path/(str(n)+'.zip'));assert recovery.verify(saved['archive'],saved['sha256'])['valid']
    check(0)
    assert action(c,rid,'hide').status_code==200;check(1);publish(c,rid);check(2)
    v=c.get(BASE+rid).json()['request']['version'];assert write(a,ACCESS+'consent',{'code':code,'version':v,'consent':consent(False),'confirmed':True}).status_code==200;check(3)
    v=c.get(BASE+rid).json()['request']['version'];assert write(a,ACCESS+'consent',{'code':code,'version':v,'consent':consent(True,statement='New customer approved words',share_rating=False),'confirmed':True}).status_code==200;check(4);publish(c,rid);check(5)
    assert action(c,rid,'revoke').status_code==200;check(6);assert action(c,rid,'reissue').status_code==200;check(7)
    assert action(c,rid,'archive').status_code==200;check(8)