From 47afbf6d037859b54474314b7b5ac7cbef33cbc6 Mon Sep 17 00:00:00 2001 From: ReimarBauer Date: Wed, 27 Sep 2023 17:21:43 +0200 Subject: [PATCH] moved user db stuff to file_manager --- mslib/mscolab/file_manager.py | 29 +++++++++++++++++++ mslib/mscolab/server.py | 19 +++++------- tests/_test_mscolab/test_file_manager.py | 37 +++++++++++++++++++++++- 3 files changed, 72 insertions(+), 13 deletions(-) diff --git a/mslib/mscolab/file_manager.py b/mslib/mscolab/file_manager.py index f6d817b1f..c16e5a10c 100644 --- a/mslib/mscolab/file_manager.py +++ b/mslib/mscolab/file_manager.py @@ -192,6 +192,35 @@ def auth_type(self, u_id, op_id): return False return perm.access_level + def modify_user(self, user, attribute=None, value=None, action=None): + if action == "create": + user_query = User.query.filter_by(emailid=str(user.emailid)).first() + if user_query is None: + db.session.add(user) + db.session.commit() + else: + return False + elif action == "delete": + user_query = User.query.filter_by(id=user.id).first() + if user_query is not None: + db.session.delete(user) + db.session.commit() + user_query = User.query.filter_by(id=user.id).first() + # on delete we return succesfull deleted + if user_query is None: + return True + user_query = User.query.filter_by(id=user.id).first() + if user_query is None: + return False + if None not in (attribute, value): + if attribute == "emailid": + user_query = User.query.filter_by(emailid=str(value)).first() + if user_query is not None: + return False + setattr(user, attribute, value) + db.session.commit() + return True + def update_operation(self, op_id, attribute, value, user): """ op_id: operation id diff --git a/mslib/mscolab/server.py b/mslib/mscolab/server.py index 8266a8389..25a38fe4e 100644 --- a/mslib/mscolab/server.py +++ b/mslib/mscolab/server.py @@ -169,9 +169,8 @@ def register_user(email, password, username): user_exists = User.query.filter_by(username=str(username)).first() if user_exists: return {"success": False, "message": "This username is already registered"} - db.session.add(user) - db.session.commit() - return {"success": True} + result = fm.modify_user(user, action="create") + return {"success": result} def verify_user(func): @@ -291,10 +290,8 @@ def confirm_email(token): if user.confirmed: return render_template('user/confirmed.html', username=user.username) else: - user.confirmed = True - user.confirmed_on = datetime.datetime.now() - db.session.add(user) - db.session.commit() + fm.modify_user(user, attribute="confirmed_on", value=datetime.datetime.now()) + fm.modify_user(user, attribute="confirmed", value=True) return render_template('user/confirmed.html', username=user.username) @@ -312,9 +309,8 @@ def delete_user(): """ # ToDo rename to delete_own_account user = g.user - db.session.delete(user) - db.session.commit() - return jsonify({"success": True}), 200 + result = fm.modify_user(user, action="delete") + return jsonify({"success": result}), 200 # Chat related routes @@ -691,8 +687,7 @@ def reset_password(token): if form.validate_on_submit(): try: user.hash_password(form.confirm_password.data) - user.confirmed = True - db.session.commit() + fm.modify_user(user, "confirmed", True) flash('Password reset Success. Please login by the user interface.', 'category_success') return render_template('user/status.html') except IOError: diff --git a/tests/_test_mscolab/test_file_manager.py b/tests/_test_mscolab/test_file_manager.py index 39b10f7b5..418491abd 100644 --- a/tests/_test_mscolab/test_file_manager.py +++ b/tests/_test_mscolab/test_file_manager.py @@ -26,10 +26,11 @@ """ from flask_testing import TestCase import os +import datetime import pytest from mslib.mscolab.conf import mscolab_settings -from mslib.mscolab.models import Operation +from mslib.mscolab.models import Operation, User from mslib.mscolab.server import APP from mslib.mscolab.file_manager import FileManager from mslib.mscolab.seed import add_user, get_user @@ -80,6 +81,40 @@ def setUp(self): def tearDown(self): pass + def test_modify_user(self): + with self.app.test_client(): + user = User("user@example.com", "user", "password") + assert user.id is None + assert User.query.filter_by(emailid=user.emailid).first() is None + # creeat the user + self.fm.modify_user(user, action="create") + user_query = User.query.filter_by(emailid=user.emailid).first() + assert user_query.id is not None + assert user_query is not None + assert user_query.confirmed is False + # cannot create a user a second time + assert self.fm.modify_user(user, action="create") is False + # confirming the user + confirm_time = datetime.datetime.now() + datetime.timedelta(days=1) + self.fm.modify_user(user_query, attribute="confirmed_on", value=confirm_time) + self.fm.modify_user(user_query, attribute="confirmed", value=True) + user_query = User.query.filter_by(id=user.id).first() + assert user_query.confirmed is True + assert user_query.confirmed_on == confirm_time + assert user_query.confirmed_on > user_query.registered_on + # deleting the user + self.fm.modify_user(user_query, action="delete") + user_query = User.query.filter_by(id=user_query.id).first() + assert user_query is None + + def test_modify_user_special_cases(self): + user1 = User("user1@example.com", "user1", "password") + user2 = User("user2@example.com", "user2", "password") + self.fm.modify_user(user1, action="create") + self.fm.modify_user(user2, action="create") + user_query1 = User.query.filter_by(emailid=user1.emailid).first() + assert self.fm.modify_user(user_query1, "emailid", user2.emailid) is False + def test_fetch_operation_creator(self): with self.app.test_client(): flight_path, operation = self._create_operation(flight_path="more_than_one")