feat(ui): added ability to replace binary file through UI, added related tests. Fixes: RCCE-19

rhodecode/apps/repository/__init__.py

@@ -446,6 +446,16 b' def includeme(config):'
         renderer='json_ext')
 
     config.add_route(
+        name='repo_files_replace_binary',
+        pattern='/{repo_name:.*?[^/]}/replace_binary/{commit_id}/{f_path:.*}',
+        repo_route=True)
+    config.add_view(
+        RepoFilesView,
+        attr='repo_files_replace_file',
+        route_name='repo_files_replace_binary', request_method='POST',
+        renderer='json_ext')
+
+    config.add_route(
         name='repo_files_create_file',
         pattern='/{repo_name:.*?[^/]}/create_file/{commit_id}/{f_path:.*}',
         repo_route=True)

rhodecode/apps/repository/tests/test_repo_files.py

@@ -924,6 +924,26 b' class TestModifyFilesWithWebInterface(ob'
         tip = repo.get_commit(commit_idx=-1)
         assert tip.message == 'I committed'
 
+    def test_replace_binary_file_view_commit_changes(self, backend, csrf_token):
+        repo = backend.create_repo()
+        backend.ensure_file(b"vcs/nodes.docx", content=b"PREVIOUS CONTENT'")
+
+        response = self.app.post(
+            route_path('repo_files_replace_binary',
+                       repo_name=repo.repo_name,
+                       commit_id=backend.default_head_id,
+                       f_path='vcs/nodes.docx'),
+            params={
+                'message': 'I committed',
+                'csrf_token': csrf_token,
+            },
+            upload_files=[('files_upload', 'vcs/nodes.docx', b'SOME CONTENT')],
+            status=200)
+        assert_session_flash(
+            response, 'Successfully committed 1 new file')
+        tip = repo.get_commit(commit_idx=-1)
+        assert tip.message == 'I committed'
+
     def test_edit_file_view_commit_changes_default_message(self, backend,
                                                            csrf_token):
         repo = backend.create_repo()
@@ -950,6 +970,48 b' class TestModifyFilesWithWebInterface(ob'
         tip = repo.get_commit(commit_idx=-1)
         assert tip.message == 'Edited file vcs/nodes.py via RhodeCode Enterprise'
 
