"""v0.20 content, resource, ownership and original-request regressions."""
import base64
from concurrent.futures import ThreadPoolExecutor
import hashlib
import io
import json
import random
import threading
import uuid
from pathlib import Path

import pillow_heif
import pytest
from fastapi import HTTPException
from fastapi.testclient import TestClient
from PIL import Image

from backend import enquiry_images as decoder, public_enquiries as intake
from test_boundaries import app, client, job, ORIGIN
from test_public_enquiries import payload, photo, submit, saved, assert_empty, PATH
from test_photos import listing, upload
from photo_format_fixtures import heif, heif_exact_size, png_exact_size


@pytest.mark.parametrize('kind', ['single', 'heif-brand', 'grid', '10bit', '12bit', 'rotated'])
def test_real_heif_variants_are_decoded_privately_with_original_facts(app, kind):
    raw = heif(kind)
    assert pillow_heif.is_supported(io.BytesIO(raw))
    source = pillow_heif.open_heif(raw)
    assert len(source) == 1
    if kind in ('10bit', '12bit'):
        assert source.info['bit_depth'] == int(kind[:-3])
    if kind == 'grid':
        assert b'grid' in raw
    owner = client(app)
    name = 'synthetic.heif' if kind == 'heif-brand' else 'synthetic.heic'
    j = saved(owner, submit(app, payload(photos=[photo(raw=raw, name=name)])))
    p = listing(owner, j).json()[0]
    assert not p['reviewed'] and p['customer_share'] is None
    url = f"/api/jobs/{j['id']}/photos/{p['id']}/image"
    result = owner.get(url)
    assert result.status_code == 200 and b'SYNTHETIC PRIVATE' not in result.content
    with Image.open(io.BytesIO(result.content)) as image:
        assert image.format == 'PNG' and image.info == {} and dict(image.getexif()) == {}
        assert image.size == ((96, 160) if kind == 'rotated' else (160, 96))
    with app.state.store.tenant('oak') as db:
        provenance = dict(db.execute('SELECT * FROM enquiry_photo_sources').fetchone())
        assert provenance['source_format'] == 'HEIF'
        assert provenance['source_bytes'] == len(raw)
        assert provenance['source_sha256'] == hashlib.sha256(raw).hexdigest()
        assert provenance['filename'] == name and provenance['job_id'] == j['id']
    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/projects').json() == []


def test_content_not_filename_or_mime_controls_decoding(app):
    owner = client(app)
    j = saved(owner, submit(app, payload(photos=[photo(raw=heif(), name='not-trusted.png')])))
    assert j['enquiry_submission']['photos'][0]['source_format'] == 'HEIF'
    assert submit(app, payload(photos=[photo(raw=b'fake HEIF content', name='fake.heic')])).status_code == 415


@pytest.mark.parametrize('kind', ['multi', 'truncated', 'header-only', 'sequence-brand', 'avif-brand'])
def test_unusable_or_multi_image_heif_is_rejected_without_partial_save(app, kind):
    raw = heif()
    if kind == 'multi':
        raw = heif('multi')
        assert len(pillow_heif.open_heif(raw)) == 2
    elif kind == 'truncated': raw = raw[:-100]
    elif kind == 'header-only': raw = raw[:28]
    elif kind == 'sequence-brand': raw = raw[:8] + b'hevc' + raw[12:]
    elif kind == 'avif-brand': raw = raw[:8] + b'avif' + raw[12:]
    response = submit(app, payload(photos=[photo(), photo(raw=raw, name='bad.heic')]))
    assert response.status_code == 415, response.text
    assert 'Photo 2' in response.json()['detail']
    assert_empty(app)


@pytest.mark.parametrize('make', [png_exact_size, heif_exact_size])
@pytest.mark.parametrize('delta', [-1, 0, 1])
def test_genuinely_readable_exact_file_size_boundaries(app, make, delta):
    raw = make(intake.MAX_FILE_BYTES + delta)
    assert len(raw) == 10 * 1024 * 1024 + delta
    if make is heif_exact_size:
        assert pillow_heif.is_supported(raw)
        assert pillow_heif.open_heif(raw).size == (160, 96)
    else:
        with Image.open(io.BytesIO(raw)) as image: image.verify()
    response = submit(app, payload(photos=[photo(raw=raw, name='boundary.heif' if make is heif_exact_size else 'boundary.png')]))
    assert response.status_code == (413 if delta > 0 else 200), response.text
    if delta > 0: assert_empty(app)


def test_five_full_allowance_files_fit_body_and_total_without_reducing_per_file_limit(app):
    raw = png_exact_size(intake.MAX_FILE_BYTES)
    data = payload(photos=[photo(raw=raw, name=f'full-allowance-{i}.png') for i in range(5)])
    assert sum(len(base64.b64decode(p['data'])) for p in data['photos']) == intake.MAX_TOTAL_BYTES == 50 * 1024 * 1024
    assert len(json.dumps(data).encode()) < intake.MAX_REQUEST_BYTES
    owner = client(app)
    j = saved(owner, submit(app, data))
    assert len(listing(owner, j).json()) == 5


