from backend.public_enquiries import TABLES as INTAKE_TABLES
from backend.customer_review_models import TABLES as CUSTOMER_TABLES
"""Four-business disappearance drill and strict historical recurrence recovery."""
from pathlib import Path
from datetime import date
import copy,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.recurrence import TABLES
from backend.store import DEMO_BUSINESSES
from test_boundaries import app,client,write,job,portal,ORIGIN
from test_identity import new_business
from test_recovery import table_rows,new_login
from test_fieldwork import captured,base,draft,docact,start,act,timer
from test_photos import reviewed
from test_recurrence import definition,generate,apply,single,revise,preview,current,bookingForTest,BASE


def run_recurrence_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-repeat','cedar-repeat']):
        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,clock,tid=captured(c,app,2);photo=reviewed(c,j)
        assert write(c,base(j)+'/logs',{'kind':'note','content':'Retained completed work','work_date':'2026-10-25','photo_ids':[photo['id']]}).status_code==200
        invoice=draft(c,j).json();assert docact(c,invoice,'issue').status_code==200
        d,j,p=definition(c,app,j,people[0]);d=generate(c,d,'2026-10-26','2026-11-30');past=copy.deepcopy(d['occurrences'][0]);app.state.recurrence_today=lambda:date(2026,10,27)
        o=d['occurrences'][1];d=apply(c,d,single(o,'override',booking={**bookingForTest(o['booking']),'start_date':'2026-11-03','end_date':'2026-11-03'}));d=apply(c,d,single(d['occurrences'][2]))
        for reason in ('First holiday','Overlapping holiday'):d=apply(c,d,{'action':'block','start_date':'2026-11-16','end_date':'2026-11-23','reason':reason})
        for ix in (3,4):d=apply(c,d,single(d['occurrences'][ix],'exclude'))
        d=apply(c,d,revise(d,rule={**d['rules'][0]['content'],'start_time':'12:00'}));assert d['occurrences'][0]==past
        assert not preview(c,d,single(past))['can_confirm']
        opened=start(c,j,[people[1]]).json()['id'];clock.advance(60);assert act(c,j,opened,'heartbeat').status_code==200
        records[bid]={'job':j,'membership':membership,'contract':d['id'],'open':opened,'timer':tid,'before':table_rows(root,bid),'photos':{p.name:p.read_bytes() for p in (root/bid/'photos').iterdir()}}
    oldowner=client(app);oldportal,code=portal(app,oldowner,records['oak']['job']);snapshot=tmp_path/'recurrence.zip';saved=recovery.backup(root,snapshot);assert recovery.verify(snapshot,saved['sha256'])['valid']
    shutil.rmtree(root);target=tmp_path/'restored';result=recovery.restore(snapshot,saved['sha256'],target,confirm_fresh=True);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
    stale=TestClient(restored,headers=ORIGIN);stale.cookies.update(oldowner.cookies);assert stale.get('/api/session').status_code==401
    stale.cookies.clear();stale.cookies.update(oldportal.cookies);assert stale.get('/api/portal/oak-and-vale/me').status_code==401
    for bid,r in records.items():
        after=table_rows(target,bid)
        for table in (*TABLES,'calendar_bookings','calendar_audit','field_logs','field_log_history','field_time_links','time_entries','job_photos','photo_history','document_sources','source_history','quotes','invoices','payments'):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'];opened=next(t for t in after['field_sessions'] if t['id']==r['open']);assert opened['state']=='interrupted'
        assert next(t for t in after['field_sessions'] if t['id']==r['timer'])==next(t for t in r['before']['field_sessions'] if t['id']==r['timer'])
        fresh,_=new_login(restored,target,'harbour' if bid=='harbour' else 'oak')
        if bid not in ('oak','harbour'):
            sw=write(fresh,'/api/business/switch',{'membership_id':r['membership']});assert sw.status_code==200;fresh.headers['x-csrf-token']=sw.json()['csrf']
        assert len(fresh.get(BASE+'/'+r['contract']).json()['occurrences'])==6
        assert timer(fresh,r['job'],r['open'])['confirmed_ms']==60000
        another=next(v for key,v in records.items() if key!=bid);assert fresh.get(BASE+'/'+another['contract']).status_code==404
    second=tmp_path/'second.zip';again=recovery.backup(target,second);assert recovery.verify(second,again['sha256'])['valid']
    return {'businesses':4,'snapshot_format':recovery.VERSION,'snapshot_files':saved['files'],'snapshot_sha256':saved['sha256'],'source_deleted_before_restore':True,'recurrence_definitions':4,'immutable_rules':8,'stable_occurrences':24,'past_visits_preserved':4,'manual_overrides_preserved':4,'manual_cancellations_preserved':4,'excluded_visits_preserved':8,'overlapping_date_blocks_preserved':8,'complete_review_runs_and_calendar_history_preserved':True,'linked_field_logs_photos_labour_and_issued_documents_unchanged':True,'private_photos':4,'two_person_converted_captures':4,'open_timers_interrupted_at_last_confirmation':4,'old_passwords_sessions_and_portal_grants_rejected':True,'no_auto_generation_labour_billing_or_payment':True,'second_generation_backup_verified':True,'browser_verified':False}


