"""Invented local records only. No original SHS runtime or external transport."""
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from datetime import datetime, timedelta
import hashlib
import json
import sqlite3
import shutil
import time
import uuid
import zipfile
import pytest
from fastapi.testclient import TestClient
from backend.app import create_app
from backend.store import Store, DEMO_BUSINESSES, initial_site, uid, password_hash, digest
from backend import recovery
from test_boundaries import app, client, write, job, quote, invoice, portal, ORIGIN


def doc(c,kind,obj):
    r=c.get('/api/documents/'+kind+'/'+obj['id']);assert r.status_code==200,r.text
    return r.json()


def act(c,kind,obj,action,reason='A documented test reason',key=None):
    return write(c,'/api/'+kind+'s/'+obj['id']+'/'+action,{'version':doc(c,kind,obj)['version'],'reason':reason},key=key)


def draft_invoice(c,total=10000):
    j=job(c);q=quote(c,j,total);assert act(c,'quote',q,'accept').status_code==200
    result=act(c,'quote',q,'invoice');assert result.status_code==200,result.text
    return j,q,result.json()


def edit_payload(document,total=None):
    s=document['snapshot']
    data={k:s[k] for k in ('items','vat_percent','notes','document_date','valid_until') if k in s}
    if 'due_date' in s:data['due_date']=s['due_date']
    if total is not None:data['items']=[{'description':'Revised work','quantity':1,'unit_pence':total}]
    return {**data,'version':document['version']}


def schedule(c,i,amounts=(3000,2000,5000)):
    r=write(c,'/api/invoices/'+i['id']+'/schedule',{'version':doc(c,'invoice',i)['version'],'stages':[{'label':'Deposit' if n==0 else 'Stage '+str(n+1),'amount_pence':a,'due_date':''} for n,a in enumerate(amounts)]})
    assert r.status_code==200,r.text
    return doc(c,'invoice',i)['schedule']


def test_draft_edit_revision_preserves_published_snapshot_and_history(app):
    c=client(app);j=job(c);q=quote(c,j)
    before=doc(c,'quote',q);payload=edit_payload(before,12345)
    assert write(c,'/api/quotes/'+q['id'],payload,method='put').status_code==200
    assert write(c,'/api/quotes/'+q['id'],payload,method='put').status_code==409
    assert act(c,'quote',q,'share').status_code==200
    sealed=doc(c,'quote',q);assert sealed['total_pence']==12345
    assert write(c,'/api/quotes/'+q['id'],edit_payload(sealed,9999),method='put').status_code==409
    p,_=portal(app,c,j)
    revised=act(c,'quote',q,'revise').json()
    assert revised['id']!=q['id'] and revised['reference']!=q['reference']
    old=doc(c,'quote',q);new=doc(c,'quote',revised)
    assert old['snapshot']==sealed['snapshot'] and old['status']=='superseded'
    assert new['revision']==2 and new['previous_id']==q['id'] and new['status']=='draft'
    assert p.get('/api/portal/oak-and-vale/documents/quote/'+revised['id']).status_code==404
    assert write(p,'/api/portal/oak-and-vale/quotes/'+q['id']+'/accept',{'version':sealed['version'],'snapshot_hash':sealed['snapshot_hash'],'confirmed':True}).status_code==409
    assert len(old['history'])==4 and new['history'][0]['action']=='revision.created'
    assert act(c,'quote',revised,'share').status_code==200
    current=doc(c,'quote',revised);key=str(uuid.uuid4())
    first=write(p,'/api/portal/oak-and-vale/quotes/'+revised['id']+'/accept',{'version':current['version'],'snapshot_hash':current['snapshot_hash'],'confirmed':True},key=key)
    assert first.status_code==200
    assert write(p,'/api/portal/oak-and-vale/quotes/'+revised['id']+'/accept',{'version':current['version'],'snapshot_hash':current['snapshot_hash'],'confirmed':True},key=key).json()==first.json()
    assert doc(c,'quote',revised)['accepted_at']


def test_expiry_and_cancelled_draft_do_not_create_portal_access(app):
    c=client(app);j=job(c);q=quote(c,j);p,_=portal(app,c,j)
    assert act(c,'quote',q,'cancel').status_code==200
    assert p.get('/api/portal/oak-and-vale/documents/quote/'+q['id']).status_code==404
    assert p.get('/api/portal/oak-and-vale/me').json()['quotes']==[]
    q=quote(c,j);payload=edit_payload(doc(c,'quote',q));payload['valid_until']='2000-01-01'
    assert write(c,'/api/quotes/'+q['id'],payload,method='put').status_code==200
    assert act(c,'quote',q,'share').status_code==409
    assert act(c,'quote',q,'accept').status_code==409


