"""Invented feedback; no live requests, email or original SHS runtime."""
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import copy,json,time,uuid
import pytest
from fastapi.testclient import TestClient
from backend.app import create_app
from backend import feedback
from backend.store import digest
from test_boundaries import app,client,write,job,portal,ORIGIN
from test_identity import add_staff
from test_documents import worker

BASE='/api/feedback/requests/'
PUBLIC='/api/public/oak-and-vale/testimonials'
ACCESS='/api/feedback/oak-and-vale/'
PORTAL='/api/portal/oak-and-vale/feedback'
PRIVATE='PRIVATE-SENTINEL-never-public'


def completed(c,j=None):
    j=j or job(c);old=c.get('/api/jobs/'+j['id']).json()
    r=write(c,'/api/jobs/'+j['id'],{'version':old['version'],'status':'completed','notes':PRIVATE,'scheduled_date':''},method='put');assert r.status_code==200,r.text
    return j

def prepare(c,j=None,**extra):
    j=j or completed(c);e=next(e for e in c.get('/api/feedback/eligibility').json() if e['id']==j['id'])
    body={k:e[k] for k in ('customer_id','customer_version','job_version')};body.update(job_id=j['id'],expires_days=14,**extra)
    result=write(c,'/api/feedback/requests',body);assert result.status_code==200,result.text
    rid=result.json()['id'];capture=c.get(BASE+rid+'/capture');assert capture.status_code==200,capture.text
    return j,rid,capture.json()['code']

def consent(allow=False,**extra):return {'allow_public':allow,'display_name':'Alex Public' if allow else '', 'statement':'Careful work <script>example</script> & clear updates.' if allow else '', 'share_rating':allow,'wording_version':1,**extra}

def payload(version=1,allow=False,**extra):
    return {'version':version,'answers':{'overall_rating':4,'quality_rating':3,'communication_rating':5,'timeliness_rating':4,'recommend':False,'comments':PRIVATE+' comments','improvement_suggestions':PRIVATE+' improvements'},'consent':consent(allow),'confirmed':True,**extra}

def respond(c,app,rid,code,allow=True,**extra):
    anon=TestClient(app,headers=ORIGIN);body={'code':code,**payload(c.get(BASE+rid).json()['request']['version'],allow,**extra)};r=write(anon,ACCESS+'submit',body);assert r.status_code==200,r.text;return anon,r

def publish(c,rid,**extra):
    p=c.get(BASE+rid+'/preview').json();r=write(c,BASE+rid+'/publish',{'version':p['version'],'projection_hash':p['projection_hash'],'confirmed':True,**extra});assert r.status_code==200,r.text;return r

def action(c,rid,verb,**extra):
    v=c.get(BASE+rid).json()['request']['version'];return write(c,BASE+rid+'/'+verb,{'version':v,'reason':'Explicit invented review',**extra})

def test_full_private_submission_exact_projection_and_honest_counts(app):
    c=client(app);j,rid,code=prepare(c);anon=TestClient(app,headers=ORIGIN)
    context=write(anon,ACCESS+'access',{'code':code}).json()
    for secret in (j['id'],j['customer_id'],'alex@customer.example','1 Example Street','Alex Demo','Fit shelves'):assert secret not in json.dumps(context)
    assert context['can_submit'] and not context['response'];respond(c,app,rid,code)
    assert anon.get(PUBLIC).json()['count']==0
    p=c.get(BASE+rid+'/preview').json();assert set(p['projection'])=={'display_name','statement','rating'};assert PRIVATE not in json.dumps(p)
    publish(c,rid);feed=anon.get(PUBLIC);assert feed.json()['count']==1 and feed.headers['cache-control']=='no-store'
    for secret in (PRIVATE,j['id'],j['customer_id'],code,'alex@customer.example','1 Example Street','recommend','improvement'):assert secret not in feed.text
    assert 'average' not in feed.json() and feed.json()['testimonials'][0]['rating']==4
    stats=c.get('/api/feedback').json()['stats'];assert stats['responses']==1 and stats['recommend_count']==0 and stats['average_overall_all_responses']==4
    assert c.get(BASE+rid).json()['response']['answers']['comments'].startswith(PRIVATE)


