"""Invented-account permission, provisioning and recovery contracts."""
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import json
import sqlite3
import time
import uuid
import zipfile
import pytest
from fastapi.testclient import TestClient
from backend.app import create_app
from backend.store import Store,digest,uid
from backend.recovery import backup,verify,restore,checksum,RecoveryError
from test_boundaries import app,client,write,job,quote,invoice,portal,ORIGIN
from test_workforce import person
from test_scheduling import booking

PASSWORD='Invented-staff-only-2026!'
EMAIL='jamie@staff.example'


def login(app,email=EMAIL,password=PASSWORD):
    c=TestClient(app,headers=ORIGIN);r=c.post('/api/login',json={'email':email,'password':password})
    assert r.status_code==200,r.text
    c.headers['x-csrf-token']=r.json()['csrf'];return c


def invitation(c,email=EMAIL,role='staff',resource_id=None):
    r=write(c,'/api/invitations',{'email':email,'role':role,'resource_id':resource_id})
    assert r.status_code==200,r.text;return r.json()


def accept(app,code,password=PASSWORD):
    return TestClient(app,headers=ORIGIN).post('/api/account/invitations/accept',json={'code':code,'password':password,'display_name':'Jamie Fictional'})


def member(c,email=EMAIL):return next(m for m in c.get('/api/memberships').json()['members'] if m['email']==email)


def update(c,m,**changes):
    return write(c,'/api/memberships/'+m['id'],{'version':m['version'],'role':m['role'],'enabled':bool(m['enabled']),'resource_id':m['resource_id'],**changes},method='put')


def add_staff(app,c,resource_id=None,role='staff',email=EMAIL):
    inv=invitation(c,email,role,resource_id);assert accept(app,inv['code']).status_code==200
    return login(app,email),inv


def new_business(c,slug='willow-repairs',key=None):
    return write(c,'/api/businesses',{'name':'Willow Repairs','slug':slug,'area':'Invented village'},key=key)


def test_third_business_empty_retry_restart_and_switch(app):
    c=client(app);j=job(c);old=TestClient(app,headers=ORIGIN);old.cookies.update(c.cookies);old.headers['x-csrf-token']=c.headers['x-csrf-token']
    key=str(uuid.uuid4());r=new_business(c,key=key);assert r.status_code==200,r.text;b=r.json()
    assert new_business(c,key=key).json()==b
    assert new_business(c,slug='changed-slug',key=key).status_code==409
    assert new_business(c).status_code==409
    assert c.get('/api/session').json()['business_id']=='oak'
    switched=write(c,'/api/business/switch',{'membership_id':b['membership_id']});assert switched.status_code==200
    assert old.get('/api/session').status_code==401
    assert c.get('/api/workspace',headers={'x-business-id':'oak'}).status_code==409
    assert write(c,'/api/jobs',{}).status_code==403  # stale CSRF
    c.headers['x-csrf-token']=switched.json()['csrf']
    assert c.get('/api/workspace').json()['jobs']==[]
    assert c.get('/api/jobs/'+j['id']).status_code==404
    assert c.get('/api/assets').json()==[]
    assert c.get('/api/team').json()==[]
    assert len(c.get('/api/memberships').json()['members'])==1
    assert c.get('/api/public/'+b['slug']).status_code==200
    state=c.get('/api/setup').json();assert not any(state['checks'].values())
    reopened=create_app(app.state.store.root);cc=login(reopened,'owner@oak.example','Oak-demo-only-2026!')
    assert len(cc.get('/api/session').json()['memberships'])==2
    assert len(list((app.state.store.root/'.provisioning').iterdir()))==0


@pytest.mark.parametrize('slug',['api','../other','Caps','double--dash','short_thing','-start','end-','ab','a'*61])
def test_business_slug_rejected(app,slug):
    assert new_business(client(app),slug).status_code==422


