"""PDF verification uses independent pypdf extraction, never a live browser."""
from concurrent.futures import ThreadPoolExecutor
import io
import json
import pytest
from pypdf import PdfReader
from fastapi.testclient import TestClient
from backend.private_pdf import render_document,fonts
from backend.sales_outputs import snapshot_hash
from test_boundaries import app,client,write,job,quote,invoice,portal,ORIGIN
from test_documents import doc,act,schedule,worker
from test_websites import read,save,publish,photo
from test_portal_access import enabled,response,BASE


def text(r):
    assert r.status_code==200,r.text if r.status_code!=200 else ''
    assert r.headers['content-type']=='application/pdf' and r.headers['cache-control']=='no-store'
    reader=PdfReader(io.BytesIO(r.content));assert not reader.is_encrypted
    return '\n'.join(page.extract_text() for page in reader.pages),reader


def test_pdf_private_cross_customer_business_role_and_lifecycle(app):
    c=client(app);other=client(app,1);j=job(c);q=quote(c,j);p,_,_=enabled(app,c,j);b=job(c);wrong,_,_=enabled(app,c,b)
    owner='/api/documents/quote/'+q['id']+'/pdf';path=BASE+'/documents/quote/'+q['id']+'/pdf'
    assert TestClient(app,headers=ORIGIN).get(owner).status_code==401
    assert other.get(owner).status_code==404
    assert p.get(path).status_code==404
    draft,_=text(c.get(owner));assert 'DRAFT' in draft
    act(c,'quote',q,'share');shared,_=text(p.get(path));assert 'DRAFT' not in shared and 'LOCAL DEMONSTRATION' in shared
    assert wrong.get(path).status_code==404 and p.get(owner).status_code==401
    assert p.get('/api/public/oak-and-vale/documents/quote/'+q['id']+'/pdf').status_code in (404,)
    write(c,'/api/customers/'+j['customer_id']+'/portal',method='delete');assert p.get(path).status_code==401


def test_fractional_pence_unicode_and_plain_text_are_exact(app):
    c=client(app);j=job(c,name='Catrin Dŵr — Llŷn',email='catrin@fiction.example')
    r=write(c,'/api/jobs/'+j['id']+'/quotes',{'items':[{'description':'Café, ŵ ŷ £ <b>literal markup</b> & shelves','quantity':'0.067','unit_pence':1750},{'description':'A second line','quantity':'2.125','unit_pence':333}],'vat_percent':20,'notes':'First line\nSecond line; no URL is fetched: https://invalid.example/image.png'});assert r.status_code==200
    q=r.json();d=doc(c,'quote',q);extracted,pdf=text(c.get('/api/documents/quote/'+q['id']+'/pdf'))
    assert d['snapshot']['line_totals_pence']==[117,708] and d['total_pence']==990
    for value in ('Catrin Dŵr','Llŷn','Café','<b>literal markup</b>','£1.17','£7.08','£9.90','Second line'):assert value in extracted,value
    assert all('/FontDescriptor' in font.get_object() or '/DescendantFonts' in font.get_object() for page in pdf.pages for name,font in page['/Resources']['/Font'].items() if name!='/F1')


def test_long_document_paginated_with_headers_footer_and_final_content(app):
    c=client(app);j=job(c,name='Catrin '+('Long surname '*6))
    r=write(c,'/api/jobs/'+j['id']+'/quotes',{'items':[{'description':f'Line {i+1:02}: '+('Careful shelving and room preparation. '*5),'quantity':'1.125','unit_pence':333} for i in range(40)],'vat_percent':5,'notes':('Long saved terms continue on their own pages. '*35)+'END OF SAVED TERMS'});assert r.status_code==200,r.text
    extracted,pdf=text(c.get('/api/documents/quote/'+r.json()['id']+'/pdf'))
    assert len(pdf.pages)>=4 and all('Page '+str(i+1) in page.extract_text() for i,page in enumerate(pdf.pages))
    assert all('Line '+str(i+1).zfill(2)+':' in extracted for i in range(40))
    assert 'END OF SAVED TERMS' in extracted and 'Current status summary' in extracted


def test_historical_brand_and_payment_wording_survive_rebranding_and_payment(app):
    c=client(app);logo=photo(c);site=read(c)['draft'];site.update(business_name='Original Demo Brand',logo_id=logo);assert save(c,site).status_code==200;publish(c)
    cfg=c.get('/api/document-settings').json();cfg.update(payment_instructions='Invented bank instruction ALPHA',quote_acknowledgement_text='Owner supplied demo wording.');assert write(c,'/api/document-settings',cfg,method='put').status_code==200
    j=job(c);q=quote(c,j);i=invoice(c,q);before=doc(c,'invoice',i)['snapshot'];p,_,_=enabled(app,c,j)
    site=read(c)['draft'];site.update(business_name='Changed Demo Brand',logo_id=None);save(c,site);publish(c)
    cfg=c.get('/api/document-settings').json();cfg['payment_instructions']='New wording BETA';write(c,'/api/document-settings',cfg,method='put')
    write(c,'/api/invoices/'+i['id']+'/payments',{'amount_pence':1234,'note':'PRIVATE RECEIPT NOTE'})
    extracted,pdf=text(p.get(BASE+'/documents/invoice/'+i['id']+'/pdf'))
    for needle in ('Original Demo Brand','ALPHA','£100.00','£12.34','£87.66','Current status summary'):assert needle in extracted
    for needle in ('Changed Demo Brand','BETA','PRIVATE RECEIPT NOTE'):assert needle not in extracted
    assert doc(c,'invoice',i)['snapshot']==before
    assert any(page['/Resources'].get('/XObject') for page in pdf.pages)


