"""Public intake uses invented contacts and synthetic pixels only."""
import base64,copy,hashlib,io,json,sqlite3,uuid
from concurrent.futures import ThreadPoolExecutor
from datetime import date,timedelta
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from PIL import Image
from backend import public_enquiries as intake
from backend import recovery
from test_boundaries import app,client,write,job,portal,ORIGIN
from test_photos import pixels,listing,record,state,share,project,pub,uploaded
from test_identity import add_staff
from test_workforce import person
from test_scheduling import booking
from test_customer_reviews import selection,preview,confirm
from test_recovery import table_rows

PATH='/api/public/oak-and-vale/enquiries'

def photo(colour='green',format='PNG',raw=None,name=None):
    return {'filename':name or 'synthetic-'+colour+'.'+format.lower(),'data':base64.b64encode(raw if raw is not None else pixels(colour,format,metadata=True)).decode()}

def payload(**changes):
    return {'name':'Alex Invented','email':'alex@intake.example','phone':'00000 000001','address':'1 Fictional Street','service':'Fictional shelf fitting','message':'Original invented visitor message.','preferred_date':intake.uk_today(),'photos':[],**changes}

def submit(application,data=None,key=None,path=PATH,raise_errors=True):
    return write(TestClient(application,headers=ORIGIN,raise_server_exceptions=raise_errors),path,data or payload(),key=key)

def saved(owner,response):
    assert response.status_code==200,response.text
    assert set(response.json())=={'received','receipt'} and response.json()['received'] is True
    result=owner.get('/api/jobs/'+response.json()['receipt']);assert result.status_code==200,result.text
    return result.json()

def assert_empty(application):
    with application.state.store.tenant('oak') as db:
        for table in ('jobs','customers','enquiry_intakes','enquiry_photo_sources','job_photos','photo_blobs','photo_history','operations'):
            assert db.execute('SELECT COUNT(*) FROM '+table).fetchone()[0]==0,table
    assert list((application.state.store.root/'oak/photos').iterdir())==[]


@pytest.mark.parametrize('format',['PNG','JPEG','WEBP'])
def test_saved_preference_and_sanitized_photo_have_exact_private_provenance(app,format):
    owner=client(app);data=payload(photos=[photo(format=format)])
    j=saved(owner,submit(app,data));original=j['enquiry_submission']
    assert original['facts']=={k:v for k,v in data.items() if k!='photos'}
    assert j['scheduled_date']=='' and j['status']=='new'
    with app.state.store.tenant('oak') as db:
        assert db.execute('SELECT COUNT(*) FROM calendar_bookings').fetchone()[0]==0
        source=dict(db.execute('SELECT * FROM enquiry_photo_sources').fetchone())
        assert source['job_id']==j['id'] and source['source_format']==format
        assert source['source_sha256']==hashlib.sha256(base64.b64decode(data['photos'][0]['data'])).hexdigest()
        assert 'synthetic' not in db.execute('SELECT result FROM operations').fetchone()[0]
    p=listing(owner,j).json()[0];assert p['origin']=='public_enquiry' and p['reviewed']==0 and p['customer_share'] is None
    url=f"/api/jobs/{j['id']}/photos/{p['id']}/image";image=owner.get(url)
    assert image.status_code==200 and b'PRIVATE EXAMPLE' not in image.content and 'no-store' in image.headers['cache-control']
    with Image.open(io.BytesIO(image.content)) as decoded:assert decoded.format=='PNG' and decoded.info=={}
    assert TestClient(app,headers=ORIGIN).get(url).status_code==401
    assert client(app,1).get(url).status_code==404
    assert owner.get('/api/public/oak-and-vale').json().get('enquiry_submission') is None
    assert owner.get('/api/assets').json()==[]
    assert owner.get('/api/public/oak-and-vale/projects').json()==[]
    with app.state.store.tenant('oak') as db:blob=db.execute('SELECT blob_id FROM job_photos').fetchone()[0]
    assert TestClient(app,headers=ORIGIN).get('/api/public/oak-and-vale/assets/'+blob).status_code==404
    assert owner.get('/api/assets/'+blob).status_code==404


