"""Deterministic tests for multi-range authenticated Archive.org downloads."""
import re

import httpx
import pytest

from app import archive_download, main, store


@pytest.fixture
def env(tmp_path, monkeypatch):
    monkeypatch.setattr(store, 'DATA', tmp_path)
    monkeypatch.setattr(store, 'DB_PATH', tmp_path / 'queue.sqlite3')
    monkeypatch.setenv('ARCHIVE_CONNECTIONS', '4')
    monkeypatch.setattr(archive_download, 'MIN_SEGMENT_BYTES', 16)
    monkeypatch.setattr(archive_download.time, 'sleep', lambda _: None)
    store.init_db()
    return tmp_path


def job():
    return main.new_job('url', 'https://archive.org/download/demo/Game.iso', 'Demo', 'raw')


class BytesStream(httpx.SyncByteStream):
    def __init__(self, data):
        self.data = data

    def __iter__(self):
        yield self.data


def server(monkeypatch, data, custom=None, etag='"release-1"'):
    requests = []

    def handler(req):
        requests.append(req)
        if custom:
            res = custom(req)
            if res is not None:
                return res
        header = req.headers.get('range')
        if not header:
            return httpx.Response(200, stream=BytesStream(data), headers={'Content-Length': str(len(data))})
        match = re.fullmatch(r'bytes=(\d+)-(\d*)', header)
        assert match, header
        start = int(match.group(1))
        end = int(match.group(2)) if match.group(2) else len(data) - 1
        assert end < len(data) and 0 <= start <= end
        return httpx.Response(206, stream=BytesStream(data[start:end + 1]), headers={
            'Content-Range': f'bytes {start}-{end}/{len(data)}',
            'Content-Length': str(end - start + 1), 'ETag': etag})

    transport = httpx.MockTransport(handler)
    monkeypatch.setattr(archive_download, '_client', lambda: httpx.Client(
        transport=transport, follow_redirects=False))
    return requests


def final_path(env, j):
    return env / 'raw' / j['id'] / 'Game.iso'


def test_segments_reconstruct_in_order_and_show_ranges(env, monkeypatch):
    data = bytes(range(128))
    requests = server(monkeypatch, data)
    j = job()
    archive_download._transfer_one(j)
    assert final_path(env, j).read_bytes() == data
    assert store.get(j['id'])['state'] == 'preparing'
    assert not (env / 'raw' / j['id'] / 'Game.iso.part.segments').exists()
    spans = [req.headers['range'] for req in requests]
    assert spans[0] == 'bytes=0-0'
    assert sorted(spans[1:]) == ['bytes=0-31', 'bytes=32-63', 'bytes=64-95', 'bytes=96-127']
    assert all(req.headers.get('if-range') == '"release-1"' for req in requests[1:])
    assert '4 połączeń HTTP Range' in (env / 'logs' / (j['id'] + '.download.log')).read_text()


def test_existing_serial_prefix_is_reused(env, monkeypatch):
    data = bytes(range(200))
    requests = server(monkeypatch, data)
    j = job()
    folder = env / 'raw' / j['id']
    folder.mkdir(parents=True)
    (folder / 'Game.iso.part').write_bytes(data[:40])
    archive_download._transfer_one(j)
    assert final_path(env, j).read_bytes() == data
    assert len(requests) == 5
    assert all(int(r.headers['range'].split('=')[1].split('-')[0]) >= 40 for r in requests[1:])


def test_interrupted_segment_resumes_only_missing_bytes(env, monkeypatch):
    data = bytes(range(128))
    failed = {'once': False}
    monkeypatch.setattr(archive_download, 'CHUNK_BYTES', 4)

    class BrokenStream(httpx.SyncByteStream):
        def __iter__(self):
            yield data[:5]
            raise httpx.ReadError('connection reset')

    def break_once(req):
        if req.headers.get('range') == 'bytes=0-31' and not failed['once']:
            failed['once'] = True
            return httpx.Response(206, stream=BrokenStream(), headers={
                'Content-Range': 'bytes 0-31/128', 'ETag': '"release-1"'})
        return None

    requests = server(monkeypatch, data, custom=break_once)
    monkeypatch.setattr(archive_download, 'MAX_ATTEMPTS', 1)
    j = job()
    with pytest.raises(archive_download.ArchiveDownloadError, match='Segment'):
        archive_download._transfer_one(j)
    assert (env / 'raw' / j['id'] / 'Game.iso.part.segments').exists()
    assert failed['once']
    requests.clear()
    archive_download._transfer_one(j)
    assert final_path(env, j).read_bytes() == data
    assert any(r.headers.get('range') == 'bytes=4-31' for r in requests)
    assert not (env / 'raw' / j['id'] / 'Game.iso.part.segments').exists()


def test_server_ignores_segment_range_and_serial_fallback(env, monkeypatch):
    data = bytes(range(128))
    def ignore(req):
        if req.headers.get('range') not in (None, 'bytes=0-0'):
            return httpx.Response(200, stream=BytesStream(data))
        return None

    requests = server(monkeypatch, data, custom=ignore)
    j = job()
    archive_download._transfer_one(j)
    assert final_path(env, j).read_bytes() == data
    assert any('range' not in r.headers for r in requests)
    assert not (env / 'raw' / j['id'] / 'Game.iso.part.segments').exists()


