API - fix error message on file uplaod + refactor - #72
This commit is contained in:
parent
aa1170f1c7
commit
bb8491f84d
@ -11,6 +11,21 @@ def get_empty_data_for_datatype(data_type: str) -> Union[str, List]:
|
||||
return '' if data_type in ['gpx', 'chart_data'] else []
|
||||
|
||||
|
||||
def display_readable_file_size(size_in_bytes: Union[float, int]) -> str:
|
||||
"""
|
||||
Return readable file size from size in bytes
|
||||
"""
|
||||
if size_in_bytes == 0:
|
||||
return '0 bytes'
|
||||
if size_in_bytes == 1:
|
||||
return '1 byte'
|
||||
for unit in [' bytes', 'KB', 'MB', 'GB', 'TB']:
|
||||
if abs(size_in_bytes) < 1024.0:
|
||||
return f'{size_in_bytes:3.1f}{unit}'
|
||||
size_in_bytes /= 1024.0
|
||||
return f'{size_in_bytes} bytes'
|
||||
|
||||
|
||||
class HttpResponse(Response):
|
||||
def __init__(
|
||||
self,
|
||||
@ -106,7 +121,19 @@ class DataNotFoundErrorResponse(HttpResponse):
|
||||
|
||||
|
||||
class PayloadTooLargeErrorResponse(GenericErrorResponse):
|
||||
def __init__(self, message: str) -> None:
|
||||
def __init__(
|
||||
self, file_type: str, file_size: Optional[int], max_size: Optional[int]
|
||||
) -> None:
|
||||
readable_file_size = (
|
||||
f'({display_readable_file_size(file_size)}) ' if file_size else ''
|
||||
)
|
||||
readable_max_size = (
|
||||
display_readable_file_size(max_size) if max_size else 'limit'
|
||||
)
|
||||
message = (
|
||||
f'Error during {file_type} upload, file size {readable_file_size}'
|
||||
f'exceeds {readable_max_size}.'
|
||||
)
|
||||
super().__init__(status_code=413, message=message, status='fail')
|
||||
|
||||
|
||||
|
@ -278,9 +278,9 @@ class TestUpdateConfig:
|
||||
) in data['message']
|
||||
|
||||
def test_it_raises_error_if_archive_max_size_equals_0(
|
||||
self, app_with_max_single_file_size: Flask, user_1_admin: User
|
||||
self, app_with_max_file_size_equals_0: Flask, user_1_admin: User
|
||||
) -> None:
|
||||
client = app_with_max_single_file_size.test_client()
|
||||
client = app_with_max_file_size_equals_0.test_client()
|
||||
resp_login = client.post(
|
||||
'/api/auth/login',
|
||||
data=json.dumps(
|
||||
|
41
fittrackee/tests/fixtures/fixtures_app.py
vendored
41
fittrackee/tests/fixtures/fixtures_app.py
vendored
@ -1,5 +1,5 @@
|
||||
import os
|
||||
from typing import Generator, Optional
|
||||
from typing import Generator, Optional, Union
|
||||
|
||||
import pytest
|
||||
|
||||
@ -11,17 +11,22 @@ from fittrackee.application.utils import update_app_config_from_database
|
||||
def get_app_config(
|
||||
with_config: Optional[bool] = False,
|
||||
max_workouts: Optional[int] = None,
|
||||
max_single_file_size: Optional[int] = None,
|
||||
max_single_file_size: Optional[Union[int, float]] = None,
|
||||
max_zip_file_size: Optional[Union[int, float]] = None,
|
||||
) -> Optional[AppConfig]:
|
||||
if with_config:
|
||||
config = AppConfig()
|
||||
config.gpx_limit_import = 10 if max_workouts is None else max_workouts
|
||||
config.max_single_file_size = (
|
||||
1 * 1024 * 1024
|
||||
if max_single_file_size is None
|
||||
else max_single_file_size
|
||||
(1 if max_single_file_size is None else max_single_file_size)
|
||||
* 1024
|
||||
* 1024
|
||||
)
|
||||
config.max_zip_file_size = (
|
||||
(10 if max_zip_file_size is None else max_zip_file_size)
|
||||
* 1024
|
||||
* 1024
|
||||
)
|
||||
config.max_zip_file_size = 1 * 1024 * 1024 * 10
|
||||
config.max_users = 100
|
||||
db.session.add(config)
|
||||
db.session.commit()
|
||||
@ -32,13 +37,19 @@ def get_app_config(
|
||||
def get_app(
|
||||
with_config: Optional[bool] = False,
|
||||
max_workouts: Optional[int] = None,
|
||||
max_single_file_size: Optional[int] = None,
|
||||
max_single_file_size: Optional[Union[int, float]] = None,
|
||||
max_zip_file_size: Optional[Union[int, float]] = None,
|
||||
) -> Generator:
|
||||
app = create_app()
|
||||
with app.app_context():
|
||||
try:
|
||||
db.create_all()
|
||||
app_db_config = get_app_config(with_config, max_workouts)
|
||||
app_db_config = get_app_config(
|
||||
with_config,
|
||||
max_workouts,
|
||||
max_single_file_size,
|
||||
max_zip_file_size,
|
||||
)
|
||||
if app_db_config:
|
||||
update_app_config_from_database(app, app_db_config)
|
||||
yield app
|
||||
@ -71,13 +82,25 @@ def app_with_max_workouts(monkeypatch: pytest.MonkeyPatch) -> Generator:
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def app_with_max_single_file_size(
|
||||
def app_with_max_file_size_equals_0(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> Generator:
|
||||
monkeypatch.setenv('EMAIL_URL', 'smtp://none:none@0.0.0.0:1025')
|
||||
yield from get_app(with_config=True, max_single_file_size=0)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def app_with_max_file_size(monkeypatch: pytest.MonkeyPatch) -> Generator:
|
||||
monkeypatch.setenv('EMAIL_URL', 'smtp://none:none@0.0.0.0:1025')
|
||||
yield from get_app(with_config=True, max_single_file_size=0.001)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def app_with_max_zip_file_size(monkeypatch: pytest.MonkeyPatch) -> Generator:
|
||||
monkeypatch.setenv('EMAIL_URL', 'smtp://none:none@0.0.0.0:1025')
|
||||
yield from get_app(with_config=True, max_zip_file_size=0.001)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def app_no_config() -> Generator:
|
||||
yield from get_app(with_config=False)
|
||||
|
@ -3,10 +3,8 @@ from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
|
||||
from fittrackee.users.utils import (
|
||||
display_readable_file_size,
|
||||
get_readable_duration,
|
||||
)
|
||||
from fittrackee.responses import display_readable_file_size
|
||||
from fittrackee.utils import get_readable_duration
|
||||
|
||||
|
||||
class TestDisplayReadableFileSize:
|
@ -851,6 +851,76 @@ class TestUserPicture:
|
||||
assert data['message'] == 'File extension not allowed.'
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_it_returns_error_if_image_size_exceeds_file_limit(
|
||||
self,
|
||||
app_with_max_file_size: Flask,
|
||||
user_1: User,
|
||||
sport_1_cycling: Sport,
|
||||
gpx_file: str,
|
||||
) -> None:
|
||||
client = app_with_max_file_size.test_client()
|
||||
resp_login = client.post(
|
||||
'/api/auth/login',
|
||||
data=json.dumps(dict(email='test@test.com', password='12345678')),
|
||||
content_type='application/json',
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
'/api/auth/picture',
|
||||
data=dict(
|
||||
file=(BytesIO(b'test_file_for_avatar' * 50), 'avatar.jpg')
|
||||
),
|
||||
headers=dict(
|
||||
content_type='multipart/form-data',
|
||||
authorization='Bearer '
|
||||
+ json.loads(resp_login.data.decode())['auth_token'],
|
||||
),
|
||||
)
|
||||
data = json.loads(response.data.decode())
|
||||
print('data', data)
|
||||
assert response.status_code == 413
|
||||
assert 'fail' in data['status']
|
||||
assert (
|
||||
'Error during picture upload, file size (1.2KB) exceeds 1.0KB.'
|
||||
in data['message']
|
||||
)
|
||||
assert 'data' not in data
|
||||
|
||||
def test_it_returns_error_if_image_size_exceeds_archive_limit(
|
||||
self,
|
||||
app_with_max_zip_file_size: Flask,
|
||||
user_1: User,
|
||||
sport_1_cycling: Sport,
|
||||
gpx_file: str,
|
||||
) -> None:
|
||||
client = app_with_max_zip_file_size.test_client()
|
||||
resp_login = client.post(
|
||||
'/api/auth/login',
|
||||
data=json.dumps(dict(email='test@test.com', password='12345678')),
|
||||
content_type='application/json',
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
'/api/auth/picture',
|
||||
data=dict(
|
||||
file=(BytesIO(b'test_file_for_avatar' * 50), 'avatar.jpg')
|
||||
),
|
||||
headers=dict(
|
||||
content_type='multipart/form-data',
|
||||
authorization='Bearer '
|
||||
+ json.loads(resp_login.data.decode())['auth_token'],
|
||||
),
|
||||
)
|
||||
data = json.loads(response.data.decode())
|
||||
print('data', data)
|
||||
assert response.status_code == 413
|
||||
assert 'fail' in data['status']
|
||||
assert (
|
||||
'Error during picture upload, file size (1.2KB) exceeds 1.0KB.'
|
||||
in data['message']
|
||||
)
|
||||
assert 'data' not in data
|
||||
|
||||
|
||||
class TestRegistrationConfiguration:
|
||||
def test_it_returns_error_if_it_exceeds_max_users(
|
||||
|
@ -514,6 +514,41 @@ class TestPostWorkoutWithGpx:
|
||||
assert data['status'] == 'fail'
|
||||
assert data['message'] == 'No file part.'
|
||||
|
||||
def test_it_returns_error_if_file_size_exceeds_limit(
|
||||
self,
|
||||
app_with_max_file_size: Flask,
|
||||
user_1: User,
|
||||
sport_1_cycling: Sport,
|
||||
gpx_file: str,
|
||||
) -> None:
|
||||
client = app_with_max_file_size.test_client()
|
||||
resp_login = client.post(
|
||||
'/api/auth/login',
|
||||
data=json.dumps(dict(email='test@test.com', password='12345678')),
|
||||
content_type='application/json',
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
'/api/workouts',
|
||||
data=dict(
|
||||
file=(BytesIO(str.encode(gpx_file)), 'example.gpx'),
|
||||
data='{"sport_id": 1}',
|
||||
),
|
||||
headers=dict(
|
||||
content_type='multipart/form-data',
|
||||
Authorization='Bearer '
|
||||
+ json.loads(resp_login.data.decode())['auth_token'],
|
||||
),
|
||||
)
|
||||
data = json.loads(response.data.decode())
|
||||
assert response.status_code == 413
|
||||
assert 'fail' in data['status']
|
||||
assert (
|
||||
'Error during workout upload, file size (3.6KB) exceeds 1.0KB.'
|
||||
in data['message']
|
||||
)
|
||||
assert 'data' not in data
|
||||
|
||||
|
||||
class TestPostWorkoutWithoutGpx:
|
||||
def test_it_adds_an_workout_without_gpx(
|
||||
@ -815,6 +850,46 @@ class TestPostWorkoutWithZipArchive:
|
||||
data = json.loads(response.data.decode())
|
||||
assert len(data['data']['workouts']) == 2
|
||||
|
||||
def test_it_returns_error_if_archive_size_exceeds_limit(
|
||||
self,
|
||||
app_with_max_zip_file_size: Flask,
|
||||
user_1: User,
|
||||
sport_1_cycling: Sport,
|
||||
) -> None:
|
||||
file_path = os.path.join(
|
||||
app_with_max_zip_file_size.root_path, 'tests/files/gpx_test.zip'
|
||||
)
|
||||
# 'gpx_test.zip' contains 3 gpx files (same data) and 1 non-gpx file
|
||||
with open(file_path, 'rb') as zip_file:
|
||||
client = app_with_max_zip_file_size.test_client()
|
||||
resp_login = client.post(
|
||||
'/api/auth/login',
|
||||
data=json.dumps(
|
||||
dict(email='test@test.com', password='12345678')
|
||||
),
|
||||
content_type='application/json',
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
'/api/workouts',
|
||||
data=dict(
|
||||
file=(zip_file, 'gpx_test.zip'), data='{"sport_id": 1}'
|
||||
),
|
||||
headers=dict(
|
||||
content_type='multipart/form-data',
|
||||
Authorization='Bearer '
|
||||
+ json.loads(resp_login.data.decode())['auth_token'],
|
||||
),
|
||||
)
|
||||
data = json.loads(response.data.decode())
|
||||
assert response.status_code == 413
|
||||
assert 'fail' in data['status']
|
||||
assert (
|
||||
'Error during workout upload, file size (2.5KB) exceeds 1.0KB.'
|
||||
in data['message']
|
||||
)
|
||||
assert 'data' not in data
|
||||
|
||||
|
||||
class TestPostAndGetWorkoutWithGpx:
|
||||
@staticmethod
|
||||
|
@ -18,17 +18,12 @@ from fittrackee.responses import (
|
||||
handle_error_and_return_response,
|
||||
)
|
||||
from fittrackee.tasks import reset_password_email
|
||||
from fittrackee.utils import get_readable_duration, verify_extension_and_size
|
||||
from fittrackee.workouts.utils_files import get_absolute_file_path
|
||||
|
||||
from .decorators import authenticate
|
||||
from .models import User
|
||||
from .utils import (
|
||||
check_passwords,
|
||||
display_readable_file_size,
|
||||
get_readable_duration,
|
||||
register_controls,
|
||||
verify_extension_and_size,
|
||||
)
|
||||
from .utils import check_passwords, register_controls
|
||||
from .utils_token import decode_user_token
|
||||
|
||||
auth_blueprint = Blueprint('auth', __name__)
|
||||
@ -524,10 +519,10 @@ def edit_picture(auth_user_id: int) -> Union[Dict, HttpResponse]:
|
||||
response_object = verify_extension_and_size('picture', request)
|
||||
except RequestEntityTooLarge as e:
|
||||
appLog.error(e)
|
||||
max_file_size = current_app.config['MAX_CONTENT_LENGTH']
|
||||
return PayloadTooLargeErrorResponse(
|
||||
'Error during picture update, file size exceeds '
|
||||
f'{display_readable_file_size(max_file_size)}.'
|
||||
file_type='picture',
|
||||
file_size=request.content_length,
|
||||
max_size=current_app.config['MAX_CONTENT_LENGTH'],
|
||||
)
|
||||
if response_object:
|
||||
return response_object
|
||||
|
@ -1,15 +1,11 @@
|
||||
import re
|
||||
from datetime import timedelta
|
||||
from typing import Optional, Tuple, Union
|
||||
from typing import Optional, Tuple
|
||||
|
||||
import humanize
|
||||
from flask import Request, current_app
|
||||
from flask import Request
|
||||
|
||||
from fittrackee.responses import (
|
||||
ForbiddenErrorResponse,
|
||||
HttpResponse,
|
||||
InvalidPayloadErrorResponse,
|
||||
PayloadTooLargeErrorResponse,
|
||||
UnauthorizedErrorResponse,
|
||||
)
|
||||
|
||||
@ -64,49 +60,6 @@ def register_controls(
|
||||
return ret
|
||||
|
||||
|
||||
def verify_extension_and_size(
|
||||
file_type: str, req: Request
|
||||
) -> Optional[HttpResponse]:
|
||||
"""
|
||||
Return error Response if file is invalid
|
||||
"""
|
||||
if 'file' not in req.files:
|
||||
return InvalidPayloadErrorResponse('No file part.', 'fail')
|
||||
|
||||
file = req.files['file']
|
||||
if file.filename == '':
|
||||
return InvalidPayloadErrorResponse('No selected file.', 'fail')
|
||||
|
||||
allowed_extensions = (
|
||||
'WORKOUT_ALLOWED_EXTENSIONS'
|
||||
if file_type == 'workout'
|
||||
else 'PICTURE_ALLOWED_EXTENSIONS'
|
||||
)
|
||||
|
||||
file_extension = (
|
||||
file.filename.rsplit('.', 1)[1].lower()
|
||||
if '.' in file.filename
|
||||
else None
|
||||
)
|
||||
max_file_size = current_app.config['max_single_file_size']
|
||||
|
||||
if not (
|
||||
file_extension
|
||||
and file_extension in current_app.config[allowed_extensions]
|
||||
):
|
||||
return InvalidPayloadErrorResponse(
|
||||
'File extension not allowed.', 'fail'
|
||||
)
|
||||
|
||||
if file_extension != 'zip' and req.content_length > max_file_size:
|
||||
return PayloadTooLargeErrorResponse(
|
||||
'Error during picture update, file size exceeds '
|
||||
f'{display_readable_file_size(max_file_size)}.'
|
||||
)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def verify_user(
|
||||
current_request: Request, verify_admin: bool
|
||||
) -> Tuple[Optional[HttpResponse], Optional[int]]:
|
||||
@ -139,35 +92,3 @@ def can_view_workout(
|
||||
if auth_user_id != workout_user_id:
|
||||
return ForbiddenErrorResponse()
|
||||
return None
|
||||
|
||||
|
||||
def display_readable_file_size(size_in_bytes: Union[float, int]) -> str:
|
||||
"""
|
||||
Return readable file size from size in bytes
|
||||
"""
|
||||
if size_in_bytes == 0:
|
||||
return '0 bytes'
|
||||
if size_in_bytes == 1:
|
||||
return '1 byte'
|
||||
for unit in [' bytes', 'KB', 'MB', 'GB', 'TB']:
|
||||
if abs(size_in_bytes) < 1024.0:
|
||||
return f'{size_in_bytes:3.1f}{unit}'
|
||||
size_in_bytes /= 1024.0
|
||||
return f'{size_in_bytes} bytes'
|
||||
|
||||
|
||||
def get_readable_duration(duration: int, locale: Optional[str] = None) -> str:
|
||||
"""
|
||||
Return readable and localized duration from duration in seconds
|
||||
"""
|
||||
if locale is None:
|
||||
locale = 'en'
|
||||
if locale != 'en':
|
||||
try:
|
||||
_t = humanize.i18n.activate(locale) # noqa
|
||||
except FileNotFoundError:
|
||||
locale = 'en'
|
||||
readable_duration = humanize.naturaldelta(timedelta(seconds=duration))
|
||||
if locale != 'en':
|
||||
humanize.i18n.deactivate()
|
||||
return readable_duration
|
||||
|
76
fittrackee/utils.py
Normal file
76
fittrackee/utils.py
Normal file
@ -0,0 +1,76 @@
|
||||
from datetime import timedelta
|
||||
from typing import Optional
|
||||
|
||||
import humanize
|
||||
from flask import Request, current_app
|
||||
|
||||
from .responses import (
|
||||
HttpResponse,
|
||||
InvalidPayloadErrorResponse,
|
||||
PayloadTooLargeErrorResponse,
|
||||
)
|
||||
|
||||
|
||||
def verify_extension_and_size(
|
||||
file_type: str, req: Request
|
||||
) -> Optional[HttpResponse]:
|
||||
"""
|
||||
Return error Response if file is invalid
|
||||
"""
|
||||
if 'file' not in req.files:
|
||||
return InvalidPayloadErrorResponse('No file part.', 'fail')
|
||||
|
||||
file = req.files['file']
|
||||
if file.filename == '':
|
||||
return InvalidPayloadErrorResponse('No selected file.', 'fail')
|
||||
|
||||
allowed_extensions = (
|
||||
'WORKOUT_ALLOWED_EXTENSIONS'
|
||||
if file_type == 'workout'
|
||||
else 'PICTURE_ALLOWED_EXTENSIONS'
|
||||
)
|
||||
|
||||
file_extension = (
|
||||
file.filename.rsplit('.', 1)[1].lower()
|
||||
if '.' in file.filename
|
||||
else None
|
||||
)
|
||||
max_file_size = current_app.config['max_single_file_size']
|
||||
|
||||
if not (
|
||||
file_extension
|
||||
and file_extension in current_app.config[allowed_extensions]
|
||||
):
|
||||
return InvalidPayloadErrorResponse(
|
||||
'File extension not allowed.', 'fail'
|
||||
)
|
||||
|
||||
if (
|
||||
file_extension != 'zip'
|
||||
and req.content_length is not None
|
||||
and req.content_length > max_file_size
|
||||
):
|
||||
return PayloadTooLargeErrorResponse(
|
||||
file_type=file_type,
|
||||
file_size=req.content_length,
|
||||
max_size=max_file_size,
|
||||
)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def get_readable_duration(duration: int, locale: Optional[str] = None) -> str:
|
||||
"""
|
||||
Return readable and localized duration from duration in seconds
|
||||
"""
|
||||
if locale is None:
|
||||
locale = 'en'
|
||||
if locale != 'en':
|
||||
try:
|
||||
_t = humanize.i18n.activate(locale) # noqa
|
||||
except FileNotFoundError:
|
||||
locale = 'en'
|
||||
readable_duration = humanize.naturaldelta(timedelta(seconds=duration))
|
||||
if locale != 'en':
|
||||
humanize.i18n.deactivate()
|
||||
return readable_duration
|
@ -7,6 +7,7 @@ from typing import Any, Dict, List, Optional, Tuple, Union
|
||||
import requests
|
||||
from flask import Blueprint, Response, current_app, request, send_file
|
||||
from sqlalchemy import exc
|
||||
from werkzeug.exceptions import RequestEntityTooLarge
|
||||
|
||||
from fittrackee import appLog, db
|
||||
from fittrackee.responses import (
|
||||
@ -16,14 +17,13 @@ from fittrackee.responses import (
|
||||
InternalServerErrorResponse,
|
||||
InvalidPayloadErrorResponse,
|
||||
NotFoundErrorResponse,
|
||||
PayloadTooLargeErrorResponse,
|
||||
handle_error_and_return_response,
|
||||
)
|
||||
from fittrackee.users.decorators import authenticate
|
||||
from fittrackee.users.utils import (
|
||||
User,
|
||||
can_view_workout,
|
||||
verify_extension_and_size,
|
||||
)
|
||||
from fittrackee.users.models import User
|
||||
from fittrackee.users.utils import can_view_workout
|
||||
from fittrackee.utils import verify_extension_and_size
|
||||
|
||||
from .models import Workout
|
||||
from .utils import (
|
||||
@ -880,7 +880,15 @@ def post_workout(auth_user_id: int) -> Union[Tuple[Dict, int], HttpResponse]:
|
||||
:statuscode 500:
|
||||
|
||||
"""
|
||||
error_response = verify_extension_and_size('workout', request)
|
||||
try:
|
||||
error_response = verify_extension_and_size('workout', request)
|
||||
except RequestEntityTooLarge as e:
|
||||
appLog.error(e)
|
||||
return PayloadTooLargeErrorResponse(
|
||||
file_type='workout',
|
||||
file_size=request.content_length,
|
||||
max_size=current_app.config['MAX_CONTENT_LENGTH'],
|
||||
)
|
||||
if error_response:
|
||||
return error_response
|
||||
|
||||
|
Loading…
Reference in New Issue
Block a user