from backend.public_enquiries import TABLES as INTAKE_TABLES
from backend.customer_review_models import TABLES as CUSTOMER_TABLES
"""Synthetic pixel fixtures; no original SHS execution, customer media or browser."""
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import copy,hashlib,io,json,shutil,sqlite3,uuid,zipfile
import pytest
from PIL import Image,PngImagePlugin
from fastapi.testclient import TestClient
from backend.app import create_app
from backend.store import Store,uid,DEMO_BUSINESSES
from backend import recovery
from backend.photo_models import ProjectContent
from test_boundaries import app,client,write,job,quote,invoice,portal,ORIGIN
from test_identity import add_staff,member,update,new_business,login
from test_workforce import person
from test_scheduling import booking
from test_recovery import table_rows,new_login


def pixels(colour='green',format='PNG',metadata=False):
    out=io.BytesIO();im=Image.new('RGB',(32,24),colour);args={}
    if metadata and format=='PNG':
        info=PngImagePlugin.PngInfo();info.add_text('Location','PRIVATE EXAMPLE LOCATION');args['pnginfo']=info
    if metadata and format=='JPEG':
        exif=Image.Exif();exif[270]='PRIVATE EXAMPLE DESCRIPTION';exif[274]=6;args['exif']=exif
    im.save(out,format,**args);return out.getvalue()


def upload(c,j,colour='green',raw=None,key=None):
    return c.post('/api/jobs/'+j['id']+'/photos',content=raw if raw is not None else pixels(colour),headers={'idempotency-key':key or str(uuid.uuid4())})

def listing(c,j,archived=False):return c.get('/api/jobs/'+j['id']+'/photos?archived='+str(archived).lower())
def record(c,j,p,archived=False):return next(v for v in listing(c,j,archived).json() if v['id']==p['id'])
def state(c,j,p,action):return write(c,'/api/jobs/'+j['id']+'/photos/'+p['id']+'/state/'+action,{'version':record(c,j,p,action=='restore')['version']})
def share(c,j,p,shared=True):return write(c,'/api/jobs/'+j['id']+'/photos/'+p['id']+'/share',{'version':record(c,j,p)['version'],'shared':shared,'caption':'Customer safe caption','alt':'A neutral synthetic green square'})
def uploaded(c,j,colour='green'):
    r=upload(c,j,colour);assert r.status_code==200,r.text;return r.json()
def reviewed(c,j,colour='green'):
    p=uploaded(c,j,colour);assert state(c,j,p,'review').status_code==200;return p
def project(c,j,p,slug='shelf-project',**extra):
    content=ProjectContent(title='Demonstration shelving',slug=slug,images=[{'id':uid(),'source_kind':'job','source_id':p['id'],'alt':'A neutral synthetic square'}],**extra).model_dump()
    r=write(c,'/api/gallery',{'source_job_id':j['id'],'content':content});assert r.status_code==200,r.text
    return {**r.json(),'content':content}
def pub(c,p,confirmed=True,key=None):
    current=c.get('/api/gallery/'+p['id']).json();return write(c,'/api/gallery/'+p['id']+'/publish',{'version':current['version'],'permission_confirmed':confirmed},key=key)
def project_action(c,p,action):
    version=c.get('/api/gallery/'+p['id']).json()['version'];return write(c,'/api/gallery/'+p['id']+'/state/'+action,{'version':version})
def edit_photo(c,j,p,**fields):
    row=record(c,j,p);payload={k:row[k] for k in ('version','caption','description','category','phase','pair_id','position')};return write(c,'/api/jobs/'+j['id']+'/photos/'+p['id'],{**payload,**fields},method='put')
def public_urls(p,slug='oak-and-vale'):
    base='/api/public/'+slug+'/projects/'+p['content']['slug'];return [base,base+'/images/'+p['content']['images'][0]['id']]


@pytest.mark.parametrize('format',['PNG','JPEG','WEBP'])
def test_valid_images_are_private_reencoded_and_metadata_removed(app,format):
    c=client(app);j=job(c);p=upload(c,j,raw=pixels(format=format,metadata=True));assert p.status_code==200,p.text;p=p.json()
    url='/api/jobs/'+j['id']+'/photos/'+p['id']+'/image';r=c.get(url);assert r.status_code==200 and 'no-store' in r.headers['cache-control']
    with Image.open(io.BytesIO(r.content)) as image:
        assert image.format=='PNG' and image.mode=='RGBA' and image.info=={}
        assert image.size==((24,32) if format=='JPEG' else (32,24))
    assert b'PRIVATE EXAMPLE' not in r.content
    assert TestClient(app,headers=ORIGIN).get(url).status_code==401
    assert client(app,1).get(url).status_code==404
    assert c.get('/api/assets').json()==[]
    assert c.get('/api/public/oak-and-vale/projects').json()==[]