def test_invoice_private_draft_issue_sealing_and_unpaid_replacement(app):
    c=client(app);j,q,i=draft_invoice(c);p,_=portal(app,c,j)
    assert p.get('/api/portal/oak-and-vale/documents/invoice/'+i['id']).status_code==404
    assert c.get('/api/workspace').json()['stats']['outstanding_pence']==0
    assert write(c,'/api/invoices/'+i['id']+'/payments',{'amount_pence':100}).status_code==409
    payload=edit_payload(doc(c,'invoice',i),12000);payload['due_date']='2030-01-01'
    assert write(c,'/api/invoices/'+i['id'],payload,method='put').status_code==200
    assert act(c,'invoice',i,'issue').status_code==200
    sealed=doc(c,'invoice',i)
    assert p.get('/api/portal/oak-and-vale/documents/invoice/'+i['id']).status_code==200
    assert write(c,'/api/invoices/'+i['id'],edit_payload(sealed,999),method='put').status_code==409
    replacement=act(c,'invoice',i,'correct').json()
    assert doc(c,'invoice',i)['snapshot']==sealed['snapshot']
    assert doc(c,'invoice',i)['status']=='void' and replacement['reference']!=i['reference']
    assert doc(c,'invoice',replacement)['replaces_id']==i['id']
    assert p.get('/api/portal/oak-and-vale/documents/invoice/'+replacement['id']).status_code==404
    portal_old=p.get('/api/portal/oak-and-vale/documents/invoice/'+i['id']).json()
    assert portal_old['balance_pence']==0 and portal_old['replacement'] is None and 'history' not in portal_old
    assert act(c,'invoice',replacement,'issue').status_code==200
    assert c.get('/api/workspace').json()['stats']['outstanding_pence']==12000
    assert c.get('/api/jobs/'+j['id']+'/costs').json()['invoiced_value_pence']==12000
    assert c.get('/api/customers/'+j['customer_id']+'/history').json()['totals']['invoiced_pence']==12000
    assert doc(c,'quote',q)['snapshot']['total_pence']==10000


def test_cancel_draft_and_void_paid_invoice_boundaries(app):
    c=client(app);j,q,i=draft_invoice(c);p,_=portal(app,c,j)
    assert act(c,'invoice',i,'cancel','').status_code==422
    assert act(c,'invoice',i,'cancel').status_code==200
    assert act(c,'invoice',i,'issue').status_code==409
    assert p.get('/api/portal/oak-and-vale/documents/invoice/'+i['id']).status_code==404
    another=act(c,'quote',q,'invoice').json();act(c,'invoice',another,'issue')
    assert write(c,'/api/invoices/'+another['id']+'/payments',{'amount_pence':100}).status_code==200
    assert act(c,'invoice',another,'void').status_code==409
    assert act(c,'invoice',another,'correct').status_code==409
    assert act(c,'quote',q,'revise').status_code==409


def test_stage_requests_never_add_money_and_receipts_allocate_once(app):
    c=client(app);j,q,i=draft_invoice(c);stages=schedule(c,i)
    assert write(c,'/api/invoices/'+i['id']+'/schedule/'+stages[0]['id']+'/request',{'version':doc(c,'invoice',i)['version']}).status_code==409
    act(c,'invoice',i,'issue');version=doc(c,'invoice',i)['version'];key=str(uuid.uuid4())
    path='/api/invoices/'+i['id']+'/schedule/'+stages[0]['id']+'/request'
    first=write(c,path,{'version':version},key=key)
    assert first.status_code==200 and write(c,path,{'version':version},key=key).json()==first.json()
    assert c.get('/api/workspace').json()['stats']['outstanding_pence']==10000
    assert doc(c,'invoice',i)['paid_pence']==0
    path='/api/invoices/'+i['id']+'/payments';key=str(uuid.uuid4());payload={'amount_pence':4000,'note':'Private reference'}
    first=write(c,path,payload,key=key);assert first.status_code==200
    assert write(c,path,payload,key=key).json()==first.json()
    current=doc(c,'invoice',i)
    assert [s['paid_pence'] for s in current['schedule']]==[3000,1000,0]
    assert [s['status'] for s in current['schedule']]==['paid','part_paid','planned']
    assert write(c,path,{'amount_pence':1001,'stage_id':stages[1]['id']}).status_code==409
    assert write(c,path,{'amount_pence':1000,'stage_id':stages[1]['id']}).status_code==200
    p,_=portal(app,c,j);public=p.get('/api/portal/oak-and-vale/documents/invoice/'+i['id']).json()
    assert public['paid_pence']==5000 and all('note' not in row for row in public['payments'])
    assert write(c,'/api/invoices/'+i['id']+'/schedule/clear',{'version':doc(c,'invoice',i)['version']}).status_code==409
    assert c.get('/api/jobs/'+j['id']+'/costs').json()['recorded_payments_pence']==5000