def test_default_private_and_no_owner_rewrite(app):
    c=client(app);_,rid,code=prepare(c);respond(c,app,rid,code,allow=False)
    p=c.get(BASE+rid+'/preview').json();assert not p['can_publish'] and p['projection'] is None
    assert write(c,BASE+rid+'/publish',{'version':p['version'],'projection_hash':'a'*64,'confirmed':True}).status_code==409
    for method in ('put','patch','delete'):assert write(c,BASE+rid,{'statement':'Invented by owner'},method=method).status_code==405
    assert c.get(PUBLIC).json()['count']==0


@pytest.mark.parametrize('change',[{'overall_rating':0},{'quality_rating':6},{'communication_rating':True},{'timeliness_rating':'4'},{'recommend':1},{'comments':'x'*4001},{'improvement_suggestions':'bad\x00text'}])
def test_invalid_answers_are_bounded_without_echo(app,change):
    c=client(app);_,rid,code=prepare(c);body=payload();body['answers'].update(change);r=write(TestClient(app,headers=ORIGIN),ACCESS+'submit',{'code':code,**body});assert r.status_code==422;assert PRIVATE not in r.text and code not in r.text;assert c.get(BASE+rid).json()['response'] is None


@pytest.mark.parametrize('choice',[consent(False,display_name='Leak'),consent(True,statement=''),consent(True,share_rating=1),consent(True,wording_version=2),consent(True,wording_version=True),consent(True,display_name='x'*101)])
def test_invalid_permission_cannot_publish(app,choice):
    c=client(app);_,rid,code=prepare(c);r=write(TestClient(app,headers=ORIGIN),ACCESS+'submit',{'code':code,**payload(consent=choice)});assert r.status_code==422


def test_feedback_credential_cannot_sign_in_or_read_financial_documents(app):
    c=client(app);j,rid,code=prepare(c);a=TestClient(app,headers=ORIGIN)
    assert a.post('/api/portal/oak-and-vale/login',json={'code':code}).status_code==401
    assert a.post('/api/portal/oak-and-vale/access/complete',json={'code':code,'password':'Invented-Only-2026!'}).status_code==401
    assert a.get('/api/portal/oak-and-vale/me').status_code==401
    assert a.get('/api/documents/quote/'+j['id']).status_code==401
    assert write(a,'/api/feedback/harbour-homeworks/access',{'code':code}).status_code==401
    assert write(a,ACCESS+'access',{'code':j['id']}).status_code==401
    assert a.get(ACCESS+'access?code='+code).status_code==405
    assert a.get('/.feedback-capture/test.json').status_code==404


def test_credentials_hashed_only_and_captures_private(app):
    c=client(app);_,rid,code=prepare(c)
    with app.state.store.tenant('oak') as db:
        assert db.execute('SELECT token_hash FROM feedback_tokens').fetchone()[0]==digest(code)
        dump='\n'.join(db.iterdump());assert code not in dump
    assert not any('feedback' in p.name for p in app.state.store.root.glob('*.json'))
    path=next((app.state.store.root/'.feedback-capture').glob('*.json'));assert path.stat().st_mode&0o777==0o600;assert path.parent.stat().st_mode&0o777==0o700
    assert c.get(BASE+rid).json()['delivery'].startswith('Captured locally')


