"""Local invented identities; never sends setup or recovery messages."""
from concurrent.futures import ThreadPoolExecutor
import json
import stat
import time
import uuid
import pytest
from fastapi.testclient import TestClient
from backend.store import Store,digest
from test_boundaries import app,client,write,job,quote,portal,ORIGIN
from test_documents import doc,act,worker

BASE='/api/portal/oak-and-vale'
PASSWORD='Fictional-portal-2026!'

def invited(c,j):
    state=c.get('/api/customers/'+j['customer_id']+'/portal-access').json()
    r=write(c,'/api/customers/'+j['customer_id']+'/portal-access/invite',{'version':state['customer_version']})
    assert r.status_code==200,r.text
    value=r.json();r=c.get('/api/customers/'+j['customer_id']+'/portal-access/captures/'+value['capture_id']);assert r.status_code==200,r.text
    return value,r.json()


def enabled(app,c,j,password=PASSWORD):
    grant,capture=invited(c,j);p=TestClient(app,headers=ORIGIN);base='/api/portal/'+grant['slug']
    r=p.post(base+'/access/complete',json={'code':capture['code'],'password':password});assert r.status_code==200,r.text
    r=p.post(base+'/access/login',json={'portal_id':capture['portal_id'],'password':password});assert r.status_code==200,r.text
    p.headers['x-csrf-token']=r.json()['csrf'];return p,grant,capture


def response(c,q,**extra):
    d=doc(c,'quote',q)
    return {'version':d['version'],'snapshot_hash':d['snapshot_hash'],'confirmed':True,**extra}


def test_exact_customer_invitations_and_private_local_capture(app):
    c=client(app);a=job(c);b=job(c);ga,ca=invited(c,a);gb,cb=invited(c,b)
    assert ca['to']==cb['to'] and ca['portal_id']!=cb['portal_id'] and ca['code']!=cb['code']
    path='/api/customers/'+a['customer_id']+'/portal-access/captures/'+ga['capture_id']
    assert TestClient(app,headers=ORIGIN).get(path).status_code==401
    assert client(app,1).get(path).status_code==404
    assert c.get('/api/customers/'+b['customer_id']+'/portal-access/captures/'+ga['capture_id']).status_code==404
    file=app.state.store.root/'.portal-capture'/(ga['capture_id']+'.json')
    assert stat.S_IMODE(file.stat().st_mode)==0o600
    assert stat.S_IMODE(file.parent.stat().st_mode)==0o700
    with app.state.store.tenant('oak') as db:
        assert all(ca['code'] not in row[0] for row in db.execute('SELECT result FROM operations'))
        assert ca['code'] not in str(db.execute('SELECT * FROM portal_events').fetchall())
        assert db.execute('SELECT token_hash FROM portal_tokens WHERE id=?',(ga['capture_id'],)).fetchone()[0]==digest(ca['code'])


def test_non_example_and_stale_invites_are_rejected(app):
    c=client(app);j=job(c,email='fiction@invalid.test');path='/api/customers/'+j['customer_id']+'/portal-access/invite'
    assert write(c,path,{'version':99}).status_code==409
    assert write(c,path,{'version':1}).status_code==422
    assert c.get('/api/customers/'+j['customer_id']+'/portal-access').json()['account'] is None


def test_setup_single_use_concurrent_atomic_and_exact_customer(app):
    c=client(app);a=job(c);b=job(c);grant,capture=invited(c,a);q=quote(c,b);act(c,'quote',q,'share')
    payload={'code':capture['code'],'password':PASSWORD}
    with ThreadPoolExecutor(max_workers=2) as pool:
        statuses=list(pool.map(lambda _:TestClient(app,headers=ORIGIN).post(BASE+'/access/complete',json=payload).status_code,range(2)))
    assert sorted(statuses)==[200,401]
    p=TestClient(app,headers=ORIGIN);r=p.post(BASE+'/access/login',json={'portal_id':capture['portal_id'],'password':PASSWORD});assert r.status_code==200
    assert [r['id'] for r in p.get(BASE+'/me').json()['jobs']]==[a['id']]
    assert p.get(BASE+'/documents/quote/'+q['id']).status_code==404
    assert c.get('/api/customers/'+a['customer_id']+'/portal-access/captures/'+grant['capture_id']).status_code==404
    assert not (app.state.store.root/'.portal-capture'/(grant['capture_id']+'.json')).exists()


