refactored ToDos of MSColab server code (#2051)

* refactored ToDos of MSColab server code

* flake8

mslib/mscolab/chat_manager.py

@@ -25,8 +25,11 @@
     limitations under the License.
 """
 import datetime
+import os
+import time
 
 import fs
+from werkzeug.utils import secure_filename
 
 from mslib.mscolab.conf import mscolab_settings
 from mslib.mscolab.models import db, Message, MessageType
@@ -93,3 +96,24 @@ def delete_message(self, message_id):
                 upload_dir.remove(fs.path.join(str(message.op_id), file_name))
         db.session.delete(message)
         db.session.commit()
+
+    def add_attachment(self, op_id, upload_folder, file, file_token):
+        with fs.open_fs('/') as home_fs:
+            file_dir = fs.path.join(upload_folder, str(op_id))
+            if '\\' not in file_dir:
+                if not home_fs.exists(file_dir):
+                    home_fs.makedirs(file_dir)
+            else:
+                file_dir = file_dir.replace('\\', '/')
+                if not os.path.exists(file_dir):
+                    os.makedirs(file_dir)
+            file_name, file_ext = file.filename.rsplit('.', 1)
+            file_name = f'{file_name}-{time.strftime("%Y%m%dT%H%M%S")}-{file_token}.{file_ext}'
+            file_name = secure_filename(file_name)
+            file_path = fs.path.join(file_dir, file_name)
+            file.save(file_path)
+            static_dir = fs.path.basename(upload_folder)
+            static_dir = static_dir.replace('\\', '/')
+            static_file_path = os.path.join(static_dir, str(op_id), file_name)
+        if os.path.exists(file_path):
+            return static_file_path

mslib/mscolab/file_manager.py

@@ -262,12 +262,11 @@ def update_operation(self, op_id, attribute, value, user):
         db.session.commit()
         return True
 
-    def delete_file(self, op_id, user):
+    def delete_operation(self, op_id, user):
         """
         op_id: operation id
         user: logged in user
         """
-        # ToDo rename to delete_operation
         if self.auth_type(user.id, op_id) != "creator":
             return False
         Permission.query.filter_by(op_id=op_id).delete()
@@ -377,14 +376,18 @@ def get_all_changes(self, op_id, user, named_version=False):
             'created_at': change.created_at.strftime("%Y-%m-%d, %H:%M:%S")
         }, changes))
 
-    def get_change_content(self, ch_id):
+    def get_change_content(self, ch_id, user):
         """
         ch_id: change id
         user: user of this request
 
         Get change related to id
         """
-        # ToDo refactor check user in op
+        ch = Change.query.filter_by(id=ch_id).first()
+        perm = Permission.query.filter_by(u_id=user.id, op_id=ch.op_id).first()
+        if perm is None:
+            return False
+
         change = Change.query.filter_by(id=ch_id).first()
         if not change:
             return False
@@ -404,13 +407,12 @@ def set_version_name(self, ch_id, op_id, u_id, version_name):
         db.session.commit()
         return True
 
-    def undo(self, ch_id, user):
+    def undo_changes(self, ch_id, user):
         """
         ch_id: change-id
         user: user of this request
 
         Undo a change
-        # ToDo rename to undo_changes
         # ToDo add a revert option, which removes only that commit's change
         """
         ch = Change.query.filter_by(id=ch_id).first()

mslib/mscolab/server.py

@@ -27,11 +27,9 @@
 import functools
 import json
 import logging
-import time
 import datetime
 import secrets
 import fs
-import os
 import socketio
 import sqlalchemy.exc
 from itsdangerous import URLSafeTimedSerializer, BadSignature
@@ -41,7 +39,6 @@
 from flask_cors import CORS
 from flask_httpauth import HTTPBasicAuth
 from validate_email import validate_email
-from werkzeug.utils import secure_filename
 
 from mslib.mscolab.conf import mscolab_settings
 from mslib.mscolab.models import Change, MessageType, User
@@ -298,13 +295,12 @@ def get_user():
     return json.dumps({'user': {'id': g.user.id, 'username': g.user.username}})
 
 
-@APP.route("/delete_user", methods=["POST"])
+@APP.route("/delete_own_account", methods=["POST"])
 @verify_user
-def delete_user():
+def delete_own_account():
     """
     delete own account
     """
-    # ToDo rename to delete_own_account
     user = g.user
     result = fm.modify_user(user, action="delete")
     return jsonify({"success": result}), 200
@@ -314,7 +310,6 @@ def delete_user():
 @APP.route("/messages", methods=["GET"])
 @verify_user
 def messages():
-    # ToDo maybe move is_member part to file_manager
     user = g.user
     op_id = request.args.get("op_id", request.form.get("op_id", None))
     if fm.is_member(user.id, op_id):
@@ -334,32 +329,18 @@ def message_attachment():
         file = request.files['file']
         message_type = MessageType(int(request.form.get("message_type")))
         user = g.user
-        # ToDo review
         users = fm.fetch_users_without_permission(int(op_id), user.id)
         if users is False:
             return jsonify({"success": False, "message": "Could not send message. No file uploaded."})
         if file is not None:
-            with fs.open_fs('/') as home_fs:
-                file_dir = fs.path.join(APP.config['UPLOAD_FOLDER'], op_id)
-                if '\\' not in file_dir:
-                    if not home_fs.exists(file_dir):
-                        home_fs.makedirs(file_dir)
-                else:
-                    file_dir = file_dir.replace('\\', '/')
-                    if not os.path.exists(file_dir):
-                        os.makedirs(file_dir)
-                file_name, file_ext = file.filename.rsplit('.', 1)
-                file_name = f'{file_name}-{time.strftime("%Y%m%dT%H%M%S")}-{file_token}.{file_ext}'
-                file_name = secure_filename(file_name)
-                file_path = fs.path.join(file_dir, file_name)
-                file.save(file_path)
-                static_dir = fs.path.basename(APP.config['UPLOAD_FOLDER'])
-                static_dir = static_dir.replace('\\', '/')
-                static_file_path = os.path.join(static_dir, op_id, file_name)
-            new_message = cm.add_message(user, static_file_path, op_id, message_type)
-            new_message_dict = get_message_dict(new_message)
-            sockio.emit('chat-message-client', json.dumps(new_message_dict))
-            return jsonify({"success": True, "path": static_file_path})
+            static_file_path = cm.add_attachment(op_id, APP.config['UPLOAD_FOLDER'], file, file_token)
+            if static_file_path is not None:
+                new_message = cm.add_message(user, static_file_path, op_id, message_type)
+                new_message_dict = get_message_dict(new_message)
+                sockio.emit('chat-message-client', json.dumps(new_message_dict))
+                return jsonify({"success": True, "path": static_file_path})
+            else:
+                return "False"
         return jsonify({"success": False, "message": "Could not send message. No file uploaded."})
     # normal use case never gets to this
     return "False"
@@ -428,9 +409,9 @@ def get_all_changes():
 @APP.route('/get_change_content', methods=['GET'])
 @verify_user
 def get_change_content():
-    # ToDo refactor see fm.get_change_content(
     ch_id = int(request.args.get('ch_id', request.form.get('ch_id', 0)))
-    result = fm.get_change_content(ch_id)
+    user = g.user
+    result = fm.get_change_content(ch_id, user)
     if result is False:
         return "False"
     return jsonify({"content": result})
@@ -470,7 +451,7 @@ def get_operations():
 def delete_operation():
     op_id = int(request.form.get('op_id', 0))
     user = g.user
-    success = fm.delete_file(op_id, user)
+    success = fm.delete_operation(op_id, user)
     if success is False:
         return jsonify({"success": False, "message": "You don't have access for this operation!"})
 
@@ -507,7 +488,6 @@ def get_operation_details():
 @APP.route('/set_last_used', methods=["POST"])
 @verify_user
 def set_last_used():
-    # ToDo refactor move to file_manager
     op_id = request.form.get('op_id', None)
     user = g.user
     days_ago = int(request.form.get('days', 0))
@@ -524,14 +504,13 @@ def set_last_used():
     return jsonify({"success": True}), 200
 
 
-@APP.route('/undo', methods=["POST"])
+@APP.route('/undo_changes', methods=["POST"])
 @verify_user
-def undo_ftml():
-    # ToDo rename to undo_changes
+def undo_changes():
     ch_id = request.form.get('ch_id', -1)
     ch_id = int(ch_id)
     user = g.user
-    result = fm.undo(ch_id, user)
+    result = fm.undo_changes(ch_id, user)
     # get op_id from change
     ch = Change.query.filter_by(id=ch_id).first()
     if result is True:

mslib/msui/mscolab.py

@@ -756,7 +756,7 @@ def delete_account(self):
             }
 
             try:
-                r = requests.post(self.mscolab_server_url + '/delete_user', data=data,
+                r = requests.post(self.mscolab_server_url + '/delete_own_account', data=data,
                                   timeout=tuple(config_loader(dataset="MSCOLAB_timeout")))
             except requests.exceptions.RequestException as e:
                 logging.error(e)

mslib/msui/mscolab_version_history.py

@@ -273,7 +273,7 @@ def handle_undo(self):
                     "token": self.token,
                     "ch_id": self.changes.currentItem().id
                 }
-                url = urljoin(self.mscolab_server_url, 'undo')
+                url = urljoin(self.mscolab_server_url, 'undo_changes')
                 r = requests.post(url, data=data, timeout=tuple(config_loader(dataset="MSCOLAB_timeout")))
                 if r.text != "False":
                     # reload windows

tests/_test_mscolab/test_chat_manager.py

@@ -24,11 +24,14 @@
     See the License for the specific language governing permissions and
     limitations under the License.
 """
+import os
+import secrets
 
 from flask_testing import TestCase
+from werkzeug.datastructures import FileStorage
 
 from mslib.mscolab.conf import mscolab_settings
-from mslib.mscolab.models import Message, MessageType
+from mslib.mscolab.models import Operation, Message, MessageType
 from mslib.mscolab.mscolab import handle_db_reset
 from mslib.mscolab.server import APP
 from mslib.mscolab.seed import add_user, get_user, add_operation, add_user_to_operation
@@ -89,3 +92,17 @@ def test_delete_messages(self):
             self.cm.delete_message(message.id)
             message = Message.query.filter(Message.id == message.id).first()
             assert message is None
+
+    def test_add_attachment(self):
+        sample_path = os.path.join(os.path.dirname(__file__), "..", "data")
+        filename = "example.csv"
+        name, ext = filename.split('.')
+        open_csv = os.path.join(sample_path, "example.csv")
+        operation = Operation.query.filter_by(path=self.operation_name).first()
+        token = secrets.token_urlsafe(16)
+        with open(open_csv, 'rb') as fp:
+            file = FileStorage(fp, filename=filename, content_type="text/csv")
+            static_path = self.cm.add_attachment(operation.id, mscolab_settings.UPLOAD_FOLDER, file, token)
+            assert name in static_path
+            assert static_path.endswith(ext)
+            assert token in static_path

tests/_test_mscolab/test_file_manager.py

@@ -238,11 +238,10 @@ def test_update_operation(self):
             assert ren_operation.id == operation.id
             assert ren_operation.path == rename_to
 
-    def test_delete_file(self):
-        # Todo rename "file" to operation
+    def test_delete_operation(self):
         with self.app.test_client():
             flight_path, operation = self._create_operation(flight_path='operation4')
-            assert self.fm.delete_file(operation.id, self.user)
+            assert self.fm.delete_operation(operation.id, self.user)
             assert Operation.query.filter_by(path=flight_path).first() is None
 
     def test_get_authorized_users(self):
@@ -279,7 +278,7 @@ def test_get_change_content(self):
             assert self.fm.save_file(operation.id, self.content1, self.user)
             assert self.fm.save_file(operation.id, self.content2, self.user)
             all_changes = self.fm.get_all_changes(operation.id, self.user)
-            assert self.fm.get_change_content(all_changes[1]["id"]) == self.content1
+            assert self.fm.get_change_content(all_changes[1]["id"], self.user) == self.content1
 
     def test_set_version_name(self):
         with self.app.test_client():
@@ -307,15 +306,15 @@ def test_undo(self):
             assert self.fm.save_file(operation.id, self.content2, self.user)
             all_changes = self.fm.get_all_changes(operation.id, self.user)
             # crestor
-            assert self.fm.undo(all_changes[1]["id"], self.user)
+            assert self.fm.undo_changes(all_changes[1]["id"], self.user)
             # check collaborator
             self.fm.add_bulk_permission(operation.id, self.user, [self.collaboratoruser.id], "collaborator")
             assert self.fm.is_collaborator(self.collaboratoruser.id, operation.id)
-            assert self.fm.undo(all_changes[1]["id"], self.collaboratoruser)
+            assert self.fm.undo_changes(all_changes[1]["id"], self.collaboratoruser)
             # check viewer
             self.fm.add_bulk_permission(operation.id, self.user, [self.vieweruser.id], "viewer")
             assert self.fm.is_viewer(self.vieweruser.id, operation.id)
-            assert self.fm.undo(all_changes[1]["id"], self.vieweruser) is False
+            assert self.fm.undo_changes(all_changes[1]["id"], self.vieweruser) is False
 
     def test_fetch_users_without_permission(self):
         with self.app.test_client():

tests/_test_mscolab/test_files.py

@@ -129,7 +129,7 @@ def test_undo(self):
             changes = Change.query.filter_by(op_id=operation.id).all()
             assert changes is not None
             assert changes[0].id == 1
-            assert self.fm.undo(changes[0].id, self.user) is True
+            assert self.fm.undo_changes(changes[0].id, self.user) is True
             assert len(self.fm.get_all_changes(operation.id, self.user)) == 3
             assert "beta" in self.fm.get_file(operation.id, self.user)
 
@@ -162,9 +162,9 @@ def test_delete_operation(self):
         with self.app.test_client():
             self._create_operation(flight_path="f3")
             op_id = get_recent_op_id(self.fm, self.user)
-            assert self.fm.delete_file(op_id, self.user2) is False
-            assert self.fm.delete_file(op_id, self.user) is True
-            assert self.fm.delete_file(op_id, self.user) is False
+            assert self.fm.delete_operation(op_id, self.user2) is False
+            assert self.fm.delete_operation(op_id, self.user) is True
+            assert self.fm.delete_operation(op_id, self.user) is False
             permissions = Permission.query.filter_by(op_id=op_id).all()
             assert len(permissions) == 0
             operations_db = Operation.query.filter_by(id=op_id).all()

tests/_test_mscolab/test_files_api.py

@@ -182,12 +182,11 @@ def test_update_operation(self):
             operation = Operation.query.filter_by(path=new_flight_path).first()
             assert operation.description == new_description
 
-    def test_delete_file(self):
-        # ToDo rename to operation
+    def test_delete_operation(self):
         with self.app.test_client():
             flight_path, operation = self._create_operation(flight_path="V10")
             assert operation.path == flight_path
-            assert self.fm.delete_file(operation.id, self.user)
+            assert self.fm.delete_operation(operation.id, self.user)
             operation = Operation.query.filter_by(path=flight_path).first()
             assert operation is None
 
@@ -206,9 +205,9 @@ def test_get_change_content(self):
             assert self.fm.save_file(operation.id, "content2", self.user)
             assert self.fm.save_file(operation.id, "content3", self.user)
             all_changes = self.fm.get_all_changes(operation.id, self.user)
-            previous_change = self.fm.get_change_content(all_changes[2]["id"])
+            previous_change = self.fm.get_change_content(all_changes[2]["id"], self.user)
             assert previous_change == "content1"
-            previous_change = self.fm.get_change_content(all_changes[1]["id"])
+            previous_change = self.fm.get_change_content(all_changes[1]["id"], self.user)
             assert previous_change == "content2"
 
     def test_set_version_name(self):

tests/_test_mscolab/test_server.py

@@ -149,11 +149,11 @@ def test_delete_user(self):
         assert add_user(self.userdata[0], self.userdata[1], self.userdata[2])
         with self.app.test_client() as test_client:
             token = self._get_token(test_client, self.userdata)
-            response = test_client.post('/delete_user', data={"token": token})
+            response = test_client.post('/delete_own_account', data={"token": token})
             assert response.status_code == 200
             data = json.loads(response.data.decode('utf-8'))
             assert data["success"] is True
-            response = test_client.post('/delete_user', data={"token": "dsdsds"})
+            response = test_client.post('/delete_own_account', data={"token": "dsdsds"})
             assert response.status_code == 200
             assert response.data.decode('utf-8') == "False"
 

tests/_test_msui/test_mscolab_version_history.py

@@ -116,7 +116,7 @@ def test_version_name_delete(self, mockbox):
         assert self.version_window.changes.currentItem().version_name is None
 
     @mock.patch("PyQt5.QtWidgets.QMessageBox.question", return_value=QtWidgets.QMessageBox.Yes)
-    def test_undo(self, mockbox):
+    def test_undo_changes(self, mockbox):
         self._change_version_filter(1)
         # make changes
         for i in range(2):

tests/utils.py

@@ -107,7 +107,7 @@ def mscolab_delete_user(app, msc_url, email, password):
         response = mscolab_login(app, msc_url, email, password)
         if response.status == '200 OK':
             data = json.loads(response.get_data(as_text=True))
-            url = urljoin(msc_url, 'delete_user')
+            url = urljoin(msc_url, 'delete_own_account')
             response = app.test_client().post(url, data=data)
             if response.status == '200 OK':
                 data = json.loads(response.get_data(as_text=True))