@pytest.mark.parametrize('choice',['omitted','blank','null'])
def test_zero_photo_old_shape_and_blank_preference_do_not_invent_dates(app,choice):
    data=payload();data.pop('photos');data.pop('preferred_date')
    if choice!='omitted':data['preferred_date']='' if choice=='blank' else None
    owner=client(app);j=saved(owner,submit(app,data))
    assert j['enquiry_submission']['facts']['preferred_date']=='' and j['enquiry_submission']['photos']==[]
    assert j['scheduled_date']=='' and listing(owner,j).json()==[]
    manual=job(owner);assert owner.get('/api/jobs/'+manual['id']).json()['enquiry_submission'] is None


@pytest.mark.parametrize('value',['2026-02-30','20261029','2026-1-01','2026-10-29T00:00:00Z','0000-01-01','10000-01-01',True,20300101,'2020-01-01'])
def test_invalid_or_past_date_is_rejected_without_saving(app,value):
    r=submit(app,payload(preferred_date=value,photos=[photo()]));assert r.status_code==422,r.text
    assert 'Preferred date' in r.json()['detail'];assert_empty(app)


def test_today_future_upper_date_and_retry_after_midnight(app,monkeypatch):
    monkeypatch.setattr(intake,'uk_today',lambda:'2026-10-29')
    owner=client(app);key=str(uuid.uuid4());data=payload(preferred_date='2026-10-29');first=submit(app,data,key)
    assert first.status_code==200
    monkeypatch.setattr(intake,'uk_today',lambda:'2026-10-30')
    assert submit(app,data,key).json()==first.json()
    assert submit(app,data).status_code==422
    j=saved(owner,submit(app,payload(preferred_date='9999-12-31')));assert j['enquiry_submission']['facts']['preferred_date']=='9999-12-31'


@pytest.mark.parametrize('bad',[{'filename':'../../escape.png','data':'AAAA'},{'filename':'C:\\photo.png','data':'AAAA'},{'filename':'bad\x00.png','data':'AAAA'},{'filename':'x'*121,'data':'AAAA'},{'filename':'bad.png','data':'not base64'},{'url':'https://example.invalid/private.png'},{'photo_id':'f'*32},{'filename':'fake.png','data':base64.b64encode(b'<svg><script>active</script></svg>').decode()}])
def test_untrusted_names_paths_references_and_content_are_rejected(app,bad):
    r=submit(app,payload(photos=[bad]));assert r.status_code in (413,415,422),r.text;assert_empty(app)


def test_sixth_file_oversize_malformed_animation_and_pixels_are_rejected(app):
    assert submit(app,payload(photos=[photo()]*6)).status_code==422
    samples=[(b'x'*(intake.MAX_FILE_BYTES+1),413),(b'x'*(intake.MAX_FILE_BYTES+4),422),(b'',422),(b'not an image',415),(pixels(format='GIF'),415),(pixels()[:25],415)]
    out=io.BytesIO();Image.new('RGB',(4000,3001)).save(out,'PNG');samples.append((out.getvalue(),415))
    out=io.BytesIO();Image.new('RGB',(12001,1)).save(out,'PNG');samples.append((out.getvalue(),415))
    out=io.BytesIO();Image.new('RGB',(8,8),'red').save(out,'PNG',save_all=True,append_images=[Image.new('RGB',(8,8),'blue')]);samples.append((out.getvalue(),415))
    for raw,status in samples:assert submit(app,payload(photos=[photo(raw=raw)])).status_code==status
    assert_empty(app)


def test_total_converted_storage_count_and_body_bounds_leave_no_enquiry(app,monkeypatch):
    monkeypatch.setattr(intake,'MAX_TOTAL_BYTES',1)
    assert submit(app,payload(photos=[photo()])).status_code==413
    monkeypatch.setattr(intake,'MAX_TOTAL_BYTES',10*1024*1024);monkeypatch.setattr(intake,'MAX_NORMALIZED_BYTES',1)
    assert submit(app,payload(photos=[photo()])).status_code==413
    monkeypatch.setattr(intake,'MAX_NORMALIZED_BYTES',20*1024*1024);monkeypatch.setattr(intake,'MAX_BUSINESS_BYTES',1)
    assert submit(app,payload(photos=[photo()])).status_code==409
    monkeypatch.setattr(intake,'MAX_BUSINESS_BYTES',128*1024*1024);monkeypatch.setattr(intake,'MAX_BUSINESS_PHOTOS',0)
    assert submit(app,payload(photos=[photo()])).status_code==409
    anonymous=TestClient(app,headers=ORIGIN)
    assert anonymous.post(PATH,content=b'x'*(intake.MAX_REQUEST_BYTES+1)).status_code==413
    assert anonymous.post('/api/jobs',content=b'x'*(2*1024*1024+1)).status_code==413
    assert_empty(app)


