daniel156161
3194e534e0
Build and Push Docker Container / build-and-push (push) Successful in 1m40s
- Recover regular multipart fields when the file part is not exposed through request.files. - Preserve expires and note values for browser uploads that need body-based file recovery. - Add parser coverage for file parts without filenames and recovered expiry fields. - Extend upload route tests to assert recovered uploads keep expiry metadata. - Bump NanoShare to 1.27.3.
147 lines
4.4 KiB
Python
147 lines
4.4 KiB
Python
import asyncio
|
|
import importlib.util
|
|
import sys
|
|
import types
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
|
|
from quart import Quart, session
|
|
from quart.datastructures import FileStorage
|
|
|
|
def load_upload_module(monkeypatch):
|
|
fake_header = types.ModuleType('my_modules.decoratory.header')
|
|
|
|
def login_required(func):
|
|
async def wrapper(*args, **kwargs):
|
|
return await func(user=session.get('user'), *args, **kwargs)
|
|
return wrapper
|
|
|
|
fake_header.login_required = login_required
|
|
|
|
fake_event = types.ModuleType('quart_common.web.wide_event')
|
|
fake_event.add_wide_event_context = lambda **kwargs: None
|
|
|
|
monkeypatch.setitem(sys.modules, 'my_modules.decoratory.header', fake_header)
|
|
monkeypatch.setitem(sys.modules, 'quart_common.web.wide_event', fake_event)
|
|
|
|
module_path = Path(__file__).resolve().parents[1] / 'routes' / 'side' / 'upload.py'
|
|
spec = importlib.util.spec_from_file_location('upload_route_under_test', module_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
class AwaitableValue:
|
|
def __init__(self, value):
|
|
self.value = value
|
|
|
|
def __await__(self):
|
|
async def get_value():
|
|
return self.value
|
|
return get_value().__await__()
|
|
|
|
class FakeRequest:
|
|
def __init__(self, form, files, body=b'', boundary=None):
|
|
self.form = AwaitableValue(form)
|
|
self.files = AwaitableValue(files)
|
|
self.mimetype = 'multipart/form-data' if boundary else ''
|
|
self.mimetype_params = {'boundary': boundary} if boundary else {}
|
|
self._body = body
|
|
|
|
async def get_data(self, cache=True):
|
|
return self._body
|
|
|
|
class FakeConvex:
|
|
def __init__(self):
|
|
self.sent = []
|
|
self.files = []
|
|
|
|
async def send_to_storage(self, data, content_type):
|
|
self.sent.append((data, content_type))
|
|
return 'storage_1'
|
|
|
|
async def send_stream_to_storage(self, stream, content_type):
|
|
data = stream.read()
|
|
self.sent.append((data, content_type))
|
|
return 'storage_1', len(data)
|
|
|
|
async def add_file(self, **kwargs):
|
|
self.files.append(kwargs)
|
|
return {'file_id': 'file_1'}
|
|
|
|
def test_upload_accepts_file_field_without_filename(monkeypatch):
|
|
async def run_test():
|
|
upload = load_upload_module(monkeypatch)
|
|
|
|
app = Quart(__name__)
|
|
app.secret_key = 'test-secret'
|
|
app.convex = FakeConvex()
|
|
app.orphan_storage_registry = None
|
|
|
|
async with app.test_request_context('/api/upload', method='POST'):
|
|
session['user'] = {'sub': 'user_1'}
|
|
|
|
uploaded = FileStorage(
|
|
stream=BytesIO(b'hello'),
|
|
filename='',
|
|
name='file',
|
|
content_type='application/octet-stream',
|
|
)
|
|
fake_request = FakeRequest(
|
|
form={'expires': '7d', 'note': 'test', 'text': ''},
|
|
files={'file': uploaded},
|
|
)
|
|
monkeypatch.setattr(upload, 'request', fake_request)
|
|
|
|
response = await upload.api_upload()
|
|
|
|
assert await response.get_json() == {'ok': True}
|
|
assert app.convex.sent == [(b'hello', 'application/octet-stream')]
|
|
assert app.convex.files[0]['file_name'].endswith('.bin')
|
|
assert app.convex.files[0]['expires_at'] is not None
|
|
|
|
asyncio.run(run_test())
|
|
|
|
def test_upload_recovers_file_field_from_multipart_body(monkeypatch):
|
|
async def run_test():
|
|
upload = load_upload_module(monkeypatch)
|
|
|
|
app = Quart(__name__)
|
|
app.secret_key = 'test-secret'
|
|
app.convex = FakeConvex()
|
|
app.orphan_storage_registry = None
|
|
|
|
boundary = '----nanoshare-test-boundary'
|
|
body = (
|
|
f'--{boundary}\r\n'
|
|
'Content-Disposition: form-data; name="file"\r\n'
|
|
'Content-Type: application/octet-stream\r\n'
|
|
'\r\n'
|
|
).encode() + b'hello recovered' + (
|
|
'\r\n'
|
|
f'--{boundary}\r\n'
|
|
'Content-Disposition: form-data; name="expires"\r\n'
|
|
'\r\n'
|
|
'7d\r\n'
|
|
f'--{boundary}--\r\n'
|
|
).encode()
|
|
|
|
async with app.test_request_context('/api/upload', method='POST'):
|
|
session['user'] = {'sub': 'user_1'}
|
|
fake_request = FakeRequest(
|
|
form={'expires': '7d', 'note': 'test', 'text': ''},
|
|
files={},
|
|
body=body,
|
|
boundary=boundary,
|
|
)
|
|
monkeypatch.setattr(upload, 'request', fake_request)
|
|
|
|
response = await upload.api_upload()
|
|
|
|
assert await response.get_json() == {'ok': True}
|
|
assert app.convex.sent == [(b'hello recovered', 'application/octet-stream')]
|
|
assert app.convex.files[0]['file_name'].endswith('.bin')
|
|
assert app.convex.files[0]['expires_at'] is not None
|
|
assert app.convex.files[0]['note'] == 'test'
|
|
|
|
asyncio.run(run_test())
|