@pytest.mark.parametrize('cause',['expired','reissue','revoke','contact'])
def test_old_setup_code_cannot_survive_changes(app,cause):
    c=client(app);j=job(c);grant,capture=invited(c,j)
    if cause=='expired':
        with app.state.store.tenant('oak') as db:db.execute('UPDATE portal_tokens SET expires=? WHERE id=?',(time.time()-1,grant['capture_id']))
    elif cause=='reissue':invited(c,j)
    elif cause=='revoke':write(c,'/api/customers/'+j['customer_id']+'/portal',method='delete')
    else:
        r=write(c,'/api/customers/'+j['customer_id'],{'version':1,'name':'Corrected Demo','email':'new@fiction.example','phone':'','address':''},method='put');assert r.status_code==200
    p=TestClient(app,headers=ORIGIN)
    assert p.post(BASE+'/access/complete',json={'code':capture['code'],'password':PASSWORD}).status_code==401


def test_password_portal_not_email_join_and_no_owner_login(app):
    c=client(app);j=job(c);p,g,cap=enabled(app,c,j)
    anon=TestClient(app,headers=ORIGIN)
    for ident in ('Alex Demo','alex@customer.example','unknown'):
        assert anon.post(BASE+'/access/login',json={'portal_id':ident,'password':PASSWORD}).status_code==401
    assert anon.post('/api/login',json={'email':'alex@customer.example','password':PASSWORD}).status_code==401
    assert p.get('/api/workspace').status_code==401 and c.get(BASE+'/me').status_code==401
    assert p.get('/api/portal/harbour-homeworks/me').status_code==401
    assert write(c,'/api/customers/'+j['customer_id']+'/portal').status_code==409
    invited(c,j)
    assert p.get(BASE+'/me').status_code==401
    assert anon.post(BASE+'/access/login',json={'portal_id':cap['portal_id'],'password':PASSWORD}).status_code==401


def test_recovery_generic_response_rotates_password_sessions_and_codes(app):
    c=client(app);j=job(c);p,g,cap=enabled(app,c,j);anon=TestClient(app,headers=ORIGIN)
    valid={'portal_id':cap['portal_id'],'email':cap['to']}
    expected=anon.post(BASE+'/access/reset-request',json=valid)
    wrong=anon.post(BASE+'/access/reset-request',json={**valid,'email':'nobody@fiction.example'})
    unknown=anon.post('/api/portal/unknown/access/reset-request',json=valid)
    assert expected.status_code==wrong.status_code==unknown.status_code==200 and expected.json()==wrong.json()==unknown.json()
    assert p.get(BASE+'/me').status_code==200
    base='/api/customers/'+j['customer_id']+'/portal-access';state=c.get(base).json();assert len(state['captures'])==1
    reset=c.get(base+'/captures/'+state['captures'][0]['id']).json();assert reset['kind']=='reset'
    assert reset['expires']-time.time()<=1800
    r=anon.post(BASE+'/access/complete',json={'code':reset['code'],'password':PASSWORD+'new'});assert r.status_code==200
    assert p.get(BASE+'/me').status_code==401
    assert anon.post(BASE+'/access/login',json={'portal_id':cap['portal_id'],'password':PASSWORD}).status_code==401
    assert anon.post(BASE+'/access/login',json={'portal_id':cap['portal_id'],'password':PASSWORD+'new'}).status_code==200
    assert anon.post(BASE+'/access/complete',json={'code':reset['code'],'password':PASSWORD}).status_code==401


def test_legacy_code_now_single_use_and_preupgrade_grant_revoked(app):
    c=client(app);j=job(c);p,code=portal(app,c,j)
    assert TestClient(app,headers=ORIGIN).post(BASE+'/login',json={'code':code}).status_code==401
    with app.state.store.tenant('oak') as db:
        db.execute('DELETE FROM portal_migrations');db.execute('UPDATE customers SET portal_hash=? WHERE id=?',(digest('old-permanent-code'),j['customer_id']))
    Store(app.state.store.root)
    assert p.get(BASE+'/me').status_code==401
    assert TestClient(app,headers=ORIGIN).post(BASE+'/login',json={'code':'old-permanent-code'}).status_code==401