def test_http_503_one_segment_retries_without_losing_other_parts(env, monkeypatch):
    data = bytes(range(128))
    failed = {'once': False}
    def temporary(req):
        if req.headers.get('range') == 'bytes=32-63' and not failed['once']:
            failed['once'] = True
            return httpx.Response(503, headers={'Retry-After': '1'})
        return None

    requests = server(monkeypatch, data, custom=temporary)
    j = job()
    archive_download._transfer_one(j)
    assert final_path(env, j).read_bytes() == data
    assert sum(req.headers.get('range') == 'bytes=32-63' for req in requests) == 2
    assert 'HTTP 503' in (env / 'logs' / (j['id'] + '.download.log')).read_text()


def test_interrupted_assembly_is_rebuilt_safely(env, monkeypatch):
    data = bytes(range(128))
    server(monkeypatch, data)
    original = archive_download._assemble_segments
    failed = {'once': False}

    def broken_assembly(jid, partial, destination, folder, manifest):
        if not failed['once']:
            failed['once'] = True
            manifest['assembling'] = True
            archive_download._write_manifest(folder, manifest)
            partial.write_bytes(b'partial-assembly')
            raise OSError('simulated crash')
        return original(jid, partial, destination, folder, manifest)

    monkeypatch.setattr(archive_download, '_assemble_segments', broken_assembly)
    j = job()
    with pytest.raises(OSError, match='simulated crash'):
        archive_download._transfer_one(j)
    archive_download._transfer_one(j)
    assert final_path(env, j).read_bytes() == data


def test_small_file_uses_single_transfer(env, monkeypatch):
    data = b'hello world'
    requests = server(monkeypatch, data)
    j = job()
    archive_download._transfer_one(j)
    assert final_path(env, j).read_bytes() == data
    assert len(requests) == 2
    assert requests[0].headers['range'] == 'bytes=0-0'
    assert 'range' not in requests[1].headers


def test_archive_connections_can_be_disabled(env, monkeypatch):
    data = bytes(range(128))
    monkeypatch.setenv('ARCHIVE_CONNECTIONS', '1')
    requests = server(monkeypatch, data)
    j = job()
    archive_download._transfer_one(j)
    assert final_path(env, j).read_bytes() == data
    assert len(requests) == 1
    assert 'range' not in requests[0].headers


def test_changed_etag_forces_clean_serial_restart(env, monkeypatch):
    data = bytes(range(128))
    def changed(req):
        if req.headers.get('range') not in (None, 'bytes=0-0'):
            value = req.headers['range'].split('=', 1)[1]
            lo, hi = [int(x) for x in value.split('-')]
            return httpx.Response(206, stream=BytesStream(data[lo:hi + 1]), headers={
                'Content-Range': f'bytes {lo}-{hi}/128', 'ETag': '"changed"'})
        return None

    requests = server(monkeypatch, data, custom=changed)
    j = job()
    # This prefix must be discarded when the origin presents conflicting validators.
    raw = env / 'raw' / j['id']
    raw.mkdir(parents=True)
    (raw / 'Game.iso.part').write_bytes(b'old-data')
    archive_download._transfer_one(j)
    assert final_path(env, j).read_bytes() == data
    assert any('range' not in req.headers for req in requests)


def test_archive_rejects_unexpected_compression(env, monkeypatch):
    def compressed(_request):
        return httpx.Response(206, headers={
            'Content-Encoding': 'gzip', 'Content-Range': 'bytes 0-0/128'})
    server(monkeypatch, bytes(range(128)), custom=compressed)
    j = job()
    with pytest.raises(archive_download.ArchiveDownloadError, match='Accept-Encoding'):
        archive_download._transfer_one(j)
    assert not final_path(env, j).exists()


def test_invalid_connection_settings_clamped(env, monkeypatch):
    monkeypatch.setenv('ARCHIVE_CONNECTIONS', '100')
    assert archive_download._connection_limit() == 50
    monkeypatch.setenv('ARCHIVE_CONNECTIONS', '-1')
    assert archive_download._connection_limit() == 1
    monkeypatch.setenv('ARCHIVE_CONNECTIONS', 'invalid')
    assert archive_download._connection_limit() == 50


def test_default_fifty_parallel_ranges_complete(env, monkeypatch):
    """The new default really runs 50 transfers, not only displaying that value."""
    monkeypatch.delenv('ARCHIVE_CONNECTIONS', raising=False)
    assert archive_download._connection_limit() == 50
    data = bytes(range(256)) * 4  # Enough to split into 50 ranges in this test
    requests = server(monkeypatch, data)
    j = job()
    archive_download._transfer_one(j)
    assert final_path(env, j).read_bytes() == data
    spans = [request.headers['range'] for request in requests]
    assert spans[0] == 'bytes=0-0'
    assert len(spans) == 51  # Range probe plus 50 segments
    assert len(set(spans[1:])) == 50
    assert store.get(j['id'])['state'] == 'preparing'
    assert '50 połączeń HTTP Range' in (env / 'logs' / (j['id'] + '.download.log')).read_text()