daniel156161
21aca1b9d7
Build and Push Docker Container / build-and-push (push) Successful in 1m38s
- Treat an existing multipart file field as upload content even when FileStorage has an empty filename. - Keep upload telemetry aligned with file field presence instead of FileStorage truthiness. - Add regression coverage for frontend-style uploads that provide a file field without a filename. - Bump NanoShare to 1.27.2 and refresh the staged lockfile.
96 lines
2.8 KiB
Python
96 lines
2.8 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):
|
|
self.form = AwaitableValue(form)
|
|
self.files = AwaitableValue(files)
|
|
|
|
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')
|
|
|
|
asyncio.run(run_test())
|