"""Typed device notes only; disposable TestClient accounts, never browser sessions."""
from concurrent.futures import ThreadPoolExecutor
import copy, json, uuid
import pytest
from fastapi.testclient import TestClient
from backend.app import create_app
from backend import recovery
from test_boundaries import app, client, write, job, ORIGIN
from test_fieldwork import setup, base
from test_identity import add_staff, member, update
from test_scheduling import booking


def access(c,j,device=None):
    r=write(c,base(j)+'/note-access',{'device_id':device or str(uuid.uuid4())})
    assert r.status_code==200,r.text
    assert 'no-store' in r.headers['cache-control']
    return r.json()


def note(c,j):
    a=access(c,j)
    return str(uuid.uuid4()), {'binding':a['binding'],'revision':1,'content':'PRIVATE durable text only','work_date':'2026-10-31'}, a


def action(c,j,op,body,verb='deliver'):
    return write(c,base(j)+'/device-notes/'+op+'/'+verb,body)


def test_lost_before_send_and_after_commit_reload_same_identity(app):
    c=client(app);j,_,_=setup(c,app);op,p,a=note(c,j)
    assert action(c,j,op,{'binding':p['binding']},'inspect').json()=={'state':'never-sent','revision':1}
    first=action(c,j,op,p);assert first.status_code==200,first.text
    reopened=create_app(app.state.store.root);c=client(reopened)
    assert access(c,j,a['binding']['device_id'])['key']==a['key']
    assert action(c,j,op,p).json()==first.json()
    assert action(c,j,op,{'binding':p['binding']},'inspect').json()==first.json()
    with reopened.state.store.tenant('oak') as db:
        assert db.execute('SELECT COUNT(*) FROM field_logs').fetchone()[0]==1
        assert db.execute('SELECT COUNT(*) FROM field_log_history').fetchone()[0]==1


def test_concurrent_duplicate_and_distinct_notes(app):
    c=client(app);other=client(app);j,_,_=setup(c,app);op,p,_=note(c,j)
    with ThreadPoolExecutor(4) as pool:results=list(pool.map(lambda n:action(c if n%2 else other,j,op,p),range(4)))
    assert all(r.status_code==200 for r in results)
    assert len({r.json()['id'] for r in results})==1
    with ThreadPoolExecutor(2) as pool:results=list(pool.map(lambda n:action(c if n else other,j,str(uuid.uuid4()),p),range(2)))
    assert all(r.status_code==200 for r in results)
    assert len(c.get(base(j)).json()['logs'])==3


def test_correction_fences_late_old_send_and_preserves_id(app):
    c=client(app);j,_,_=setup(c,app);op,p,_=note(c,j)
    with app.state.store.tenant('oak') as db:db.execute('UPDATE jobs SET version=version+1 WHERE id=?',(j['id'],))
    assert action(c,j,op,p).status_code==409
    correction={'binding':p['binding'],'revision':2}
    first=action(c,j,op,correction,'revise');assert first.status_code==200
    assert action(c,j,op,correction,'revise').json()==first.json()
    assert action(c,j,op,p).status_code==409
    fresh=access(c,j,p['binding']['device_id']);p.update(binding=fresh['binding'],revision=2,content='Corrected retained note')
    r=action(c,j,op,p);assert r.status_code==200,r.text
    assert action(c,j,op,p).json()==r.json()
    assert len(c.get(base(j)).json()['logs'])==1


def test_correction_after_lost_committed_reply_returns_original_receipt(app):
    c=client(app);j,_,_=setup(c,app);op,p,_=note(c,j);saved=action(c,j,op,p).json()
    assert action(c,j,op,{'binding':p['binding'],'revision':2},'revise').json()==saved
    p['content']='Different text';assert action(c,j,op,p).status_code==409
    assert len(c.get(base(j)).json()['logs'])==1