def test_repeat_preparation_reissue_expiry_and_revocation(app,monkeypatch):
    c=client(app);j,rid,code=prepare(c);e=next(x for x in c.get('/api/feedback/eligibility').json() if x['id']==j['id']);assert not e['eligible']
    with pytest.raises(AssertionError):prepare(c,j,reason='Second request deliberately')
    anon=TestClient(app,headers=ORIGIN);clock=time.time();monkeypatch.setattr(feedback.time,'time',lambda:clock+15*86400)
    assert write(anon,ACCESS+'access',{'code':code}).status_code==401
    # Login again because the simulated wall clock also expires account sessions.
    c=client(app);assert action(c,rid,'reissue').status_code==200
    new=c.get(BASE+rid+'/capture').json()['code'];assert new!=code
    assert write(anon,ACCESS+'submit',{'code':code,**payload(version=2)}).status_code==401
    assert write(anon,ACCESS+'access',{'code':new}).status_code==200
    assert action(c,rid,'revoke').status_code==200
    assert write(anon,ACCESS+'access',{'code':new}).status_code==401
    with pytest.raises(AssertionError):prepare(c,j)
    _,second,_=prepare(c,j,reason='Another reviewed attempt');assert second!=rid
    assert action(c,rid,'reissue').status_code==409


def test_duplicate_identity_portal_exact_customer_and_histories(app):
    c=client(app);j1,r1,code1=prepare(c);j2,r2,code2=prepare(c);p1,_=portal(app,c,j1);p2,_=portal(app,c,j2)
    assert [r['id'] for r in p1.get(PORTAL).json()]==[r1];assert [r['id'] for r in p2.get(PORTAL).json()]==[r2]
    assert p2.get(PORTAL+'/'+r1).status_code==404
    assert write(p2,PORTAL+'/'+r1+'/submit',payload()).status_code==404
    assert write(p1,PORTAL+'/'+r1+'/submit',payload(allow=True)).status_code==200
    assert write(p2,PORTAL+'/'+r1+'/consent',{'version':2,'consent':consent(False),'confirmed':True}).status_code==404
    assert write(p1,PORTAL+'/'+r1+'/consent',{'version':2,'consent':consent(False),'confirmed':True}).status_code==200
    assert PRIVATE not in json.dumps(p2.get(PORTAL).json());assert len(p1.get(PORTAL+'/'+r1).json()['history'])==3
    assert j1['customer_id']!=j2['customer_id'];assert code1!=code2


@pytest.mark.parametrize('who',['anonymous','staff','customer','foreign_owner'])
def test_all_new_owner_routes_valid_shaped_boundaries(app,who):
    c=client(app);j,rid,code=prepare(c);respond(c,app,rid,code);p=c.get(BASE+rid+'/preview').json()
    actors={'anonymous':lambda:TestClient(app,headers=ORIGIN),'staff':lambda:add_staff(app,c)[0],'customer':lambda:portal(app,c,j)[0],'foreign_owner':lambda:client(app,1)};a=actors[who]();expected=404 if who=='foreign_owner' else 403 if who=='staff' else 401
    for path in (BASE+rid,BASE+rid+'/capture',BASE+rid+'/preview'):assert a.get(path).status_code==expected,path
    for verb in ('reissue','revoke','archive','hide','publish'):
        body={'version':p['version'],'projection_hash':p['projection_hash'],'confirmed':True} if verb=='publish' else {'version':p['version'],'reason':'Invented boundary check'}
        assert write(a,BASE+rid+'/'+verb,body).status_code==expected,verb
    data={'job_id':j['id'],'customer_id':j['customer_id'],'job_version':2,'customer_version':1,'expires_days':14}
    assert write(a,'/api/feedback/requests',data).status_code==expected
    for path in ('/api/feedback','/api/feedback/eligibility'):
        r=a.get(path);assert r.status_code==(200 if who=='foreign_owner' else expected)
        assert rid not in r.text and PRIVATE not in r.text


@pytest.mark.parametrize('who',['anonymous','staff','owner','foreign_customer'])
def test_portal_routes_need_exact_portal_access(app,who):
    c=client(app);j,rid,_=prepare(c)
    a=TestClient(app,headers=ORIGIN) if who=='anonymous' else add_staff(app,c)[0] if who=='staff' else c if who=='owner' else portal(app,client(app,1),completed(client(app,1)),slug='harbour-homeworks')[0]
    for path in (PORTAL,PORTAL+'/'+rid):assert a.get(path).status_code==401
    for verb,body in (('submit',payload()),('consent',{'version':1,'consent':consent(False),'confirmed':True})):assert write(a,PORTAL+'/'+rid+'/'+verb,body).status_code==401