@pytest.mark.parametrize('raw,status',[(b'',413),(b'<svg/>',415),(b'not-image',415),(pixels(format='GIF'),415),(b'a'*(2*1024*1024+1),413)])
def test_unsupported_or_oversize_upload_rejected(app,raw,status):
    c=client(app);j=job(c);r=upload(c,j,raw=raw);assert r.status_code==status,r.text
    assert listing(c,j).json()==[] and not list((app.state.store.root/'oak/photos').iterdir())


def test_animation_and_pixel_limit(app):
    c=client(app);j=job(c);out=io.BytesIO();a=Image.new('RGB',(8,8),'red');a.save(out,'PNG',save_all=True,append_images=[Image.new('RGB',(8,8),'blue')],duration=10)
    assert upload(c,j,raw=out.getvalue()).status_code==415
    out=io.BytesIO();Image.new('RGB',(4000,3001)).save(out,'PNG');assert upload(c,j,raw=out.getvalue()).status_code==415


def test_upload_idempotency_duplicate_concurrency_and_different_file(app):
    a=client(app);b=client(app);j=job(a);key=str(uuid.uuid4())
    with ThreadPoolExecutor(2) as pool:results=list(pool.map(lambda c:upload(c,j,key=key),[a,b]))
    assert all(r.status_code==200 for r in results);assert results[0].json()==results[1].json()
    assert upload(a,j,'blue',key=key).status_code==409
    assert upload(b,j).json()['duplicate']
    assert len(listing(a,j).json())==1 and len(list((app.state.store.root/'oak/photos').iterdir()))==1
    j2=job(a);assert uploaded(a,j2)['id']!=results[0].json()['id']


def test_upload_failure_removes_only_owned_new_file(app,monkeypatch):
    from backend import photos
    c=client(app);j=job(c);uploaded(c,j);before=set((app.state.store.root/'oak/photos').iterdir())
    def broken(*args):raise OSError('Synthetic audit failure')
    monkeypatch.setattr(photos,'photo_snapshot',broken)
    with pytest.raises(OSError):upload(c,j,'blue')
    assert set((app.state.store.root/'oak/photos').iterdir())==before and len(listing(c,j).json())==1


def test_staff_exact_assignment_own_edits_and_permission_revocation(app):
    c=client(app);j=job(c);other=job(c);team=person(c);booking(c,j,team);s,_=add_staff(app,c,resource_id=team['id']);p=uploaded(s,j)
    assert listing(s,other).status_code==404 and upload(s,other).status_code==404
    assert edit_photo(s,j,p,caption='Private staff submission').status_code==200
    ownerphoto=uploaded(c,j,'blue');assert edit_photo(s,j,ownerphoto,caption='No').status_code==403
    assert state(s,j,p,'review').status_code==403
    assert state(c,j,p,'review').status_code==200
    assert edit_photo(s,j,p,caption='No').status_code==403
    assert listing(s,j,True).status_code==403
    assert 'uploaded_by' not in listing(s,j).json()[0]
    url='/api/jobs/'+j['id']+'/photos/'+p['id']+'/image';assert s.get(url).status_code==200
    assert update(c,member(c),resource_id=None).status_code==200
    assert listing(s,j).status_code==401 and s.get(url).status_code==401
    s=login(app);assert listing(s,j).status_code==404 and s.get(url).status_code==404 and upload(s,j,'red').status_code==404


def test_disabled_staff_and_archived_resource_cannot_read_bytes(app):
    c=client(app);j=job(c);team=person(c);booking(c,j,team,start_date='2020-11-02',end_date='2020-11-02');s,_=add_staff(app,c,resource_id=team['id']);p=uploaded(s,j);url='/api/jobs/'+j['id']+'/photos/'+p['id']+'/image'
    assert write(c,'/api/team/'+team['id']+'/archive',{'version':1}).status_code==200
    assert s.get(url).status_code==404
    assert update(c,member(c),enabled=False).status_code==200
    assert s.get(url).status_code==401