def test_schedule_validation_lock_and_clear_before_use(app):
    c=client(app);j,q,i=draft_invoice(c);version=1
    url='/api/invoices/'+i['id']+'/schedule'
    for amount in (-1,0,9999,10001,True,10000.5):
        assert write(c,url,{'version':version,'stages':[{'label':'Deposit','amount_pence':amount}]}).status_code==422
    schedule(c,i)
    assert write(c,'/api/invoices/'+i['id'],edit_payload(doc(c,'invoice',i),9000),method='put').status_code==409
    assert write(c,url+'/clear',{'version':doc(c,'invoice',i)['version']}).status_code==200
    assert write(c,'/api/invoices/'+i['id'],edit_payload(doc(c,'invoice',i),9000),method='put').status_code==200
    schedule(c,i,(4000,5000));act(c,'invoice',i,'issue')
    write(c,'/api/invoices/'+i['id']+'/payments',{'amount_pence':1})
    assert write(c,url,{'version':doc(c,'invoice',i)['version'],'stages':[{'label':'Final','amount_pence':9000}]}).status_code==409


def test_voided_invoice_closes_existing_stage_requests_in_portal(app):
    c=client(app);j,q,i=draft_invoice(c);stages=schedule(c,i);act(c,'invoice',i,'issue')
    path='/api/invoices/'+i['id']+'/schedule/'+stages[0]['id']+'/request'
    assert write(c,path,{'version':doc(c,'invoice',i)['version']}).status_code==200
    assert act(c,'invoice',i,'void').status_code==200
    p,_=portal(app,c,j);view=p.get('/api/portal/oak-and-vale/documents/invoice/'+i['id']).json()
    assert view['balance_pence']==0 and all(s['status']=='void' and s['balance_pence']==0 for s in view['schedule'])
    assert write(c,path,{'version':doc(c,'invoice',i)['version']}).status_code==409


def test_fractional_quantities_exact_line_rounding_and_vat(app):
    c=client(app);j=job(c)
    payload={'items':[{'description':'2.5 units','quantity':'2.5','unit_pence':1234},{'description':'Half-penny line','quantity':'0.005','unit_pence':100}],'vat_percent':20}
    r=write(c,'/api/jobs/'+j['id']+'/quotes',payload);assert r.status_code==200,r.text
    s=doc(c,'quote',r.json())['snapshot']
    assert s['line_totals_pence']==[3085,1] and s['subtotal_pence']==3086 and s['vat_pence']==617 and s['total_pence']==3703
    for quantity in (True,'0','1.0001','1e2','10000.001'):
        payload['items'][0]['quantity']=quantity
        assert write(c,'/api/jobs/'+j['id']+'/quotes',payload).status_code==422