def test_provision_concurrency_same_and_distinct_slugs(app):
    a=client(app);b=client(app)
    with ThreadPoolExecutor(2) as pool:results=list(pool.map(new_business,[a,b]))
    assert sorted(r.status_code for r in results)==[200,409]
    with ThreadPoolExecutor(2) as pool:results=list(pool.map(lambda args:new_business(*args),[(a,'birch-repairs'),(b,'cedar-repairs')]))
    assert all(r.status_code==200 for r in results)
    assert len({r.json()['business_id'] for r in results})==2
    with app.state.store.registry() as db:assert db.execute('SELECT COUNT(*) FROM businesses').fetchone()[0]==5


@pytest.mark.parametrize('point',['staged','renamed','before_commit'])
def test_failed_provision_rolls_back_then_retry_succeeds(app,point):
    c=client(app);before={p.name for p in app.state.store.root.iterdir()};key=str(uuid.uuid4())
    def fault(p):
        if p==point:raise OSError('Injected local failure')
    app.state.identity.provision_hook=fault
    with pytest.raises(OSError):new_business(c,key=key)
    with app.state.store.registry() as db:
        assert db.execute('SELECT COUNT(*) FROM businesses').fetchone()[0]==2
        assert db.execute('SELECT COUNT(*) FROM registry_operations').fetchone()[0]==0
    assert not list((app.state.store.root/'.provisioning').iterdir())
    assert {p.name for p in app.state.store.root.iterdir()}-before=={'.provisioning'}
    app.state.identity.provision_hook=lambda p:None
    assert new_business(c,key=key).status_code==200


def test_crash_marker_cleanup_only_matching_uncommitted_data(app):
    store=app.state.store;bid=uid();pending=store.root/'.provisioning';pending.mkdir()
    with store.registry() as db:dataset=db.execute('SELECT dataset_id FROM dataset_identity').fetchone()[0]
    store.initialise_folder(store.root/bid,{'id':bid,'name':'Crash Fiction','area':'Test','email':'','colour':'#245648'},dataset)
    (pending/(bid+'.json')).write_text(json.dumps({'business_id':bid,'dataset_id':dataset}))
    Store(store.root);assert not (store.root/bid).exists();assert not list(pending.iterdir())
    r=new_business(client(app)).json();(pending/(r['business_id']+'.json')).write_text(json.dumps({'business_id':r['business_id'],'dataset_id':dataset}))
    Store(store.root);assert (store.root/r['business_id']/'business.sqlite3').is_file()


def test_default_deny_complete_inventory_and_every_owner_endpoint(app):
    owner=client(app);staff,_=add_staff(app,owner)
    policy=json.loads((Path(__file__).parents[1]/'docs/access-policy.json').read_text())
    inventory={(r['method'],r['path']):r['access'] for r in policy['routes']}
    for route in app.routes:
        for method in route.methods:assert (method,route.path) in inventory
    for row in policy['routes']:
        if row['access']=='owner':
            path=row['path']
            import re
            path=re.sub(r'\{[^}]+\}','not-a-record',path)
            r=staff.request(row['method'],path,json={'deliberately':'invalid'} if row['method']!='GET' else None)
            assert r.status_code==403,(row,r.status_code,r.text)
    @app.get('/api/unlisted-test-endpoint')
    def rogue():return {'secret':'must not run'}
    assert owner.get('/api/unlisted-test-endpoint').status_code==403
    assert staff.get('/api/unlisted-test-endpoint').status_code==403
    assert owner.get('/api/staff/work').status_code==403


def test_staff_assignments_projection_notes_and_revoked_resource(app):
    c=client(app);p=person(c);other=person(c,name='Jamie Demo')
    j=job(c);hidden=job(c);cal=booking(c,j,p);booking(c,hidden,other)
    staff,_=add_staff(app,c,p['id']);before=staff.get('/api/staff/work').json()
    assert [x['id'] for x in before['jobs']]==[j['id']]
    raw=json.dumps(before)
    for secret in ('email','notes','hourly_cost','resource_snapshot','Private invented booking note','Two shelves'):assert secret not in raw
    key=str(uuid.uuid4());r=write(staff,'/api/staff/jobs/'+j['id']+'/updates',{'note':'Invented visit completed'},key=key)
    assert r.status_code==200 and write(staff,'/api/staff/jobs/'+j['id']+'/updates',{'note':'Invented visit completed'},key=key).json()==r.json()
    assert len(c.get('/api/jobs/'+j['id']+'/staff-updates').json())==1
    assert write(staff,'/api/staff/jobs/'+hidden['id']+'/updates',{'note':'Not assigned'}).status_code==404
    assert write(c,'/api/calendar/'+cal['id']+'/archive',{'version':1}).status_code==200
    assert write(c,'/api/team/'+p['id']+'/archive',{'version':1}).status_code==200
    assert staff.get('/api/staff/work').json()['jobs']==[]
    assert write(staff,'/api/staff/jobs/'+j['id']+'/updates',{'note':'No longer active'}).status_code==404


