from backend.public_enquiries import TABLES as INTAKE_TABLES
from backend.customer_review_models import TABLES as CUSTOMER_TABLES
"""Format 8: passwords/codes die; exact customer and sealed history survive."""
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
from test_boundaries import app,client,write,job,quote,ORIGIN
from test_documents import doc,act,schedule
from test_portal_access import invited,enabled,response,PASSWORD
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_portal_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-portal','cedar-portal')):
        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
        cfg=c.get('/api/document-settings').json();cfg.update(payment_instructions='Invented '+slug+' instructions',quote_prefix='LOCAL-Q',invoice_prefix='LOCAL-I',include_year=True);assert write(c,'/api/document-settings',cfg,method='put').status_code==200
        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,typed_name='Alex Demo',acknowledgement='Invented review only')).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
        assert write(c,'/api/invoices/'+inv['id']+'/payments',{'amount_pence':1,'stage_id':stage['id']}).status_code==200
        recurring,_,_=definition(c,application,j,people[0]);generate(c,recurring)
        twin=job(c);twin_grant,twin_capture=invited(c,twin);assert twin_capture['to']==cap['to'] and twin_capture['portal_id']!=cap['portal_id']
        assert p.get(base+'/me').json()['jobs'][0]['id']==j['id'] and len(p.get(base+'/me').json()['jobs'])==1
        assert TestClient(application,headers=ORIGIN).post(base+'/access/reset-request',json={'portal_id':cap['portal_id'],'email':cap['to']}).status_code==200
        state=c.get('/api/customers/'+j['customer_id']+'/portal-access').json();reset=c.get('/api/customers/'+j['customer_id']+'/portal-access/captures/'+state['captures'][0]['id']).json()
        records[bid]={'slug':slug,'member':membership,'customer':j['customer_id'],'portal':p,'portal_id':cap['portal_id'],'reset':reset['code'],'pending':twin_capture['code'],'invoice':inv,'before':table_rows(root,bid),'photos':{p.name:p.read_bytes() for p in (root/bid/'photos').iterdir()}}
    archive=tmp_path/'portal.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():
            assert r['reset'].encode() not in all_bytes and r['pending'].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','sales_settings','sales_settings_history','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')
    for bid,r in records.items():
        after=table_rows(target,bid)
        for table in unchanged:assert after[table]==r['before'][table],table
        assert not after['portal_tokens'] and all(a['state']=='revoked' and a['password']=='' for a in after['portal_accounts'])
        assert after['portal_events'][:len(r['before']['portal_events'])]==r['before']['portal_events']
        assert {p.name:p.read_bytes() for p in (target/bid/'photos').iterdir()}==r['photos']
        p=TestClient(restored,headers=ORIGIN);p.cookies.update(r['portal'].cookies);base='/api/portal/'+r['slug']
        assert p.get(base+'/me').status_code==401
        assert p.post(base+'/access/login',json={'portal_id':r['portal_id'],'password':PASSWORD}).status_code==401
        for code in (r['reset'],r['pending']):assert p.post(base+'/access/complete',json={'code':code,'password':PASSWORD}).status_code==401
        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('/api/documents/invoice/'+foreign['invoice']['id']+'/pdf').status_code==404
        assert r['portal_id'] not in (target/'RESTORED-LOGIN.txt').read_text()
    second=recovery.backup(target,tmp_path/'second.zip');assert recovery.verify(second['archive'],second['sha256'])['valid']
    return {'businesses':4,'exact_customers':8,'same_name_and_email_stay_separate':True,'password_accounts':4,'pending_accounts':4,'quote_responses':4,'issued_invoices':4,'sealed_payment_requests':4,'private_photos':4,'recurrence_contracts':4,'field_source_and_calendar_history_preserved':True,'source_deleted_before_restore':True,'all_old_portal_passwords_sessions_setup_and_reset_codes_rejected':True,'captures_and_password_hashes_excluded_from_archive':True,'historical_pdf_available_to_recovered_owner_only':True,'no_customer_credentials_in_restored_login_file':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_portal_document_and_field_recovery(tmp_path):
    report=run_portal_drill(tmp_path);(tmp_path/'portal-recovery-drill.json').write_text(json.dumps(report,indent=2)+'\n')


@pytest.mark.parametrize('tamper',['account_history','response_hash','response_customer','response_actor','response_wording','request_amount','request_invoice','settings_history','settings_actor','portal_secret'])
def test_portal_document_recovery_rejects_inconsistent_records(app,tmp_path,tamper):
    c=client(app);j=job(c);p,g,cap=enabled(app,c,j);q=quote(c,j);act(c,'quote',q,'share');write(p,'/api/portal/oak-and-vale/quotes/'+q['id']+'/accept',response(c,q));inv=act(c,'quote',q,'invoice').json();act(c,'invoice',inv,'issue');stage=schedule(c,inv)[0];write(c,'/api/invoices/'+inv['id']+'/schedule/'+stage['id']+'/request',{'version':doc(c,'invoice',inv)['version']})
    cfg=c.get('/api/document-settings').json();cfg['payment_instructions']='Invented';write(c,'/api/document-settings',cfg,method='put');other=job(c)
    with app.state.store.tenant('oak') as db:
        if tamper=='account_history':db.execute("UPDATE portal_accounts SET delivery_email='wrong@fiction.example'")
        elif tamper.startswith('response_'):
            r=db.execute('SELECT evidence FROM quote_responses').fetchone();v=json.loads(r[0])
            if tamper=='response_hash':v['snapshot_hash']='0'*64
            if tamper=='response_actor':v['actor']='portal:'+other['customer_id']
            if tamper=='response_wording':v['owner_wording']='Invented after the response'
            if tamper=='response_customer':db.execute('UPDATE quote_responses SET customer_id=?',(other['customer_id'],))
            db.execute('UPDATE quote_responses SET evidence=?',(json.dumps(v),))
        elif tamper.startswith('request_'):
            v=json.loads(db.execute('SELECT snapshot FROM payment_request_snapshots').fetchone()[0]);v['requested_remaining_pence']=1 if tamper=='request_amount' else v['requested_remaining_pence']
            if tamper=='request_invoice':v['invoice_id']='a'*32
            db.execute('UPDATE payment_request_snapshots SET snapshot=?',(json.dumps(v),))
        elif tamper=='settings_history':db.execute('DELETE FROM sales_settings_history')
        elif tamper=='settings_actor':db.execute("UPDATE sales_settings_history SET actor='harbour'")
        else:
            v=json.loads(db.execute("SELECT snapshot FROM portal_events WHERE action='setup.completed'").fetchone()[0]);v['password']='secret';db.execute("UPDATE portal_events SET snapshot=? WHERE action='setup.completed'",(json.dumps(v),))
    with pytest.raises((recovery.RecoveryError,ValueError,KeyError)):recovery.backup(app.state.store.root,tmp_path/'bad.zip')
    assert not (tmp_path/'bad.zip').exists()


def test_populated_format_seven_retains_records_and_adds_empty_portal_history(tmp_path):
    root=tmp_path/'old';app=create_app(root);c=client(app);j=job(c);q=quote(c,j);act(c,'quote',q,'share');d,_,_=definition(c,app,j);generate(c,d)
    archive=tmp_path/'v8.zip';recovery.backup(root,archive)
    def mutate(files,meta):
        meta['version']=7;meta['prototype_version']='0.10.0'
        from backend.portal_access import TABLES as PORTAL
        from backend.sales_outputs import TABLES as SALES
        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:
                from backend.feedback_models import TABLES as FEEDBACK_TABLES
                for table in INTAKE_TABLES+CUSTOMER_TABLES+WORKFLOW_TABLES+FEEDBACK_TABLES+PORTAL+SALES:db.execute('DROP TABLE '+table)
            files[name]=path.read_bytes();meta['files'][name].update(bytes=len(files[name]),sha256=hashlib.sha256(files[name]).hexdigest())
    legacy=tmp_path/'format7.zip';sha=rewrite_archive(archive,legacy,mutate);assert recovery.verify(legacy,sha)['valid'];target=tmp_path/'restored';recovery.restore(legacy,sha,target,confirm_fresh=True)
    before=table_rows(root,'oak');after=table_rows(target,'oak')
    for table in ('quotes','quote_links','recurring_contracts','recurrence_rules','recurrence_occurrences','recurrence_runs'):assert before[table]==after[table]
    assert not after['portal_accounts'] and not after['quote_responses'] and not after['payment_request_snapshots']