def test_retry_stale_and_concurrent_submission_have_one_original(app):
    c=client(app);_,rid,code=prepare(c);a=TestClient(app,headers=ORIGIN);key=str(uuid.uuid4());body={'code':code,**payload(allow=True)}
    with ThreadPoolExecutor(max_workers=2) as pool:rs=list(pool.map(lambda _:write(TestClient(app,headers=ORIGIN),ACCESS+'submit',body,key=key),range(2)))
    assert [r.status_code for r in rs]==[200,200];assert rs[0].json()==rs[1].json()
    assert write(a,ACCESS+'submit',body).status_code==409
    assert write(a,ACCESS+'submit',{**body,'confirmed':1}).status_code==422
    altered=copy.deepcopy(body);altered['answers']['comments']='Different'
    assert write(a,ACCESS+'submit',altered,key=key).status_code==409
    assert len(c.get(BASE+rid).json()['history'])==2
    assert action(c,rid,'revoke').status_code==200
    assert write(a,ACCESS+'submit',body,key=key).status_code==401


def test_different_concurrent_submission_and_consent_single_winner(app):
    c=client(app);_,rid,code=prepare(c)
    bodies=[{'code':code,**payload(allow=True)} for _ in range(2)];bodies[1]['answers']['comments']='second private answer'
    with ThreadPoolExecutor(max_workers=2) as pool:rs=list(pool.map(lambda body:write(TestClient(app,headers=ORIGIN),ACCESS+'submit',body),bodies))
    assert sorted(r.status_code for r in rs)==[200,409]
    changes=[{'code':code,'version':2,'consent':consent(True,statement='Newly approved statement'),'confirmed':True},{'code':code,'version':2,'consent':consent(False),'confirmed':True}]
    with ThreadPoolExecutor(max_workers=2) as pool:rs=list(pool.map(lambda body:write(TestClient(app,headers=ORIGIN),ACCESS+'consent',body),changes))
    assert sorted(r.status_code for r in rs)==[200,409];assert len(c.get(BASE+rid).json()['history'])==3


def test_withdrawal_stale_publication_website_restore_and_new_consent(app):
    c=client(app);j,rid,code=prepare(c);a,_=respond(c,app,rid,code);original=c.get(BASE+rid).json()['response'];preview=c.get(BASE+rid+'/preview').json();publish(c,rid)
    pub=c.get(BASE+rid).json();v=pub['request']['version']
    assert write(a,ACCESS+'consent',{'code':code,'version':v,'consent':consent(False),'confirmed':True}).status_code==200
    assert c.get(PUBLIC).json()['count']==0
    assert write(c,BASE+rid+'/publish',{'version':preview['version'],'projection_hash':preview['projection_hash'],'confirmed':True}).status_code==409
    site=c.get('/api/site').json();assert write(c,'/api/site/restore',{'version':site['version'],'revision_id':site['history'][0]['id']}).status_code==200
    assert write(c,'/api/site/publish',{'version':c.get('/api/site').json()['version']}).status_code==200
    assert c.get(PUBLIC).json()['count']==0
    v=c.get(BASE+rid).json()['request']['version'];assert write(a,ACCESS+'consent',{'code':code,'version':v,'consent':consent(True,statement='New exact customer wording',share_rating=False),'confirmed':True}).status_code==200
    assert c.get(PUBLIC).json()['count']==0;publish(c,rid);feed=c.get(PUBLIC).json();assert feed['testimonials'][0]['rating'] is None and feed['testimonials'][0]['statement']=='New exact customer wording'
    assert c.get(BASE+rid).json()['response']==original
    assert action(c,rid,'hide').status_code==200;assert c.get(PUBLIC).json()['count']==0
    assert c.get('/api/feedback?state=published').json()['requests']==[]


