from backend.public_enquiries import TABLES as INTAKE_TABLES
from backend.customer_review_models import TABLES as CUSTOMER_TABLES
"""Disposable invented businesses: field history survives loss; clocks never resume."""
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,portal,ORIGIN
from test_identity import new_business,add_staff,member,update
from test_workforce import person,entry
from test_scheduling import booking
from test_photos import reviewed
from test_recovery import table_rows,new_login
from test_fieldwork import setup,FakeClock,base,start,act,timer,review,convert,captured,suggested,draft,doc,docact

FIELD_TABLES=('source_history','document_sources','field_material_links','field_time_links','field_history','field_resource_locks','field_intervals','field_sessions','field_log_history','field_logs')

def run_field_drill(tmp_path):
    root=tmp_path/'fictional';app=create_app(root);owner=client(app);records={}
    for n,slug in enumerate(['oak-and-vale','harbour-homeworks','willow-field','cedar-field']):
        if n<2:c=client(app,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
        j,people,_=setup(c,app,2);clock=FakeClock('2026-10-24T22:59:30+00:00');app.state.field_clock=clock
        photo=reviewed(c,j);log=write(c,base(j)+'/logs',{'kind':'materials','content':'Private observed fixings','quantity':'1.125','unit':'box','work_date':'2026-10-24','photo_ids':[photo['id']]}).json()
        assert write(c,base(j)+'/logs/'+log['id']+'/state/review',{'version':1,'reason':'Owner checked the observation'}).status_code==200
        assert write(c,base(j)+'/logs/'+log['id']+'/material',{'version':2,'reason':'Reviewed explicit internal cost','cost_price':'1.25','status':'received'}).status_code==200
        tid=start(c,j,people).json()['id'];clock.advance(90);assert act(c,j,tid,'stop').status_code==200 and review(c,j,tid).status_code==200 and convert(c,j,tid).status_code==200
        corrected=[{'start':'2026-10-24T22:59:30+00:00','end':'2026-10-24T23:00:30+00:00'}]
        assert review(c,j,tid,corrected).status_code==200 and convert(c,j,tid).status_code==200
        assert write(c,'/api/team/'+people[0]['id'],{'version':1,'name':'Renamed worker','role':'Lead','hourly_cost':'99.00','colour':'#245648'},method='put').status_code==200
        assert write(c,'/api/team/'+people[0]['id']+'/archive',{'version':2}).status_code==200
        q=draft(c,j,kind='quote').json();assert docact(c,q,'share').status_code==200 and docact(c,q,'accept').status_code==200
        made=docact(c,q,'invoice');assert made.status_code==200;i={'kind':'invoice',**made.json()};assert docact(c,i,'issue').status_code==200
        made=docact(c,i,'correct');assert made.status_code==200;replacement={'kind':'invoice',**made.json()};assert docact(c,replacement,'issue').status_code==200
        entry(c,j,people[1],work_date='2026-10-22',duration_minutes=15)
        manual=[s for s in suggested(c,j)['sources'] if s['source_kind']=='time'];reserved=draft(c,j,manual).json()
        third=person(c,name='Running capture worker');booking(c,j,third);staff,_=add_staff(app,c,resource_id=third['id'],email='field-'+bid+'@staff.example')
        open_id=start(staff,j,[third]).json()['id'];clock.advance(60);assert act(staff,j,open_id,'heartbeat').status_code==200
        assert update(c,member(c,'field-'+bid+'@staff.example'),enabled=False).status_code==200
        records[bid]={'job':j,'membership':membership,'open_id':open_id,'timer':tid,'reserved':reserved,'replacement':replacement,'before':table_rows(root,bid),'photos':{p.name:p.read_bytes() for p in (root/bid/'photos').iterdir()}}
    original_owner=client(app);old_portal,code=portal(app,original_owner,records['oak']['job']);snap=tmp_path/'field.zip'
    saved=recovery.backup(root,snap);assert recovery.verify(snap,saved['sha256'])['valid'];assert saved['files']==9
    shutil.rmtree(root);target=tmp_path/'recovered';result=recovery.restore(snap,saved['sha256'],target,confirm_fresh=True)
    report=json.loads(Path(result['report']).read_text());assert report['source_format']==recovery.VERSION
    restored=create_app(target);anonymous=TestClient(restored,headers=ORIGIN)
    assert anonymous.post('/api/login',json={'email':'owner@oak.example','password':'Oak-demo-only-2026!'}).status_code==401
    assert anonymous.post('/api/portal/oak-and-vale/login',json={'code':code}).status_code==401
    old=TestClient(restored,headers=ORIGIN);old.cookies.update(original_owner.cookies);assert old.get('/api/session').status_code==401
    oldp=TestClient(restored,headers=ORIGIN);oldp.cookies.update(old_portal.cookies);assert oldp.get('/api/portal/oak-and-vale/me').status_code==401
    for bid,r in records.items():
        after=table_rows(target,bid)
        for table in ('field_logs','field_log_history','field_material_links','field_time_links','document_sources','source_history','time_entries','job_materials','team_resources','quotes','invoices','payments','job_photos','photo_history'):assert after[table]==r['before'][table],table
        assert {p.name:p.read_bytes() for p in (target/bid/'photos').iterdir()}==r['photos']
        assert not after['field_resource_locks']
        row=next(t for t in after['field_sessions'] if t['id']==r['open_id']);original=next(t for t in r['before']['field_sessions'] if t['id']==r['open_id']);assert row['state']=='interrupted' and row['version']==original['version']+1
        for t in after['field_sessions']:
            if t['id']!=r['open_id']:assert t==next(b for b in r['before']['field_sessions'] if b['id']==t['id'])
        interval=next(i for i in after['field_intervals'] if i['session_id']==r['open_id']);assert interval['ended_at']==interval['last_seen_at']
        assert after['field_history'][:-1]==r['before']['field_history'] and after['field_history'][-1]['action']=='interrupted'
        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['membership']});assert switched.status_code==200;fresh.headers['x-csrf-token']=switched.json()['csrf']
        assert timer(fresh,r['job'],r['open_id'])['confirmed_ms']==60000
        assert timer(fresh,r['job'],r['timer'])['labour_preview'][0]['hourly_cost_pence']==2000
        assert draft(fresh,r['job']).status_code==409
        assert len(doc(fresh,r['replacement'])['source_review'])==2
        assert fresh.get(base(next(v['job'] for key,v in records.items() if key!=bid))).status_code==404
    with sqlite3.connect(target/'registry.sqlite3') as db:assert db.execute("SELECT COUNT(*) FROM memberships WHERE role='staff' AND enabled=0 AND resource_id IS NOT NULL").fetchone()[0]==4
    second=tmp_path/'second.zip';again=recovery.backup(target,second);assert recovery.verify(second,again['sha256'])['valid']
    return {'businesses':4,'source_deleted_before_restore':True,'snapshot_format':recovery.VERSION,'snapshot_files':saved['files'],'snapshot_sha256':saved['sha256'],'private_photos_preserved':4,'field_log_material_links_preserved':4,'converted_timers_preserved':4,'corrected_conversion_generations_preserved':8,'active_generated_time_entries':16,'archived_generated_time_entries':16,'retained_manual_time_entries':4,'open_timers_interrupted_at_last_confirmation':4,'no_running_resource_locks_after_restore':True,'no_new_labour_invoices_or_payments_on_restore':True,'raw_intervals_review_history_saved_rates_and_named_people_preserved':True,'london_midnight_splits_preserved':True,'quote_invoice_replacement_and_reserved_sources_preserved':True,'archived_resources_and_disabled_staff_links_preserved':True,'exact_business_customer_job_photo_bindings_preserved':True,'old_credentials_sessions_and_portal_grants_rejected':True,'second_generation_backup_verified':True,'browser_verified':False}