@pytest.mark.parametrize('utc_stamp,business_date',[
    ('2026-06-30T23:30:00+00:00','2026-07-01'),
    ('2026-03-29T23:30:00+00:00','2026-03-30'),
    ('2026-10-24T23:30:00+00:00','2026-10-25'),
    ('2026-10-25T00:30:00+00:00','2026-10-25'),
    ('2026-10-25T01:30:00+00:00','2026-10-25'),
    ('2026-12-01T23:30:00+00:00','2026-12-01'),
])
def test_london_document_expiry_defaults_and_receipt_date_on_utc_host(app,monkeypatch,utc_stamp,business_date):
    from backend import documents
    stamp=datetime.fromisoformat(utc_stamp)
    class FrozenDatetime(datetime):
        @classmethod
        def now(cls,tz=None):
            # A naive host clock is UTC; production code must request its zone.
            return stamp.astimezone(tz) if tz is not None else stamp.replace(tzinfo=None)
    monkeypatch.setattr(documents,'datetime',FrozenDatetime)
    assert documents.today()==business_date
    c=client(app);j=job(c);q=quote(c,j)
    assert doc(c,'quote',q)['snapshot']['document_date']==business_date
    yesterday=(datetime.fromisoformat(business_date)-timedelta(days=1)).date().isoformat()
    payload=edit_payload(doc(c,'quote',q));payload['valid_until']=yesterday
    assert write(c,'/api/quotes/'+q['id'],payload,method='put').status_code==200
    assert act(c,'quote',q,'accept').status_code==409
    payload=edit_payload(doc(c,'quote',q));payload['valid_until']=business_date
    assert write(c,'/api/quotes/'+q['id'],payload,method='put').status_code==200
    assert act(c,'quote',q,'share').status_code==200
    p,_=portal(app,c,j)
    assert write(p,'/api/portal/oak-and-vale/quotes/'+q['id']+'/accept',{'version':doc(c,'quote',q)['version'],'snapshot_hash':doc(c,'quote',q)['snapshot_hash'],'confirmed':True}).status_code==200
    created=act(c,'quote',q,'invoice').json();act(c,'invoice',created,'issue')
    assert doc(c,'invoice',created)['snapshot']['document_date']==business_date
    path='/api/invoices/'+created['id']+'/payments'
    tomorrow=(datetime.fromisoformat(business_date)+timedelta(days=1)).date().isoformat()
    assert write(c,path,{'amount_pence':100,'paid_on':tomorrow}).status_code==422
    assert write(c,path,{'amount_pence':100,'paid_on':business_date}).status_code==200
    assert write(c,path,{'amount_pence':100}).status_code==200
    current=doc(c,'invoice',created)
    assert [p['paid_on'] for p in current['payments']]==[business_date,business_date]
    correction_path=path+'/'+current['payments'][0]['id']+'/correct'
    correction={'version':current['version'],'recording_error':True,'reason':'Date correction example','replacement':{'amount_pence':100,'paid_on':tomorrow}}
    assert write(c,correction_path,correction).status_code==422
    assert doc(c,'invoice',created)['paid_pence']==200
    assert not any(p['reversed'] for p in doc(c,'invoice',created)['payments'])


def test_document_and_receipt_dates_reject_invalid_or_timestamp_input(app):
    c=client(app);j,q,i=draft_invoice(c)
    invalid=('2026-02-30','20261025','2026-10-25T01:30:00Z')
    for value in invalid:
        payload=edit_payload(doc(c,'invoice',i));payload['due_date']=value
        assert write(c,'/api/invoices/'+i['id'],payload,method='put').status_code==422
        payload=edit_payload(doc(c,'quote',q));payload['document_date']=value
        assert write(c,'/api/quotes/'+q['id'],payload,method='put').status_code==422
        assert write(c,'/api/invoices/'+i['id']+'/payments',{'amount_pence':100,'paid_on':value}).status_code==422


def test_payment_error_correction_atomic_retry_and_preserved_ledger(app):
    c=client(app);j,q,i=draft_invoice(c);schedule(c,i);act(c,'invoice',i,'issue')
    receipt=write(c,'/api/invoices/'+i['id']+'/payments',{'amount_pence':4500}).json()
    url='/api/invoices/'+i['id']+'/payments/'+receipt['id']+'/correct'
    version=doc(c,'invoice',i)['version']
    payload={'version':version,'reason':'Wrong amount was entered','recording_error':True,'replacement':{'amount_pence':11000}}
    assert write(c,url,payload).status_code==409
    assert doc(c,'invoice',i)['paid_pence']==4500 and doc(c,'invoice',i)['version']==version
    payload['replacement']={'amount_pence':2000};key=str(uuid.uuid4())
    first=write(c,url,payload,key=key);assert first.status_code==200
    assert write(c,url,payload,key=key).json()==first.json()
    current=doc(c,'invoice',i)
    assert current['paid_pence']==2000 and len(current['payments'])==2
    assert current['payments'][0]['reversed']==1
    assert [s['paid_pence'] for s in current['schedule']]==[2000,0,0]
    assert write(c,url,{**payload,'version':current['version']}).status_code==409
    totals=c.get('/api/customers/'+j['customer_id']+'/history').json()['totals']
    assert totals['received_pence']==2000 and totals['outstanding_pence']==8000
    other=first.json()['replacement']['id']
    payload={'version':current['version'],'reason':'Duplicate fictional entry','recording_error':False}
    path='/api/invoices/'+i['id']+'/payments/'+other+'/correct'
    assert write(c,path,payload).status_code==422
    payload['recording_error']=True
    assert write(c,path,payload).status_code==200
    assert act(c,'invoice',i,'void').status_code==200
    assert c.get('/api/workspace').json()['stats']['received_pence']==0