@pytest.mark.parametrize('change',['scope','device','job','business','actor','disabled','expired','archived','transferred','removed_assignment','archived_resource','deleted'])
def test_current_authority_even_after_confirmation(app,change):
    owner=client(app);j,people,_=setup(owner,app);booking(owner,j,people[0]);c,_=add_staff(app,owner,resource_id=people[0]['id'])
    op,p,_=note(c,j);assert action(c,j,op,p).status_code==200
    if change=='scope':p['binding']['scope']='f'*64
    elif change=='device':p['binding']['device_id']=str(uuid.uuid4())
    elif change=='job':j,_,_=setup(owner,app)
    elif change=='business':c=client(app,1)
    elif change=='actor':c=owner
    elif change=='disabled':
        with app.state.store.registry() as db:db.execute("UPDATE memberships SET enabled=0 WHERE role='staff'")
    elif change=='expired':
        with app.state.store.registry() as db:db.execute("UPDATE sessions SET expires=0 WHERE subject<>'oak'")
    elif change=='archived':
        with app.state.store.tenant('oak') as db:db.execute("UPDATE jobs SET status='completed' WHERE id=?",(j['id'],))
    elif change=='transferred':
        other=job(owner,name='Different exact customer',email='other@example.test')
        with app.state.store.tenant('oak') as db:db.execute('UPDATE jobs SET customer_id=(SELECT customer_id FROM jobs WHERE id=?) WHERE id=?',(other['id'],j['id']))
    elif change=='removed_assignment':
        with app.state.store.tenant('oak') as db:db.execute('UPDATE calendar_bookings SET archived=1 WHERE job_id=?',(j['id'],))
    elif change=='archived_resource':
        with app.state.store.tenant('oak') as db:db.execute('UPDATE team_resources SET archived=1 WHERE id=?',(people[0]['id'],))
    elif change=='deleted':j={'id':'a'*32}
    for verb,body in [('deliver',p),('inspect',{'binding':p['binding']}),('revise',{'binding':p['binding'],'revision':2})]:
        assert action(c,j,op,body,verb).status_code in (401,403,404,409),change


@pytest.mark.parametrize('extra',[{'endpoint':'/api/payments'},{'photo_ids':['a'*32]},{'kind':'materials'},{'timer':'resume'},{'revision':True},{'work_date':'invalid'},{'content':''}])
def test_rejects_untyped_work_without_creating_a_log(app,extra):
    c=client(app);j,_,_=setup(c,app);op,p,_=note(c,j);p.update(extra)
    r=action(c,j,op,p);assert r.status_code==422,r.text
    assert not c.get(base(j)).json()['logs']


def test_scope_separation_and_password_rotation(app):
    c=client(app);j,_,_=setup(c,app);op,p,a=note(c,j);other=access(c,j)
    assert other['binding']['scope']!=a['binding']['scope'] and other['key']!=a['key']
    with app.state.store.registry() as db:db.execute("UPDATE users SET password=password||'rotated' WHERE id='oak'")
    # Current session still exists in this injected fixture; the key boundary alone changes.
    rotated=access(c,j,p['binding']['device_id']);assert rotated['key']!=a['key']
    assert action(c,j,op,p).status_code==403


def test_recovery_keeps_logs_history_and_receipts_but_cannot_unlock_old_device_notes(app,tmp_path):
    from test_recovery import new_login
    c=client(app);j,_,_=setup(c,app);op,p,a=note(c,j);result=action(c,j,op,p).json()
    archive=tmp_path/'device-notes.zip';b=recovery.backup(app.state.store.root,archive)
    assert recovery.verify(archive,b['sha256'])['valid']
    target=tmp_path/'restored';recovery.restore(archive,b['sha256'],target,confirm_fresh=True)
    recovered=create_app(target);after,_=new_login(recovered,target,'oak')
    assert access(after,j,p['binding']['device_id'])['key']!=a['key']
    assert action(after,j,op,p).status_code==403
    with recovered.state.store.tenant('oak') as db:
        assert db.execute('SELECT id FROM field_logs').fetchone()[0]==result['id']
        assert db.execute('SELECT COUNT(*) FROM field_log_history').fetchone()[0]==1
        assert db.execute('SELECT result FROM operations WHERE key=?',(op,)).fetchone()
    second=tmp_path/'again.zip';b=recovery.backup(target,second);assert recovery.verify(second,b['sha256'])['valid']


def test_unauthenticated_and_missing_csrf_cannot_unlock(app):
    c=client(app);j,_,_=setup(c,app);p={'device_id':str(uuid.uuid4())}
    assert TestClient(app,headers=ORIGIN).post(base(j)+'/note-access',json=p).status_code==401
    c.headers.pop('x-csrf-token');assert c.post(base(j)+'/note-access',json=p).status_code==403