def test_four_business_field_loss_restore_and_interrupted_clocks(tmp_path):
    proof=run_field_drill(tmp_path)
    (tmp_path/'field-recovery-drill.json').write_text(json.dumps(proof,indent=2)+'\n')


@pytest.mark.parametrize('tamper',['customer','raw_interval','locks','rate','time_link','time_cost','history','membership','source_line','source_snapshot','source_history'])
def test_field_backup_rejects_inconsistent_provenance(app,tmp_path,tamper):
    c=client(app);j,people,clock,tid=captured(c,app);draft(c,j);other=job(c)
    with app.state.store.tenant('oak') as db:
        if tamper=='customer':db.execute('UPDATE field_sessions SET customer_id=?',(other['customer_id'],))
        if tamper=='raw_interval':db.execute("UPDATE field_intervals SET ended_at='2026-10-25T01:00:00.000+00:00'")
        if tamper=='locks':db.execute('INSERT INTO field_resource_locks VALUES(?,?)',(people[0]['id'],tid))
        if tamper=='rate':
            resources=json.loads(db.execute('SELECT resources FROM field_sessions').fetchone()[0]);resources[0]['hourly_cost_pence']+=1;db.execute('UPDATE field_sessions SET resources=?',(json.dumps(resources),))
        if tamper=='time_link':db.execute('DELETE FROM field_time_links')
        if tamper=='time_cost':db.execute('UPDATE time_entries SET cost_pence=cost_pence+1')
        if tamper=='history':db.execute('DELETE FROM field_history WHERE id=(SELECT MIN(id) FROM field_history)')
        if tamper=='membership':db.execute("UPDATE field_history SET membership_id='foreign'")
        if tamper=='source_line':db.execute('UPDATE document_sources SET line_index=30')
        if tamper=='source_snapshot':
            source=json.loads(db.execute('SELECT source_snapshot FROM document_sources').fetchone()[0]);source['time_entries'][0]['cost_pence']+=1;db.execute('UPDATE document_sources SET source_snapshot=?',(json.dumps(source),))
        if tamper=='source_history':db.execute('DELETE FROM source_history')
    with pytest.raises((recovery.RecoveryError,ValueError)):recovery.backup(app.state.store.root,tmp_path/'bad.zip')
    assert not (tmp_path/'bad.zip').exists()