def worker(c,app):
    w=TestClient(app,headers=dict(c.headers));w.cookies.update(c.cookies);return w


def test_concurrent_stage_receipts_and_same_key_cannot_double_apply(app):
    c=client(app);_,_,i=draft_invoice(c);stages=schedule(c,i);act(c,'invoice',i,'issue')
    path='/api/invoices/'+i['id']+'/payments'
    with ThreadPoolExecutor(max_workers=4) as pool:
        codes=list(pool.map(lambda _:write(worker(c,app),path,{'amount_pence':3000,'stage_id':stages[0]['id']}).status_code,range(4)))
    assert sorted(codes)==[200,409,409,409]
    key=str(uuid.uuid4())
    with ThreadPoolExecutor(max_workers=4) as pool:
        responses=list(pool.map(lambda _:write(worker(c,app),path,{'amount_pence':7000},key=key),range(4)))
    assert all(r.status_code==200 for r in responses) and len({r.json()['id'] for r in responses})==1
    result=doc(c,'invoice',i);assert result['paid_pence']==10000 and len(result['payments'])==2


def test_concurrent_quote_revision_acceptance_and_invoice_numbering(app):
    c=client(app);j=job(c);q=quote(c,j);act(c,'quote',q,'share');version=doc(c,'quote',q)['version']
    with ThreadPoolExecutor(max_workers=2) as pool:
        outcomes=list(pool.map(lambda action:write(worker(c,app),'/api/quotes/'+q['id']+'/'+action,{'version':version,'reason':'Changed project scope'}).status_code,['accept','revise']))
    assert sorted(outcomes)==[200,409]
    accepted=quote(c,j);act(c,'quote',accepted,'accept');version=doc(c,'quote',accepted)['version']
    with ThreadPoolExecutor(max_workers=4) as pool:
        results=list(pool.map(lambda _:write(worker(c,app),'/api/quotes/'+accepted['id']+'/invoice',{'version':version}).json(),range(4)))
    assert len({r['id'] for r in results})==1
    def make(_):
        return write(worker(c,app),'/api/jobs/'+j['id']+'/invoices',{'items':[{'description':'Separate scope','quantity':1,'unit_pence':100}]}).json()
    with ThreadPoolExecutor(max_workers=6) as pool:results=list(pool.map(make,range(6)))
    assert len({r['reference'] for r in results})==6
    restarted=create_app(app.state.store.root);after=client(restarted)
    next_doc=write(after,'/api/jobs/'+j['id']+'/invoices',{'items':[{'description':'Next work','quantity':1,'unit_pence':100}]}).json()
    assert next_doc['reference'] not in {r['reference'] for r in results}


@pytest.mark.parametrize('action',['issue','cancel','void','correct'])
def test_cross_business_document_actions_and_csrf(app,action):
    a=client(app);b=client(app,1);_,_,i=draft_invoice(a)
    path='/api/invoices/'+i['id']+'/'+action
    assert write(b,path,{'version':1,'reason':'Test cross-business access'}).status_code==404
    assert b.get('/api/documents/invoice/'+i['id']).status_code==404
    assert a.post(path,json={'version':1,'reason':'Test CSRF boundary'},headers={'x-csrf-token':''}).status_code==403
    assert a.post(path,json={'version':1},headers={'origin':'https://wrong.example'}).status_code==403


def test_customer_scope_retry_actor_and_stage_ownership(app):
    c=client(app);j,q,i=draft_invoice(c);stages=schedule(c,i);act(c,'invoice',i,'issue')
    other=job(c);otherq=quote(c,other);otheri=invoice(c,otherq)
    p,_=portal(app,c,other)
    assert p.get('/api/portal/oak-and-vale/documents/invoice/'+i['id']).status_code==404
    assert write(p,'/api/invoices/'+i['id']+'/payments',{'amount_pence':1}).status_code==401
    assert write(c,'/api/invoices/'+otheri['id']+'/payments',{'amount_pence':1,'stage_id':stages[0]['id']}).status_code==404
    grant=quote(c,j);act(c,'quote',grant,'share');owner_portal,_=portal(app,c,j);version=doc(c,'quote',grant)['version'];key=str(uuid.uuid4())
    path='/api/portal/oak-and-vale/quotes/'+grant['id']+'/accept'
    assert write(owner_portal,path,{'version':version,'snapshot_hash':doc(c,'quote',grant)['snapshot_hash'],'confirmed':True},key=key).status_code==200
    assert write(p,path,{'version':version,'snapshot_hash':doc(c,'quote',grant)['snapshot_hash'],'confirmed':True},key=key).status_code==409


