API & Client - user can delete his account - fix #17

fittrackee_api/fittrackee_api/tests/conftest.py

@@ -117,6 +117,15 @@ def user_2():
     return user
 
 
+@pytest.fixture()
+def user_2_admin():
+    user = User(username='toto', email='toto@toto.com', password='87654321')
+    user.admin = True
+    db.session.add(user)
+    db.session.commit()
+    return user
+
+
 @pytest.fixture()
 def user_3():
     user = User(username='sam', email='sam@test.com', password='12345678')

fittrackee_api/fittrackee_api/tests/test_users_api.py

@@ -1,4 +1,5 @@
 import json
+from io import BytesIO
 
 from fittrackee_api.users.models import User
 
@@ -409,3 +410,186 @@ def test_it_returns_error_if_user_can_not_change_admin_rights(
     assert response.status_code == 403
     assert 'error' in data['status']
     assert 'You do not have permissions.' in data['message']
+
+
+def test_user_can_delete_its_own_account(app, user_1):
+    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/users/test',
+        headers=dict(
+            Authorization='Bearer '
+            + json.loads(resp_login.data.decode())['auth_token']
+        ),
+    )
+
+    assert response.status_code == 204
+
+
+def test_user_with_activity_can_delete_its_own_account(
+    app, user_1, 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'],
+        ),
+    )
+    response = client.delete(
+        '/api/users/test',
+        headers=dict(
+            Authorization='Bearer '
+            + json.loads(resp_login.data.decode())['auth_token']
+        ),
+    )
+
+    assert response.status_code == 204
+
+
+def test_user_with_picture_can_delete_its_own_account(
+    app, user_1, 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/auth/picture',
+        data=dict(file=(BytesIO(b'avatar'), 'avatar.png')),
+        headers=dict(
+            content_type='multipart/form-data',
+            authorization='Bearer '
+            + json.loads(resp_login.data.decode())['auth_token'],
+        ),
+    )
+    response = client.delete(
+        '/api/users/test',
+        headers=dict(
+            Authorization='Bearer '
+            + json.loads(resp_login.data.decode())['auth_token']
+        ),
+    )
+
+    assert response.status_code == 204
+
+
+def test_user_can_not_delete_another_user_account(app, user_1, 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.delete(
+        '/api/users/toto',
+        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_it_returns_error_when_deleting_non_existing_user(app, user_1):
+    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/users/not_existing',
+        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 'User does not exist.' in data['message']
+
+
+def test_admin_can_delete_another_user_account(app, user_1_admin, user_2):
+    client = app.test_client()
+    resp_login = client.post(
+        '/api/auth/login',
+        data=json.dumps(dict(email='admin@example.com', password='12345678')),
+        content_type='application/json',
+    )
+    response = client.delete(
+        '/api/users/toto',
+        headers=dict(
+            Authorization='Bearer '
+            + json.loads(resp_login.data.decode())['auth_token']
+        ),
+    )
+
+    assert response.status_code == 204
+
+
+def test_admin_can_delete_its_own_account(app, user_1_admin, user_2_admin):
+    client = app.test_client()
+    resp_login = client.post(
+        '/api/auth/login',
+        data=json.dumps(dict(email='admin@example.com', password='12345678')),
+        content_type='application/json',
+    )
+    response = client.delete(
+        '/api/users/admin',
+        headers=dict(
+            Authorization='Bearer '
+            + json.loads(resp_login.data.decode())['auth_token']
+        ),
+    )
+
+    assert response.status_code == 204
+
+
+def test_admin_can_not_delete_its_own_account_if_no_other_admin(
+    app, user_1_admin, user_2
+):
+    client = app.test_client()
+    resp_login = client.post(
+        '/api/auth/login',
+        data=json.dumps(dict(email='admin@example.com', password='12345678')),
+        content_type='application/json',
+    )
+    response = client.delete(
+        '/api/users/admin',
+        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 can not delete your account, no other user has admin rights.'
+        in data['message']
+    )

fittrackee_api/fittrackee_api/users/users.py

@@ -1,9 +1,12 @@
+import os
+import shutil
+
 from fittrackee_api import appLog, db
 from flask import Blueprint, jsonify, request, send_file
 from sqlalchemy import exc
 
 from ..activities.utils_files import get_absolute_file_path
-from .models import User
+from .models import Activity, User
 from .utils import authenticate, authenticate_as_admin
 
 users_blueprint = Blueprint('users', __name__)
@@ -270,7 +273,7 @@ def update_user(auth_user_id, user_name):
       }
 
     :param integer auth_user_id: authenticate user id (from JSON Web Token)
-    :param integer user_name: user name
+    :param string user_name: user name
 
     :<json boolean admin: does the user have administrator rights
 
@@ -316,6 +319,111 @@ def update_user(auth_user_id, user_name):
     return jsonify(response_object), code
 
 
+@users_blueprint.route('/users/<user_name>', methods=['DELETE'])
+@authenticate
+def delete_activity(auth_user_id, user_name):
+    """
+    Delete a user account
+    - a user can only delete his own account
+    - an admin can delete all accounts except his account if he's the only
+      one admin
+
+    **Example request**:
+
+    .. sourcecode:: http
+
+      DELETE /api/users/john_doe HTTP/1.1
+      Content-Type: application/json
+
+    **Example response**:
+
+    .. sourcecode:: http
+
+      HTTP/1.1 204 NO CONTENT
+      Content-Type: application/json
+
+    :param integer auth_user_id: authenticate user id (from JSON Web Token)
+    :param string user_name: user name
+
+    :reqheader Authorization: OAuth 2.0 Bearer Token
+
+    :statuscode 204: user account deleted
+    :statuscode 401:
+        - Provide a valid auth token.
+        - Signature expired. Please log in again.
+        - Invalid token. Please log in again.
+    :statuscode 403:
+        - You do not have permissions.
+        - You can not delete your account, no other user has admin rights.
+    :statuscode 404:
+        - User does not exist.
+    :statuscode 500: Error. Please try again or contact the administrator.
+
+    """
+    try:
+        auth_user = User.query.filter_by(id=auth_user_id).first()
+        user = User.query.filter_by(username=user_name).first()
+        if user:
+            if user.id != auth_user_id and not auth_user.admin:
+                response_object = {
+                    'status': 'error',
+                    'message': 'You do not have permissions.',
+                }
+                return response_object, 403
+            if (
+                user.admin is True
+                and User.query.filter_by(admin=True).count() == 1
+            ):
+                response_object = {
+                    'status': 'error',
+                    'message': (
+                        'You can not delete your account, '
+                        'no other user has admin rights.'
+                    ),
+                }
+                return response_object, 403
+            for activity in Activity.query.filter_by(user_id=user.id).all():
+                db.session.delete(activity)
+                db.session.flush()
+            user_picture = user.picture
+            db.session.delete(user)
+            db.session.commit()
+            if user_picture:
+                picture_path = get_absolute_file_path(user.picture)
+                if os.path.isfile(picture_path):
+                    os.remove(picture_path)
+            shutil.rmtree(
+                get_absolute_file_path(f'activities/{user.id}'),
+                ignore_errors=True,
+            )
+            shutil.rmtree(
+                get_absolute_file_path(f'pictures/{user.id}'),
+                ignore_errors=True,
+            )
+            response_object = {'status': 'no content'}
+            code = 204
+        else:
+            response_object = {
+                'status': 'not found',
+                'message': 'User does not exist.',
+            }
+            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
+
+
 @users_blueprint.route('/ping', methods=['GET'])
 def ping_pong():
     """ health check endpoint