API: delete an activity

This commit is contained in:
Sam 2018-05-09 18:54:30 +02:00
parent 8d65b91022
commit 58b69878c4
4 changed files with 194 additions and 5 deletions

View File

@ -1,4 +1,5 @@
import json
import os
from flask import Blueprint, jsonify, request
from mpwo_api import appLog, db
@ -57,8 +58,7 @@ def get_activity(auth_user_id, activity_id):
@activities_blueprint.route(
'/activities/<int:activity_id>/gpx',
methods=['GET']
'/activities/<int:activity_id>/gpx', methods=['GET']
)
@authenticate
def get_activity_gpx(auth_user_id, activity_id):
@ -196,3 +196,43 @@ def post_activity_no_gpx(auth_user_id):
'message': 'Error during activity save.'
}
return jsonify(response_object), 500
@activities_blueprint.route(
'/activities/<int:activity_id>', methods=['DELETE']
)
@authenticate
def delete_activity(auth_user_id, activity_id):
"""Delete an activity"""
try:
activity = Activity.query.filter_by(id=activity_id).first()
if activity:
gpx_filepath = activity.gpx
db.session.delete(activity)
db.session.commit()
if gpx_filepath:
os.remove(gpx_filepath)
response_object = {
'status': 'no content'
}
code = 204
else:
response_object = {
'status': 'not found',
'data': {
'activities': []
}
}
code = 404
except (exc.IntegrityError, exc.OperationalError, ValueError, OSError) \
as e:
db.session.rollback()
appLog.error(e)
response_object = {
'status': 'error',
'message': 'Error. Please try again or contact the administrator.'
}
code = 500
return jsonify(response_object), code

View File

@ -140,7 +140,6 @@ def update_sport(auth_user_id, sport_id):
@authenticate_as_admin
def delete_sport(auth_user_id, sport_id):
"""Delete a sport"""
sports_list = []
try:
sport = Sport.query.filter_by(id=sport_id).first()
if sport:
@ -154,7 +153,7 @@ def delete_sport(auth_user_id, sport_id):
response_object = {
'status': 'not found',
'data': {
'sports': sports_list
'sports': []
}
}
code = 404

View File

@ -1,8 +1,11 @@
import datetime
import json
import os
from io import BytesIO
from mpwo_api.tests.utils import add_activity, add_sport, add_user
from mpwo_api.tests.utils import (
add_activity, add_sport, add_user, get_gpx_filepath
)
from mpwo_api.tests.utils_gpx import gpx_file
@ -473,3 +476,145 @@ def test_get_an_activity_with_gpx(app):
assert data['data']['activities'][0]['moving'] == '0:04:10'
assert data['data']['activities'][0]['pauses'] is None
assert data['data']['activities'][0]['with_gpx'] is True
def test_delete_an_activity_with_gpx(app):
add_user('test', 'test@test.com', '12345678')
add_sport('cycling')
client = app.test_client()
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='test@test.com',
password='12345678'
)),
content_type='application/json'
)
client.post(
'/api/activities',
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']
)
)
response = client.delete(
'/api/activities/1',
headers=dict(
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
assert response.status_code == 204
def test_delete_an_activity_wo_gpx(app):
add_user('test', 'test@test.com', '12345678')
add_sport('cycling')
add_activity(
1,
1,
datetime.datetime.strptime('01/01/2018', '%d/%m/%Y'),
datetime.timedelta(seconds=1024))
client = app.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.delete(
'/api/activities/1',
headers=dict(
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
assert response.status_code == 204
def test_delete_an_activity_no_activityy(app):
add_user('test', 'test@test.com', '12345678')
client = app.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.delete(
'/api/activities/9999',
headers=dict(
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 404
assert 'not found' in data['status']
def test_delete_an_activity_with_gpx_invalid_file(app):
add_user('test', 'test@test.com', '12345678')
add_sport('cycling')
client = app.test_client()
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='test@test.com',
password='12345678'
)),
content_type='application/json'
)
client.post(
'/api/activities',
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']
)
)
gpx_filepath = get_gpx_filepath(1)
os.remove(gpx_filepath)
response = client.delete(
'/api/activities/1',
headers=dict(
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 500
assert 'error' in data['status']
assert 'Error. Please try again or contact the administrator.' \
in data['message']

View File

@ -52,3 +52,8 @@ def add_activity(user_id, sport_id, activity_date, duration):
db.session.add(activity)
db.session.commit()
return activity
def get_gpx_filepath(activity_id):
activity = Activity.query.filter_by(id=activity_id).first()
return activity.gpx