+    def test_replace_binary_file_content_with_content_that_not_belong_to_original_type(self, backend, csrf_token):
+        repo = backend.create_repo()
+        backend.ensure_file(b"vcs/sheet.xlsx", content=b"PREVIOUS CONTENT'")
+
+        response = self.app.post(
+            route_path('repo_files_replace_binary',
+                       repo_name=repo.repo_name,
+                       commit_id=backend.default_head_id,
+                       f_path='vcs/sheet.xlsx'),
+            params={
+                'message': 'I committed',
+                'csrf_token': csrf_token,
+            },
+            upload_files=[('files_upload', 'vcs/sheet.docx', b'SOME CONTENT')],
+            status=200)
+        assert response.json['error'] == "file extension of uploaded file doesn't match an original file's extension"
+
+    @pytest.mark.parametrize("replacement_files, expected_error", [
+        ([], 'missing files'),
+        (
+                [('files_upload', 'vcs/node1.docx', b'SOME CONTENT'),
+                 ('files_upload', 'vcs/node2.docx', b'SOME CONTENT')],
+                'too many files for replacement'),
+    ])
+    def test_replace_binary_with_wrong_amount_of_content_sources(self, replacement_files, expected_error, backend,
+                                                                 csrf_token):
+        repo = backend.create_repo()
+        backend.ensure_file(b"vcs/node.docx", content=b"PREVIOUS CONTENT'")
+
+        response = self.app.post(
+            route_path('repo_files_replace_binary',
+                       repo_name=repo.repo_name,
+                       commit_id=backend.default_head_id,
+                       f_path='vcs/node.docx'),
+            params={
+                'message': 'I committed',
+                'csrf_token': csrf_token,
+            },
+            upload_files=replacement_files,
+            status=200)
+        assert response.json['error'] == expected_error
+
     def test_delete_file_view(self, backend):
         self.app.get(
             route_path('repo_files_remove_file',

rhodecode/apps/repository/views/repo_files.py

@@ -1341,6 +1341,9 b' class RepoFilesView(RepoAppView):'
 
         self._ensure_not_locked()
 
+        # Check if we need to use this page to upload binary
+        upload_binary = str2bool(self.request.params.get('upload_binary', False))
+
         c.commit = self._get_commit_or_redirect(commit_id, redirect_after=False)
         if c.commit is None:
             c.commit = EmptyCommit(alias=self.rhodecode_vcs_repo.alias)
@@ -1357,8 +1360,10 b' class RepoFilesView(RepoAppView):'
         self.forbid_non_head(is_head, f_path, commit_id=commit_id)
         self.check_branch_permission(_branch_name, commit_id=commit_id)
 
-        c.default_message = (_('Added file via RhodeCode Enterprise'))
+        c.default_message = (_('Added file via RhodeCode Enterprise')) \
+            if not upload_binary else (_('Edited file {} via RhodeCode Enterprise').format(f_path))
         c.f_path = f_path.lstrip('/')  # ensure not relative path
+        c.replace_binary = upload_binary
 
         return self._get_template_context(c)
 
@@ -1584,3 +1589,120 b' class RepoFilesView(RepoAppView):'
             'error': None,
             'redirect_url': default_redirect_url
         }
+
+    @LoginRequired()
+    @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
+    @CSRFRequired()
+    def repo_files_replace_file(self):
+        _ = self.request.translate
+        c = self.load_default_context()
+        commit_id, f_path = self._get_commit_and_path()
+
+        self._ensure_not_locked()
+
+        c.commit = self._get_commit_or_redirect(commit_id, redirect_after=False)
+        if c.commit is None:
+            c.commit = EmptyCommit(alias=self.rhodecode_vcs_repo.alias)
+
+        if self.rhodecode_vcs_repo.is_empty():
+            default_redirect_url = h.route_path(
+                'repo_summary', repo_name=self.db_repo_name)
+        else:
+            default_redirect_url = h.route_path(
+                'repo_commit', repo_name=self.db_repo_name, commit_id='tip')
+
+        if self.rhodecode_vcs_repo.is_empty():
+            # for empty repository we cannot check for current branch, we rely on
+            # c.commit.branch instead
+            _branch_name, _sha_commit_id, is_head = c.commit.branch, '', True
+        else:
+            _branch_name, _sha_commit_id, is_head = \
+                self._is_valid_head(commit_id, self.rhodecode_vcs_repo,
+                                    landing_ref=self.db_repo.landing_ref_name)
+
+        error = self.forbid_non_head(is_head, f_path, json_mode=True)
+        if error:
+            return {
+                'error': error,
+                'redirect_url': default_redirect_url
+            }
+        error = self.check_branch_permission(_branch_name, json_mode=True)
+        if error:
+            return {
+                'error': error,
+                'redirect_url': default_redirect_url
+            }
+
+        c.default_message = (_('Edited file {} via RhodeCode Enterprise').format(f_path))
+        c.f_path = f_path
+
+        r_post = self.request.POST
+
+        message = c.default_message
+        user_message = r_post.getall('message')
+        if isinstance(user_message, list) and user_message:
+            # we take the first from duplicated results if it's not empty
+            message = user_message[0] if user_message[0] else message
+
+        data_for_replacement = r_post.getall('files_upload') or []
+        if (objects_count := len(data_for_replacement)) > 1:
+            return {
+                'error': 'too many files for replacement',
+                'redirect_url': default_redirect_url
+            }
+        elif not objects_count:
+            return {
+                'error': 'missing files',
+                'redirect_url': default_redirect_url
+            }
+
+        content = data_for_replacement[0].file
+        retrieved_filename = data_for_replacement[0].filename
+
+        if retrieved_filename.split('.')[-1] != f_path.split('.')[-1]:
+            return {
+                'error': 'file extension of uploaded file doesn\'t match an original file\'s extension',
+                'redirect_url': default_redirect_url
+            }
+
+        author = self._rhodecode_db_user.full_contact
+
+        try:
+            commit = ScmModel().update_binary_node(
+                user=self._rhodecode_db_user.user_id,
+                repo=self.db_repo,
+                message=message,
+                node={
+                    'content': content,
+                    'file_path': f_path.encode(),
+                },
+                parent_commit=c.commit,
+                author=author,
+            )
+
+            h.flash(_('Successfully committed 1 new file'), category='success')
+
+            default_redirect_url = h.route_path(
+                'repo_commit', repo_name=self.db_repo_name, commit_id=commit.raw_id)
+
+        except (NodeError, NodeAlreadyExistsError) as e:
+            error = h.escape(e)
+            h.flash(error, category='error')
+
+            return {
+                'error': error,
+                'redirect_url': default_redirect_url
+            }
+        except Exception:
+            log.exception('Error occurred during commit')
+            error = _('Error occurred during commit')
+            h.flash(error, category='error')
+            return {
+                'error': error,
+                'redirect_url': default_redirect_url
+            }
+
+        return {
+            'error': None,
+            'redirect_url': default_redirect_url
+        }

rhodecode/model/scm.py

@@ -255,6 +255,31 b' class ScmModel(BaseModel):'
             all_repos, repos_path=self.repos_path, order_by=sort_key)
         return repo_iter
 
