moved user db stuff to file_manager

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

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:

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("[email protected]", "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("[email protected]", "user1", "password")
+        user2 = User("[email protected]", "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")