def test_five_photo_retry_concurrency_content_binding_and_separate_matching_contacts(app):
    owner=client(app);key=str(uuid.uuid4());data=payload(photos=[photo(c) for c in ['red','blue','green','yellow','purple']])
    with ThreadPoolExecutor(2) as pool:results=list(pool.map(lambda _:submit(app,data,key),range(2)))
    first=saved(owner,results[0]);assert results[0].json()==results[1].json()
    assert len(listing(owner,first).json())==5
    for changed in [payload(**{**data,'message':'A changed message'}),payload(**{**data,'preferred_date':'9999-12-31'}),payload(**{**data,'photos':list(reversed(data['photos']))})]:assert submit(app,changed,key).status_code==409
    altered=copy.deepcopy(data);altered['photos'][0]=photo('black')
    assert submit(app,altered,key).status_code==409
    second=saved(owner,submit(app,data));assert second['id']!=first['id'] and second['customer_id']!=first['customer_id']
    assert {p['id'] for p in listing(owner,first).json()}.isdisjoint(p['id'] for p in listing(owner,second).json())
    other=submit(app,data,key,path='/api/public/harbour-homeworks/enquiries');foreign=saved(client(app,1),other)
    assert foreign['id'] not in (first['id'],second['id'])
    assert owner.get('/api/jobs/'+foreign['id']).status_code==404
    with app.state.store.tenant('oak') as db:assert db.execute('SELECT COUNT(*) FROM jobs').fetchone()[0]==2 and db.execute('SELECT COUNT(*) FROM photo_blobs').fetchone()[0]==10


@pytest.mark.parametrize('failure',['second_file','operation_record'])
def test_mid_operation_failure_rolls_back_files_and_rows_and_retries_once(app,monkeypatch,failure):
    key=str(uuid.uuid4());data=payload(photos=[photo('red'),photo('blue')]);real=intake.write_photo
    if failure=='second_file':
        def broken(store,bid,blob,body,created):
            real(store,bid,blob,body,created)
            if len(created)==2:raise OSError('Injected disk failure')
        monkeypatch.setattr(intake,'write_photo',broken)
    else:
        with app.state.store.tenant('oak') as db:db.execute("CREATE TRIGGER fail_operation BEFORE INSERT ON operations BEGIN SELECT RAISE(ABORT,'Injected transaction failure'); END")
    assert submit(app,data,key,raise_errors=False).status_code==500
    assert_empty(app)
    if failure=='second_file':monkeypatch.setattr(intake,'write_photo',real)
    else:
        with app.state.store.tenant('oak') as db:db.execute('DROP TRIGGER fail_operation')
    result=submit(app,data,key);assert result.status_code==200
    assert submit(app,data,key).json()==result.json()
    assert len(list((app.state.store.root/'oak/photos').iterdir()))==2


@pytest.mark.parametrize('image_format',['PNG','HEIF'])
def test_new_visitor_photo_requires_review_for_assigned_staff_and_explicit_sharing(app,image_format):
    from photo_format_fixtures import heif
    image=photo(raw=heif(),name='synthetic.heic') if image_format=='HEIF' else photo()
    owner=client(app);j=saved(owner,submit(app,payload(photos=[image])));p=listing(owner,j).json()[0]
    resource=person(owner);booking(owner,j,resource);staff,_=add_staff(app,owner,resource_id=resource['id'])
    assert listing(staff,j).json()==[]
    path=f"/api/jobs/{j['id']}/photos/{p['id']}/image"
    assert staff.get(path).status_code==404
    customer,_=portal(app,owner,j)
    assert customer.get('/api/portal/oak-and-vale/jobs/'+j['id']+'/photos').json()==[]
    assert share(owner,j,p).status_code==409
    gallery=project(owner,j,p);assert pub(owner,gallery).status_code==409
    assert state(owner,j,p,'review').status_code==200
    assert len(listing(staff,j).json())==1 and staff.get(path).status_code==200
    assert customer.get('/api/portal/oak-and-vale/jobs/'+j['id']+'/photos').json()==[]
    assert pub(owner,gallery,False).status_code==422
    assert share(owner,j,p).status_code==200
    assert len(customer.get('/api/portal/oak-and-vale/jobs/'+j['id']+'/photos').json())==1
    assert state(owner,j,p,'archive').status_code==200 and staff.get(path).status_code==404
    assert state(owner,j,p,'restore').status_code==200 and staff.get(path).status_code==404