def test_correction_racing_new_payment_never_uses_a_stale_balance(app):
    c=client(app);_,_,i=draft_invoice(c);schedule(c,i);act(c,'invoice',i,'issue')
    receipt=write(c,'/api/invoices/'+i['id']+'/payments',{'amount_pence':4000}).json()
    version=doc(c,'invoice',i)['version']
    paths=[('/api/invoices/'+i['id']+'/payments/'+receipt['id']+'/correct',{'version':version,'reason':'Wrong amount in demo','recording_error':True,'replacement':{'amount_pence':6000}}),('/api/invoices/'+i['id']+'/payments',{'amount_pence':5000})]
    with ThreadPoolExecutor(max_workers=2) as pool:
        codes=list(pool.map(lambda item:write(worker(c,app),*item).status_code,paths))
    assert sorted(codes)==[200,409]
    result=doc(c,'invoice',i)
    assert result['paid_pence'] in (6000,9000)
    assert sum(s['paid_pence'] for s in result['schedule'])==result['paid_pence']


def test_void_racing_payment_has_one_valid_winner(app):
    c=client(app);_,_,i=draft_invoice(c);act(c,'invoice',i,'issue');version=doc(c,'invoice',i)['version']
    paths=[('/api/invoices/'+i['id']+'/void',{'version':version,'reason':'Wrong invoice issued'}),('/api/invoices/'+i['id']+'/payments',{'amount_pence':1000})]
    with ThreadPoolExecutor(max_workers=2) as pool:
        codes=list(pool.map(lambda item:write(worker(c,app),*item).status_code,paths))
    assert sorted(codes)==[200,409]
    result=doc(c,'invoice',i)
    assert (result['status'],result['paid_pence']) in (('void',0),('issued',1000))


def test_schedule_change_racing_receipt_preserves_exact_allocations(app):
    c=client(app);_,_,i=draft_invoice(c);schedule(c,i);act(c,'invoice',i,'issue');version=doc(c,'invoice',i)['version']
    paths=[('/api/invoices/'+i['id']+'/schedule',{'version':version,'stages':[{'label':'First','amount_pence':5000},{'label':'Final','amount_pence':5000}]}),('/api/invoices/'+i['id']+'/payments',{'amount_pence':4000})]
    with ThreadPoolExecutor(max_workers=2) as pool:
        codes=list(pool.map(lambda item:write(worker(c,app),*item).status_code,paths))
    assert codes[1]==200 and codes[0] in (200,409)
    result=doc(c,'invoice',i)
    assert result['paid_pence']==4000 and sum(s['paid_pence'] for s in result['schedule'])==4000
    assert sum(s['amount_pence'] for s in result['schedule'])==10000


@pytest.mark.parametrize('value',[True,False,1.5,'100',-1,0,100_000_001])
def test_receipts_require_positive_bounded_integer_pence(app,value):
    c=client(app);_,_,i=draft_invoice(c);act(c,'invoice',i,'issue')
    assert write(c,'/api/invoices/'+i['id']+'/payments',{'amount_pence':value}).status_code==422


def test_recovery_rejects_cross_customer_document_link_even_with_valid_foreign_key(app,tmp_path):
    c=client(app);j,q,i=draft_invoice(c);other=job(c)
    with app.state.store.tenant('oak') as db:db.execute('UPDATE invoices SET customer_id=? WHERE id=?',(other['customer_id'],i['id']))
    with pytest.raises(ValueError,match='ownership'):
        recovery.backup(app.state.store.root,tmp_path/'bad.zip')
    assert not (tmp_path/'bad.zip').exists()