def test_historical_manual_and_material_sources_remain_valid_after_correction(app,tmp_path):
    from test_materials import material
    c=client(app);j,people,clock=setup(c,app);t=entry(c,j,people[0]);m=material(c,j,status='received');d=draft(c,j).json()
    assert write(c,'/api/time/'+t['id']+'/archive',{'version':1}).status_code==200
    assert write(c,'/api/materials/'+m['id']+'/archive',{'version':1}).status_code==200
    assert docact(c,d,'issue').status_code==409
    saved=recovery.backup(app.state.store.root,tmp_path/'valid.zip');assert recovery.verify(tmp_path/'valid.zip',saved['sha256'])['valid']


def test_pinned_format_five_restores_populated_records_with_empty_field_tables(tmp_path):
    root=tmp_path/'old';app=create_app(root);c=client(app);j=job(c);reviewed(c,j);p=person(c);entry(c,j,p)
    snapshot=tmp_path/'current.zip';recovery.backup(root,snapshot);legacy=tmp_path/'format5.zip'
    with zipfile.ZipFile(snapshot) as z:files={name:z.read(name) for name in z.namelist()}
    meta=json.loads(files.pop('manifest.json'));meta['version']=5;meta['prototype_version']='0.8.0'
    for name,data in list(files.items()):
        if not name.startswith('businesses/') or not name.endswith('business.sqlite3'):continue
        dbfile=tmp_path/(name.split('/')[1]+'.sqlite3');dbfile.write_bytes(data)
        with sqlite3.connect(dbfile) as db:
            from backend.recurrence import TABLES
            from backend.portal_access import TABLES as PORTAL_TABLES
            from backend.sales_outputs import TABLES as SALES_TABLES
            from backend.feedback_models import TABLES as FEEDBACK_TABLES
            for table in INTAKE_TABLES+CUSTOMER_TABLES+WORKFLOW_TABLES+FEEDBACK_TABLES+PORTAL_TABLES+SALES_TABLES+TABLES+FIELD_TABLES:db.execute('DROP TABLE '+table)
        files[name]=dbfile.read_bytes();meta['files'][name].update(bytes=len(files[name]),sha256=hashlib.sha256(files[name]).hexdigest())
    with zipfile.ZipFile(legacy,'w') as z:
        for name,data in files.items():z.writestr(name,data)
        z.writestr('manifest.json',json.dumps(meta))
    target=tmp_path/'restored';result=recovery.restore(legacy,recovery.checksum(legacy),target,confirm_fresh=True);assert json.loads(Path(result['report']).read_text())['source_format']==5
    after=table_rows(target,'oak');assert len(after['jobs'])==len(after['job_photos'])==len(after['time_entries'])==1
    assert all(not after[table] for table in FIELD_TABLES)