@pytest.mark.parametrize('change',['job','customer','archive'])
def test_record_changes_preserve_evidence_hide_and_kill_codes(app,change):
    c=client(app);j,rid,code=prepare(c);respond(c,app,rid,code);publish(c,rid);old=c.get(BASE+rid).json()['response']
    if change=='job':
        old_job=c.get('/api/jobs/'+j['id']).json();assert write(c,'/api/jobs/'+j['id'],{'version':old_job['version'],'status':'in_progress','notes':old_job['notes'],'scheduled_date':''},method='put').status_code==200
    elif change=='customer':
        row=next(r for r in c.get('/api/customers').json() if r['id']==j['customer_id']);body={k:row[k] for k in ('version','name','email','phone','address')};body['phone']='123';assert write(c,'/api/customers/'+j['customer_id'],body,method='put').status_code==200
    else:assert action(c,rid,'archive').status_code==200
    assert c.get(PUBLIC).json()['count']==0;new=c.get(BASE+rid).json();assert new['response']==old and new['consent']['state']=='suspended'
    assert write(TestClient(app,headers=ORIGIN),ACCESS+'access',{'code':code}).status_code==401
    assert action(c,rid,'reissue').status_code==409
    assert len(new['history'])==4


def test_static_public_page_escapes_content_without_private_metadata(tmp_path):
    root=Path(__file__).parents[1];application=create_app(tmp_path/'data',root/'frontend/dist');c=client(application);_,rid,code=prepare(c);respond(c,application,rid,code);publish(c,rid)
    a=TestClient(application);html=a.get('/site/oak-and-vale/reviews/selected');assert html.status_code==200
    assert '&lt;script&gt;example&lt;/script&gt;' in html.text and '<script>example' not in html.text and PRIVATE not in html.text and rid not in html.text and code not in html.text
    assert 'Selected customer testimonials' in html.text and 'aggregateRating' not in html.text
    assert html.headers['cache-control']=='no-store';action(c,rid,'archive');assert 'No testimonials have been published.' in a.get('/site/oak-and-vale/reviews/selected').text


def test_portal_withdrawal_available_after_code_expiry_and_archive(app):
    c=client(app);j,rid,code=prepare(c);respond(c,app,rid,code);publish(c,rid);p,_=portal(app,c,j);action(c,rid,'archive')
    v=p.get(PORTAL+'/'+rid).json()['version'];assert write(p,PORTAL+'/'+rid+'/consent',{'version':v,'consent':consent(False),'confirmed':True}).status_code==200
    assert p.get(PORTAL+'/'+rid).json()['consent']['state']=='withdrawn';assert c.get(PUBLIC).json()['count']==0


def test_concurrent_publication_and_withdrawal_never_leave_withdrawn_text_public(app):
    c=client(app);_,rid,code=prepare(c);respond(c,app,rid,code);preview=c.get(BASE+rid+'/preview').json();second=worker(c,app)
    with ThreadPoolExecutor(max_workers=2) as pool:
        publish_future=pool.submit(write,second,BASE+rid+'/publish',{'version':preview['version'],'projection_hash':preview['projection_hash'],'confirmed':True})
        withdraw_future=pool.submit(write,TestClient(app,headers=ORIGIN),ACCESS+'consent',{'code':code,'version':preview['version'],'consent':consent(False),'confirmed':True})
        results=[publish_future.result(),withdraw_future.result()]
    assert sorted(r.status_code for r in results)==[200,409]
    # A stale withdrawal must be shown for review, not silently discarded. A fresh explicit retry wins.
    v=c.get(BASE+rid).json()['request']['version'];assert write(TestClient(app,headers=ORIGIN),ACCESS+'consent',{'code':code,'version':v,'consent':consent(False),'confirmed':True}).status_code==200
    assert c.get(PUBLIC).json()['count']==0