def test_unlinked_staff_never_match_name_or_email(app):
    c=client(app);p=person(c);j=job(c);booking(c,j,p)
    staff,_=add_staff(app,c)
    assert staff.get('/api/staff/work').json()=={'jobs':[],'bookings':[],'resource_linked':False}
    assert update(c,member(c),resource_id=p['id']).status_code==200
    assert staff.get('/api/session').status_code==401
    assert login(app).get('/api/staff/work').json()['jobs'][0]['id']==j['id']


def test_membership_disable_role_change_and_last_owner(app):
    c=client(app);staff,_=add_staff(app,c)
    m=member(c);assert update(c,m,enabled=False).status_code==200
    assert staff.get('/api/session').status_code==401
    assert staff.post('/api/login',json={'email':EMAIL,'password':PASSWORD}).status_code==401
    assert update(c,m,enabled=True).status_code==409
    assert update(c,member(c),enabled=True,role='owner').status_code==200
    second=login(app);assert second.get('/api/workspace').status_code==200
    assert update(c,member(c),role='staff').status_code==200
    assert second.get('/api/workspace').status_code==401
    owner=member(c,'owner@oak.example')
    assert update(c,owner,role='staff').status_code==409
    assert update(c,owner,enabled=False).status_code==409
    assert c.get('/api/session').status_code==200
    assert len(c.get('/api/memberships/audit').json())>=4


def test_two_owners_cannot_concurrently_remove_last_owner(app):
    a=client(app);b,_=add_staff(app,a,role='owner');ma=member(a,'owner@oak.example');mb=member(a)
    with ThreadPoolExecutor(2) as pool:results=list(pool.map(lambda item:update(*item,role='staff').status_code,[(a,ma),(b,mb)]))
    assert sorted(results)==[200,409]


def test_invites_scope_expiry_reissue_reuse_and_existing_password(app):
    c=client(app);b=client(app,1);inv=invitation(c)
    assert write(b,'/api/invitations/'+inv['id'],method='delete').status_code==404
    with app.state.store.registry() as db:
        row=db.execute('SELECT token_hash FROM access_tokens WHERE id=?',(inv['id'],)).fetchone();assert row[0]==digest(inv['code']) and row[0]!=inv['code']
        db.execute('UPDATE access_tokens SET expires=? WHERE id=?',(time.time()-1,inv['id']))
    assert accept(app,inv['code']).status_code==400
    inv=invitation(c);latest=invitation(c);assert accept(app,inv['code']).status_code==400
    assert accept(app,latest['code']).status_code==200
    assert accept(app,latest['code']).status_code==400
    inv2=invitation(b)
    assert accept(app,inv2['code'],'Different-test-password!').status_code==400
    assert accept(app,inv2['code']).status_code==200
    staff=login(app);assert len(staff.get('/api/session').json()['memberships'])==2
    m=member(c);assert update(c,m,enabled=False).status_code==200
    assert invitation(b,email='second@staff.example')['code']
    assert write(c,'/api/invitations',{'email':EMAIL}).status_code==409
    assert all('code' not in r and 'token_hash' not in r for r in c.get('/api/memberships').json()['invitations'])


