Alexander/FitTrackee
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

API: update an activity

Browse Source
This commit is contained in:
Sam 2018-05-09 20:45:01 +02:00
parent 608319386b
commit 32777061ac
3 changed files with 294 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
mpwo_api/mpwo_api/activities/activities.py
View File

@ -8,8 +8,8 @@ from sqlalchemy import exc
from ..users.utils import authenticate, verify_extension
from .models import Activity
from .utils import (
create_activity_with_gpx, create_activity_wo_gpx, get_file_path,
get_gpx_info
create_activity_with_gpx, create_activity_wo_gpx, edit_activity_wo_gpx,
get_file_path, get_gpx_info
)
activities_blueprint = Blueprint('activities', __name__)
@ -198,6 +198,61 @@ def post_activity_no_gpx(auth_user_id):
return jsonify(response_object), 500
@activities_blueprint.route('/activities/<int:activity_id>', methods=['PATCH'])
@authenticate
def update_activity(auth_user_id, activity_id):
"""Update an activity"""
activity_data = request.get_json()
if not activity_data or activity_data.get('sport_id') is None \
or activity_data.get('duration') is None \
or activity_data.get('distance') is None \
or activity_data.get('activity_date') is None:
response_object = {
'status': 'error',
'message': 'Invalid payload.'
}
return jsonify(response_object), 400
try:
activity = Activity.query.filter_by(id=activity_id).first()
if activity:
if activity.gpx:
response_object = {
'status': 'error',
'message': 'You can not modify an activity with gpx file. '
'Please delete and re-import the gpx file.'
}
code = 500
return jsonify(response_object), code
activity = edit_activity_wo_gpx(activity, activity_data)
db.session.commit()
response_object = {
'status': 'success',
'data': {
'activities': [activity.serialize()]
}
}
code = 200
else:
response_object = {
'status': 'not found',
'data': {
'activities': []
}
}
code = 404
except (exc.IntegrityError, exc.OperationalError, ValueError) 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
@activities_blueprint.route(
'/activities/<int:activity_id>', methods=['DELETE']
)

13
mpwo_api/mpwo_api/activities/utils.py
View File

@ -44,6 +44,19 @@ def create_activity_wo_gpx(auth_user_id, activity_data):
return new_activity
def edit_activity_wo_gpx(activity, activity_data):
activity.sport_id = activity_data.get('sport_id')
activity.activity_date = datetime.strptime(
activity_data.get('activity_date'), '%Y-%m-%d %H:%M')
activity.duration = timedelta(seconds=activity_data.get('duration'))
activity.moving = activity.duration
activity.distance = activity_data.get('distance')
activity.ave_speed = activity.distance / (
activity.duration.seconds / 3600)
activity.max_speed = activity.ave_speed
return activity
def get_gpx_info(gpx_file):
gpx_data = {'filename': gpx_file}

224
mpwo_api/mpwo_api/tests/test_activities.py
View File

@ -478,6 +478,230 @@ def test_get_an_activity_with_gpx(app):
assert data['data']['activities'][0]['with_gpx'] is True
def test_edit_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.patch(
'/api/activities/1',
content_type='application/json',
data=json.dumps(dict(
sport_id=1,
duration=3600,
activity_date='2018-05-15 14:05',
distance=10
)),
headers=dict(
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 200
assert 'success' in data['status']
assert len(data['data']['activities']) == 1
assert 'creation_date' in data['data']['activities'][0]
assert data['data']['activities'][0]['activity_date'] == 'Tue, 15 May 2018 14:05:00 GMT' # noqa
assert data['data']['activities'][0]['user_id'] == 1
assert data['data']['activities'][0]['sport_id'] == 1
assert data['data']['activities'][0]['duration'] == '1:00:00'
assert data['data']['activities'][0]['ascent'] is None
assert data['data']['activities'][0]['ave_speed'] == 10.0
assert data['data']['activities'][0]['descent'] is None
assert data['data']['activities'][0]['distance'] == 10.0
assert data['data']['activities'][0]['max_alt'] is None
assert data['data']['activities'][0]['max_speed'] == 10.0
assert data['data']['activities'][0]['min_alt'] is None
assert data['data']['activities'][0]['moving'] == '1:00:00'
assert data['data']['activities'][0]['pauses'] is None
assert data['data']['activities'][0]['with_gpx'] is False
def test_edit_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.patch(
'/api/activities/1',
content_type='application/json',
data=json.dumps(dict(
sport_id=1,
duration=3600,
activity_date='2018-05-15 14:05',
distance=10
)),
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 ('You can not modify an activity with gpx file. '
'Please delete and re-import the gpx file.') in data['message']
def test_edit_an_activity_wo_gpx_invalid_payload(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.patch(
'/api/activities/1',
content_type='application/json',
data=json.dumps(dict(
sport_id=1,
activity_date='2018-05-15 14:05',
distance=10
)),
headers=dict(
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 400
assert 'error' in data['status']
assert 'Invalid payload.' in data['message']
def test_edit_an_activity_wo_gpx_incorrect_data(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.patch(
'/api/activities/1',
content_type='application/json',
data=json.dumps(dict(
sport_id=1,
duration=3600,
activity_date='15/2018',
distance=10
)),
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']
def test_edit_an_activity_no_activity(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'
)
response = client.patch(
'/api/activities/1',
content_type='application/json',
data=json.dumps(dict(
sport_id=1,
duration=3600,
activity_date='2018-05-15 14:05',
distance=10
)),
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']
assert len(data['data']['activities']) == 0
def test_delete_an_activity_with_gpx(app):
add_user('test', 'test@test.com', '12345678')
add_sport('cycling')