def test_large_encoded_pixels_resize_and_five_images_can_exceed_old_converted_total(app):
    pixels = random.Random(20).randbytes(2048 * 1536 * 3)
    original = Image.frombytes('RGB', (2048, 1536), pixels)
    output = io.BytesIO(); original.save(output, 'PNG')
    raw = output.getvalue()
    assert 2 * 1024 * 1024 < len(raw) < intake.MAX_FILE_BYTES
    data = payload(photos=[photo(raw=raw, name=f'noisy-synthetic-{i}.png') for i in range(5)])
    owner = client(app); j = saved(owner, submit(app, data))
    with app.state.store.tenant('oak') as db:
        blobs = list(db.execute('SELECT bytes,width,height FROM photo_blobs'))
        assert len(blobs) == 5 and sum(b['bytes'] for b in blobs) > 20 * 1024 * 1024
        assert all((b['width'], b['height']) == (1800, 1350) for b in blobs)


@pytest.mark.parametrize('bound', ['time', 'memory', 'output', 'exit'])
def test_supervised_decoder_failure_saves_nothing_and_can_retry(app, monkeypatch, tmp_path, bound):
    data = payload(photos=[photo(raw=heif(), name='synthetic.heic')])
    key = str(uuid.uuid4())
    with monkeypatch.context() as m:
        if bound == 'time': m.setattr(decoder, 'DECODE_SECONDS', 0)
        elif bound == 'memory': m.setattr(decoder, 'MAX_DECODER_RSS', 1)
        elif bound == 'output': m.setattr(decoder, 'MAX_OUTPUT_BYTES', 16)
        else:
            child = tmp_path / 'failing-decoder.py'
            child.write_text('raise SystemExit(7)\n')
            m.setattr(decoder, '_worker', child)
        assert submit(app, data, key).status_code == 415
        assert_empty(app)
    first = submit(app, data, key)
    assert first.status_code == 200
    assert submit(app, data, key).json() == first.json()


def test_new_format_retry_keeps_original_after_cache_pruning_and_changed_content_conflicts(app):
    key = str(uuid.uuid4()); data = payload(photos=[photo(raw=heif(), name='synthetic.heic')])
    with ThreadPoolExecutor(2) as pool:
        results = list(pool.map(lambda _: submit(app, data, key), range(2)))
    assert results[0].status_code == 200 and results[0].json() == results[1].json()
    with app.state.store.tenant('oak') as db: db.execute('DELETE FROM operations')
    assert submit(app, data, key).json() == results[0].json()
    assert submit(app, payload(photos=[photo(raw=heif('grid'), name='synthetic.heic')]), key).status_code == 409
    with app.state.store.tenant('oak') as db:
        assert db.execute('SELECT COUNT(*) FROM jobs').fetchone()[0] == 1
        assert db.execute('SELECT COUNT(*) FROM photo_blobs').fetchone()[0] == 1


def test_admission_bounds_large_public_bodies_and_recovers_capacity(app, monkeypatch):
    active = 0; lock = threading.Lock(); ready = threading.Event(); release = threading.Event()
    original = intake.prepare_photos
    def blocked(photos):
        nonlocal active
        with lock:
            active += 1
            if active == 2: ready.set()
        assert release.wait(10)
        return original(photos)
    with monkeypatch.context() as m:
        m.setattr(intake, 'prepare_photos', blocked)
        with ThreadPoolExecutor(2) as pool:
            pending = [pool.submit(submit, app, payload()) for _ in range(2)]
            try:
                assert ready.wait(10)
                assert submit(app, payload()).status_code == 503
            finally: release.set()
            assert [p.result().status_code for p in pending] == [200, 200]
    assert submit(app, payload()).status_code == 200


@pytest.mark.parametrize('failure', ['second_file', 'operation_record'])
def test_heif_storage_failure_rolls_back_only_its_files_and_retries_exactly(app, monkeypatch, failure):
    key = str(uuid.uuid4())
    data = payload(photos=[photo(raw=heif('single'), name='first.heic'), photo(raw=heif('grid'), name='second.heif')])
    real = intake.write_photo
    with monkeypatch.context() as m:
        if failure == 'second_file':
            def broken(store, bid, blob, body, created):
                real(store, bid, blob, body, created)
                if len(created) == 2: raise OSError('Synthetic disk failure')
            m.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,'Synthetic transaction failure'); END")
        assert submit(app, data, key, raise_errors=False).status_code == 500
        assert_empty(app)
    if failure == 'operation_record':
        with app.state.store.tenant('oak') as db: db.execute('DROP TRIGGER fail_operation')
    first = submit(app, data, key)
    assert first.status_code == 200 and submit(app, data, key).json() == first.json()
    assert len(list((app.state.store.root/'oak/photos').iterdir())) == 2


def test_enquiry_decoder_does_not_expand_authenticated_photo_contract(app):
    owner = client(app); j = job(owner)
    assert upload(owner, j, raw=heif()).status_code == 415
    assert upload(owner, j, raw=png_exact_size(2*1024*1024+1)).status_code == 413
    assert listing(owner, j).json() == []


def test_slow_initial_reader_receives_whole_image_without_relaxing_limits(app, monkeypatch, tmp_path):
    # Delay reading beyond several monitor ticks, with input larger than a pipe.
    worker = tmp_path / 'delayed-image-reader.py'
    worker.write_text('import time, runpy\ntime.sleep(0.2)\nrunpy.run_path(' + repr(str(decoder._worker)) + ', run_name="__main__")\n')
    monkeypatch.setattr(decoder, '_worker', worker)
    raw = png_exact_size(intake.MAX_FILE_BYTES)
    response = submit(app, payload(photos=[photo(raw=raw, name='delayed-readable.png')]))
    assert response.status_code == 200, response.text
    assert decoder.DECODE_SECONDS == 20 and decoder.MAX_DECODER_RSS == 768 * 1024 * 1024