def test_invitation_concurrent_consume_and_resource_ownership(app):
    c=client(app);b=client(app,1);p=person(b)
    assert write(c,'/api/invitations',{'email':EMAIL,'resource_id':p['id']}).status_code==422
    p=person(c);inv=invitation(c,resource_id=p['id'])
    with ThreadPoolExecutor(2) as pool:results=list(pool.map(lambda _:accept(app,inv['code']).status_code,range(2)))
    assert sorted(results)==[200,400]
    assert write(c,'/api/invitations',{'email':'other@staff.example','resource_id':p['id']}).status_code==409
    assert update(c,member(c),enabled=False).status_code==200
    assert write(c,'/api/invitations',{'email':'other@staff.example','resource_id':p['id']}).status_code==409
    assert update(c,member(c),resource_id=None).status_code==200
    assert write(c,'/api/invitations',{'email':'other@staff.example','resource_id':p['id']}).status_code==200


def reset_request(app,email=EMAIL):return TestClient(app,headers=ORIGIN).post('/api/account/reset/request',json={'email':email})


def reset_code(app,email=EMAIL):
    captures=[json.loads(p.read_text()) for p in (app.state.store.root/'.local-capture').glob('*.json')]
    return next(c['code'] for c in captures if c['email']==email)


def test_reset_generic_hashed_capture_reuse_and_cross_business_revocation(app):
    owner=client(app);staff,_=add_staff(app,owner);b=client(app,1);inv=invitation(b);assert accept(app,inv['code']).status_code==200
    other=login(app);m=next(m for m in other.get('/api/session').json()['memberships'] if m['business_id']=='harbour')
    switch=write(other,'/api/business/switch',{'membership_id':m['id']});other.headers['x-csrf-token']=switch.json()['csrf']
    unknown=reset_request(app,'unknown@staff.example');known=reset_request(app)
    assert unknown.status_code==known.status_code==200 and unknown.json()==known.json()
    code=reset_code(app)
    with app.state.store.registry() as db:assert db.execute("SELECT token_hash FROM access_tokens WHERE kind='reset'").fetchone()[0]==digest(code)
    root=app.state.store.root
    assert ((root/'.local-capture').stat().st_mode&0o777)==0o700
    assert all((p.stat().st_mode&0o777)==0o600 for p in (root/'.local-capture').iterdir())
    anon=TestClient(app,headers=ORIGIN);payload={'code':code,'new_password':'New-invented-password-2026!'}
    assert anon.post('/api/account/reset/complete',json=payload).status_code==200
    assert anon.post('/api/account/reset/complete',json=payload).status_code==400
    assert staff.get('/api/session').status_code==other.get('/api/session').status_code==401
    assert not list((root/'.local-capture').iterdir())
    assert login(app,password=payload['new_password']).get('/api/session').status_code==200


def test_expired_superseded_and_password_change_reset_codes(app):
    owner=client(app);staff,_=add_staff(app,owner);reset_request(app);old=reset_code(app)
    reset_request(app);new=reset_code(app);anon=TestClient(app,headers=ORIGIN)
    def complete(code):return anon.post('/api/account/reset/complete',json={'code':code,'new_password':'Updated-fictional-2026!'})
    assert old!=new and complete(old).status_code==400
    with app.state.store.registry() as db:db.execute("UPDATE access_tokens SET expires=? WHERE kind='reset'",(time.time()-1,))
    assert complete(new).status_code==400
    reset_request(app);new=reset_code(app)
    assert write(staff,'/api/password',{'current_password':PASSWORD,'new_password':'Changed-fictional-2026!'}).status_code==200
    assert complete(new).status_code==400
    assert staff.get('/api/session').status_code==401


def test_reset_capture_failure_and_rate_fail_closed(app,monkeypatch):
    c=client(app);add_staff(app,c)
    from backend import recovery
    def fail_capture(*a):raise OSError('Injected disk failure')
    monkeypatch.setattr(recovery,'private_write',fail_capture)
    assert reset_request(app).json()==reset_request(app,'missing@staff.example').json()
    with app.state.store.registry() as db:assert db.execute("SELECT COUNT(*) FROM access_tokens WHERE kind='reset' AND used=0").fetchone()[0]==0
    for _ in range(4):assert reset_request(app).status_code==200
    assert reset_request(app).status_code==429