def test_pair_order_archive_restore_history_and_version_conflicts(app):
    c=client(app);j=job(c);a=uploaded(c,j);b=uploaded(c,j,'blue')
    assert edit_photo(c,j,a,caption='Private A',phase='before',pair_id='wall',position=8).status_code==200
    assert edit_photo(c,j,b,phase='before',pair_id='wall').status_code==409
    assert edit_photo(c,j,b,phase='after',pair_id='wall',position=2).status_code==200
    assert listing(c,j).json()[0]['id']==b['id']
    assert write(c,'/api/jobs/'+j['id']+'/photos/'+a['id']+'/state/review',{'version':1}).status_code==409
    assert state(c,j,a,'review').status_code==200;assert share(c,j,a).status_code==200
    assert state(c,j,a,'archive').status_code==200;assert upload(c,j).status_code==409
    assert state(c,j,a,'restore').status_code==200
    row=record(c,j,a);assert row['customer_share'] is None and row['reviewed']==0
    history=c.get('/api/jobs/'+j['id']+'/photos/'+a['id']+'/history').json();assert len(history)==6 and history[0]['action']=='photo.restore'


def test_exact_customer_sharing_is_separate_from_private_metadata(app):
    c=client(app);j=job(c,name='Private customer name');j2=job(c);p=uploaded(c,j);pc,_=portal(app,c,j);other,_=portal(app,c,j2)
    base='/api/portal/oak-and-vale/jobs/'+j['id']+'/photos';url=base+'/'+p['id']+'/image'
    assert pc.get(base).json()==[] and pc.get(url).status_code==404
    assert share(c,j,p).status_code==409
    assert edit_photo(c,j,p,caption='PRIVATE CAPTION',description='PRIVATE NOTE').status_code==200
    assert state(c,j,p,'review').status_code==200;assert share(c,j,p).status_code==200
    row=pc.get(base).json();assert row==[{'id':p['id'],'caption':'Customer safe caption','alt':'A neutral synthetic green square'}]
    assert pc.get(url).status_code==200 and other.get(base).status_code==404 and other.get(url).status_code==404
    assert share(c,j,p,False).status_code==200 and pc.get(url).status_code==404
    assert share(c,j,p).status_code==200
    assert write(c,'/api/customers/'+j['customer_id']+'/portal',method='delete').status_code==200
    assert pc.get(url).status_code==401


@pytest.mark.parametrize('archived',[False,True])
def test_customer_relink_blocked_by_photo_history(app,archived):
    c=client(app);j=job(c);j2=job(c);p=uploaded(c,j)
    if archived:assert state(c,j,p,'archive').status_code==200
    r=write(c,'/api/jobs/'+j['id']+'/customer-link-preview',{'customer_id':j2['customer_id'],'customer_version':1,'job_version':1});assert r.status_code==410


@pytest.mark.parametrize('confirmed',[False,1,'true',None])
def test_publication_requires_strict_explicit_acknowledgement(app,confirmed):
    c=client(app);j=job(c);p=project(c,j,reviewed(c,j));assert pub(c,p,confirmed).status_code==422
    assert c.get(public_urls(p)[0]).status_code==404


def test_publication_review_separate_projection_and_history(app):
    c=client(app);j=job(c,name='PRIVATE CUSTOMER');ph=uploaded(c,j);assert edit_photo(c,j,ph,caption='PRIVATE CAPTION',description='PRIVATE LOCATION').status_code==200
    p=project(c,j,ph);assert pub(c,p).status_code==409
    assert state(c,j,ph,'review').status_code==200;key=str(uuid.uuid4());r=pub(c,p,key=key);assert r.status_code==200,r.text
    # Retry exact saved request, not a new version.
    assert write(c,'/api/gallery/'+p['id']+'/publish',{'version':1,'permission_confirmed':True},key=key).json()==r.json()
    anonymous=TestClient(app,headers=ORIGIN);body=anonymous.get(public_urls(p)[0]);assert body.status_code==200
    for secret in ('PRIVATE CUSTOMER','PRIVATE CAPTION','PRIVATE LOCATION',j['id'],j['customer_id'],ph['id'],'source_id','uploaded_by','membership_id'):assert secret not in body.text
    assert anonymous.get(public_urls(p)[1]).status_code==200
    assert state(c,j,ph,'archive').status_code==409
    history=c.get('/api/gallery/'+p['id']).json()['history'];assert history[0]['permission_confirmed']==1
    assert project_action(c,p,'withdraw').status_code==200
    assert all(anonymous.get(url).status_code==404 for url in public_urls(p))
    assert anonymous.get('/api/public/oak-and-vale/projects').json()==[]
    assert state(c,j,ph,'archive').status_code==200
    old='/api/gallery/'+p['id']+'/history/'+str(history[0]['id'])+'/images/'+p['content']['images'][0]['id']
    assert c.get(old).status_code==200 and anonymous.get(old).status_code==401
    assert project_action(c,p,'archive').status_code==200;assert project_action(c,p,'restore').status_code==200
    assert c.get('/api/gallery/'+p['id']).json()['published'] is None


