engineering/unrhodecode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

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

Browse source
This commit is contained in:
Serhii Ilin 2024-01-25 16:32:51 +02:00
parent 79aeb82ee8
commit 174ccfff79
7 changed files with 293 additions and 68 deletions
Download patch file Download diff file Expand all files Collapse all files

10
rhodecode/apps/repository/__init__.py
View file

@ -445,6 +445,16 @@ def includeme(config):
route_name='repo_files_upload_file', request_method='POST',
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:.*}',

62
rhodecode/apps/repository/tests/test_repo_files.py
View file

@ -924,6 +924,26 @@ class TestModifyFilesWithWebInterface(object):
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 @@ class TestModifyFilesWithWebInterface(object):
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',

124
rhodecode/apps/repository/views/repo_files.py
View file

@ -1341,6 +1341,9 @@ 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 @@ 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 @@ 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
}

115
rhodecode/model/scm.py
View file

@ -255,6 +255,31 @@ 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 @@ class ScmModel(BaseModel):
:returns: new committed commit
"""
user, scm_instance, message, commiter, author, imc = self.initialize_inmemory_vars(
user, repo, message, author)
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
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 @@ 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)
user, scm_instance, message, commiter, author, imc = self.initialize_inmemory_vars(
user, repo, message, author)
message = safe_str(message)
commiter = user.full_contact
author = safe_str(author) if author else commiter
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 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 @@ 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 @@ 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 @@ 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
parent_commit, parents = self.get_parent_commits(parent_commit, scm_instance)
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 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))

2
rhodecode/templates/files/files_source.mako
View file

@ -36,7 +36,7 @@
%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)}">

47
rhodecode/templates/files/files_upload.mako
View file

@ -49,7 +49,11 @@
</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 @@
% 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>
</div>
</li>
</ul>
</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 @@
<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>

1
rhodecode/tests/routes.py
View file

@ -75,6 +75,7 @@ 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",