def test_setup_resume_requires_real_local_progress(app):
    c=client(app);state=c.get('/api/setup').json()
    p={'version':1,'reviewed_publication_id':state['publication_id'],'solo_confirmed':True,'guide_read':True}
    assert write(c,'/api/setup',p,method='put').status_code==200
    assert write(c,'/api/setup',p,method='put').status_code==409
    c=client(create_app(app.state.store.root));state=c.get('/api/setup').json()
    assert state['checks']['website_reviewed'] and state['checks']['team_choice'] and state['checks']['guide_read']
    assert not state['checks']['first_job'] and not state['checks']['issued_invoice']
    site=c.get('/api/site').json();assert write(c,'/api/site/publish',{'version':site['version']}).status_code==200
    assert not c.get('/api/setup').json()['checks']['website_reviewed']
    j=job(c);q=quote(c,j);write(c,'/api/quotes/'+q['id']+'/share',{'version':1})
    state=c.get('/api/setup').json();assert state['checks']['first_job'] and state['checks']['shared_quote']
    assert set(state['external'].values())=={'not available'}


def test_customer_cookie_and_cross_tenant_memberships_never_authorize_staff(app):
    c=client(app);p,_=portal(app,c,job(c));staff,_=add_staff(app,c)
    assert p.get('/api/staff/work').status_code==401
    assert staff.get('/api/portal/oak-and-vale/me').status_code==401
    other=client(app,1);m=member(other,'owner@harbour.example')
    assert write(c,'/api/business/switch',{'membership_id':m['id']}).status_code==404
    assert update(c,m,enabled=False).status_code==404
    assert 'owner@harbour.example' not in c.get('/api/session').text


def test_four_business_recovery_preserves_inactive_links_never_auth_tokens(app,tmp_path):
    c=client(app);new_business(c);new_business(c,'fourth-workshop');p=person(c);j=job(c);cal=booking(c,j,p)
    staff,invite=add_staff(app,c,p['id']);write(staff,'/api/staff/jobs/'+j['id']+'/updates',{'note':'Invented saved note'})
    reset_request(app);reset=reset_code(app);pending=invitation(c,'pending@staff.example')
    assert update(c,member(c),enabled=False).status_code==200
    assert write(c,'/api/calendar/'+cal['id']+'/archive',{'version':1}).status_code==200
    assert write(c,'/api/team/'+p['id']+'/archive',{'version':1}).status_code==200
    with app.state.store.registry() as db:
        db.execute('UPDATE users SET disabled=1 WHERE email=?',(EMAIL,))
        memberships=[tuple(r) for r in db.execute('SELECT * FROM memberships ORDER BY id')]
        password_hashes=[r[0] for r in db.execute('SELECT password FROM users')]
    archive=tmp_path/'four.zip';saved=backup(app.state.store.root,archive);assert len(verify(archive,saved['sha256'])['businesses'])==4
    with zipfile.ZipFile(archive) as z:
        raw=b''.join(z.read(n) for n in z.namelist())
        for secret in [reset,pending['code'],invite['code'],digest(reset),digest(pending['code']),*password_hashes]:assert secret.encode() not in raw
        assert not any('capture' in n for n in z.namelist())
    target=tmp_path/'restored';restore(archive,saved['sha256'],target,confirm_fresh=True)
    with sqlite3.connect(target/'registry.sqlite3') as db:
        assert db.execute('SELECT * FROM memberships ORDER BY id').fetchall()==memberships
        for table in ('sessions','access_tokens','registry_operations'):assert db.execute('SELECT COUNT(*) FROM '+table).fetchone()[0]==0
    restored=create_app(target);anon=TestClient(restored,headers=ORIGIN)
    assert anon.post('/api/account/reset/complete',json={'code':reset,'new_password':'New-fictional-2026!'}).status_code==400
    assert accept(restored,pending['code']).status_code==400
    assert EMAIL not in (target/'RESTORED-LOGIN.txt').read_text()
    assert (target/'RESTORED-LOGIN.txt').read_text().count('owner@oak.example')==1
    with sqlite3.connect(target/'registry.sqlite3') as db:assert db.execute('SELECT disabled FROM users WHERE email=?',(EMAIL,)).fetchone()[0]==1
    second=backup(target,tmp_path/'second.zip');assert verify(second['archive'],second['sha256'])['valid']