def test_four_business_recurrence_loss_drill(tmp_path):
    report=run_recurrence_drill(tmp_path);(tmp_path/'recurrence-recovery-drill.json').write_text(json.dumps(report,indent=2)+'\n')


@pytest.mark.parametrize('tamper',['customer','state','rule_snapshot','original_rule','block_link','history_missing','history_immutable','run_operation','run_result','membership','booking_history'])
def test_recurrence_tampering_rejected(app,tmp_path,tamper):
    c=client(app);d,j,p=definition(c,app);d=generate(c,d);other,j2,p2=definition(c,app);other=generate(c,other);d=apply(c,d,{'action':'block','start_date':'2026-11-02','end_date':'2026-11-02','reason':'Review block'})
    with app.state.store.tenant('oak') as db:
        if tamper=='customer':db.execute('UPDATE recurring_contracts SET customer_id=? WHERE id=?',(j2['customer_id'],d['id']))
        if tamper=='state':db.execute("UPDATE recurrence_occurrences SET state='active' WHERE state='excluded'")
        if tamper=='rule_snapshot':db.execute("UPDATE recurrence_occurrences SET rule_snapshot='{}'")
        if tamper=='original_rule':db.execute('UPDATE recurrence_occurrences SET original_rule_id=? WHERE contract_id=?',(other['rules'][0]['id'],d['id']))
        if tamper=='block_link':db.execute('UPDATE recurrence_occurrences SET exclusion_ids=? WHERE contract_id=?',(json.dumps([d['blocks'][0]['id']]),other['id']))
        if tamper=='history_missing':db.execute("DELETE FROM recurrence_history WHERE kind='occurrence'")
        if tamper=='history_immutable':
            h=db.execute("SELECT id,after_value FROM recurrence_history WHERE kind='contract' LIMIT 1").fetchone();v=json.loads(h[1]);v['customer_id']=j2['customer_id'];db.execute('UPDATE recurrence_history SET after_value=? WHERE id=?',(json.dumps(v),h[0]))
        if tamper=='run_operation':db.execute("UPDATE recurrence_runs SET operation='{}'")
        if tamper=='run_result':
            h=db.execute('SELECT id,result FROM recurrence_runs LIMIT 1').fetchone();v=json.loads(h[1]);v['changed_occurrence_ids']=[];db.execute('UPDATE recurrence_runs SET result=? WHERE id=?',(json.dumps(v),h[0]))
        if tamper=='membership':db.execute("UPDATE recurrence_rules SET membership_id=?",('a'*32,))
        if tamper=='booking_history':db.execute("UPDATE calendar_bookings SET title='Changed outside review'")
    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_pinned_format_six_restores_with_empty_recurrence(tmp_path):
    root=tmp_path/'old';app=create_app(root);c=client(app);j,people,clock,tid=captured(c,app);reviewed(c,j);assert draft(c,j).status_code==200
    snap=tmp_path/'current.zip';recovery.backup(root,snap)
    with zipfile.ZipFile(snap) as z:files={name:z.read(name) for name in z.namelist()}
    meta=json.loads(files.pop('manifest.json'));meta['version']=6;meta['prototype_version']='0.9.0'
    for name,data in list(files.items()):
        if not name.startswith('businesses/') or not name.endswith('business.sqlite3'):continue
        local=tmp_path/(name.split('/')[1]+'.sqlite3');local.write_bytes(data)
        with sqlite3.connect(local) as db:
            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:db.execute('DROP TABLE '+table)
        files[name]=local.read_bytes();meta['files'][name].update(bytes=len(files[name]),sha256=hashlib.sha256(files[name]).hexdigest())
    legacy=tmp_path/'format6.zip'
    with zipfile.ZipFile(legacy,'w') as z:
        z.writestr('manifest.json',json.dumps(meta))
        for name,data in files.items():z.writestr(name,data)
    sha=hashlib.sha256(legacy.read_bytes()).hexdigest();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');assert all(not after[t] for t in TABLES)
    for table in ('field_sessions','field_intervals','field_history','time_entries','document_sources','source_history','job_photos'):assert before[table]==after[table]