def make_legacy(root):
    """Build invented v0.4 data using the exact schema pinned at its checkpoint."""
    schema=json.loads((Path(__file__).parents[1]/'docs/recovery-schema-v1.json').read_text())
    root.mkdir();dataset=uid();records={};stamp='2026-10-20T09:00:00+00:00'
    def database(path,definitions):
        db=sqlite3.connect(path)
        for kind in ('table','index'):
            for row in definitions:
                if row[0]==kind:db.execute(row[3])
        return db
    registry=database(root/'registry.sqlite3',schema['registry'])
    registry.execute('INSERT INTO dataset_identity VALUES(1,?)',(dataset,))
    for b in DEMO_BUSINESSES:
        registry.execute('INSERT INTO businesses VALUES(?,?,?,1)',(b['id'],b['slug'],b['name']))
        registry.execute('INSERT INTO users VALUES(?,?,?,?)',(b['id'],b['id'],b['email'],password_hash(b['password'])))
        folder=root/b['id'];folder.mkdir();(folder/'assets').mkdir()
        db=database(folder/'business.sqlite3',schema['tenant'])
        cid,jid,qid,iid,pid=(uid() for _ in range(5));records[b['id']]={'customer':cid,'job':jid,'quote':qid,'invoice':iid,'payment':pid}
        db.execute('INSERT INTO storage_identity VALUES(1,?,?)',(b['id'],dataset))
        customer={'id':cid,'name':'Legacy Example','email':'legacy@customer.example','phone':'','address':'Example lane'}
        db.execute('INSERT INTO customers(id,name,email,phone,address,portal_hash,portal_version,created_at) VALUES(?,?,?,?,?,?,1,?)',(cid,customer['name'],customer['email'],'',customer['address'],digest('legacy-local-portal-code'),stamp))
        db.execute("INSERT INTO jobs(id,customer_id,service,message,status,source,created_at) VALUES(?,?, 'Legacy job','Fictional historical work','new','staff',?)",(jid,cid,stamp))
        site=initial_site(b);content={'business':{k:site[k] for k in ('business_name','email','phone','colour')},'customer':customer,'service':'Legacy job','items':[{'description':'Legacy work','quantity':1,'unit_pence':10000}],'vat_percent':0,'notes':'Original sealed content','subtotal_pence':10000,'vat_pence':0,'total_pence':10000}
        db.execute('INSERT INTO quotes VALUES(?,?,?,?,?,?,?,?,?)',(qid,'Q-0042',jid,cid,'accepted',json.dumps(content),10000,3,stamp))
        db.execute('INSERT INTO invoices VALUES(?,?,?,?,?,?,?,?)',(iid,'I-0042',qid,jid,cid,json.dumps(content),10000,stamp))
        db.execute('INSERT INTO payments VALUES(?,?,?,?,?)',(pid,iid,2500,'Legacy receipt',stamp))
        legacy_key=str(uuid.uuid4());records[b['id']]['legacy_operation_key']=legacy_key
        old_body={'amount_pence':2500,'note':'Legacy receipt'}
        old_fingerprint=digest('/api/invoices/'+iid+'/payments'+json.dumps(old_body,sort_keys=True))
        db.execute('INSERT INTO operations VALUES(?,?,?)',(legacy_key,old_fingerprint,json.dumps({'id':pid,'paid_pence':2500})))
        db.execute('INSERT INTO site VALUES(1,?,?,1,?)',(json.dumps(site),json.dumps(site),stamp))
        db.execute('INSERT INTO site_history(content,created_at) VALUES(?,?)',(json.dumps(site),stamp))
        db.commit();db.close()
    registry.execute('INSERT INTO sessions VALUES(?,?,?,?,?,?,?)',(digest('legacy-staff-session'),'oak','oak','staff','legacy-csrf',time.time()+3600,0))
    registry.commit();registry.close()
    return dataset,records


def test_v04_upgrade_retains_documents_and_staff_session_but_revokes_permanent_portal_code(tmp_path):
    root=tmp_path/'legacy';_,records=make_legacy(root)
    iid=records['oak']['invoice']
    with sqlite3.connect(root/'oak/business.sqlite3') as db:original=db.execute('SELECT snapshot FROM invoices WHERE id=?',(iid,)).fetchone()[0]
    migrated=create_app(root);c=client(migrated)
    old=doc(c,'invoice',{'id':iid})
    assert old['status']=='issued' and old['paid_pence']==2500 and old['balance_pence']==7500 and old['issued_at']
    assert json.dumps(old['snapshot'])==original
    retry=write(c,'/api/invoices/'+iid+'/payments',{'amount_pence':2500,'note':'Legacy receipt'},key=records['oak']['legacy_operation_key'])
    assert retry.status_code==409 and doc(c,'invoice',{'id':iid})['paid_pence']==2500
    p=TestClient(migrated,headers=ORIGIN);r=p.post('/api/portal/oak-and-vale/login',json={'code':'legacy-local-portal-code'});assert r.status_code==401
    p,_=portal(migrated,c,{'customer_id':old['customer_id']})
    assert p.get('/api/portal/oak-and-vale/me').json()['invoices'][0]['id']==iid
    assert p.get('/api/portal/oak-and-vale/documents/invoice/'+iid).status_code==200
    legacy=TestClient(migrated,headers=ORIGIN);legacy.cookies.set('shs_demo_staff','legacy-staff-session')
    assert legacy.get('/api/session').status_code==200
    Store(root);assert doc(c,'invoice',{'id':iid})['history']==old['history']
    q=quote(c,{'id':records['oak']['job']});assert q['reference']=='Q-0043'
    fresh=invoice(c,q);assert fresh['reference']=='I-0043'