def test_stage_request_snapshot_and_current_balance_do_not_duplicate_invoice(app):
    c=client(app);j=job(c);q=quote(c,j);i=invoice(c,q);stages=schedule(c,i);p,_,_=enabled(app,c,j);stage=stages[0]
    path='/api/invoices/'+i['id']+'/schedule/'+stage['id'];portal_path=BASE+'/invoices/'+i['id']+'/schedule/'+stage['id']+'/pdf'
    assert c.get(path+'/pdf').status_code==409
    r=write(c,path+'/request',{'version':doc(c,'invoice',i)['version']});assert r.status_code==200,r.text
    with app.state.store.tenant('oak') as db:original=db.execute('SELECT snapshot FROM payment_request_snapshots WHERE stage_id=?',(stage['id'],)).fetchone()[0]
    write(c,'/api/invoices/'+i['id']+'/payments',{'amount_pence':1000,'stage_id':stage['id']})
    extracted,_=text(p.get(portal_path))
    for needle in ('Payment request','Not another invoice','£30.00','£20.00','£90.00','Current status summary'):assert needle in extracted,needle
    with app.state.store.tenant('oak') as db:
        assert db.execute('SELECT snapshot FROM payment_request_snapshots WHERE stage_id=?',(stage['id'],)).fetchone()[0]==original
        assert db.execute('SELECT COUNT(*) FROM invoices').fetchone()[0]==1
    unrelated=invoice(c,quote(c,job(c)));assert c.get('/api/invoices/'+unrelated['id']+'/schedule/'+stage['id']+'/pdf').status_code==404


def test_existing_legacy_request_requires_explicit_owner_freeze(app):
    c=client(app);j=job(c);i=invoice(c,quote(c,j));stage=schedule(c,i)[0];path='/api/invoices/'+i['id']+'/schedule/'+stage['id']
    with app.state.store.tenant('oak') as db:db.execute("UPDATE payment_stages SET requested_at='2026-10-27T12:00:00+00:00' WHERE id=?",(stage['id'],))
    assert c.get(path+'/pdf').status_code==409
    assert write(c,path+'/request',{'version':doc(c,'invoice',i)['version']}).status_code==200
    extracted,_=text(c.get(path+'/pdf'));assert 'owner-reviewed legacy request' in extracted and '2026-10-27' in extracted


def test_numbering_concurrent_settings_preserves_all_references(app):
    c=client(app);j=job(c);first=quote(c,j);cfg=c.get('/api/document-settings').json();cfg.update(quote_prefix='DEMO-Q',invoice_prefix='DEMO-I',include_year=True,padding=5)
    assert write(c,'/api/document-settings',cfg,method='put').status_code==200
    def create(_):return quote(worker(c,app),j)['reference']
    with ThreadPoolExecutor(max_workers=4) as pool:refs=list(pool.map(create,range(8)))
    assert len(set(refs))==8 and all(ref.startswith('DEMO-Q-') for ref in refs)
    assert sorted(int(ref.split('-')[-1]) for ref in refs)==list(range(2,10))
    assert doc(c,'quote',first)['reference']=='Q-0001'
    assert write(c,'/api/document-settings',cfg,method='put').status_code==409
    other=client(app,1);assert quote(other,job(other))['reference']=='Q-0001'


@pytest.mark.parametrize('patch',[{'quote_prefix':'http://bad'},{'quote_prefix':'I'},{'padding':2},{'include_year':'yes'},{'payment_instructions':'Bad\x00text'}])
def test_document_settings_reject_unsafe_or_ambiguous_values(app,patch):
    c=client(app);cfg=c.get('/api/document-settings').json()
    assert write(c,'/api/document-settings',{**cfg,**patch},method='put').status_code==422


def test_unsupported_pdf_glyph_is_explicit_without_silent_corruption(app):
    c=client(app);j=job(c);r=write(c,'/api/jobs/'+j['id']+'/quotes',{'items':[{'description':'Unrenderable \U0001f9f0','quantity':1,'unit_pence':100}]});assert r.status_code==200
    r=c.get('/api/documents/quote/'+r.json()['id']+'/pdf');assert r.status_code==422 and 'unsupported' in r.json()['detail']


def test_pdf_download_uses_no_arbitrary_image_url(app):
    c=client(app);j=job(c);q=quote(c,j)
    with app.state.store.tenant('oak') as db:
        d=doc(c,'quote',q)['snapshot'];d['business']['logo_id']='https://invalid.example/private.png';db.execute('UPDATE quotes SET snapshot=? WHERE id=?',(json.dumps(d),q['id']))
    assert c.get('/api/documents/quote/'+q['id']+'/pdf').status_code==409