def test_gallery_draft_publish_pairs_source_scope_slug_and_image_identity(app):
    c=client(app);j=job(c);a=reviewed(c,j);b=reviewed(c,j,'blue');p=project(c,j,a);row=c.get('/api/gallery/'+p['id']).json();content=copy.deepcopy(row['draft']);content['images'][0].update(phase='before',pair_id='shelves')
    def save(content):return write(c,'/api/gallery/'+p['id'],{'version':c.get('/api/gallery/'+p['id']).json()['version'],'content':content},method='put')
    assert save(content).status_code==200;assert pub(c,p).status_code==422
    content['images'].append({'id':uid(),'source_kind':'job','source_id':b['id'],'caption':'','alt':'Blue square','phase':'after','pair_id':'shelves'});assert save(content).status_code==200;assert pub(c,p).status_code==200
    wrong=copy.deepcopy(content);wrong['images'][0]['source_id']=b['id'];wrong['images']=wrong['images'][:1];assert save(wrong).status_code==422
    other=job(c);foreign=uploaded(c,other,'red');wrong=copy.deepcopy(content);wrong['images'][0]['source_id']=foreign['id'];wrong['images'][0]['id']=uid();assert save(wrong).status_code==404
    assert write(c,'/api/gallery',{'source_job_id':j['id'],'content':content}).status_code==409
    assert write(c,'/api/gallery',{'source_job_id':other['id'],'content':ProjectContent(title='Other',slug=p['content']['slug']).model_dump()}).status_code==409


def test_unlisted_is_public_preview_is_private_and_cross_business_ids_fail(app):
    c=client(app);j=job(c);p=project(c,j,reviewed(c,j),in_index=False);assert pub(c,p).status_code==200
    anon=TestClient(app,headers=ORIGIN);other=client(app,1);assert anon.get('/api/public/oak-and-vale/projects').json()==[]
    assert anon.get(public_urls(p)[0]).status_code==200
    for path in ['/api/gallery/'+p['id'],'/api/gallery/'+p['id']+'/preview','/api/gallery/'+p['id']+'/preview/images/'+p['content']['images'][0]['id']]:
        assert anon.get(path).status_code==401 and other.get(path).status_code==404
    assert other.get('/api/public/harbour-home/projects/'+p['content']['slug']).status_code==404


def test_public_shell_xss_sitemap_withdraw_and_service_associations(tmp_path):
    app=create_app(tmp_path/'demo',static_dir=Path(__file__).parents[1]/'frontend/dist');c=client(app);j=job(c);p=project(c,j,reviewed(c,j),service_slugs=['carpentry']);current=c.get('/api/gallery/'+p['id']).json()
    # Use the saved published service slug, not an assumed business default.
    service=c.get('/api/public/oak-and-vale').json()['services'][0]['slug'];content=current['draft'];content.update(title='<script>alert(1)</script>',description='Private review has "quotes" & text',service_slugs=[service]);assert write(c,'/api/gallery/'+p['id'],{'version':1,'content':content},method='put').status_code==200
    assert pub(c,p).status_code==200
    r=c.get('/site/oak-and-vale/work/projects/'+p['content']['slug']);assert r.status_code==200 and '&lt;script&gt;' in r.text and '<script>alert(1)</script>' not in r.text and '<noscript>' in r.text and 'noindex' in r.text
    assert p['content']['slug'] in c.get('/site/oak-and-vale/sitemap.xml').text
    assert c.get(public_urls(p)[0]).json()['project']['service_slugs']==[service]
    assert project_action(c,p,'withdraw').status_code==200
    assert c.get('/site/oak-and-vale/work/projects/'+p['content']['slug']).status_code==404
    assert p['content']['slug'] not in c.get('/site/oak-and-vale/sitemap.xml').text