def test_recovery_format1_migrates_only_fresh_restored_copy(tmp_path):
    source=tmp_path/'legacy';dataset,records=make_legacy(source)
    meta={'format':recovery.FORMAT,'version':1,'prototype_version':'0.4.0','created_at':'2026-10-20T09: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/'v1.zip'
    with zipfile.ZipFile(archive,'w') as z:
        for b in DEMO_BUSINESSES:
            path=source/b['id']/'business.sqlite3'
            with sqlite3.connect(path) as db:db.execute('UPDATE customers SET portal_hash=NULL')
            raw=path.read_bytes();name='businesses/'+b['id']+'/business.sqlite3'
            meta['files'][name]={'bytes':len(raw),'sha256':hashlib.sha256(raw).hexdigest(),'business_id':b['id'],'kind':'database'};z.writestr(name,raw)
        z.writestr('manifest.json',json.dumps(meta))
    checksum=hashlib.sha256(archive.read_bytes()).hexdigest();assert recovery.verify(archive,checksum)['valid']
    target=tmp_path/'restored';recovery.restore(archive,checksum,target,confirm_fresh=True)
    with sqlite3.connect(target/'oak/business.sqlite3') as db:
        row=db.execute('SELECT status,total_pence FROM invoices WHERE id=?',(records['oak']['invoice'],)).fetchone()
        assert row==('issued',10000) and db.execute('SELECT COUNT(*) FROM payment_reversals').fetchone()[0]==0
    with sqlite3.connect(source/'oak/business.sqlite3') as db:assert 'status' not in [r[1] for r in db.execute('PRAGMA table_info(invoices)')]
    current=recovery.backup(target,tmp_path/'v2.zip');assert recovery.verify(current['archive'],current['sha256'])['valid']
    with zipfile.ZipFile(current['archive']) as z:assert json.loads(z.read('manifest.json'))['version']==recovery.VERSION


def test_document_lifecycle_survives_loss_and_recovery(app,tmp_path):
    expected={}
    for n,bid in enumerate(('oak','harbour')):
        c=client(app,n);j,q,i=draft_invoice(c);schedule(c,i);act(c,'invoice',i,'issue')
        receipt=write(c,'/api/invoices/'+i['id']+'/payments',{'amount_pence':4000}).json()
        assert write(c,'/api/invoices/'+i['id']+'/payments/'+receipt['id']+'/correct',{'version':doc(c,'invoice',i)['version'],'reason':'Correct fictional amount','recording_error':True,'replacement':{'amount_pence':2000}}).status_code==200
        expected[bid]=doc(c,'invoice',i)
    backup=recovery.backup(app.state.store.root,tmp_path/'current.zip');target=tmp_path/'fresh'
    assert app.state.store.root.is_relative_to(tmp_path)
    shutil.rmtree(app.state.store.root)
    assert not app.state.store.root.exists()
    recovery.restore(backup['archive'],backup['sha256'],target,confirm_fresh=True)
    restored_store=Store(target)
    for bid,document in expected.items():
        with restored_store.tenant(bid) as db:
            from backend.documents import get_document
            restored=get_document(db,'invoice',document['id'])
            assert restored==document
    proof={'scope':'invented local data only','source_deleted_before_restore':True,'businesses':{bid:{'invoice_status':d['status'],'total_pence':d['total_pence'],'net_received_pence':d['paid_pence'],'balance_pence':d['balance_pence'],'retained_receipts':len(d['payments']),'reversed_receipts':sum(p['reversed'] for p in d['payments']),'stage_paid_pence':[s['paid_pence'] for s in d['schedule']],'document_history_entries':len(d['history']),'restored_document_equals_saved_document':True} for bid,d in expected.items()},'browser_verified':False}
    (tmp_path/'document-lifecycle-drill.json').write_text(json.dumps(proof,indent=2)+'\n')