256 lines
9.7 KiB
Python
256 lines
9.7 KiB
Python
"""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()
|