def test_standalone_library_project_retains_document_logo(app):
    c=client(app);aid=c.post('/api/assets',content=pixels()).json()['id'];site=c.get('/api/site').json();site['draft']['logo_id']=aid
    assert write(c,'/api/site',{'version':site['version'],'content':site['draft']},method='put').status_code==200
    assert write(c,'/api/site/publish',{'version':2}).status_code==200
    j=job(c);q=quote(c,j)
    content=ProjectContent(title='Library project',slug='library',images=[{'id':uid(),'source_kind':'website','source_id':aid,'alt':'Synthetic library image'}]).model_dump();p=write(c,'/api/gallery',{'content':content}).json();assert pub(c,p).status_code==200
    assert project_action(c,p,'withdraw').status_code==200
    assert c.get('/api/documents/quote/'+q['id']+'/logo').status_code==200 and c.get('/api/public/oak-and-vale/assets/'+aid).status_code==200


def test_publication_concurrency_and_history_restore_does_not_republish(app):
    c=client(app);other=client(app);j=job(c);p=project(c,j,reviewed(c,j));path='/api/gallery/'+p['id']
    with ThreadPoolExecutor(2) as pool:rs=list(pool.map(lambda who:write(who,path+'/publish',{'version':1,'permission_confirmed':True}),[c,other]))
    assert sorted(r.status_code for r in rs)==[200,409]
    revision=c.get(path).json()['history'][0]['id'];assert project_action(c,p,'withdraw').status_code==200
    r=write(c,path+'/restore-draft',{'version':3,'revision_id':revision});assert r.status_code==200
    assert c.get(path).json()['published'] is None and c.get(public_urls(p)[0]).status_code==404