def test_invitation_retry_contains_no_code_and_one_capture(app):
    c=client(app);j=job(c);key=str(uuid.uuid4());path='/api/customers/'+j['customer_id']+'/portal-access/invite'
    a=write(c,path,{'version':1},key=key);b=write(c,path,{'version':1},key=key)
    assert a.status_code==b.status_code==200 and a.json()==b.json() and 'code' not in a.json()
    assert len(c.get('/api/customers/'+j['customer_id']+'/portal-access').json()['captures'])==1


@pytest.mark.parametrize('action',['accept','decline'])
def test_valid_foreign_customer_response_rejected_separately_from_bad_payload(app,action):
    c=client(app);a=job(c);b=job(c);q=quote(c,b);act(c,'quote',q,'share');p,_,_=enabled(app,c,a)
    before=doc(c,'quote',q);path=BASE+'/quotes/'+q['id']+'/'+action
    assert write(p,path,response(c,q)).status_code==404
    assert write(p,path,{'version':before['version']}).status_code==422
    after=doc(c,'quote',q);assert after==before


def test_quote_concurrent_opposite_responses_save_one_exact_snapshot(app):
    c=client(app);j=job(c);q=quote(c,j);act(c,'quote',q,'share');p,_,_=enabled(app,c,j)
    payload=response(c,q,typed_name='Catrin Dŵr',acknowledgement='Local sample acknowledgement only.')
    with ThreadPoolExecutor(max_workers=2) as pool:
        statuses=list(pool.map(lambda action:write(worker(p,app),BASE+'/quotes/'+q['id']+'/'+action,payload).status_code,['accept','decline']))
    assert sorted(statuses)==[200,409]
    after=doc(c,'quote',q);e=after['response']['evidence']
    assert e['snapshot']==after['snapshot'] and e['snapshot_hash']==payload['snapshot_hash'] and e['quote_version']==2 and e['typed_name']=='Catrin Dŵr'
    assert e['actor']=='portal:'+j['customer_id'] and e['confirmed'] is True
    assert len([h for h in after['history'] if h['action'] in ('accepted','declined')])==1


@pytest.mark.parametrize('patch,status',[({'confirmed':False},422),({'snapshot_hash':'0'*64},409),({'version':1},409)])
def test_quote_response_requires_reviewed_content_and_current_version(app,patch,status):
    c=client(app);j=job(c);q=quote(c,j);act(c,'quote',q,'share');p,_,_=enabled(app,c,j)
    assert write(p,BASE+'/quotes/'+q['id']+'/accept',response(c,q,**patch)).status_code==status
    assert doc(c,'quote',q)['response'] is None


def test_decline_then_revision_preserves_response_and_hides_private_revision(app):
    c=client(app);j=job(c);q=quote(c,j);act(c,'quote',q,'share');p,_,_=enabled(app,c,j)
    assert write(p,BASE+'/quotes/'+q['id']+'/decline',response(c,q)).status_code==200
    before=doc(c,'quote',q)['response'];revision=act(c,'quote',q,'revise').json()
    assert doc(c,'quote',q)['response']==before and doc(c,'quote',q)['status']=='superseded'
    assert p.get(BASE+'/documents/quote/'+revision['id']).status_code==404
    assert write(p,BASE+'/quotes/'+q['id']+'/accept',response(c,q)).status_code==409


def test_unknown_password_portal_login_is_rate_limited(app):
    p=TestClient(app,headers=ORIGIN)
    results=[p.post(BASE+'/access/login',json={'portal_id':'unknown','password':'wrong'}).status_code for _ in range(13)]
    assert results==[401]*12+[429]

def test_declined_source_quote_keeps_reservation_until_owner_revision_or_cancel(app):
    from test_fieldwork import captured,draft,suggested
    c=client(app);j,people,clock,tid=captured(c,app);q=draft(c,j,kind='quote').json();act(c,'quote',q,'share');p,_,_=enabled(app,c,j)
    assert write(p,BASE+'/quotes/'+q['id']+'/decline',response(c,q)).status_code==200
    assert all(s['claim']['document_id']==q['id'] for s in suggested(c,j)['sources'])
    assert draft(c,j).status_code==409
    revision=act(c,'quote',q,'revise').json();assert all(s['claim']['document_id']==revision['id'] for s in suggested(c,j)['sources'])
    assert act(c,'quote',revision,'cancel').status_code==200
    assert all(s['claim'] is None for s in suggested(c,j)['sources'])
    assert draft(c,j).status_code==200