API: only activity owner can view his activity

This commit is contained in:
Sam 2018-06-15 10:50:37 +02:00
parent 2a5070aa8a
commit ac7c7a41eb
6 changed files with 360 additions and 3 deletions

View File

@ -7,7 +7,9 @@ from fittrackee_api import appLog, db
from flask import Blueprint, current_app, jsonify, request, send_file
from sqlalchemy import exc
from ..users.utils import User, authenticate, verify_extension
from ..users.utils import (
User, authenticate, verify_extension, can_view_activity
)
from .models import Activity
from .utils import (
ActivityException, create_activity, edit_activity, get_chart_data,
@ -99,6 +101,11 @@ def get_activity(auth_user_id, activity_id):
activities_list = []
if activity:
response_object, code = can_view_activity(
auth_user_id, activity.user_id)
if response_object:
return jsonify(response_object), code
activities_list.append(activity.serialize())
status = 'success'
code = 200
@ -120,6 +127,10 @@ def get_activity_data(auth_user_id, activity_id, data_type):
activity = Activity.query.filter_by(id=activity_id).first()
content = ''
if activity:
response_object, code = can_view_activity(
auth_user_id, activity.user_id)
if response_object:
return jsonify(response_object), code
if not activity.gpx or activity.gpx == '':
response_object = {
'status': 'fail',
@ -313,6 +324,11 @@ def update_activity(auth_user_id, activity_id):
try:
activity = Activity.query.filter_by(id=activity_id).first()
if activity:
response_object, code = can_view_activity(
auth_user_id, activity.user_id)
if response_object:
return jsonify(response_object), code
activity = edit_activity(activity, activity_data, auth_user_id)
db.session.commit()
response_object = {
@ -350,6 +366,11 @@ def delete_activity(auth_user_id, activity_id):
try:
activity = Activity.query.filter_by(id=activity_id).first()
if activity:
response_object, code = can_view_activity(
auth_user_id, activity.user_id)
if response_object:
return jsonify(response_object), code
db.session.delete(activity)
db.session.commit()
response_object = {

View File

@ -582,6 +582,34 @@ def test_get_an_activity(
assert '0:17:04' == data['data']['activities'][0]['duration']
def test_get_an_activity_different_user(
app, user_1, user_2, sport_1_cycling, activity_cycling_user_2
):
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.get(
'/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 == 403
assert 'error' in data['status']
assert 'You do not have permissions.' in data['message']
def test_get_activities_per_page(
app, user_1, sport_1_cycling, seven_activities_user_1
):
@ -652,7 +680,7 @@ def test_get_an_activity_invalid_id(app, user_1):
assert len(data['data']['activities']) == 0
def test_get_an_activity_no_actvity_no_gpx(app, user_1):
def test_get_an_activity_no_activity_no_gpx(app, user_1):
client = app.test_client()
resp_login = client.post(
'/api/auth/login',

View File

@ -236,6 +236,67 @@ def test_get_an_activity_with_gpx(app, user_1, sport_1_cycling, gpx_file):
assert data['message'] == 'internal error.'
def test_get_an_activity_with_gpx_different_user(
app, user_1, user_2, sport_1_cycling, gpx_file):
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']
)
)
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='toto@toto.com',
password='87654321'
)),
content_type='application/json'
)
response = client.get(
'/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 == 403
assert 'error' in data['status']
assert 'You do not have permissions.' in data['message']
response = client.get(
'/api/activities/1/gpx',
headers=dict(
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 403
assert 'error' in data['status']
assert 'You do not have permissions.' in data['message']
def test_get_chart_data_activty_with_gpx(
app, user_1, sport_1_cycling, gpx_file
):
@ -277,6 +338,54 @@ def test_get_chart_data_activty_with_gpx(
assert data['data']['chart_data'] != ''
def test_get_chart_data_activty_with_gpx_different_user(
app, user_1, user_2, sport_1_cycling, gpx_file
):
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']
)
)
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='toto@toto.com',
password='87654321'
)),
content_type='application/json'
)
response = client.get(
'/api/activities/1/chart_data',
headers=dict(
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 403
assert 'error' in data['status']
assert 'You do not have permissions.' in data['message']
def test_add_an_activity_with_gpx_without_name(
app, user_1, sport_1_cycling, gpx_file_wo_name
):

View File

@ -110,6 +110,59 @@ def test_edit_an_activity_with_gpx(
assert data['data']['activities'][0]['notes'] == 'test notes'
def test_edit_an_activity_with_gpx_different_user(
app, user_1, user_2, sport_1_cycling, sport_2_running, gpx_file
):
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']
)
)
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='toto@toto.com',
password='87654321'
)),
content_type='application/json'
)
response = client.patch(
'/api/activities/1',
content_type='application/json',
data=json.dumps(dict(
sport_id=2,
title="Activity test",
)),
headers=dict(
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 403
assert 'error' in data['status']
assert 'You do not have permissions.' in data['message']
def test_edit_an_activity_with_gpx_partial(
app, user_1, sport_1_cycling, sport_2_running, gpx_file
):
@ -451,6 +504,66 @@ def test_edit_an_activity_wo_gpx(
assert records[3]['value'] == 8.0
def test_edit_an_activity_wo_gpx_different_user(
app, user_1, user_2, sport_1_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/no_gpx',
content_type='application/json',
data=json.dumps(dict(
sport_id=1,
duration=3600,
activity_date='2018-05-14 14:05',
distance=7,
title='Activity test'
)),
headers=dict(
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='toto@toto.com',
password='87654321'
)),
content_type='application/json'
)
response = client.patch(
'/api/activities/1',
content_type='application/json',
data=json.dumps(dict(
sport_id=2,
duration=3600,
activity_date='2018-05-15 15:05',
distance=8,
title='Activity test'
)),
headers=dict(
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 403
assert 'error' in data['status']
assert 'You do not have permissions.' in data['message']
def test_edit_an_activity_wo_gpx_timezone(
app, user_1_paris, sport_1_cycling, sport_2_running
):

View File

@ -45,6 +45,54 @@ def test_delete_an_activity_with_gpx(app, user_1, sport_1_cycling, gpx_file):
assert response.status_code == 204
def test_delete_an_activity_with_gpx_different_user(
app, user_1, user_2, sport_1_cycling, gpx_file):
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']
)
)
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='toto@toto.com',
password='87654321'
)),
content_type='application/json'
)
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 == 403
assert 'error' in data['status']
assert 'You do not have permissions.' in data['message']
def test_delete_an_activity_wo_gpx(
app, user_1, sport_1_cycling, activity_cycling_user_1
):
@ -68,7 +116,35 @@ def test_delete_an_activity_wo_gpx(
assert response.status_code == 204
def test_delete_an_activity_no_activityy(app, user_1):
def test_delete_an_activity_wo_gpx_different_user(
app, user_1, user_2, sport_1_cycling, activity_cycling_user_1
):
client = app.test_client()
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='toto@toto.com',
password='87654321'
)),
content_type='application/json'
)
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 == 403
assert 'error' in data['status']
assert 'You do not have permissions.' in data['message']
def test_delete_an_activity_no_activity(app, user_1):
client = app.test_client()
resp_login = client.post(
'/api/auth/login',

View File

@ -104,3 +104,13 @@ def authenticate_as_admin(f):
return f(resp, *args, **kwargs)
return decorated_function
def can_view_activity(auth_user_id, activity_user_id):
if auth_user_id != activity_user_id:
response_object = {
'status': 'error',
'message': 'You do not have permissions.'
}
return response_object, 403
return None, None