def run_media_drill(tmp_path):
    root=tmp_path/'fictional';app=create_app(root);owner=client(app);records={}
    for n,slug in enumerate(['oak-and-vale','harbour-home','willow-repairs','cedar-repairs']):
        if n<2:c=client(app,n);bid=DEMO_BUSINESSES[n]['id'];slug=DEMO_BUSINESSES[n]['slug']
        else:
            b=new_business(owner,slug).json();bid=b['business_id'];r=write(owner,'/api/business/switch',{'membership_id':b['membership_id']});owner.headers['x-csrf-token']=r.json()['csrf'];c=owner
        j=job(c,name='Synthetic media owner');a=reviewed(c,j);assert share(c,j,a).status_code==200;p=project(c,j,a)
        after=reviewed(c,j,'blue');p['content']['images'][0].update(phase='before',pair_id='example')
        p['content']['images'].append({'id':uid(),'source_kind':'job','source_id':after['id'],'alt':'Synthetic blue after image','caption':'Reviewed after','phase':'after','pair_id':'example'})
        assert write(c,'/api/gallery/'+p['id'],{'version':1,'content':p['content']},method='put').status_code==200;assert pub(c,p).status_code==200
        archived=uploaded(c,j,'red');assert state(c,j,archived,'archive').status_code==200
        private=uploaded(c,j,'yellow')
        logo=c.post('/api/assets',content=pixels('purple')).json()['id'];site=c.get('/api/site').json();site['draft']['logo_id']=logo
        assert write(c,'/api/site',{'version':site['version'],'content':site['draft']},method='put').status_code==200
        assert write(c,'/api/site/publish',{'version':site['version']+1}).status_code==200
        q=quote(c,j)
        team=person(c);booking(c,j,team);staff,_=add_staff(app,c,resource_id=team['id'],email='staff-'+bid+'@fiction.example')
        assert update(c,member(c,'staff-'+bid+'@fiction.example'),enabled=False).status_code==200
        records[bid]={'slug':slug,'job':j,'photo':a,'project':p,'quote':q,'logo':logo,'before':table_rows(root,bid)}
    old_portal,code=portal(app,client(app),records['oak']['job']);old_owner=client(app)
    snap=tmp_path/'media.zip';saved=recovery.backup(root,snap);assert recovery.verify(snap,saved['sha256'])['valid'];assert saved['files']==25  # registry + four DBs + sixteen private images + four logos
    source_images={bid:{p.name:p.read_bytes() for p in (root/bid/'photos').iterdir()} for bid in records}
    shutil.rmtree(root);target=tmp_path/'recovered';result=recovery.restore(snap,saved['sha256'],target,confirm_fresh=True);assert json.loads(Path(result['report']).read_text())['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
    for bid,r in records.items():
        after=table_rows(target,bid)
        for table in ('photo_blobs','job_photos','photo_history','gallery_projects','gallery_history'):assert after[table]==r['before'][table]
        assert {p.name:p.read_bytes() for p in (target/bid/'photos').iterdir()}==source_images[bid]
        assert anonymous.get(public_urls(r['project'],r['slug'])[1]).status_code==200
    fresh,_=new_login(restored,target,'oak');assert listing(fresh,records['oak']['job']).status_code==200
    assert fresh.get('/api/jobs/'+records['harbour']['job']['id']+'/photos').status_code==404
    assert fresh.get('/api/documents/quote/'+records['oak']['quote']['id']+'/logo').status_code==200
    old=TestClient(restored,headers=ORIGIN);old.cookies.update(old_owner.cookies);assert old.get('/api/session').status_code==401
    oldpc=TestClient(restored,headers=ORIGIN);oldpc.cookies.update(old_portal.cookies);assert oldpc.get('/api/portal/oak-and-vale/me').status_code==401
    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';second_saved=recovery.backup(target,second);assert recovery.verify(second,second_saved['sha256'])['valid']
    return {'businesses':4,'source_deleted_before_restore':True,'private_photos':16,'retained_logos':4,'all_media_bytes_and_five_media_tables_preserved':True,'private_shared_published_and_archived_states_preserved':True,'before_after_public_projects':4,'disabled_staff_resource_links_preserved':True,'exact_customer_and_business_binding_preserved':True,'historical_document_logo_read_verified':True,'old_passwords_sessions_and_customer_grants_rejected':True,'second_generation_backup_verified':True,'snapshot_format':recovery.VERSION,'snapshot_files':saved['files'],'snapshot_sha256':saved['sha256'],'browser_verified':False}


def test_four_business_media_loss_restore_and_old_credentials_rejected(tmp_path):
    run_media_drill(tmp_path)


def test_format_four_backup_restores_into_empty_photo_schema(tmp_path):
    # Rebuild pinned v0.7 tables using trusted local schema; no supplied source runtime.
    root=tmp_path/'old';Store(root)
    with sqlite3.connect(root/'registry.sqlite3') as db:dataset=db.execute('SELECT dataset_id FROM dataset_identity').fetchone()[0];db.execute("UPDATE users SET password=''")
    for bid in ('oak','harbour'):
        with sqlite3.connect(root/bid/'business.sqlite3') 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
            from backend.workflow_recovery import TABLES as WORKFLOW_TABLES
            for table in INTAKE_TABLES+CUSTOMER_TABLES+WORKFLOW_TABLES+FEEDBACK_TABLES+PORTAL_TABLES+SALES_TABLES+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','gallery_history','gallery_projects','photo_history','job_photos','photo_blobs'):db.execute('DROP TABLE '+table)
    meta={'format':recovery.FORMAT,'version':4,'prototype_version':'0.7.0','dataset_id':dataset,'created_at':'2026-10-27','auth_policy':recovery.AUTH_POLICY,'businesses':[{'id':b['id'],'slug':b['slug'],'name':b['name'],'enabled':1} for b in DEMO_BUSINESSES],'domains':[],'files':{}}
    files={'registry.sqlite3':(root/'registry.sqlite3',None,'identity'),**{'businesses/'+b+'/business.sqlite3':(root/b/'business.sqlite3',b,'database') for b in ('oak','harbour')}};archive=tmp_path/'format4.zip'
    with zipfile.ZipFile(archive,'w') as z:
        for name,(path,bid,kind) in files.items():
            data=path.read_bytes();z.writestr(name,data);meta['files'][name]={'bytes':len(data),'sha256':hashlib.sha256(data).hexdigest(),'business_id':bid,'kind':kind}
        z.writestr('manifest.json',json.dumps(meta))
    target=tmp_path/'restored';result=recovery.restore(archive,recovery.checksum(archive),target,confirm_fresh=True);assert json.loads(Path(result['report']).read_text())['source_format']==4
    for bid in ('oak','harbour'):
        with sqlite3.connect(target/bid/'business.sqlite3') as db:assert db.execute('SELECT COUNT(*) FROM job_photos').fetchone()[0]==0


@pytest.mark.parametrize('tamper',['bytes','binding','history','membership','published','orphan'])
def test_media_recovery_rejects_inconsistent_state(app,tmp_path,tamper):
    c=client(app);j=job(c);p=reviewed(c,j);g=project(c,j,p);assert pub(c,g).status_code==200;root=app.state.store.root
    other_customer=job(c)['customer_id'] if tamper=='binding' else None
    if tamper=='bytes':next((root/'oak/photos').iterdir()).write_bytes(pixels('blue'))
    else:
        with app.state.store.tenant('oak') as db:
            if tamper=='binding':db.execute('UPDATE job_photos SET customer_id=?',(other_customer,))
            if tamper=='history':db.execute('DELETE FROM photo_history WHERE id=(SELECT MIN(id) FROM photo_history)')
            if tamper=='membership':db.execute("UPDATE gallery_history SET membership_id='foreign'")
            if tamper=='published':db.execute('UPDATE gallery_history SET permission_confirmed=0')
            if tamper=='orphan':db.execute('INSERT INTO photo_blobs SELECT ?,sha256,bytes,width,height,created_at FROM photo_blobs',(uid(),))
    with pytest.raises((recovery.RecoveryError,ValueError)):recovery.backup(root,tmp_path/'bad.zip')

def test_old_upload_retry_cannot_bypass_removed_booking(app):
    c=client(app);j=job(c);team=person(c);book=booking(c,j,team);s,_=add_staff(app,c,resource_id=team['id']);key=str(uuid.uuid4());assert upload(s,j,key=key).status_code==200
    assert write(c,'/api/calendar/'+book['id']+'/archive',{'version':1}).status_code==200
    assert upload(s,j,key=key).status_code==404


def test_photo_symlink_denied_and_business_context_enforced(app,tmp_path):
    c=client(app);j=job(c);p=uploaded(c,j);url='/api/jobs/'+j['id']+'/photos/'+p['id']+'/image'
    assert c.get(url,headers={'x-business-id':'harbour'}).status_code==409
    path=next((app.state.store.root/'oak/photos').iterdir());saved=tmp_path/'saved.png';path.rename(saved);path.symlink_to(saved)
    with pytest.raises(ValueError):c.get(url)


def test_restored_gallery_snapshot_does_not_include_new_unsaved_text(app):
    c=client(app);j=job(c);p=project(c,j,reviewed(c,j));assert pub(c,p).status_code==200
    row=c.get('/api/gallery/'+p['id']).json();row['draft']['title']='New private draft title';assert write(c,'/api/gallery/'+p['id'],{'version':row['version'],'content':row['draft']},method='put').status_code==200
    assert 'New private draft title' not in c.get(public_urls(p)[0]).text
    assert c.get('/api/gallery/'+p['id']+'/preview').json()['title']=='New private draft title'

def test_relink_does_not_move_shared_photos_or_staff_notes(app):
    c=client(app);j=job(c);other=job(c);p=reviewed(c,j);assert share(c,j,p).status_code==200
    payload={'customer_id':other['customer_id'],'customer_version':1,'job_version':1,'confirmed':True}
    assert write(c,'/api/jobs/'+j['id']+'/customer-link',payload).status_code==410
    # Remove the original scheduling assignment after a staff note was saved.
    work=job(c);team=person(c);book=booking(c,work,team);s,_=add_staff(app,c,resource_id=team['id']);assert write(s,'/api/staff/jobs/'+work['id']+'/updates',{'note':'Private original customer work note'}).status_code==200
    assert write(c,'/api/calendar/'+book['id']+'/archive',{'version':1}).status_code==200
    # The stricter existing booking-history guard also remains intact.
    assert write(c,'/api/jobs/'+work['id']+'/customer-link',payload).status_code==410
    with app.state.store.tenant('oak') as db:
        assert db.execute('SELECT customer_id FROM jobs WHERE id=?',(j['id'],)).fetchone()[0]==j['customer_id']
        assert db.execute('SELECT customer_id FROM job_photos WHERE id=?',(p['id'],)).fetchone()[0]==j['customer_id']