def test_customer_transfer_keeps_original_intake_and_clears_photo_permission(app,tmp_path):
    owner=client(app);j=saved(owner,submit(app,payload(photos=[photo()])));p=listing(owner,j).json()[0];original=j['enquiry_submission']
    assert state(owner,j,p,'review').status_code==200 and share(owner,j,p).status_code==200
    target=job(owner,name='Different exact record');assert confirm(owner,preview(owner,selection(owner,j,target))).status_code==200
    current=owner.get('/api/jobs/'+j['id']).json()
    assert current['customer_id']==target['customer_id'] and current['enquiry_submission']==original
    after=listing(owner,current).json()[0];assert after['reviewed']==0 and after['customer_share'] is None
    result=recovery.backup(app.state.store.root,tmp_path/'transfer.zip');assert recovery.verify(result['archive'],result['sha256'])['valid']


def test_submission_cannot_supply_identity_or_website_asset_references(app):
    for changes in [{'customer_id':'a'*32},{'job_id':'b'*32},{'business_id':'harbour'},{'photos':[{'asset_id':'c'*32}]}]:
        assert submit(app,payload(**changes)).status_code==422
    assert_empty(app)


def test_public_rate_bound_precedes_image_decode(app,monkeypatch):
    for _ in range(20):assert submit(app,payload(preferred_date='')).status_code==200
    monkeypatch.setattr(intake,'prepare_photos',lambda _:pytest.fail('Rate-limited images must not be decoded'))
    assert submit(app,payload(photos=[photo()])).status_code==429


def test_matching_contact_portal_cannot_read_another_enquirys_shared_photo(app):
    owner=client(app);first=saved(owner,submit(app,payload(photos=[photo('red')])));second=saved(owner,submit(app,payload(photos=[photo('blue')])) )
    p=listing(owner,second).json()[0];assert state(owner,second,p,'review').status_code==200 and share(owner,second,p).status_code==200
    customer,_=portal(app,owner,first)
    base='/api/portal/oak-and-vale/jobs/'+second['id']+'/photos'
    assert customer.get(base).status_code==404 and customer.get(base+'/'+p['id']+'/image').status_code==404
    assert customer.get('/api/jobs/'+second['id']).status_code==401


def test_failure_cleanup_preserves_preexisting_private_records_and_files(app,monkeypatch):
    owner=client(app);old=job(owner);uploaded(owner,old)
    before=table_rows(app.state.store.root,'oak');files={p.name:p.read_bytes() for p in (app.state.store.root/'oak/photos').iterdir()};real=intake.write_photo
    def fail_after_write(store,bid,blob,body,created):
        real(store,bid,blob,body,created);raise OSError('Injected failure after own file creation')
    monkeypatch.setattr(intake,'write_photo',fail_after_write)
    assert submit(app,payload(photos=[photo('purple')]),raise_errors=False).status_code==500
    assert table_rows(app.state.store.root,'oak')==before
    assert {p.name:p.read_bytes() for p in (app.state.store.root/'oak/photos').iterdir()}==files


def test_jpeg_gps_and_orientation_metadata_are_removed_from_saved_pixels(app):
    exif=Image.Exif();exif[270]='SYNTHETIC PRIVATE LOCATION';exif[274]=6
    exif[34853]={1:'N',2:(1.0,2.0,3.0),3:'E',4:(4.0,5.0,6.0)}
    output=io.BytesIO();Image.new('RGB',(40,24),'teal').save(output,'JPEG',exif=exif)
    with Image.open(io.BytesIO(output.getvalue())) as source:assert source.getexif().get_ifd(34853)
    owner=client(app);j=saved(owner,submit(app,payload(photos=[photo(raw=output.getvalue(),name='synthetic-gps.jpg')])));p=listing(owner,j).json()[0]
    result=owner.get('/api/jobs/'+j['id']+'/photos/'+p['id']+'/image')
    assert b'SYNTHETIC PRIVATE LOCATION' not in result.content
    with Image.open(io.BytesIO(result.content)) as clean:assert clean.size==(24,40) and clean.info=={} and dict(clean.getexif())=={}