def test_registry_credential_and_foreign_resource_corruption_rejected(app,tmp_path):
    from test_recovery import rewrite_archive
    c=client(app);add_staff(app,c);archive=tmp_path/'good.zip';backup(app.state.store.root,archive)
    for bad_kind in ('credential','session','resource','last-owner'):
        def mutate(files,meta):
            path=tmp_path/(bad_kind+'.sqlite3');path.write_bytes(files['registry.sqlite3'])
            with sqlite3.connect(path) as db:
                if bad_kind=='credential':db.execute("UPDATE users SET password='forbidden-credential'")
                elif bad_kind=='session':db.execute("INSERT INTO sessions VALUES('forbidden','oak','oak','staff','csrf',99999999999,0)")
                elif bad_kind=='resource':db.execute("UPDATE memberships SET resource_id=? WHERE user_id<>'oak' AND business_id='oak'",(uid(),))
                else:db.execute("UPDATE memberships SET role='staff' WHERE business_id='oak'")
            files['registry.sqlite3']=path.read_bytes();meta['files']['registry.sqlite3'].update(bytes=len(files['registry.sqlite3']),sha256=checksum(path))
        bad=tmp_path/(bad_kind+'.zip');sha=rewrite_archive(archive,bad,mutate)
        with pytest.raises(RecoveryError):verify(bad,sha)


def test_missing_registered_storage_fails_closed(app):
    import shutil
    c=client(app);bid=new_business(c).json()['business_id'];shutil.rmtree(app.state.store.root/bid)
    with pytest.raises(ValueError,match='storage is missing'):Store(app.state.store.root)
    assert not (app.state.store.root/bid).exists()


def test_simultaneous_same_operation_provisions_once(app):
    a=client(app);b=client(app);key=str(uuid.uuid4())
    with ThreadPoolExecutor(2) as pool:results=list(pool.map(lambda c:new_business(c,key=key),[a,b]))
    assert all(r.status_code==200 for r in results) and results[0].json()==results[1].json()
    with app.state.store.registry() as db:assert db.execute('SELECT COUNT(*) FROM businesses').fetchone()[0]==3


def test_reset_consumed_once_concurrently_and_disabled_member_stays_disabled(app):
    c=client(app);staff,_=add_staff(app,c);reset_request(app);code=reset_code(app)
    def consume(_):return TestClient(app,headers=ORIGIN).post('/api/account/reset/complete',json={'code':code,'new_password':'Reset-concurrent-fictional!'}).status_code
    with ThreadPoolExecutor(2) as pool:assert sorted(pool.map(consume,range(2)))==[200,400]
    reset_request(app);code=reset_code(app);assert update(c,member(c),enabled=False).status_code==200
    assert consume(0)==400
    with app.state.store.registry() as db:assert db.execute("SELECT enabled FROM memberships WHERE user_id=(SELECT id FROM users WHERE email=?)",(EMAIL,)).fetchone()[0]==0


def test_removing_owner_revokes_their_unaccepted_invites(app):
    c=client(app);second,_=add_staff(app,c,role='owner');inv=invitation(c,'waiting@staff.example')
    assert update(second,member(second,'owner@oak.example'),role='staff').status_code==200
    assert accept(app,inv['code']).status_code==400


def test_explicit_resource_change_revokes_sessions_and_owner_only_audit(app):
    c=client(app);r1=person(c);r2=person(c,name='Other');staff,_=add_staff(app,c,r1['id'])
    assert update(c,member(c),resource_id=r2['id']).status_code==200
    assert staff.get('/api/session').status_code==401
    assert login(app).get('/api/memberships/audit').status_code==403
    assert not any('password' in row['detail'] or 'code' in row['detail'] for row in c.get('/api/memberships/audit').json())