def test_capture_failure_rolls_back_request_and_no_delivery_claim(app,monkeypatch):
    c=client(app);j=completed(c)
    def failed(*args):raise OSError('Invented local disk failure')
    monkeypatch.setattr(feedback,'capture_path',failed)
    with pytest.raises(OSError):prepare(c,j)
    with app.state.store.tenant('oak') as db:
        for table in ('feedback_requests','feedback_tokens','feedback_history'):assert db.execute('SELECT COUNT(*) FROM '+table).fetchone()[0]==0


def test_exact_job_customer_match_and_real_email_not_capturable(app):
    c=client(app);j1=completed(c);j2=completed(c);foreign=completed(client(app,1))
    base={'job_id':j1['id'],'customer_id':j1['customer_id'],'job_version':2,'customer_version':1,'expires_days':14}
    for data in ({**base,'customer_id':j2['customer_id']},{**base,'customer_id':foreign['customer_id']},{**base,'job_id':foreign['id']}):assert write(c,'/api/feedback/requests',data).status_code==404
    assert write(c,'/api/feedback/requests',{**base,'job_version':1}).status_code==409
    outside=completed(c,job(c,email='invented@invalid.test'));base.update(job_id=outside['id'],customer_id=outside['customer_id'])
    assert write(c,'/api/feedback/requests',base).status_code==422


def test_noop_contact_edit_preserves_permission_and_job_relink_is_blocked(app):
    c=client(app);j,rid,code=prepare(c);respond(c,app,rid,code);publish(c,rid)
    row=next(r for r in c.get('/api/customers').json() if r['id']==j['customer_id']);body={k:row[k] for k in ('version','name','email','phone','address')};assert write(c,'/api/customers/'+j['customer_id'],body,method='put').status_code==200
    assert c.get(PUBLIC).json()['count']==1
    old=c.get('/api/jobs/'+j['id']).json();write(c,'/api/jobs/'+j['id'],{'version':old['version'],'status':'new','notes':'','scheduled_date':''},method='put');other=job(c)
    assert write(c,'/api/jobs/'+j['id']+'/customer-link-preview',{'customer_id':other['customer_id'],'customer_version':1,'job_version':3}).status_code==410


def test_recommendation_and_average_use_all_saved_responses_not_curated_subset(app):
    c=client(app);_,a,ca=prepare(c);respond(c,app,a,ca,answers={**payload()['answers'],'overall_rating':5,'recommend':False});publish(c,a)
    _,b,cb=prepare(c);respond(c,app,b,cb,allow=False,answers={**payload()['answers'],'overall_rating':1,'recommend':True});action(c,b,'archive')
    stats=c.get('/api/feedback?state=published').json()['stats'];assert stats['responses']==2 and stats['average_overall_all_responses']==3 and stats['recommend_count']==1
    feed=c.get(PUBLIC).json();assert feed['count']==1 and set(feed)=={'label','selection_notice','count','testimonials'}


def test_review_page_does_not_shadow_editable_customer_page(tmp_path):
    root=Path(__file__).parents[1];application=create_app(tmp_path/'data',root/'frontend/dist');c=client(application)
    site=c.get('/api/site').json();content=site['draft'];page=copy.deepcopy(content['pages'][1]);page.update(id='reviews-page',slug='reviews',title='Our editable review information',nav_label='Review information',visible=True);content['pages'].append(page)
    content['pages'][0]['sections'][0].update(link='/reviews/selected',link_label='Selected customer testimonials')
    assert write(c,'/api/site',{'version':site['version'],'content':content},method='put').status_code==200
    assert write(c,'/api/site/publish',{'version':c.get('/api/site').json()['version']}).status_code==200
    a=TestClient(application);assert 'Our editable review information' in a.get('/site/oak-and-vale/reviews').text
    assert 'Selected customer testimonials' in a.get('/site/oak-and-vale/reviews/selected').text
    assert 'No testimonials have been published.' in a.get('/site/oak-and-vale/reviews/selected').text