+    @staticmethod
+    def get_parent_commits(parent_commit, scm_instance):
+        if not parent_commit:
+            parent_commit = EmptyCommit(alias=scm_instance.alias)
+
+        if isinstance(parent_commit, EmptyCommit):
+            # EmptyCommit means we're editing empty repository
+            parents = None
+        else:
+            parents = [parent_commit]
+        return parent_commit, parents
+
+    def initialize_inmemory_vars(self, user, repo, message, author):
+        """
+        Initialize node specific objects for further usage
+        """
+        user = self._get_user(user)
+        scm_instance = repo.scm_instance(cache=False)
+        message = safe_str(message)
+        commiter = user.full_contact
+        author = safe_str(author) if author else commiter
+        imc = scm_instance.in_memory_commit
+
+        return user, scm_instance, message, commiter, author, imc
+
     def get_repo_groups(self, all_groups=None):
         if all_groups is None:
             all_groups = RepoGroup.query()\
@@ -763,24 +788,10 b' class ScmModel(BaseModel):'
 
         :returns: new committed commit
         """
-
-        user = self._get_user(user)
-        scm_instance = repo.scm_instance(cache=False)
-
-        message = safe_str(message)
-        commiter = user.full_contact
-        author = safe_str(author) if author else commiter
+        user, scm_instance, message, commiter, author, imc = self.initialize_inmemory_vars(
+            user, repo, message, author)
 
-        imc = scm_instance.in_memory_commit
-
-        if not parent_commit:
-            parent_commit = EmptyCommit(alias=scm_instance.alias)
-
-        if isinstance(parent_commit, EmptyCommit):
-            # EmptyCommit means we're editing empty repository
-            parents = None
-        else:
-            parents = [parent_commit]
+        parent_commit, parents = self.get_parent_commits(parent_commit, scm_instance)
 
         upload_file_types = (io.BytesIO, io.BufferedRandom)
         processed_nodes = []
@@ -827,23 +838,10 b' class ScmModel(BaseModel):'
 
     def update_nodes(self, user, repo, message, nodes, parent_commit=None,
                      author=None, trigger_push_hook=True):
-        user = self._get_user(user)
-        scm_instance = repo.scm_instance(cache=False)
-
-        message = safe_str(message)
-        commiter = user.full_contact
-        author = safe_str(author) if author else commiter
-
-        imc = scm_instance.in_memory_commit
+        user, scm_instance, message, commiter, author, imc = self.initialize_inmemory_vars(
+            user, repo, message, author)
 
-        if not parent_commit:
-            parent_commit = EmptyCommit(alias=scm_instance.alias)
-
-        if isinstance(parent_commit, EmptyCommit):
-            # EmptyCommit means we we're editing empty repository
-            parents = None
-        else:
-            parents = [parent_commit]
+        parent_commit, parents = self.get_parent_commits(parent_commit, scm_instance)
 
         # add multiple nodes
         for _filename, data in nodes.items():
@@ -890,6 +888,39 b' class ScmModel(BaseModel):'
 
         return tip
 
+    def update_binary_node(self, user, repo, message, node, parent_commit=None, author=None):
+        user, scm_instance, message, commiter, author, imc = self.initialize_inmemory_vars(
+            user, repo, message, author)
+
+        parent_commit, parents = self.get_parent_commits(parent_commit, scm_instance)
+
+        file_path = node.get('file_path')
+        if isinstance(raw_content := node.get('content'), (io.BytesIO, io.BufferedRandom)):
+            content = raw_content.read()
+        else:
+            raise Exception("Wrong content was provided")
+        file_node = FileNode(file_path, content=content)
+        imc.change(file_node)
+
+        try:
+            tip = imc.commit(message=message,
+                             author=author,
+                             parents=parents,
+                             branch=parent_commit.branch)
+        except NodeNotChangedError:
+            raise
+        except Exception as e:
+            log.exception("Unexpected exception during call to imc.commit")
+            raise IMCCommitError(str(e))
+        finally:
+            self.mark_for_invalidation(repo.repo_name)
+
+        hooks_utils.trigger_post_push_hook(
+            username=user.username, action='push_local', hook_type='post_push',
+            repo_name=repo.repo_name, repo_type=scm_instance.alias,
+            commit_ids=[tip.raw_id])
+        return tip
+
     def delete_nodes(self, user, repo, message, nodes, parent_commit=None,
                      author=None, trigger_push_hook=True):
         """