def make_v06(root):
    """Exact pinned v0.6 SQL with copied synthetic records; never runs old code."""
    current=create_app(root/'new-schema-source');c=client(current);j=job(c);quote(c,j)
    legacy=root/'legacy';legacy.mkdir()
    pinned=json.loads((Path(__file__).parents[1]/'docs/recovery-schema-v3.json').read_text())
    for part in ['registry','oak','harbour']:
        relative='registry.sqlite3' if part=='registry' else part+'/business.sqlite3'
        destination=legacy/relative;destination.parent.mkdir(exist_ok=True)
        if part!='registry':(destination.parent/'assets').mkdir()
        with sqlite3.connect(destination) as output,sqlite3.connect(current.state.store.root/relative) as source:
            sql=pinned['registry' if part=='registry' else 'tenant']
            for row in sorted(sql,key=lambda r:r[0]!='table'):output.execute(row[3])
            for kind,table,_,_ in sql:
                if kind!='table':continue
                columns=[r[1] for r in output.execute('PRAGMA table_info('+table+')')]
                fields=','.join(columns);rows=source.execute('SELECT '+fields+' FROM '+table).fetchall()
                output.executemany('INSERT INTO '+table+'('+fields+') VALUES('+','.join('?' for _ in columns)+')',rows)
    return legacy,j


def test_v06_existing_data_migrates_memberships_once_and_v3_restore(tmp_path):
    root,j=make_v06(tmp_path);before=checksum(root/'oak/business.sqlite3')
    with sqlite3.connect(root/'registry.sqlite3') as db:dataset=db.execute('SELECT dataset_id FROM dataset_identity').fetchone()[0]
    from backend import recovery
    from backend.store import DEMO_BUSINESSES
    meta={'format':recovery.FORMAT,'version':3,'prototype_version':'0.6.0','created_at':'2026-10-27T12:00:00+00:00','dataset_id':dataset,'auth_policy':recovery.LEGACY_AUTH_POLICY,'businesses':[{'id':b['id'],'slug':b['slug'],'name':b['name'],'enabled':1} for b in DEMO_BUSINESSES],'domains':[],'files':{}}
    archive=tmp_path/'v3.zip'
    with zipfile.ZipFile(archive,'w') as z:
        for b in DEMO_BUSINESSES:
            path=root/b['id']/'business.sqlite3';name='businesses/'+b['id']+'/business.sqlite3'
            meta['files'][name]={'bytes':path.stat().st_size,'sha256':checksum(path),'business_id':b['id'],'kind':'database'};z.write(path,name)
        z.writestr('manifest.json',json.dumps(meta))
    sha=checksum(archive);assert verify(archive,sha)['valid'];restore(archive,sha,tmp_path/'restored-v3',confirm_fresh=True)
    assert checksum(root/'oak/business.sqlite3')==before
    app=create_app(root);c=client(app);assert c.get('/api/jobs/'+j['id']).status_code==200
    members=c.get('/api/memberships').json()['members'];assert len(members)==1 and members[0]['role']=='owner' and members[0]['resource_id'] is None
    Store(root);assert c.get('/api/memberships').json()['members']==members
    fresh=backup(root,tmp_path/'now-v4.zip');assert verify(fresh['archive'],fresh['sha256'])['valid']


def test_mock_capture_cli_prints_only_private_path(app):
    import subprocess,sys
    c=client(app);add_staff(app,c);reset_request(app);code=reset_code(app)
    result=subprocess.run([sys.executable,'account_recovery.py','--data-dir',str(app.state.store.root),'--email',EMAIL],capture_output=True,text=True)
    assert result.returncode==0 and code not in result.stdout+result.stderr
    assert Path(result.stdout.strip()).is_file()


def test_disabled_owner_membership_can_be_disabled_without_removing_active_owner(app):
    c=client(app);add_staff(app,c,role='owner')
    with app.state.store.registry() as db:db.execute('UPDATE users SET disabled=1 WHERE email=?',(EMAIL,))
    assert update(c,member(c),enabled=False).status_code==200
    assert update(c,member(c,'owner@oak.example'),enabled=False).status_code==409


def test_provision_identity_collision_does_not_delete_existing_business(app,monkeypatch):
    from backend import identity
    c=client(app);created=new_business(c).json();folder=app.state.store.root/created['business_id'];before=checksum(folder/'business.sqlite3')
    monkeypatch.setattr(identity,'uid',lambda:created['business_id'])
    assert new_business(c,'different-new-name').status_code==409
    assert checksum(folder/'business.sqlite3')==before
    with app.state.store.registry() as db:assert db.execute('SELECT COUNT(*) FROM businesses').fetchone()[0]==3