117 lines
3.6 KiB
Python
117 lines
3.6 KiB
Python
import threading
|
|
from time import monotonic
|
|
|
|
import httpx
|
|
from flask import Flask
|
|
|
|
from upload.backend.session import add_file, cleanup, configure, create_session, get_files
|
|
from upload.backend.session import state as session_state
|
|
from upload.backend.upload_refs.blueprint import create_upload_refs_blueprint
|
|
|
|
|
|
def build_app(transport, **config):
|
|
app = Flask(__name__)
|
|
app.config['TESTING'] = True
|
|
app.config['UPLOAD_HTTPX_TRANSPORT'] = transport
|
|
app.register_blueprint(create_upload_refs_blueprint({
|
|
'vmUploadPrefix': 'https://vm.test/buffer/',
|
|
'httpxTransport': transport,
|
|
'pullRetries': 3,
|
|
'pullRetryDelay': 0,
|
|
**config,
|
|
}))
|
|
return app
|
|
|
|
|
|
def test_session_limit_accepts_exact_boundary_and_rejects_overflow():
|
|
sid = create_session()
|
|
previous = session_state.MAX_SESSION_BYTES
|
|
configure(max_session_bytes=10)
|
|
try:
|
|
assert add_file(sid, 'exact.bin', b'x' * 10) is True
|
|
assert add_file(sid, 'overflow.bin', b'x') is False
|
|
assert len(get_files(sid)) == 1
|
|
finally:
|
|
configure(max_session_bytes=previous)
|
|
cleanup(sid)
|
|
|
|
|
|
def test_upload_refs_rejects_actual_oversize_and_deletes_buffer():
|
|
storage = {'oversize': b'123456789'}
|
|
deleted = []
|
|
|
|
def handler(request):
|
|
if request.method == 'GET':
|
|
return httpx.Response(200, content=storage['oversize'])
|
|
if request.method == 'DELETE':
|
|
deleted.append(request.url.path)
|
|
storage.clear()
|
|
return httpx.Response(204)
|
|
return httpx.Response(405)
|
|
|
|
transport = httpx.MockTransport(handler)
|
|
app = build_app(transport, maxFileBytes=8)
|
|
sid = create_session()
|
|
try:
|
|
with app.test_client() as client:
|
|
response = client.post('/api/upload_refs', json={
|
|
'session': sid,
|
|
'files': [{
|
|
'name': 'oversize.txt',
|
|
'size': 8,
|
|
'url': 'https://vm.test/buffer/oversize',
|
|
}],
|
|
})
|
|
assert response.status_code == 200
|
|
assert response.get_json()['added'] == 0
|
|
assert get_files(sid) == []
|
|
assert deleted == ['/buffer/oversize']
|
|
assert storage == {}
|
|
finally:
|
|
cleanup(sid)
|
|
|
|
|
|
def test_upload_refs_retries_transient_pull_and_deletes_after_success():
|
|
attempts = []
|
|
deleted = []
|
|
|
|
def handler(request):
|
|
if request.method == 'GET':
|
|
attempts.append(request.url.path)
|
|
if len(attempts) == 1:
|
|
return httpx.Response(503)
|
|
return httpx.Response(200, content=b'retry-content')
|
|
if request.method == 'DELETE':
|
|
deleted.append(request.url.path)
|
|
return httpx.Response(204)
|
|
return httpx.Response(405)
|
|
|
|
transport = httpx.MockTransport(handler)
|
|
app = build_app(transport)
|
|
with app.test_client() as client:
|
|
response = client.post('/api/upload_refs', json={'files': [{
|
|
'name': 'retry.txt',
|
|
'size': 13,
|
|
'url': 'https://vm.test/buffer/retry',
|
|
}]})
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert data['added'] == 1
|
|
assert len(attempts) == 2
|
|
assert deleted == ['/buffer/retry']
|
|
cleanup(data['session'])
|
|
|
|
|
|
def test_session_ttl_removes_session():
|
|
previous = session_state.TTL_SECONDS
|
|
configure(ttl_seconds=0.05)
|
|
sid = create_session()
|
|
try:
|
|
deadline = monotonic() + 1
|
|
while monotonic() < deadline and get_files(sid) is not None:
|
|
threading.Event().wait(0.01)
|
|
assert get_files(sid) is None
|
|
finally:
|
|
configure(ttl_seconds=previous)
|
|
cleanup(sid)
|