@@ -908,8 +939,8 b' class ScmModel(BaseModel):'
         :returns: new commit after deletion
         """
 
-        user = self._get_user(user)
-        scm_instance = repo.scm_instance(cache=False)
+        user, scm_instance, message, commiter, author, imc = self.initialize_inmemory_vars(
+            user, repo, message, author)
 
         processed_nodes = []
         for f_path in nodes:
@@ -919,20 +950,8 b' class ScmModel(BaseModel):'
             content = nodes[f_path].get('content')
             processed_nodes.append((safe_bytes(f_path), content))
 
-        message = safe_str(message)
-        commiter = user.full_contact
-        author = safe_str(author) if author else commiter
-
-        imc = scm_instance.in_memory_commit
+        parent_commit, parents = self.get_parent_commits(parent_commit, scm_instance)
 
-        if not parent_commit:
-            parent_commit = EmptyCommit(alias=scm_instance.alias)
-
-        if isinstance(parent_commit, EmptyCommit):
-            # EmptyCommit means we we're editing empty repository
-            parents = None
-        else:
-            parents = [parent_commit]
         # add multiple nodes
         for path, content in processed_nodes:
             imc.remove(FileNode(path, content=content))

rhodecode/templates/files/files_source.mako

@@ -36,7 +36,7 b''
               %if c.on_branch_head and c.branch_or_raw_id:
                   ## binary files are delete only
                   % if c.file.is_binary:
-                    ${h.link_to(_('Edit'), '#Edit', class_="btn btn-default disabled tooltip", title=_('Editing binary files not allowed'))}
+                    ${h.link_to(_('Replace'), h.route_path('repo_files_upload_file', repo_name=c.repo_name,commit_id=c.commit.raw_id,f_path=c.f_path, _query={'upload_binary': 'true'}), class_="btn btn-default active tooltip", title=_('You can replace content of your binary file'))}
                     ${h.link_to(_('Delete'), h.route_path('repo_files_remove_file',repo_name=c.repo_name,commit_id=c.branch_or_raw_id,f_path=c.f_path, _query=query),class_="btn btn-danger")}
                   % else:
                     <a  class="btn btn-default" href="${h.route_path('repo_files_edit_file',repo_name=c.repo_name,commit_id=c.branch_or_raw_id,f_path=c.f_path, _query=query)}">

rhodecode/templates/files/files_upload.mako

@@ -49,7 +49,11 b''
     </div>
 
     <div class="edit-file-title">
-        <span class="title-heading">${_('Upload new file')} @ <code>${h.show_id(c.commit)}</code></span>
+        % if c.replace_binary:
+            <span class="title-heading">${_('Replace content of')} <b>${c.f_path}</b> @ <code>${h.show_id(c.commit)}</code></span>
+        % else:
+            <span class="title-heading">${_('Upload new file')} @ <code>${h.show_id(c.commit)}</code></span>
+        % endif
         % if c.commit.branch:
         <span class="tag branchtag">
             <i class="icon-branch"></i> ${c.commit.branch}
@@ -57,24 +61,27 b''
         % endif
     </div>
 
-    <% form_url = h.route_path('repo_files_upload_file', repo_name=c.repo_name, commit_id=c.commit.raw_id, f_path=c.f_path) %>
-    ##${h.secure_form(form_url, id='eform', enctype="multipart/form-data", request=request)}
-    <div class="edit-file-fieldset">
-        <div class="path-items">
-            <ul class="tooltip" title="Repository path to store uploaded files. To change it, navigate to different path and click upload from there.">
-            <li class="breadcrumb-path">
-                <div>
-                    <a href="${h.route_path('repo_files', repo_name=c.repo_name, commit_id=c.commit.raw_id, f_path='')}"><i class="icon-home"></i></a> /
-                    <a href="${h.route_path('repo_files', repo_name=c.repo_name, commit_id=c.commit.raw_id, f_path=c.f_path)}">${c.f_path}</a>${('/' if c.f_path else '')}
-                </div>
-            </li>
-            <li class="location-path">
+    % if not c.replace_binary:
+        <% form_url = h.route_path('repo_files_upload_file', repo_name=c.repo_name, commit_id=c.commit.raw_id, f_path=c.f_path) %>
+        <div class="edit-file-fieldset">
+            <div class="path-items">
+                <ul class="tooltip" title="Repository path to store uploaded files. To change it, navigate to different path and click upload from there.">
+                <li class="breadcrumb-path">
+                    <div>
+                        <a href="${h.route_path('repo_files', repo_name=c.repo_name, commit_id=c.commit.raw_id, f_path='')}"><i class="icon-home"></i></a> /
+                        <a href="${h.route_path('repo_files', repo_name=c.repo_name, commit_id=c.commit.raw_id, f_path=c.f_path)}">${c.f_path}</a>${('/' if c.f_path else '')}
+                    </div>
+                </li>
+                <li class="location-path">
 
-            </li>
-            </ul>
+                </li>
+                </ul>
+            </div>
+
         </div>
-
-    </div>
+    % else:
+        <% form_url = h.route_path('repo_files_replace_binary', repo_name=c.repo_name, commit_id=c.commit.raw_id, f_path=c.f_path) %>
+    % endif
 
     <div class="upload-form table">
         <div>
@@ -83,7 +90,11 b''
                 <div class="dropzone-pure">
                     <div class="dz-message">
                     <i class="icon-upload" style="font-size:36px"></i></br>
-                    ${_("Drag'n Drop files here or")} <span class="link">${_('Choose your files')}</span>.<br>
+                    % if not c.replace_binary:
+                        ${_("Drag'n Drop files here or")} <span class="link">${_('Choose your files')}</span>.<br>
+                    % else:
+                        ${_("Drag'n Drop file here or")} <span class="link">${_('Choose your file')}</span>.<br>
+                    % endif
                     </div>
                 </div>
 

rhodecode/tests/routes.py

@@ -75,6 +75,7 b' def get_url_defs():'
         "repo_files_add_file": "/{repo_name}/add_file/{commit_id}/{f_path}",
         "repo_files_upload_file": "/{repo_name}/upload_file/{commit_id}/{f_path}",
         "repo_files_create_file": "/{repo_name}/create_file/{commit_id}/{f_path}",
+        "repo_files_replace_binary": "/{repo_name}/replace_binary/{commit_id}/{f_path}",
         "repo_nodetree_full": "/{repo_name}/nodetree_full/{commit_id}/{f_path}",
         "repo_nodetree_full:default_path": "/{repo_name}/nodetree_full/{commit_id}/",
         "journal": ADMIN_PREFIX + "/journal",