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

rhodecode/apps/repository/__init__.py

@@ -1,1225 +1,1235 b''
 # Copyright (C) 2016-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 from rhodecode.apps._base import add_route_with_slash
 
 
 def includeme(config):
     from rhodecode.apps.repository.views.repo_artifacts import RepoArtifactsView
     from rhodecode.apps.repository.views.repo_audit_logs import AuditLogsView
     from rhodecode.apps.repository.views.repo_automation import RepoAutomationView
     from rhodecode.apps.repository.views.repo_bookmarks import RepoBookmarksView
     from rhodecode.apps.repository.views.repo_branch_permissions import RepoSettingsBranchPermissionsView
     from rhodecode.apps.repository.views.repo_branches import RepoBranchesView
     from rhodecode.apps.repository.views.repo_caches import RepoCachesView
     from rhodecode.apps.repository.views.repo_changelog import RepoChangelogView
     from rhodecode.apps.repository.views.repo_checks import RepoChecksView
     from rhodecode.apps.repository.views.repo_commits import RepoCommitsView
     from rhodecode.apps.repository.views.repo_compare import RepoCompareView
     from rhodecode.apps.repository.views.repo_feed import RepoFeedView
     from rhodecode.apps.repository.views.repo_files import RepoFilesView
     from rhodecode.apps.repository.views.repo_forks import RepoForksView
     from rhodecode.apps.repository.views.repo_maintainance import RepoMaintenanceView
     from rhodecode.apps.repository.views.repo_permissions import RepoSettingsPermissionsView
     from rhodecode.apps.repository.views.repo_pull_requests import RepoPullRequestsView
     from rhodecode.apps.repository.views.repo_review_rules import RepoReviewRulesView
     from rhodecode.apps.repository.views.repo_settings import RepoSettingsView
     from rhodecode.apps.repository.views.repo_settings_advanced import RepoSettingsAdvancedView
     from rhodecode.apps.repository.views.repo_settings_fields import RepoSettingsFieldsView
     from rhodecode.apps.repository.views.repo_settings_issue_trackers import RepoSettingsIssueTrackersView
     from rhodecode.apps.repository.views.repo_settings_remote import RepoSettingsRemoteView
     from rhodecode.apps.repository.views.repo_settings_vcs import RepoSettingsVcsView
     from rhodecode.apps.repository.views.repo_strip import RepoStripView
     from rhodecode.apps.repository.views.repo_summary import RepoSummaryView
     from rhodecode.apps.repository.views.repo_tags import RepoTagsView
 
     # repo creating checks, special cases that aren't repo routes
     config.add_route(
         name='repo_creating',
         pattern='/{repo_name:.*?[^/]}/repo_creating')
     config.add_view(
         RepoChecksView,
         attr='repo_creating',
         route_name='repo_creating', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_creating.mako')
 
     config.add_route(
         name='repo_creating_check',
         pattern='/{repo_name:.*?[^/]}/repo_creating_check')
     config.add_view(
         RepoChecksView,
         attr='repo_creating_check',
         route_name='repo_creating_check', request_method='GET',
         renderer='json_ext')
 
     # Summary
     # NOTE(marcink): one additional route is defined in very bottom, catch
     # all pattern
     config.add_route(
         name='repo_summary_explicit',
         pattern='/{repo_name:.*?[^/]}/summary', repo_route=True)
     config.add_view(
         RepoSummaryView,
         attr='summary',
         route_name='repo_summary_explicit', request_method='GET',
         renderer='rhodecode:templates/summary/summary.mako')
 
     config.add_route(
         name='repo_summary_commits',
         pattern='/{repo_name:.*?[^/]}/summary-commits', repo_route=True)
     config.add_view(
         RepoSummaryView,
         attr='summary_commits',
         route_name='repo_summary_commits', request_method='GET',
         renderer='rhodecode:templates/summary/summary_commits.mako')
     
     # Commits
     config.add_route(
         name='repo_commit',
         pattern='/{repo_name:.*?[^/]}/changeset/{commit_id}', repo_route=True)
     config.add_view(
         RepoCommitsView,
         attr='repo_commit_show',
         route_name='repo_commit', request_method='GET',
         renderer=None)
 
     config.add_route(
         name='repo_commit_children',
         pattern='/{repo_name:.*?[^/]}/changeset_children/{commit_id}', repo_route=True)
     config.add_view(
         RepoCommitsView,
         attr='repo_commit_children',
         route_name='repo_commit_children', request_method='GET',
         renderer='json_ext', xhr=True)
 
     config.add_route(
         name='repo_commit_parents',
         pattern='/{repo_name:.*?[^/]}/changeset_parents/{commit_id}', repo_route=True)
     config.add_view(
         RepoCommitsView,
         attr='repo_commit_parents',
         route_name='repo_commit_parents', request_method='GET',
         renderer='json_ext')
 
     config.add_route(
         name='repo_commit_raw',
         pattern='/{repo_name:.*?[^/]}/changeset-diff/{commit_id}', repo_route=True)
     config.add_view(
         RepoCommitsView,
         attr='repo_commit_raw',
         route_name='repo_commit_raw', request_method='GET',
         renderer=None)
 
     config.add_route(
         name='repo_commit_patch',
         pattern='/{repo_name:.*?[^/]}/changeset-patch/{commit_id}', repo_route=True)
     config.add_view(
         RepoCommitsView,
         attr='repo_commit_patch',
         route_name='repo_commit_patch', request_method='GET',
         renderer=None)
 
     config.add_route(
         name='repo_commit_download',
         pattern='/{repo_name:.*?[^/]}/changeset-download/{commit_id}', repo_route=True)
     config.add_view(
         RepoCommitsView,
         attr='repo_commit_download',
         route_name='repo_commit_download', request_method='GET',
         renderer=None)
 
     config.add_route(
         name='repo_commit_data',
         pattern='/{repo_name:.*?[^/]}/changeset-data/{commit_id}', repo_route=True)
     config.add_view(
         RepoCommitsView,
         attr='repo_commit_data',
         route_name='repo_commit_data', request_method='GET',
         renderer='json_ext', xhr=True)
 
     config.add_route(
         name='repo_commit_comment_create',
         pattern='/{repo_name:.*?[^/]}/changeset/{commit_id}/comment/create', repo_route=True)
     config.add_view(
         RepoCommitsView,
         attr='repo_commit_comment_create',
         route_name='repo_commit_comment_create', request_method='POST',
         renderer='json_ext')
 
     config.add_route(
         name='repo_commit_comment_preview',
         pattern='/{repo_name:.*?[^/]}/changeset/{commit_id}/comment/preview', repo_route=True)
     config.add_view(
         RepoCommitsView,
         attr='repo_commit_comment_preview',
         route_name='repo_commit_comment_preview', request_method='POST',
         renderer='string', xhr=True)
 
     config.add_route(
         name='repo_commit_comment_history_view',
         pattern='/{repo_name:.*?[^/]}/changeset/{commit_id}/comment/{comment_id}/history_view/{comment_history_id}', repo_route=True)
     config.add_view(
         RepoCommitsView,
         attr='repo_commit_comment_history_view',
         route_name='repo_commit_comment_history_view', request_method='POST',
         renderer='string', xhr=True)
 
     config.add_route(
         name='repo_commit_comment_attachment_upload',
         pattern='/{repo_name:.*?[^/]}/changeset/{commit_id}/comment/attachment_upload', repo_route=True)
     config.add_view(
         RepoCommitsView,
         attr='repo_commit_comment_attachment_upload',
         route_name='repo_commit_comment_attachment_upload', request_method='POST',
         renderer='json_ext', xhr=True)
 
     config.add_route(
         name='repo_commit_comment_delete',
         pattern='/{repo_name:.*?[^/]}/changeset/{commit_id}/comment/{comment_id}/delete', repo_route=True)
     config.add_view(
         RepoCommitsView,
         attr='repo_commit_comment_delete',
         route_name='repo_commit_comment_delete', request_method='POST',
         renderer='json_ext')
 
     config.add_route(
         name='repo_commit_comment_edit',
         pattern='/{repo_name:.*?[^/]}/changeset/{commit_id}/comment/{comment_id}/edit', repo_route=True)
     config.add_view(
         RepoCommitsView,
         attr='repo_commit_comment_edit',
         route_name='repo_commit_comment_edit', request_method='POST',
         renderer='json_ext')
 
     # still working url for backward compat.
     config.add_route(
         name='repo_commit_raw_deprecated',
         pattern='/{repo_name:.*?[^/]}/raw-changeset/{commit_id}', repo_route=True)
     config.add_view(
         RepoCommitsView,
         attr='repo_commit_raw',
         route_name='repo_commit_raw_deprecated', request_method='GET',
         renderer=None)
 
     # Files
     config.add_route(
         name='repo_archivefile',
         pattern='/{repo_name:.*?[^/]}/archive/{fname:.*}', repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_archivefile',
         route_name='repo_archivefile', request_method='GET',
         renderer=None)
 
     config.add_route(
         name='repo_files_diff',
         pattern='/{repo_name:.*?[^/]}/diff/{f_path:.*}', repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_files_diff',
         route_name='repo_files_diff', request_method='GET',
         renderer=None)
 
     config.add_route(  # legacy route to make old links work
         name='repo_files_diff_2way_redirect',
         pattern='/{repo_name:.*?[^/]}/diff-2way/{f_path:.*}', repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_files_diff_2way_redirect',
         route_name='repo_files_diff_2way_redirect', request_method='GET',
         renderer=None)
 
     config.add_route(
         name='repo_files',
         pattern='/{repo_name:.*?[^/]}/files/{commit_id}/{f_path:.*}', repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_files',
         route_name='repo_files', request_method='GET',
         renderer=None)
 
     config.add_route(
         name='repo_files:default_path',
         pattern='/{repo_name:.*?[^/]}/files/{commit_id}/', repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_files',
         route_name='repo_files:default_path', request_method='GET',
         renderer=None)
 
     config.add_route(
         name='repo_files:default_commit',
         pattern='/{repo_name:.*?[^/]}/files', repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_files',
         route_name='repo_files:default_commit', request_method='GET',
         renderer=None)
 
     config.add_route(
         name='repo_files:rendered',
         pattern='/{repo_name:.*?[^/]}/render/{commit_id}/{f_path:.*}', repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_files',
         route_name='repo_files:rendered', request_method='GET',
         renderer=None)
 
     config.add_route(
         name='repo_files:annotated',
         pattern='/{repo_name:.*?[^/]}/annotate/{commit_id}/{f_path:.*}', repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_files',
         route_name='repo_files:annotated', request_method='GET',
         renderer=None)
 
     config.add_route(
         name='repo_files:annotated_previous',
         pattern='/{repo_name:.*?[^/]}/annotate-previous/{commit_id}/{f_path:.*}', repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_files_annotated_previous',
         route_name='repo_files:annotated_previous', request_method='GET',
         renderer=None)
 
     config.add_route(
         name='repo_nodetree_full',
         pattern='/{repo_name:.*?[^/]}/nodetree_full/{commit_id}/{f_path:.*}', repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_nodetree_full',
         route_name='repo_nodetree_full', request_method='GET',
         renderer=None, xhr=True)
 
     config.add_route(
         name='repo_nodetree_full:default_path',
         pattern='/{repo_name:.*?[^/]}/nodetree_full/{commit_id}/', repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_nodetree_full',
         route_name='repo_nodetree_full:default_path', request_method='GET',
         renderer=None, xhr=True)
 
     config.add_route(
         name='repo_files_nodelist',
         pattern='/{repo_name:.*?[^/]}/nodelist/{commit_id}/{f_path:.*}', repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_nodelist',
         route_name='repo_files_nodelist', request_method='GET',
         renderer='json_ext', xhr=True)
 
     config.add_route(
         name='repo_file_raw',
         pattern='/{repo_name:.*?[^/]}/raw/{commit_id}/{f_path:.*}', repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_file_raw',
         route_name='repo_file_raw', request_method='GET',
         renderer=None)
 
     config.add_route(
         name='repo_file_download',
         pattern='/{repo_name:.*?[^/]}/download/{commit_id}/{f_path:.*}', repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_file_download',
         route_name='repo_file_download', request_method='GET',
         renderer=None)
 
     config.add_route(  # backward compat to keep old links working
         name='repo_file_download:legacy',
         pattern='/{repo_name:.*?[^/]}/rawfile/{commit_id}/{f_path:.*}',
         repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_file_download',
         route_name='repo_file_download:legacy', request_method='GET',
         renderer=None)
 
     config.add_route(
         name='repo_file_history',
         pattern='/{repo_name:.*?[^/]}/history/{commit_id}/{f_path:.*}', repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_file_history',
         route_name='repo_file_history', request_method='GET',
         renderer='json_ext')
 
     config.add_route(
         name='repo_file_authors',
         pattern='/{repo_name:.*?[^/]}/authors/{commit_id}/{f_path:.*}', repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_file_authors',
         route_name='repo_file_authors', request_method='GET',
         renderer='rhodecode:templates/files/file_authors_box.mako')
 
     config.add_route(
         name='repo_files_check_head',
         pattern='/{repo_name:.*?[^/]}/check_head/{commit_id}/{f_path:.*}',
         repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_files_check_head',
         route_name='repo_files_check_head', request_method='POST',
         renderer='json_ext', xhr=True)
 
     config.add_route(
         name='repo_files_remove_file',
         pattern='/{repo_name:.*?[^/]}/remove_file/{commit_id}/{f_path:.*}',
         repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_files_remove_file',
         route_name='repo_files_remove_file', request_method='GET',
         renderer='rhodecode:templates/files/files_delete.mako')
 
     config.add_route(
         name='repo_files_delete_file',
         pattern='/{repo_name:.*?[^/]}/delete_file/{commit_id}/{f_path:.*}',
         repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_files_delete_file',
         route_name='repo_files_delete_file', request_method='POST',
         renderer=None)
 
     config.add_route(
         name='repo_files_edit_file',
         pattern='/{repo_name:.*?[^/]}/edit_file/{commit_id}/{f_path:.*}',
         repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_files_edit_file',
         route_name='repo_files_edit_file', request_method='GET',
         renderer='rhodecode:templates/files/files_edit.mako')
 
     config.add_route(
         name='repo_files_update_file',
         pattern='/{repo_name:.*?[^/]}/update_file/{commit_id}/{f_path:.*}',
         repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_files_update_file',
         route_name='repo_files_update_file', request_method='POST',
         renderer=None)
 
     config.add_route(
         name='repo_files_add_file',
         pattern='/{repo_name:.*?[^/]}/add_file/{commit_id}/{f_path:.*}',
         repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_files_add_file',
         route_name='repo_files_add_file', request_method='GET',
         renderer='rhodecode:templates/files/files_add.mako')
 
     config.add_route(
         name='repo_files_upload_file',
         pattern='/{repo_name:.*?[^/]}/upload_file/{commit_id}/{f_path:.*}',
         repo_route=True)
     config.add_view(
         RepoFilesView,
         attr='repo_files_add_file',
         route_name='repo_files_upload_file', request_method='GET',
         renderer='rhodecode:templates/files/files_upload.mako')
     config.add_view( # POST creates
         RepoFilesView,
         attr='repo_files_upload_file',
         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:.*}',
         repo_route=True)
     config.add_view( # POST creates
         RepoFilesView,
         attr='repo_files_create_file',
         route_name='repo_files_create_file', request_method='POST',
         renderer=None)
 
     # Refs data
     config.add_route(
         name='repo_refs_data',
         pattern='/{repo_name:.*?[^/]}/refs-data', repo_route=True)
     config.add_view(
         RepoSummaryView,
         attr='repo_refs_data',
         route_name='repo_refs_data', request_method='GET',
         renderer='json_ext')
 
     config.add_route(
         name='repo_refs_changelog_data',
         pattern='/{repo_name:.*?[^/]}/refs-data-changelog', repo_route=True)
     config.add_view(
         RepoSummaryView,
         attr='repo_refs_changelog_data',
         route_name='repo_refs_changelog_data', request_method='GET',
         renderer='json_ext')
 
     config.add_route(
         name='repo_stats',
         pattern='/{repo_name:.*?[^/]}/repo_stats/{commit_id}', repo_route=True)
     config.add_view(
         RepoSummaryView,
         attr='repo_stats',
         route_name='repo_stats', request_method='GET',
         renderer='json_ext')
 
     # Commits
     config.add_route(
         name='repo_commits',
         pattern='/{repo_name:.*?[^/]}/commits', repo_route=True)
     config.add_view(
         RepoChangelogView,
         attr='repo_changelog',
         route_name='repo_commits', request_method='GET',
         renderer='rhodecode:templates/commits/changelog.mako')
     # old routes for backward compat
     config.add_view(
         RepoChangelogView,
         attr='repo_changelog',
         route_name='repo_changelog', request_method='GET',
         renderer='rhodecode:templates/commits/changelog.mako')
 
     config.add_route(
         name='repo_commits_elements',
         pattern='/{repo_name:.*?[^/]}/commits_elements', repo_route=True)
     config.add_view(
         RepoChangelogView,
         attr='repo_commits_elements',
         route_name='repo_commits_elements', request_method=('GET', 'POST'),
         renderer='rhodecode:templates/commits/changelog_elements.mako',
         xhr=True)
 
     config.add_route(
         name='repo_commits_elements_file',
         pattern='/{repo_name:.*?[^/]}/commits_elements/{commit_id}/{f_path:.*}', repo_route=True)
     config.add_view(
         RepoChangelogView,
         attr='repo_commits_elements',
         route_name='repo_commits_elements_file', request_method=('GET', 'POST'),
         renderer='rhodecode:templates/commits/changelog_elements.mako',
         xhr=True)
 
     config.add_route(
         name='repo_commits_file',
         pattern='/{repo_name:.*?[^/]}/commits/{commit_id}/{f_path:.*}', repo_route=True)
     config.add_view(
         RepoChangelogView,
         attr='repo_changelog',
         route_name='repo_commits_file', request_method='GET',
         renderer='rhodecode:templates/commits/changelog.mako')
     # old routes for backward compat
     config.add_view(
         RepoChangelogView,
         attr='repo_changelog',
         route_name='repo_changelog_file', request_method='GET',
         renderer='rhodecode:templates/commits/changelog.mako')
 
     # Changelog (old deprecated name for commits page)
     config.add_route(
         name='repo_changelog',
         pattern='/{repo_name:.*?[^/]}/changelog', repo_route=True)
     config.add_route(
         name='repo_changelog_file',
         pattern='/{repo_name:.*?[^/]}/changelog/{commit_id}/{f_path:.*}', repo_route=True)
 
     # Compare
     config.add_route(
         name='repo_compare_select',
         pattern='/{repo_name:.*?[^/]}/compare', repo_route=True)
     config.add_view(
         RepoCompareView,
         attr='compare_select',
         route_name='repo_compare_select', request_method='GET',
         renderer='rhodecode:templates/compare/compare_diff.mako')
 
     config.add_route(
         name='repo_compare',
         pattern='/{repo_name:.*?[^/]}/compare/{source_ref_type}@{source_ref:.*?}...{target_ref_type}@{target_ref:.*?}', repo_route=True)
     config.add_view(
         RepoCompareView,
         attr='compare',
         route_name='repo_compare', request_method='GET',
         renderer=None)
 
     # Tags
     config.add_route(
         name='tags_home',
         pattern='/{repo_name:.*?[^/]}/tags', repo_route=True)
     config.add_view(
         RepoTagsView,
         attr='tags',
         route_name='tags_home', request_method='GET',
         renderer='rhodecode:templates/tags/tags.mako')
 
     # Branches
     config.add_route(
         name='branches_home',
         pattern='/{repo_name:.*?[^/]}/branches', repo_route=True)
     config.add_view(
         RepoBranchesView,
         attr='branches',
         route_name='branches_home', request_method='GET',
         renderer='rhodecode:templates/branches/branches.mako')
 
     # Bookmarks
     config.add_route(
         name='bookmarks_home',
         pattern='/{repo_name:.*?[^/]}/bookmarks', repo_route=True)
     config.add_view(
         RepoBookmarksView,
         attr='bookmarks',
         route_name='bookmarks_home', request_method='GET',
         renderer='rhodecode:templates/bookmarks/bookmarks.mako')
 
     # Forks
     config.add_route(
         name='repo_fork_new',
         pattern='/{repo_name:.*?[^/]}/fork', repo_route=True,
         repo_forbid_when_archived=True,
         repo_accepted_types=['hg', 'git'])
     config.add_view(
         RepoForksView,
         attr='repo_fork_new',
         route_name='repo_fork_new', request_method='GET',
         renderer='rhodecode:templates/forks/forks.mako')
 
     config.add_route(
         name='repo_fork_create',
         pattern='/{repo_name:.*?[^/]}/fork/create', repo_route=True,
         repo_forbid_when_archived=True,
         repo_accepted_types=['hg', 'git'])
     config.add_view(
         RepoForksView,
         attr='repo_fork_create',
         route_name='repo_fork_create', request_method='POST',
         renderer='rhodecode:templates/forks/fork.mako')
 
     config.add_route(
         name='repo_forks_show_all',
         pattern='/{repo_name:.*?[^/]}/forks', repo_route=True,
         repo_accepted_types=['hg', 'git'])
     config.add_view(
         RepoForksView,
         attr='repo_forks_show_all',
         route_name='repo_forks_show_all', request_method='GET',
         renderer='rhodecode:templates/forks/forks.mako')
     
     config.add_route(
         name='repo_forks_data',
         pattern='/{repo_name:.*?[^/]}/forks/data', repo_route=True,
         repo_accepted_types=['hg', 'git'])
     config.add_view(
         RepoForksView,
         attr='repo_forks_data',
         route_name='repo_forks_data', request_method='GET',
         renderer='json_ext', xhr=True)
 
     # Pull Requests
     config.add_route(
         name='pullrequest_show',
         pattern=r'/{repo_name:.*?[^/]}/pull-request/{pull_request_id:\d+}',
         repo_route=True)
     config.add_view(
         RepoPullRequestsView,
         attr='pull_request_show',
         route_name='pullrequest_show', request_method='GET',
         renderer='rhodecode:templates/pullrequests/pullrequest_show.mako')
 
     config.add_route(
         name='pullrequest_show_all',
         pattern='/{repo_name:.*?[^/]}/pull-request',
         repo_route=True, repo_accepted_types=['hg', 'git'])
     config.add_view(
         RepoPullRequestsView,
         attr='pull_request_list',
         route_name='pullrequest_show_all', request_method='GET',
         renderer='rhodecode:templates/pullrequests/pullrequests.mako')
 
     config.add_route(
         name='pullrequest_show_all_data',
         pattern='/{repo_name:.*?[^/]}/pull-request-data',
         repo_route=True, repo_accepted_types=['hg', 'git'])
     config.add_view(
         RepoPullRequestsView,
         attr='pull_request_list_data',
         route_name='pullrequest_show_all_data', request_method='GET',
         renderer='json_ext', xhr=True)
 
     config.add_route(
         name='pullrequest_repo_refs',
         pattern='/{repo_name:.*?[^/]}/pull-request/refs/{target_repo_name:.*?[^/]}',
         repo_route=True)
     config.add_view(
         RepoPullRequestsView,
         attr='pull_request_repo_refs',
         route_name='pullrequest_repo_refs', request_method='GET',
         renderer='json_ext', xhr=True)
 
     config.add_route(
         name='pullrequest_repo_targets',
         pattern='/{repo_name:.*?[^/]}/pull-request/repo-targets',
         repo_route=True)
     config.add_view(
         RepoPullRequestsView,
         attr='pullrequest_repo_targets',
         route_name='pullrequest_repo_targets', request_method='GET',
         renderer='json_ext', xhr=True)
 
     config.add_route(
         name='pullrequest_new',
         pattern='/{repo_name:.*?[^/]}/pull-request/new',
         repo_route=True, repo_accepted_types=['hg', 'git'],
         repo_forbid_when_archived=True)
     config.add_view(
         RepoPullRequestsView,
         attr='pull_request_new',
         route_name='pullrequest_new', request_method='GET',
         renderer='rhodecode:templates/pullrequests/pullrequest.mako')
 
     config.add_route(
         name='pullrequest_create',
         pattern='/{repo_name:.*?[^/]}/pull-request/create',
         repo_route=True, repo_accepted_types=['hg', 'git'],
         repo_forbid_when_archived=True)
     config.add_view(
         RepoPullRequestsView,
         attr='pull_request_create',
         route_name='pullrequest_create', request_method='POST',
         renderer=None)
 
     config.add_route(
         name='pullrequest_update',
         pattern=r'/{repo_name:.*?[^/]}/pull-request/{pull_request_id:\d+}/update',
         repo_route=True, repo_forbid_when_archived=True)
     config.add_view(
         RepoPullRequestsView,
         attr='pull_request_update',
         route_name='pullrequest_update', request_method='POST',
         renderer='json_ext')
 
     config.add_route(
         name='pullrequest_merge',
         pattern=r'/{repo_name:.*?[^/]}/pull-request/{pull_request_id:\d+}/merge',
         repo_route=True, repo_forbid_when_archived=True)
     config.add_view(
         RepoPullRequestsView,
         attr='pull_request_merge',
         route_name='pullrequest_merge', request_method='POST',
         renderer='json_ext')
 
     config.add_route(
         name='pullrequest_delete',
         pattern=r'/{repo_name:.*?[^/]}/pull-request/{pull_request_id:\d+}/delete',
         repo_route=True, repo_forbid_when_archived=True)
     config.add_view(
         RepoPullRequestsView,
         attr='pull_request_delete',
         route_name='pullrequest_delete', request_method='POST',
         renderer='json_ext')
 
     config.add_route(
         name='pullrequest_comment_create',
         pattern=r'/{repo_name:.*?[^/]}/pull-request/{pull_request_id:\d+}/comment',
         repo_route=True)
     config.add_view(
         RepoPullRequestsView,
         attr='pull_request_comment_create',
         route_name='pullrequest_comment_create', request_method='POST',
         renderer='json_ext')
 
     config.add_route(
         name='pullrequest_comment_edit',
         pattern=r'/{repo_name:.*?[^/]}/pull-request/{pull_request_id:\d+}/comment/{comment_id}/edit',
         repo_route=True, repo_accepted_types=['hg', 'git'])
     config.add_view(
         RepoPullRequestsView,
         attr='pull_request_comment_edit',
         route_name='pullrequest_comment_edit', request_method='POST',
         renderer='json_ext')
 
     config.add_route(
         name='pullrequest_comment_delete',
         pattern=r'/{repo_name:.*?[^/]}/pull-request/{pull_request_id:\d+}/comment/{comment_id}/delete',
         repo_route=True, repo_accepted_types=['hg', 'git'])
     config.add_view(
         RepoPullRequestsView,
         attr='pull_request_comment_delete',
         route_name='pullrequest_comment_delete', request_method='POST',
         renderer='json_ext')
 
     config.add_route(
         name='pullrequest_comments',
         pattern=r'/{repo_name:.*?[^/]}/pull-request/{pull_request_id:\d+}/comments',
         repo_route=True)
     config.add_view(
         RepoPullRequestsView,
         attr='pullrequest_comments',
         route_name='pullrequest_comments', request_method='POST',
         renderer='string_html', xhr=True)
 
     config.add_route(
         name='pullrequest_todos',
         pattern=r'/{repo_name:.*?[^/]}/pull-request/{pull_request_id:\d+}/todos',
         repo_route=True)
     config.add_view(
         RepoPullRequestsView,
         attr='pullrequest_todos',
         route_name='pullrequest_todos', request_method='POST',
         renderer='string_html', xhr=True)
 
     config.add_route(
         name='pullrequest_drafts',
         pattern=r'/{repo_name:.*?[^/]}/pull-request/{pull_request_id:\d+}/drafts',
         repo_route=True)
     config.add_view(
         RepoPullRequestsView,
         attr='pullrequest_drafts',
         route_name='pullrequest_drafts', request_method='POST',
         renderer='string_html', xhr=True)
 
     # Artifacts, (EE feature)
     config.add_route(
         name='repo_artifacts_list',
         pattern='/{repo_name:.*?[^/]}/artifacts', repo_route=True)
     config.add_view(
         RepoArtifactsView,
         attr='repo_artifacts',
         route_name='repo_artifacts_list', request_method='GET',
         renderer='rhodecode:templates/artifacts/artifact_list.mako')
 
     # Settings
     config.add_route(
         name='edit_repo',
         pattern='/{repo_name:.*?[^/]}/settings', repo_route=True)
     config.add_view(
         RepoSettingsView,
         attr='edit_settings',
         route_name='edit_repo', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
     # update is POST on edit_repo
     config.add_view(
         RepoSettingsView,
         attr='edit_settings_update',
         route_name='edit_repo', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     # Settings advanced
     config.add_route(
         name='edit_repo_advanced',
         pattern='/{repo_name:.*?[^/]}/settings/advanced', repo_route=True)
     config.add_view(
         RepoSettingsAdvancedView,
         attr='edit_advanced',
         route_name='edit_repo_advanced', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
     
     config.add_route(
         name='edit_repo_advanced_archive',
         pattern='/{repo_name:.*?[^/]}/settings/advanced/archive', repo_route=True)
     config.add_view(
         RepoSettingsAdvancedView,
         attr='edit_advanced_archive',
         route_name='edit_repo_advanced_archive', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
     
     config.add_route(
         name='edit_repo_advanced_delete',
         pattern='/{repo_name:.*?[^/]}/settings/advanced/delete', repo_route=True)
     config.add_view(
         RepoSettingsAdvancedView,
         attr='edit_advanced_delete',
         route_name='edit_repo_advanced_delete', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     config.add_route(
         name='edit_repo_advanced_locking',
         pattern='/{repo_name:.*?[^/]}/settings/advanced/locking', repo_route=True)
     config.add_view(
         RepoSettingsAdvancedView,
         attr='edit_advanced_toggle_locking',
         route_name='edit_repo_advanced_locking', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     config.add_route(
         name='edit_repo_advanced_journal',
         pattern='/{repo_name:.*?[^/]}/settings/advanced/journal', repo_route=True)
     config.add_view(
         RepoSettingsAdvancedView,
         attr='edit_advanced_journal',
         route_name='edit_repo_advanced_journal', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
     
     config.add_route(
         name='edit_repo_advanced_fork',
         pattern='/{repo_name:.*?[^/]}/settings/advanced/fork', repo_route=True)
     config.add_view(
         RepoSettingsAdvancedView,
         attr='edit_advanced_fork',
         route_name='edit_repo_advanced_fork', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     config.add_route(
         name='edit_repo_advanced_hooks',
         pattern='/{repo_name:.*?[^/]}/settings/advanced/hooks', repo_route=True)
     config.add_view(
         RepoSettingsAdvancedView,
         attr='edit_advanced_install_hooks',
         route_name='edit_repo_advanced_hooks', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     # Caches
     config.add_route(
         name='edit_repo_caches',
         pattern='/{repo_name:.*?[^/]}/settings/caches', repo_route=True)
     config.add_view(
         RepoCachesView,
         attr='repo_caches',
         route_name='edit_repo_caches', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
     config.add_view(
         RepoCachesView,
         attr='repo_caches_purge',
         route_name='edit_repo_caches', request_method='POST')
 
     # Permissions
     config.add_route(
         name='edit_repo_perms',
         pattern='/{repo_name:.*?[^/]}/settings/permissions', repo_route=True)
     config.add_view(
         RepoSettingsPermissionsView,
         attr='edit_permissions',
         route_name='edit_repo_perms', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
     config.add_view(
         RepoSettingsPermissionsView,
         attr='edit_permissions_update',
         route_name='edit_repo_perms', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     config.add_route(
         name='edit_repo_perms_set_private',
         pattern='/{repo_name:.*?[^/]}/settings/permissions/set_private', repo_route=True)
     config.add_view(
         RepoSettingsPermissionsView,
         attr='edit_permissions_set_private_repo',
         route_name='edit_repo_perms_set_private', request_method='POST',
         renderer='json_ext')
 
     # Permissions Branch (EE feature)
     config.add_route(
         name='edit_repo_perms_branch',
         pattern='/{repo_name:.*?[^/]}/settings/branch_permissions', repo_route=True)
     config.add_view(
         RepoSettingsBranchPermissionsView,
         attr='branch_permissions',
         route_name='edit_repo_perms_branch', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     config.add_route(
         name='edit_repo_perms_branch_delete',
         pattern='/{repo_name:.*?[^/]}/settings/branch_permissions/{rule_id}/delete',
         repo_route=True)
     ## Only implemented in EE
 
     # Maintenance
     config.add_route(
         name='edit_repo_maintenance',
         pattern='/{repo_name:.*?[^/]}/settings/maintenance', repo_route=True)
     config.add_view(
         RepoMaintenanceView,
         attr='repo_maintenance',
         route_name='edit_repo_maintenance', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     config.add_route(
         name='edit_repo_maintenance_execute',
         pattern='/{repo_name:.*?[^/]}/settings/maintenance/execute', repo_route=True)
     config.add_view(
         RepoMaintenanceView,
         attr='repo_maintenance_execute',
         route_name='edit_repo_maintenance_execute', request_method='GET',
         renderer='json', xhr=True)
 
     # Fields
     config.add_route(
         name='edit_repo_fields',
         pattern='/{repo_name:.*?[^/]}/settings/fields', repo_route=True)
     config.add_view(
         RepoSettingsFieldsView,
         attr='repo_field_edit',
         route_name='edit_repo_fields', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     config.add_route(
         name='edit_repo_fields_create',
         pattern='/{repo_name:.*?[^/]}/settings/fields/create', repo_route=True)
     config.add_view(
         RepoSettingsFieldsView,
         attr='repo_field_create',
         route_name='edit_repo_fields_create', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     config.add_route(
         name='edit_repo_fields_delete',
         pattern='/{repo_name:.*?[^/]}/settings/fields/{field_id}/delete', repo_route=True)
     config.add_view(
         RepoSettingsFieldsView,
         attr='repo_field_delete',
         route_name='edit_repo_fields_delete', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     # quick actions: locking
     config.add_route(
         name='repo_settings_quick_actions',
         pattern='/{repo_name:.*?[^/]}/settings/quick-action', repo_route=True)
     config.add_view(
         RepoSettingsView,
         attr='repo_settings_quick_actions',
         route_name='repo_settings_quick_actions', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     # Remote
     config.add_route(
         name='edit_repo_remote',
         pattern='/{repo_name:.*?[^/]}/settings/remote', repo_route=True)
     config.add_view(
         RepoSettingsRemoteView,
         attr='repo_remote_edit_form',
         route_name='edit_repo_remote', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     config.add_route(
         name='edit_repo_remote_pull',
         pattern='/{repo_name:.*?[^/]}/settings/remote/pull', repo_route=True)
     config.add_view(
         RepoSettingsRemoteView,
         attr='repo_remote_pull_changes',
         route_name='edit_repo_remote_pull', request_method='POST',
         renderer=None)
 
     config.add_route(
         name='edit_repo_remote_push',
         pattern='/{repo_name:.*?[^/]}/settings/remote/push', repo_route=True)
 
     # Statistics
     config.add_route(
         name='edit_repo_statistics',
         pattern='/{repo_name:.*?[^/]}/settings/statistics', repo_route=True)
     config.add_view(
         RepoSettingsView,
         attr='edit_statistics_form',
         route_name='edit_repo_statistics', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
     
     config.add_route(
         name='edit_repo_statistics_reset',
         pattern='/{repo_name:.*?[^/]}/settings/statistics/update', repo_route=True)
     config.add_view(
         RepoSettingsView,
         attr='repo_statistics_reset',
         route_name='edit_repo_statistics_reset', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     # Issue trackers
     config.add_route(
         name='edit_repo_issuetracker',
         pattern='/{repo_name:.*?[^/]}/settings/issue_trackers', repo_route=True)
     config.add_view(
         RepoSettingsIssueTrackersView,
         attr='repo_issuetracker',
         route_name='edit_repo_issuetracker', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
     
     config.add_route(
         name='edit_repo_issuetracker_test',
         pattern='/{repo_name:.*?[^/]}/settings/issue_trackers/test', repo_route=True)
     config.add_view(
         RepoSettingsIssueTrackersView,
         attr='repo_issuetracker_test',
         route_name='edit_repo_issuetracker_test', request_method='POST',
         renderer='string', xhr=True)
 
     config.add_route(
         name='edit_repo_issuetracker_delete',
         pattern='/{repo_name:.*?[^/]}/settings/issue_trackers/delete', repo_route=True)
     config.add_view(
         RepoSettingsIssueTrackersView,
         attr='repo_issuetracker_delete',
         route_name='edit_repo_issuetracker_delete', request_method='POST',
         renderer='json_ext', xhr=True)
 
     config.add_route(
         name='edit_repo_issuetracker_update',
         pattern='/{repo_name:.*?[^/]}/settings/issue_trackers/update', repo_route=True)
     config.add_view(
         RepoSettingsIssueTrackersView,
         attr='repo_issuetracker_update',
         route_name='edit_repo_issuetracker_update', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     # VCS Settings
     config.add_route(
         name='edit_repo_vcs',
         pattern='/{repo_name:.*?[^/]}/settings/vcs', repo_route=True)
     config.add_view(
         RepoSettingsVcsView,
         attr='repo_vcs_settings',
         route_name='edit_repo_vcs', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     config.add_route(
         name='edit_repo_vcs_update',
         pattern='/{repo_name:.*?[^/]}/settings/vcs/update', repo_route=True)
     config.add_view(
         RepoSettingsVcsView,
         attr='repo_settings_vcs_update',
         route_name='edit_repo_vcs_update', request_method='POST',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     # svn pattern
     config.add_route(
         name='edit_repo_vcs_svn_pattern_delete',
         pattern='/{repo_name:.*?[^/]}/settings/vcs/svn_pattern/delete', repo_route=True)
     config.add_view(
         RepoSettingsVcsView,
         attr='repo_settings_delete_svn_pattern',
         route_name='edit_repo_vcs_svn_pattern_delete', request_method='POST',
         renderer='json_ext', xhr=True)
 
     # Repo Review Rules (EE feature)
     config.add_route(
         name='repo_reviewers',
         pattern='/{repo_name:.*?[^/]}/settings/review/rules', repo_route=True)
     config.add_view(
         RepoReviewRulesView,
         attr='repo_review_rules',
         route_name='repo_reviewers', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     config.add_route(
         name='repo_default_reviewers_data',
         pattern='/{repo_name:.*?[^/]}/settings/review/default-reviewers', repo_route=True)
     config.add_view(
         RepoReviewRulesView,
         attr='repo_default_reviewers_data',
         route_name='repo_default_reviewers_data', request_method='GET',
         renderer='json_ext')
 
     # Repo Automation (EE feature)
     config.add_route(
         name='repo_automation',
         pattern='/{repo_name:.*?[^/]}/settings/automation', repo_route=True)
     config.add_view(
         RepoAutomationView,
         attr='repo_automation',
         route_name='repo_automation', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     # Strip
     config.add_route(
         name='edit_repo_strip',
         pattern='/{repo_name:.*?[^/]}/settings/strip', repo_route=True)
     config.add_view(
         RepoStripView,
         attr='strip',
         route_name='edit_repo_strip', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     config.add_route(
         name='strip_check',
         pattern='/{repo_name:.*?[^/]}/settings/strip_check', repo_route=True)
     config.add_view(
         RepoStripView,
         attr='strip_check',
         route_name='strip_check', request_method='POST',
         renderer='json', xhr=True)
 
     config.add_route(
         name='strip_execute',
         pattern='/{repo_name:.*?[^/]}/settings/strip_execute', repo_route=True)
     config.add_view(
         RepoStripView,
         attr='strip_execute',
         route_name='strip_execute', request_method='POST',
         renderer='json', xhr=True)
 
     # Audit logs
     config.add_route(
         name='edit_repo_audit_logs',
         pattern='/{repo_name:.*?[^/]}/settings/audit_logs', repo_route=True)
     config.add_view(
         AuditLogsView,
         attr='repo_audit_logs',
         route_name='edit_repo_audit_logs', request_method='GET',
         renderer='rhodecode:templates/admin/repos/repo_edit.mako')
 
     # ATOM/RSS Feed, shouldn't contain slashes for outlook compatibility
     config.add_route(
         name='rss_feed_home',
         pattern='/{repo_name:.*?[^/]}/feed-rss', repo_route=True)
     config.add_view(
         RepoFeedView,
         attr='rss',
         route_name='rss_feed_home', request_method='GET', renderer=None)
 
     config.add_route(
         name='rss_feed_home_old',
         pattern='/{repo_name:.*?[^/]}/feed/rss', repo_route=True)
     config.add_view(
         RepoFeedView,
         attr='rss',
         route_name='rss_feed_home_old', request_method='GET', renderer=None)
 
     config.add_route(
         name='atom_feed_home',
         pattern='/{repo_name:.*?[^/]}/feed-atom', repo_route=True)
     config.add_view(
         RepoFeedView,
         attr='atom',
         route_name='atom_feed_home', request_method='GET', renderer=None)
 
     config.add_route(
         name='atom_feed_home_old',
         pattern='/{repo_name:.*?[^/]}/feed/atom', repo_route=True)
     config.add_view(
         RepoFeedView,
         attr='atom',
         route_name='atom_feed_home_old', request_method='GET', renderer=None)
 
     # NOTE(marcink): needs to be at the end for catch-all
     add_route_with_slash(
         config,
         name='repo_summary',
         pattern='/{repo_name:.*?[^/]}', repo_route=True)
     config.add_view(
         RepoSummaryView,
         attr='summary',
         route_name='repo_summary', request_method='GET',
         renderer='rhodecode:templates/summary/summary.mako')
     
     # TODO(marcink): there's no such route??
     config.add_view(
         RepoSummaryView,
         attr='summary',
         route_name='repo_summary_slash', request_method='GET',
         renderer='rhodecode:templates/summary/summary.mako')
  No newline at end of file

rhodecode/apps/repository/tests/test_repo_files.py

@@ -1,1090 +1,1152 b''
 
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import os
 
 import mock
 import pytest
 from collections import OrderedDict
 
 from rhodecode.apps.repository.tests.test_repo_compare import ComparePage
 from rhodecode.apps.repository.views.repo_files import RepoFilesView, get_archive_name, get_path_sha
 from rhodecode.lib import helpers as h
 from rhodecode.lib.ext_json import json
 from rhodecode.lib.str_utils import safe_str
 from rhodecode.lib.vcs import nodes
 from rhodecode.lib.vcs.conf import settings
 from rhodecode.model.db import Session, Repository
 
 from rhodecode.tests import assert_session_flash
 from rhodecode.tests.fixture import Fixture
 from rhodecode.tests.routes import route_path
 
 
 fixture = Fixture()
 
 
 def get_node_history(backend_type):
     return {
         'hg': json.loads(fixture.load_resource('hg_node_history_response.json')),
         'git': json.loads(fixture.load_resource('git_node_history_response.json')),
         'svn': json.loads(fixture.load_resource('svn_node_history_response.json')),
     }[backend_type]
 
 
 def assert_files_in_response(response, files, params):
     template = (
         'href="/%(repo_name)s/files/%(commit_id)s/%(name)s"')
     _assert_items_in_response(response, files, template, params)
 
 
 def assert_dirs_in_response(response, dirs, params):
     template = (
         'href="/%(repo_name)s/files/%(commit_id)s/%(name)s"')
     _assert_items_in_response(response, dirs, template, params)
 
 
 def _assert_items_in_response(response, items, template, params):
     for item in items:
         item_params = {'name': item}
         item_params.update(params)
         response.mustcontain(template % item_params)
 
 
 def assert_timeago_in_response(response, items, params):
     for item in items:
         response.mustcontain(h.age_component(params['date']))
 
 
 @pytest.mark.usefixtures("app")
 class TestFilesViews(object):
 
     def test_show_files(self, backend):
         response = self.app.get(
             route_path('repo_files',
                        repo_name=backend.repo_name,
                        commit_id='tip', f_path='/'))
         commit = backend.repo.get_commit()
 
         params = {
             'repo_name': backend.repo_name,
             'commit_id': commit.raw_id,
             'date': commit.date
         }
         assert_dirs_in_response(response, ['docs', 'vcs'], params)
         files = [
             '.gitignore',
             '.hgignore',
             '.hgtags',
             # TODO: missing in Git
             # '.travis.yml',
             'MANIFEST.in',
             'README.rst',
             # TODO: File is missing in svn repository
             # 'run_test_and_report.sh',
             'setup.cfg',
             'setup.py',
             'test_and_report.sh',
             'tox.ini',
         ]
         assert_files_in_response(response, files, params)
         assert_timeago_in_response(response, files, params)
 
     def test_show_files_links_submodules_with_absolute_url(self, backend_hg):
         repo = backend_hg['subrepos']
         response = self.app.get(
             route_path('repo_files',
                        repo_name=repo.repo_name,
                        commit_id='tip', f_path='/'))
         assert_response = response.assert_response()
         assert_response.contains_one_link(
             'absolute-path @ 000000000000', 'http://example.com/absolute-path')
 
     def test_show_files_links_submodules_with_absolute_url_subpaths(
             self, backend_hg):
         repo = backend_hg['subrepos']
         response = self.app.get(
             route_path('repo_files',
                        repo_name=repo.repo_name,
                        commit_id='tip', f_path='/'))
         assert_response = response.assert_response()
         assert_response.contains_one_link(
             'subpaths-path @ 000000000000',
             'http://sub-base.example.com/subpaths-path')
 
     @pytest.mark.xfail_backends("svn", reason="Depends on branch support")
     def test_files_menu(self, backend):
         new_branch = "temp_branch_name"
         commits = [
             {'message': 'a'},
             {'message': 'b', 'branch': new_branch}
         ]
         backend.create_repo(commits)
         backend.repo.landing_rev = f"branch:{new_branch}"
         Session().commit()
 
         # get response based on tip and not new commit
         response = self.app.get(
             route_path('repo_files',
                        repo_name=backend.repo_name,
                        commit_id='tip', f_path='/'))
 
         # make sure Files menu url is not tip but new commit
         landing_rev = backend.repo.landing_ref_name
         files_url = route_path('repo_files:default_path',
                                repo_name=backend.repo_name,
                                commit_id=landing_rev, params={'at': landing_rev})
 
         assert landing_rev != 'tip'
         response.mustcontain(f'<li class="active"><a class="menulink" href="{files_url}">')
 
     def test_show_files_commit(self, backend):
         commit = backend.repo.get_commit(commit_idx=32)
 
         response = self.app.get(
             route_path('repo_files',
                        repo_name=backend.repo_name,
                        commit_id=commit.raw_id, f_path='/'))
 
         dirs = ['docs', 'tests']
         files = ['README.rst']
         params = {
             'repo_name': backend.repo_name,
             'commit_id': commit.raw_id,
         }
         assert_dirs_in_response(response, dirs, params)
         assert_files_in_response(response, files, params)
 
     def test_show_files_different_branch(self, backend):
         branches = dict(
             hg=(150, ['git']),
             # TODO: Git test repository does not contain other branches
             git=(633, ['master']),
             # TODO: Branch support in Subversion
             svn=(150, [])
         )
         idx, branches = branches[backend.alias]
         commit = backend.repo.get_commit(commit_idx=idx)
         response = self.app.get(
             route_path('repo_files',
                        repo_name=backend.repo_name,
                        commit_id=commit.raw_id, f_path='/'))
 
         assert_response = response.assert_response()
         for branch in branches:
             assert_response.element_contains('.tags .branchtag', branch)
 
     def test_show_files_paging(self, backend):
         repo = backend.repo
         indexes = [73, 92, 109, 1, 0]
         idx_map = [(rev, repo.get_commit(commit_idx=rev).raw_id)
                    for rev in indexes]
 
         for idx in idx_map:
             response = self.app.get(
             route_path('repo_files',
                        repo_name=backend.repo_name,
                        commit_id=idx[1], f_path='/'))
 
             response.mustcontain("""r%s:%s""" % (idx[0], idx[1][:8]))
 
     def test_file_source(self, backend):
         commit = backend.repo.get_commit(commit_idx=167)
         response = self.app.get(
             route_path('repo_files',
                        repo_name=backend.repo_name,
                        commit_id=commit.raw_id, f_path='vcs/nodes.py'))
 
         msgbox = """<div class="commit">%s</div>"""
         response.mustcontain(msgbox % (commit.message, ))
 
         assert_response = response.assert_response()
         if commit.branch:
             assert_response.element_contains(
                 '.tags.tags-main .branchtag', commit.branch)
         if commit.tags:
             for tag in commit.tags:
                 assert_response.element_contains('.tags.tags-main .tagtag', tag)
 
     def test_file_source_annotated(self, backend):
         response = self.app.get(
             route_path('repo_files:annotated',
                        repo_name=backend.repo_name,
                        commit_id='tip', f_path='vcs/nodes.py'))
         expected_commits = {
             'hg': 'r356',
             'git': 'r345',
             'svn': 'r208',
         }
         response.mustcontain(expected_commits[backend.alias])
 
     def test_file_source_authors(self, backend):
         response = self.app.get(
             route_path('repo_file_authors',
                        repo_name=backend.repo_name,
                        commit_id='tip', f_path='vcs/nodes.py'))
         expected_authors = {
             'hg': ('Marcin Kuzminski', 'Lukasz Balcerzak'),
             'git': ('Marcin Kuzminski', 'Lukasz Balcerzak'),
             'svn': ('marcin', 'lukasz'),
         }
 
         for author in expected_authors[backend.alias]:
             response.mustcontain(author)
 
     def test_file_source_authors_with_annotation(self, backend):
         response = self.app.get(
             route_path('repo_file_authors',
                        repo_name=backend.repo_name,
                        commit_id='tip', f_path='vcs/nodes.py',
                        params=dict(annotate=1)))
         expected_authors = {
             'hg': ('Marcin Kuzminski', 'Lukasz Balcerzak'),
             'git': ('Marcin Kuzminski', 'Lukasz Balcerzak'),
             'svn': ('marcin', 'lukasz'),
         }
 
         for author in expected_authors[backend.alias]:
             response.mustcontain(author)
 
     def test_file_source_history(self, backend, xhr_header):
         response = self.app.get(
             route_path('repo_file_history',
                        repo_name=backend.repo_name,
                        commit_id='tip', f_path='vcs/nodes.py'),
             extra_environ=xhr_header)
         assert get_node_history(backend.alias) == json.loads(response.body)
 
     def test_file_source_history_svn(self, backend_svn, xhr_header):
         simple_repo = backend_svn['svn-simple-layout']
         response = self.app.get(
             route_path('repo_file_history',
                        repo_name=simple_repo.repo_name,
                        commit_id='tip', f_path='trunk/example.py'),
             extra_environ=xhr_header)
 
         expected_data = json.loads(
             fixture.load_resource('svn_node_history_branches.json'))
 
         assert expected_data == response.json
 
     def test_file_source_history_with_annotation(self, backend, xhr_header):
         response = self.app.get(
             route_path('repo_file_history',
                        repo_name=backend.repo_name,
                        commit_id='tip', f_path='vcs/nodes.py',
                        params=dict(annotate=1)),
 
             extra_environ=xhr_header)
         assert get_node_history(backend.alias) == json.loads(response.body)
 
     def test_tree_search_top_level(self, backend, xhr_header):
         commit = backend.repo.get_commit(commit_idx=173)
         response = self.app.get(
             route_path('repo_files_nodelist',
                        repo_name=backend.repo_name,
                        commit_id=commit.raw_id, f_path='/'),
             extra_environ=xhr_header)
         assert 'nodes' in response.json
         assert {'name': 'docs', 'type': 'dir'} in response.json['nodes']
 
     def test_tree_search_missing_xhr(self, backend):
         self.app.get(
             route_path('repo_files_nodelist',
                        repo_name=backend.repo_name,
                        commit_id='tip', f_path='/'),
             status=404)
 
     def test_tree_search_at_path(self, backend, xhr_header):
         commit = backend.repo.get_commit(commit_idx=173)
         response = self.app.get(
             route_path('repo_files_nodelist',
                        repo_name=backend.repo_name,
                        commit_id=commit.raw_id, f_path='/docs'),
             extra_environ=xhr_header)
         assert 'nodes' in response.json
         nodes = response.json['nodes']
         assert {'name': 'docs/api', 'type': 'dir'} in nodes
         assert {'name': 'docs/index.rst', 'type': 'file'} in nodes
 
     def test_tree_search_at_path_2nd_level(self, backend, xhr_header):
         commit = backend.repo.get_commit(commit_idx=173)
         response = self.app.get(
             route_path('repo_files_nodelist',
                        repo_name=backend.repo_name,
                        commit_id=commit.raw_id, f_path='/docs/api'),
             extra_environ=xhr_header)
         assert 'nodes' in response.json
         nodes = response.json['nodes']
         assert {'name': 'docs/api/index.rst', 'type': 'file'} in nodes
 
     def test_tree_search_at_path_missing_xhr(self, backend):
         self.app.get(
             route_path('repo_files_nodelist',
                        repo_name=backend.repo_name,
                        commit_id='tip', f_path='/docs'),
             status=404)
 
     def test_nodetree(self, backend, xhr_header):
         commit = backend.repo.get_commit(commit_idx=173)
         response = self.app.get(
             route_path('repo_nodetree_full',
                        repo_name=backend.repo_name,
                        commit_id=commit.raw_id, f_path='/'),
             extra_environ=xhr_header)
 
         assert_response = response.assert_response()
 
         for attr in ['data-commit-id', 'data-date', 'data-author']:
             elements = assert_response.get_elements('[{}]'.format(attr))
             assert len(elements) > 1
 
             for element in elements:
                 assert element.get(attr)
 
     def test_nodetree_if_file(self, backend, xhr_header):
         commit = backend.repo.get_commit(commit_idx=173)
         response = self.app.get(
             route_path('repo_nodetree_full',
                        repo_name=backend.repo_name,
                        commit_id=commit.raw_id, f_path='README.rst'),
             extra_environ=xhr_header)
         assert response.text == ''
 
     def test_nodetree_wrong_path(self, backend, xhr_header):
         commit = backend.repo.get_commit(commit_idx=173)
         response = self.app.get(
             route_path('repo_nodetree_full',
                        repo_name=backend.repo_name,
                        commit_id=commit.raw_id, f_path='/dont-exist'),
             extra_environ=xhr_header)
 
         err = 'error: There is no file nor ' \
               'directory at the given path'
         assert err in response.text
 
     def test_nodetree_missing_xhr(self, backend):
         self.app.get(
             route_path('repo_nodetree_full',
                        repo_name=backend.repo_name,
                        commit_id='tip', f_path='/'),
             status=404)
 
 
 @pytest.mark.usefixtures("app", "autologin_user")
 class TestRawFileHandling(object):
 
     def test_download_file(self, backend):
         commit = backend.repo.get_commit(commit_idx=173)
         response = self.app.get(
             route_path('repo_file_download',
                        repo_name=backend.repo_name,
                        commit_id=commit.raw_id, f_path='vcs/nodes.py'),)
 
         assert response.content_disposition == 'attachment; filename="nodes.py"; filename*=UTF-8\'\'nodes.py'
         assert response.content_type == "text/x-python"
 
     def test_download_file_wrong_cs(self, backend):
         raw_id = u'ERRORce30c96924232dffcd24178a07ffeb5dfc'
 
         response = self.app.get(
             route_path('repo_file_download',
                        repo_name=backend.repo_name,
                        commit_id=raw_id, f_path='vcs/nodes.svg'),
             status=404)
 
         msg = """No such commit exists for this repository"""
         response.mustcontain(msg)
 
     def test_download_file_wrong_f_path(self, backend):
         commit = backend.repo.get_commit(commit_idx=173)
         f_path = 'vcs/ERRORnodes.py'
 
         response = self.app.get(
             route_path('repo_file_download',
                        repo_name=backend.repo_name,
                        commit_id=commit.raw_id, f_path=f_path),
             status=404)
 
         msg = (
             "There is no file nor directory at the given path: "
             "`%s` at commit %s" % (f_path, commit.short_id))
         response.mustcontain(msg)
 
     def test_file_raw(self, backend):
         commit = backend.repo.get_commit(commit_idx=173)
         response = self.app.get(
             route_path('repo_file_raw',
                        repo_name=backend.repo_name,
                        commit_id=commit.raw_id, f_path='vcs/nodes.py'),)
 
         assert response.content_type == "text/plain"
 
     def test_file_raw_binary(self, backend):
         commit = backend.repo.get_commit()
         response = self.app.get(
             route_path('repo_file_raw',
                        repo_name=backend.repo_name,
                        commit_id=commit.raw_id,
                        f_path='docs/theme/ADC/static/breadcrumb_background.png'),)
 
         assert response.content_disposition == 'inline'
 
     def test_raw_file_wrong_cs(self, backend):
         raw_id = u'ERRORcce30c96924232dffcd24178a07ffeb5dfc'
 
         response = self.app.get(
             route_path('repo_file_raw',
                        repo_name=backend.repo_name,
                        commit_id=raw_id, f_path='vcs/nodes.svg'),
             status=404)
 
         msg = """No such commit exists for this repository"""
         response.mustcontain(msg)
 
     def test_raw_wrong_f_path(self, backend):
         commit = backend.repo.get_commit(commit_idx=173)
         f_path = 'vcs/ERRORnodes.py'
         response = self.app.get(
             route_path('repo_file_raw',
                        repo_name=backend.repo_name,
                        commit_id=commit.raw_id, f_path=f_path),
             status=404)
 
         msg = (
             "There is no file nor directory at the given path: "
             "`%s` at commit %s" % (f_path, commit.short_id))
         response.mustcontain(msg)
 
     def test_raw_svg_should_not_be_rendered(self, backend):
         backend.create_repo()
         backend.ensure_file(b"xss.svg")
         response = self.app.get(
             route_path('repo_file_raw',
                        repo_name=backend.repo_name,
                        commit_id='tip', f_path='xss.svg'),)
         # If the content type is image/svg+xml then it allows to render HTML
         # and malicious SVG.
         assert response.content_type == "text/plain"
 
 
 @pytest.mark.usefixtures("app")
 class TestRepositoryArchival(object):
 
     def test_archival(self, backend):
         backend.enable_downloads()
         commit = backend.repo.get_commit(commit_idx=173)
 
         for a_type, content_type, extension in settings.ARCHIVE_SPECS:
             path_sha = get_path_sha('/')
             filename = get_archive_name(backend.repo_id, backend.repo_name, commit_sha=commit.short_id, ext=extension, path_sha=path_sha)
 
             fname = commit.raw_id + extension
             response = self.app.get(
                 route_path('repo_archivefile',
                            repo_name=backend.repo_name,
                            fname=fname))
 
             assert response.status == '200 OK'
             headers = [
                 ('Content-Disposition', f'attachment; filename={filename}'),
                 ('Content-Type', content_type),
             ]
 
             for header in headers:
                 assert header in list(response.headers.items())
 
     def test_archival_no_hash(self, backend):
         backend.enable_downloads()
         commit = backend.repo.get_commit(commit_idx=173)
         for a_type, content_type, extension in settings.ARCHIVE_SPECS:
             path_sha = get_path_sha('/')
             filename = get_archive_name(backend.repo_id, backend.repo_name, commit_sha=commit.short_id, ext=extension, path_sha=path_sha, with_hash=False)
 
             fname = commit.raw_id + extension
             response = self.app.get(
                 route_path('repo_archivefile',
                            repo_name=backend.repo_name,
                            fname=fname, params={'with_hash': 0}))
 
             assert response.status == '200 OK'
             headers = [
                 ('Content-Disposition', f'attachment; filename={filename}'),
                 ('Content-Type', content_type),
             ]
 
             for header in headers:
                 assert header in list(response.headers.items())
 
     def test_archival_at_path(self, backend):
         backend.enable_downloads()
         commit = backend.repo.get_commit(commit_idx=190)
         at_path = 'vcs'
 
         for a_type, content_type, extension in settings.ARCHIVE_SPECS:
             path_sha = get_path_sha(at_path)
             filename = get_archive_name(backend.repo_id, backend.repo_name, commit_sha=commit.short_id, ext=extension, path_sha=path_sha)
 
             fname = commit.raw_id + extension
             response = self.app.get(
                 route_path('repo_archivefile',
                            repo_name=backend.repo_name,
                            fname=fname, params={'at_path': at_path}))
 
             assert response.status == '200 OK'
             headers = [
                 ('Content-Disposition', f'attachment; filename={filename}'),
                 ('Content-Type', content_type),
             ]
 
             for header in headers:
                 assert header in list(response.headers.items())
 
     @pytest.mark.parametrize('arch_ext',[
         'tar', 'rar', 'x', '..ax', '.zipz', 'tar.gz.tar'])
     def test_archival_wrong_ext(self, backend, arch_ext):
         backend.enable_downloads()
         commit = backend.repo.get_commit(commit_idx=173)
 
         fname = commit.raw_id + '.' + arch_ext
 
         response = self.app.get(
             route_path('repo_archivefile',
                        repo_name=backend.repo_name,
                        fname=fname))
         response.mustcontain(
             'Unknown archive type for: `{}`'.format(fname))
 
     @pytest.mark.parametrize('commit_id', [
         '00x000000', 'tar', 'wrong', '@$@$42413232', '232dffcd'])
     def test_archival_wrong_commit_id(self, backend, commit_id):
         backend.enable_downloads()
         fname = f'{commit_id}.zip'
 
         response = self.app.get(
             route_path('repo_archivefile',
                        repo_name=backend.repo_name,
                        fname=fname))
         response.mustcontain('Unknown commit_id')
 
 
 @pytest.mark.usefixtures("app")
 class TestFilesDiff(object):
 
     @pytest.mark.parametrize("diff", ['diff', 'download', 'raw'])
     def test_file_full_diff(self, backend, diff):
         commit1 = backend.repo.get_commit(commit_idx=-1)
         commit2 = backend.repo.get_commit(commit_idx=-2)
 
         response = self.app.get(
             route_path('repo_files_diff',
                        repo_name=backend.repo_name,
                        f_path='README'),
             params={
                 'diff1': commit2.raw_id,
                 'diff2': commit1.raw_id,
                 'fulldiff': '1',
                 'diff': diff,
             })
 
         if diff == 'diff':
             # use redirect since this is OLD view redirecting to compare page
             response = response.follow()
 
         # It's a symlink to README.rst
         response.mustcontain('README.rst')
         response.mustcontain('No newline at end of file')
 
     def test_file_binary_diff(self, backend):
         commits = [
             {'message': 'First commit'},
             {'message': 'Commit with binary',
              'added': [nodes.FileNode(b'file.bin', content='\0BINARY\0')]},
         ]
         repo = backend.create_repo(commits=commits)
 
         response = self.app.get(
             route_path('repo_files_diff',
                        repo_name=backend.repo_name,
                        f_path='file.bin'),
             params={
                 'diff1': repo.get_commit(commit_idx=0).raw_id,
                 'diff2': repo.get_commit(commit_idx=1).raw_id,
                 'fulldiff': '1',
                 'diff': 'diff',
             })
         # use redirect since this is OLD view redirecting to compare page
         response = response.follow()
         response.mustcontain('Collapse 1 commit')
         file_changes = (1, 0, 0)
 
         compare_page = ComparePage(response)
         compare_page.contains_change_summary(*file_changes)
 
         if backend.alias == 'svn':
             response.mustcontain('new file 10644')
             # TODO(marcink): SVN doesn't yet detect binary changes
         else:
             response.mustcontain('new file 100644')
             response.mustcontain('binary diff hidden')
 
     def test_diff_2way(self, backend):
         commit1 = backend.repo.get_commit(commit_idx=-1)
         commit2 = backend.repo.get_commit(commit_idx=-2)
         response = self.app.get(
             route_path('repo_files_diff_2way_redirect',
                        repo_name=backend.repo_name,
                        f_path='README'),
             params={
                 'diff1': commit2.raw_id,
                 'diff2': commit1.raw_id,
             })
         # use redirect since this is OLD view redirecting to compare page
         response = response.follow()
 
         # It's a symlink to README.rst
         response.mustcontain('README.rst')
         response.mustcontain('No newline at end of file')
 
     def test_requires_one_commit_id(self, backend, autologin_user):
         response = self.app.get(
             route_path('repo_files_diff',
                        repo_name=backend.repo_name,
                        f_path='README.rst'),
             status=400)
         response.mustcontain(
             'Need query parameter', 'diff1', 'diff2', 'to generate a diff.')
 
     def test_returns_no_files_if_file_does_not_exist(self, vcsbackend):
         repo = vcsbackend.repo
         response = self.app.get(
             route_path('repo_files_diff',
                        repo_name=repo.name,
                        f_path='does-not-exist-in-any-commit'),
             params={
                 'diff1': repo[0].raw_id,
                 'diff2': repo[1].raw_id
             })
 
         response = response.follow()
         response.mustcontain('No files')
 
     def test_returns_redirect_if_file_not_changed(self, backend):
         commit = backend.repo.get_commit(commit_idx=-1)
         response = self.app.get(
             route_path('repo_files_diff_2way_redirect',
                        repo_name=backend.repo_name,
                        f_path='README'),
             params={
                 'diff1': commit.raw_id,
                 'diff2': commit.raw_id,
             })
 
         response = response.follow()
         response.mustcontain('No files')
         response.mustcontain('No commits in this compare')
 
     def test_supports_diff_to_different_path_svn(self, backend_svn):
         #TODO: check this case
         return
 
         repo = backend_svn['svn-simple-layout'].scm_instance()
         commit_id_1 = '24'
         commit_id_2 = '26'
 
         response = self.app.get(
             route_path('repo_files_diff',
                        repo_name=backend_svn.repo_name,
                        f_path='trunk/example.py'),
             params={
                 'diff1': 'tags/v0.2/example.py@' + commit_id_1,
                 'diff2': commit_id_2,
             })
 
         response = response.follow()
         response.mustcontain(
             # diff contains this
             "Will print out a useful message on invocation.")
 
         # Note: Expecting that we indicate the user what's being compared
         response.mustcontain("trunk/example.py")
         response.mustcontain("tags/v0.2/example.py")
 
     def test_show_rev_redirects_to_svn_path(self, backend_svn):
         #TODO: check this case
         return
 
         repo = backend_svn['svn-simple-layout'].scm_instance()
         commit_id = repo[-1].raw_id
 
         response = self.app.get(
             route_path('repo_files_diff',
                        repo_name=backend_svn.repo_name,
                        f_path='trunk/example.py'),
             params={
                 'diff1': 'branches/argparse/example.py@' + commit_id,
                 'diff2': commit_id,
             },
             status=302)
         response = response.follow()
         assert response.headers['Location'].endswith(
             'svn-svn-simple-layout/files/26/branches/argparse/example.py')
 
     def test_show_rev_and_annotate_redirects_to_svn_path(self, backend_svn):
         #TODO: check this case
         return
 
         repo = backend_svn['svn-simple-layout'].scm_instance()
         commit_id = repo[-1].raw_id
         response = self.app.get(
             route_path('repo_files_diff',
                        repo_name=backend_svn.repo_name,
                        f_path='trunk/example.py'),
             params={
                 'diff1': 'branches/argparse/example.py@' + commit_id,
                 'diff2': commit_id,
                 'show_rev': 'Show at Revision',
                 'annotate': 'true',
             },
             status=302)
         response = response.follow()
         assert response.headers['Location'].endswith(
             'svn-svn-simple-layout/annotate/26/branches/argparse/example.py')
 
 
 @pytest.mark.usefixtures("app", "autologin_user")
 class TestModifyFilesWithWebInterface(object):
 
     def test_add_file_view(self, backend):
         self.app.get(
             route_path('repo_files_add_file',
                        repo_name=backend.repo_name,
                        commit_id='tip', f_path='/')
             )
 
     @pytest.mark.xfail_backends("svn", reason="Depends on online editing")
     def test_add_file_into_repo_missing_content(self, backend, csrf_token):
         backend.create_repo()
         filename = 'init.py'
         response = self.app.post(
             route_path('repo_files_create_file',
                        repo_name=backend.repo_name,
                        commit_id='tip', f_path='/'),
             params={
                 'content': "",
                 'filename': filename,
                 'csrf_token': csrf_token,
             },
             status=302)
         expected_msg = 'Successfully committed new file `{}`'.format(os.path.join(filename))
         assert_session_flash(response, expected_msg)
 
     def test_add_file_into_repo_missing_filename(self, backend, csrf_token):
         commit_id = backend.repo.get_commit().raw_id
         response = self.app.post(
             route_path('repo_files_create_file',
                        repo_name=backend.repo_name,
                        commit_id=commit_id, f_path='/'),
             params={
                 'content': "foo",
                 'csrf_token': csrf_token,
             },
             status=302)
 
         assert_session_flash(response, 'No filename specified')
 
     def test_add_file_into_repo_errors_and_no_commits(
             self, backend, csrf_token):
         repo = backend.create_repo()
         # Create a file with no filename, it will display an error but
         # the repo has no commits yet
         response = self.app.post(
             route_path('repo_files_create_file',
                        repo_name=repo.repo_name,
                        commit_id='tip', f_path='/'),
             params={
                 'content': "foo",
                 'csrf_token': csrf_token,
             },
             status=302)
 
         assert_session_flash(response, 'No filename specified')
 
         # Not allowed, redirect to the summary
         redirected = response.follow()
         summary_url = h.route_path('repo_summary', repo_name=repo.repo_name)
 
         # As there are no commits, displays the summary page with the error of
         # creating a file with no filename
 
         assert redirected.request.path == summary_url
 
     @pytest.mark.parametrize("filename, clean_filename", [
         ('/abs/foo', 'abs/foo'),
         ('../rel/foo', 'rel/foo'),
         ('file/../foo/foo', 'file/foo/foo'),
     ])
     def test_add_file_into_repo_bad_filenames(self, filename, clean_filename, backend, csrf_token):
         repo = backend.create_repo()
         commit_id = repo.get_commit().raw_id
 
         response = self.app.post(
             route_path('repo_files_create_file',
                        repo_name=repo.repo_name,
                        commit_id=commit_id, f_path='/'),
             params={
                 'content': "foo",
                 'filename': filename,
                 'csrf_token': csrf_token,
             },
             status=302)
 
         expected_msg = 'Successfully committed new file `{}`'.format(clean_filename)
         assert_session_flash(response, expected_msg)
 
     @pytest.mark.parametrize("cnt, filename, content", [
         (1, 'foo.txt', "Content"),
         (2, 'dir/foo.rst', "Content"),
         (3, 'dir/foo-second.rst', "Content"),
         (4, 'rel/dir/foo.bar', "Content"),
     ])
     def test_add_file_into_empty_repo(self, cnt, filename, content, backend, csrf_token):
         repo = backend.create_repo()
         commit_id = repo.get_commit().raw_id
         response = self.app.post(
             route_path('repo_files_create_file',
                        repo_name=repo.repo_name,
                        commit_id=commit_id, f_path='/'),
             params={
                 'content': content,
                 'filename': filename,
                 'csrf_token': csrf_token,
             },
             status=302)
 
         expected_msg = 'Successfully committed new file `{}`'.format(filename)
         assert_session_flash(response, expected_msg)
 
     def test_edit_file_view(self, backend):
         response = self.app.get(
             route_path('repo_files_edit_file',
                        repo_name=backend.repo_name,
                        commit_id=backend.default_head_id,
                        f_path='vcs/nodes.py'),
             status=200)
         response.mustcontain("Module holding everything related to vcs nodes.")
 
     def test_edit_file_view_not_on_branch(self, backend):
         repo = backend.create_repo()
         backend.ensure_file(b"vcs/nodes.py")
 
         response = self.app.get(
             route_path('repo_files_edit_file',
                        repo_name=repo.repo_name,
                        commit_id='tip',
                        f_path='vcs/nodes.py'),
             status=302)
         assert_session_flash(
             response, 'Cannot modify file. Given commit `tip` is not head of a branch.')
 
     def test_edit_file_view_commit_changes(self, backend, csrf_token):
         repo = backend.create_repo()
         backend.ensure_file(b"vcs/nodes.py", content=b"print 'hello'")
 
         response = self.app.post(
             route_path('repo_files_update_file',
                        repo_name=repo.repo_name,
                        commit_id=backend.default_head_id,
                        f_path='vcs/nodes.py'),
             params={
                 'content': "print 'hello world'",
                 'message': 'I committed',
                 'filename': "vcs/nodes.py",
                 'csrf_token': csrf_token,
             },
             status=302)
         assert_session_flash(
             response, 'Successfully committed changes to file `vcs/nodes.py`')
         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()
         backend.ensure_file(b"vcs/nodes.py", content=b"print 'hello'")
 
         commit_id = (
             backend.default_branch_name or
             backend.repo.scm_instance().commit_ids[-1])
 
         response = self.app.post(
             route_path('repo_files_update_file',
                        repo_name=repo.repo_name,
                        commit_id=commit_id,
                        f_path='vcs/nodes.py'),
             params={
                 'content': "print 'hello world'",
                 'message': '',
                 'filename': "vcs/nodes.py",
                 'csrf_token': csrf_token,
             },
             status=302)
         assert_session_flash(
             response, 'Successfully committed changes to file `vcs/nodes.py`')
         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',
                        repo_name=backend.repo_name,
                        commit_id=backend.default_head_id,
                        f_path='vcs/nodes.py'),
             status=200)
 
     def test_delete_file_view_not_on_branch(self, backend):
         repo = backend.create_repo()
         backend.ensure_file(b'vcs/nodes.py')
 
         response = self.app.get(
             route_path('repo_files_remove_file',
                        repo_name=repo.repo_name,
                        commit_id='tip',
                        f_path='vcs/nodes.py'),
             status=302)
         assert_session_flash(
             response, 'Cannot modify file. Given commit `tip` is not head of a branch.')
 
     def test_delete_file_view_commit_changes(self, backend, csrf_token):
         repo = backend.create_repo()
         backend.ensure_file(b"vcs/nodes.py")
 
         response = self.app.post(
             route_path('repo_files_delete_file',
                        repo_name=repo.repo_name,
                        commit_id=backend.default_head_id,
                        f_path='vcs/nodes.py'),
             params={
                 'message': 'i committed',
                 'csrf_token': csrf_token,
             },
             status=302)
         assert_session_flash(
             response, 'Successfully deleted file `vcs/nodes.py`')
 
 
 @pytest.mark.usefixtures("app")
 class TestFilesViewOtherCases(object):
 
     def test_access_empty_repo_redirect_to_summary_with_alert_write_perms(
             self, backend_stub, autologin_regular_user, user_regular,
             user_util):
 
         repo = backend_stub.create_repo()
         user_util.grant_user_permission_to_repo(
             repo, user_regular, 'repository.write')
         response = self.app.get(
             route_path('repo_files',
                        repo_name=repo.repo_name,
                        commit_id='tip', f_path='/'))
 
         repo_file_add_url = route_path(
             'repo_files_add_file',
             repo_name=repo.repo_name,
             commit_id=0, f_path='')
         add_new = f'<a class="alert-link" href="{repo_file_add_url}">add a new file</a>'
 
         repo_file_upload_url = route_path(
             'repo_files_upload_file',
             repo_name=repo.repo_name,
             commit_id=0, f_path='')
         upload_new = f'<a class="alert-link" href="{repo_file_upload_url}">upload a new file</a>'
 
         assert_session_flash(
             response,
             'There are no files yet. Click here to %s or %s.' % (add_new, upload_new)
             )
 
     def test_access_empty_repo_redirect_to_summary_with_alert_no_write_perms(
             self, backend_stub, autologin_regular_user):
         repo = backend_stub.create_repo()
         # init session for anon user
         route_path('repo_summary', repo_name=repo.repo_name)
 
         repo_file_add_url = route_path(
             'repo_files_add_file',
             repo_name=repo.repo_name,
             commit_id=0, f_path='')
 
         response = self.app.get(
             route_path('repo_files',
                        repo_name=repo.repo_name,
                        commit_id='tip', f_path='/'))
 
         assert_session_flash(response, no_=repo_file_add_url)
 
     @pytest.mark.parametrize('file_node', [
         b'archive/file.zip',
         b'diff/my-file.txt',
         b'render.py',
         b'render',
         b'remove_file',
         b'remove_file/to-delete.txt',
     ])
     def test_file_names_equal_to_routes_parts(self, backend, file_node):
         backend.create_repo()
         backend.ensure_file(file_node)
 
         self.app.get(
             route_path('repo_files',
                        repo_name=backend.repo_name,
                        commit_id='tip', f_path=safe_str(file_node)),
             status=200)
 
 
 class TestAdjustFilePathForSvn(object):
     """
     SVN specific adjustments of node history in RepoFilesView.
     """
 
     def test_returns_path_relative_to_matched_reference(self):
         repo = self._repo(branches=['trunk'])
         self.assert_file_adjustment('trunk/file', 'file', repo)
 
     def test_does_not_modify_file_if_no_reference_matches(self):
         repo = self._repo(branches=['trunk'])
         self.assert_file_adjustment('notes/file', 'notes/file', repo)
 
     def test_does_not_adjust_partial_directory_names(self):
         repo = self._repo(branches=['trun'])
         self.assert_file_adjustment('trunk/file', 'trunk/file', repo)
 
     def test_is_robust_to_patterns_which_prefix_other_patterns(self):
         repo = self._repo(branches=['trunk', 'trunk/new', 'trunk/old'])
         self.assert_file_adjustment('trunk/new/file', 'file', repo)
 
     def assert_file_adjustment(self, f_path, expected, repo):
         result = RepoFilesView.adjust_file_path_for_svn(f_path, repo)
         assert result == expected
 
     def _repo(self, branches=None):
         repo = mock.Mock()
         repo.branches = OrderedDict((name, '0') for name in branches or [])
         repo.tags = {}
         return repo

rhodecode/apps/repository/views/repo_files.py

@@ -1,1586 +1,1708 b''
 # Copyright (C) 2011-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import itertools
 import logging
 import os
 import collections
 import urllib.request
 import urllib.parse
 import urllib.error
 import pathlib
 
 from pyramid.httpexceptions import HTTPNotFound, HTTPBadRequest, HTTPFound
 
 from pyramid.renderers import render
 from pyramid.response import Response
 
 import rhodecode
 from rhodecode.apps._base import RepoAppView
 
 
 from rhodecode.lib import diffs, helpers as h, rc_cache
 from rhodecode.lib import audit_logger
 from rhodecode.lib.hash_utils import sha1_safe
 from rhodecode.lib.rc_cache.archive_cache import get_archival_cache_store, get_archival_config, ReentrantLock
 from rhodecode.lib.str_utils import safe_bytes, convert_special_chars
 from rhodecode.lib.view_utils import parse_path_ref
 from rhodecode.lib.exceptions import NonRelativePathError
 from rhodecode.lib.codeblocks import (
     filenode_as_lines_tokens, filenode_as_annotated_lines_tokens)
 from rhodecode.lib.utils2 import convert_line_endings, detect_mode
 from rhodecode.lib.type_utils import str2bool
 from rhodecode.lib.str_utils import safe_str, safe_int
 from rhodecode.lib.auth import (
     LoginRequired, HasRepoPermissionAnyDecorator, CSRFRequired)
 from rhodecode.lib.vcs import path as vcspath
 from rhodecode.lib.vcs.backends.base import EmptyCommit
 from rhodecode.lib.vcs.conf import settings
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.lib.vcs.exceptions import (
     RepositoryError, CommitDoesNotExistError, EmptyRepositoryError,
     ImproperArchiveTypeError, VCSError, NodeAlreadyExistsError,
     NodeDoesNotExistError, CommitError, NodeError)
 
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.db import Repository
 
 log = logging.getLogger(__name__)
 
 
 def get_archive_name(db_repo_id, db_repo_name, commit_sha, ext, subrepos=False, path_sha='', with_hash=True):
     # original backward compat name of archive
     clean_name = safe_str(convert_special_chars(db_repo_name).replace('/', '_'))
 
     # e.g vcsserver-id-abcd-sub-1-abcfdef-archive-all.zip
     # vcsserver-id-abcd-sub-0-abcfdef-COMMIT_SHA-PATH_SHA.zip
     id_sha = sha1_safe(str(db_repo_id))[:4]
     sub_repo = 'sub-1' if subrepos else 'sub-0'
     commit = commit_sha if with_hash else 'archive'
     path_marker = (path_sha if with_hash else '') or 'all'
     archive_name = f'{clean_name}-id-{id_sha}-{sub_repo}-{commit}-{path_marker}{ext}'
 
     return archive_name
 
 
 def get_path_sha(at_path):
     return safe_str(sha1_safe(at_path)[:8])
 
 
 def _get_archive_spec(fname):
     log.debug('Detecting archive spec for: `%s`', fname)
 
     fileformat = None
     ext = None
     content_type = None
     for a_type, content_type, extension in settings.ARCHIVE_SPECS:
 
         if fname.endswith(extension):
             fileformat = a_type
             log.debug('archive is of type: %s', fileformat)
             ext = extension
             break
 
     if not fileformat:
         raise ValueError()
 
     # left over part of whole fname is the commit
     commit_id = fname[:-len(ext)]
 
     return commit_id, ext, fileformat, content_type
 
 
 class RepoFilesView(RepoAppView):
 
     @staticmethod
     def adjust_file_path_for_svn(f_path, repo):
         """
         Computes the relative path of `f_path`.
 
         This is mainly based on prefix matching of the recognized tags and
         branches in the underlying repository.
         """
         tags_and_branches = itertools.chain(
             repo.branches.keys(),
             repo.tags.keys())
         tags_and_branches = sorted(tags_and_branches, key=len, reverse=True)
 
         for name in tags_and_branches:
             if f_path.startswith(f'{name}/'):
                 f_path = vcspath.relpath(f_path, name)
                 break
         return f_path
 
     def load_default_context(self):
         c = self._get_local_tmpl_context(include_app_defaults=True)
         c.rhodecode_repo = self.rhodecode_vcs_repo
         c.enable_downloads = self.db_repo.enable_downloads
         return c
 
     def _ensure_not_locked(self, commit_id='tip'):
         _ = self.request.translate
 
         repo = self.db_repo
         if repo.enable_locking and repo.locked[0]:
             h.flash(_('This repository has been locked by %s on %s')
                     % (h.person_by_id(repo.locked[0]),
                     h.format_date(h.time_to_datetime(repo.locked[1]))),
                     'warning')
             files_url = h.route_path(
                 'repo_files:default_path',
                 repo_name=self.db_repo_name, commit_id=commit_id)
             raise HTTPFound(files_url)
 
     def forbid_non_head(self, is_head, f_path, commit_id='tip', json_mode=False):
         _ = self.request.translate
 
         if not is_head:
             message = _('Cannot modify file. '
                         'Given commit `{}` is not head of a branch.').format(commit_id)
             h.flash(message, category='warning')
 
             if json_mode:
                 return message
 
             files_url = h.route_path(
                 'repo_files', repo_name=self.db_repo_name, commit_id=commit_id,
                 f_path=f_path)
             raise HTTPFound(files_url)
 
     def check_branch_permission(self, branch_name, commit_id='tip', json_mode=False):
         _ = self.request.translate
 
         rule, branch_perm = self._rhodecode_user.get_rule_and_branch_permission(
             self.db_repo_name, branch_name)
         if branch_perm and branch_perm not in ['branch.push', 'branch.push_force']:
             message = _('Branch `{}` changes forbidden by rule {}.').format(
                 h.escape(branch_name), h.escape(rule))
             h.flash(message, 'warning')
 
             if json_mode:
                 return message
 
             files_url = h.route_path(
                 'repo_files:default_path', repo_name=self.db_repo_name, commit_id=commit_id)
 
             raise HTTPFound(files_url)
 
     def _get_commit_and_path(self):
         default_commit_id = self.db_repo.landing_ref_name
         default_f_path = '/'
 
         commit_id = self.request.matchdict.get(
             'commit_id', default_commit_id)
         f_path = self._get_f_path(self.request.matchdict, default_f_path)
         return commit_id, f_path
 
     def _get_default_encoding(self, c):
         enc_list = getattr(c, 'default_encodings', [])
         return enc_list[0] if enc_list else 'UTF-8'
 
     def _get_commit_or_redirect(self, commit_id, redirect_after=True):
         """
         This is a safe way to get commit. If an error occurs it redirects to
         tip with proper message
 
         :param commit_id: id of commit to fetch
         :param redirect_after: toggle redirection
         """
         _ = self.request.translate
 
         try:
             return self.rhodecode_vcs_repo.get_commit(commit_id)
         except EmptyRepositoryError:
             if not redirect_after:
                 return None
 
             add_new = upload_new = ""
             if h.HasRepoPermissionAny(
                     'repository.write', 'repository.admin')(self.db_repo_name):
                 _url = h.route_path(
                     'repo_files_add_file',
                     repo_name=self.db_repo_name, commit_id=0, f_path='')
                 add_new = h.link_to(
                     _('add a new file'), _url, class_="alert-link")
 
                 _url_upld = h.route_path(
                     'repo_files_upload_file',
                     repo_name=self.db_repo_name, commit_id=0, f_path='')
                 upload_new = h.link_to(
                     _('upload a new file'), _url_upld, class_="alert-link")
 
             h.flash(h.literal(
                 _('There are no files yet. Click here to %s or %s.') % (add_new, upload_new)), category='warning')
             raise HTTPFound(
                 h.route_path('repo_summary', repo_name=self.db_repo_name))
 
         except (CommitDoesNotExistError, LookupError) as e:
             msg = _('No such commit exists for this repository. Commit: {}').format(commit_id)
             h.flash(msg, category='error')
             raise HTTPNotFound()
         except RepositoryError as e:
             h.flash(h.escape(safe_str(e)), category='error')
             raise HTTPNotFound()
 
     def _get_filenode_or_redirect(self, commit_obj, path, pre_load=None):
         """
         Returns file_node, if error occurs or given path is directory,
         it'll redirect to top level path
         """
         _ = self.request.translate
 
         try:
             file_node = commit_obj.get_node(path, pre_load=pre_load)
             if file_node.is_dir():
                 raise RepositoryError('The given path is a directory')
         except CommitDoesNotExistError:
             log.exception('No such commit exists for this repository')
             h.flash(_('No such commit exists for this repository'), category='error')
             raise HTTPNotFound()
         except RepositoryError as e:
             log.warning('Repository error while fetching filenode `%s`. Err:%s', path, e)
             h.flash(h.escape(safe_str(e)), category='error')
             raise HTTPNotFound()
 
         return file_node
 
     def _is_valid_head(self, commit_id, repo, landing_ref):
         branch_name = sha_commit_id = ''
         is_head = False
         log.debug('Checking if commit_id `%s` is a head for %s.', commit_id, repo)
 
         for _branch_name, branch_commit_id in repo.branches.items():
             # simple case we pass in branch name, it's a HEAD
             if commit_id == _branch_name:
                 is_head = True
                 branch_name = _branch_name
                 sha_commit_id = branch_commit_id
                 break
             # case when we pass in full sha commit_id, which is a head
             elif commit_id == branch_commit_id:
                 is_head = True
                 branch_name = _branch_name
                 sha_commit_id = branch_commit_id
                 break
 
         if h.is_svn(repo) and not repo.is_empty():
             # Note: Subversion only has one head.
             if commit_id == repo.get_commit(commit_idx=-1).raw_id:
                 is_head = True
                 return branch_name, sha_commit_id, is_head
 
         # checked branches, means we only need to try to get the branch/commit_sha
         if repo.is_empty():
             is_head = True
             branch_name = landing_ref
             sha_commit_id = EmptyCommit().raw_id
         else:
             commit = repo.get_commit(commit_id=commit_id)
             if commit:
                 branch_name = commit.branch
                 sha_commit_id = commit.raw_id
 
         return branch_name, sha_commit_id, is_head
 
     def _get_tree_at_commit(self, c, commit_id, f_path, full_load=False, at_rev=None):
 
         repo_id = self.db_repo.repo_id
         force_recache = self.get_recache_flag()
 
         cache_seconds = safe_int(
             rhodecode.CONFIG.get('rc_cache.cache_repo.expiration_time'))
         cache_on = not force_recache and cache_seconds > 0
         log.debug(
             'Computing FILE TREE for repo_id %s commit_id `%s` and path `%s`'
             'with caching: %s[TTL: %ss]' % (
                 repo_id, commit_id, f_path, cache_on, cache_seconds or 0))
 
         cache_namespace_uid = f'repo.{rc_cache.FILE_TREE_CACHE_VER}.{repo_id}'
         region = rc_cache.get_or_create_region('cache_repo', cache_namespace_uid)
 
         @region.conditional_cache_on_arguments(namespace=cache_namespace_uid, condition=cache_on)
         def compute_file_tree(_name_hash, _repo_id, _commit_id, _f_path, _full_load, _at_rev):
             log.debug('Generating cached file tree at for repo_id: %s, %s, %s',
                       _repo_id, _commit_id, _f_path)
 
             c.full_load = _full_load
             return render(
                 'rhodecode:templates/files/files_browser_tree.mako',
                 self._get_template_context(c), self.request, _at_rev)
 
         return compute_file_tree(
             self.db_repo.repo_name_hash, self.db_repo.repo_id, commit_id, f_path, full_load, at_rev)
 
     def create_pure_path(self, *parts):
         # Split paths and sanitize them, removing any ../ etc
         sanitized_path = [
             x for x in pathlib.PurePath(*parts).parts
             if x not in ['.', '..']]
 
         pure_path = pathlib.PurePath(*sanitized_path)
         return pure_path
 
     def _is_lf_enabled(self, target_repo):
         lf_enabled = False
 
         lf_key_for_vcs_map = {
             'hg': 'extensions_largefiles',
             'git': 'vcs_git_lfs_enabled'
         }
 
         lf_key_for_vcs = lf_key_for_vcs_map.get(target_repo.repo_type)
 
         if lf_key_for_vcs:
             lf_enabled = self._get_repo_setting(target_repo, lf_key_for_vcs)
 
         return lf_enabled
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator(
         'repository.read', 'repository.write', 'repository.admin')
     def repo_archivefile(self):
         # archive cache config
         from rhodecode import CONFIG
         _ = self.request.translate
         self.load_default_context()
         default_at_path = '/'
         fname = self.request.matchdict['fname']
         subrepos = self.request.GET.get('subrepos') == 'true'
         with_hash = str2bool(self.request.GET.get('with_hash', '1'))
         at_path = self.request.GET.get('at_path') or default_at_path
 
         if not self.db_repo.enable_downloads:
             return Response(_('Downloads disabled'))
 
         try:
             commit_id, ext, fileformat, content_type = \
                 _get_archive_spec(fname)
         except ValueError:
             return Response(_('Unknown archive type for: `{}`').format(
                 h.escape(fname)))
 
         try:
             commit = self.rhodecode_vcs_repo.get_commit(commit_id)
         except CommitDoesNotExistError:
             return Response(_('Unknown commit_id {}').format(
                 h.escape(commit_id)))
         except EmptyRepositoryError:
             return Response(_('Empty repository'))
 
         # we used a ref, or a shorter version, lets redirect client ot use explicit hash
         if commit_id != commit.raw_id:
             fname=f'{commit.raw_id}{ext}'
             raise HTTPFound(self.request.current_route_path(fname=fname))
 
         try:
             at_path = commit.get_node(at_path).path or default_at_path
         except Exception:
             return Response(_('No node at path {} for this repository').format(h.escape(at_path)))
 
         path_sha = get_path_sha(at_path)
 
         # used for cache etc, consistent unique archive name
         archive_name_key = get_archive_name(
             self.db_repo.repo_id, self.db_repo_name, commit_sha=commit.short_id, ext=ext, subrepos=subrepos,
             path_sha=path_sha, with_hash=True)
 
         if not with_hash:
             path_sha = ''
 
         # what end client gets served
         response_archive_name = get_archive_name(
             self.db_repo.repo_id, self.db_repo_name, commit_sha=commit.short_id, ext=ext, subrepos=subrepos,
             path_sha=path_sha, with_hash=with_hash)
 
         # remove extension from our archive directory name
         archive_dir_name = response_archive_name[:-len(ext)]
 
         archive_cache_disable = self.request.GET.get('no_cache')
 
         d_cache = get_archival_cache_store(config=CONFIG)
 
         # NOTE: we get the config to pass to a call to lazy-init the SAME type of cache on vcsserver
         d_cache_conf = get_archival_config(config=CONFIG)
 
         reentrant_lock_key = archive_name_key + '.lock'
         with ReentrantLock(d_cache, reentrant_lock_key):
             # This is also a cache key
             use_cached_archive = False
             if archive_name_key in d_cache and not archive_cache_disable:
                 reader, tag = d_cache.get(archive_name_key, read=True, tag=True, retry=True)
                 use_cached_archive = True
                 log.debug('Found cached archive as key=%s tag=%s, serving archive from cache reader=%s',
                           archive_name_key, tag, reader.name)
             else:
                 reader = None
                 log.debug('Archive with key=%s is not yet cached, creating one now...', archive_name_key)
 
             # generate new archive, as previous was not found in the cache
             if not reader:
 
                 try:
                     commit.archive_repo(archive_name_key, archive_dir_name=archive_dir_name,
                                         kind=fileformat, subrepos=subrepos,
                                         archive_at_path=at_path, cache_config=d_cache_conf)
                 except ImproperArchiveTypeError:
                     return _('Unknown archive type')
 
             reader, tag = d_cache.get(archive_name_key, read=True, tag=True, retry=True)
 
         if not reader:
             raise ValueError('archive cache reader is empty, failed to fetch file from distributed archive cache')
 
         def archive_iterator(_reader, block_size: int = 4096*512):
             # 4096 * 64 = 64KB
             while 1:
                 data = _reader.read(block_size)
                 if not data:
                     break
                 yield data
 
         response = Response(app_iter=archive_iterator(reader))
         response.content_disposition = f'attachment; filename={response_archive_name}'
         response.content_type = str(content_type)
 
         try:
             return response
         finally:
             # store download action
             audit_logger.store_web(
                 'repo.archive.download', action_data={
                     'user_agent': self.request.user_agent,
                     'archive_name': archive_name_key,
                     'archive_spec': fname,
                     'archive_cached': use_cached_archive},
                 user=self._rhodecode_user,
                 repo=self.db_repo,
                 commit=True
             )
 
     def _get_file_node(self, commit_id, f_path):
         if commit_id not in ['', None, 'None', '0' * 12, '0' * 40]:
             commit = self.rhodecode_vcs_repo.get_commit(commit_id=commit_id)
             try:
                 node = commit.get_node(f_path)
                 if node.is_dir():
                     raise NodeError(f'{node} path is a {type(node)} not a file')
             except NodeDoesNotExistError:
                 commit = EmptyCommit(
                     commit_id=commit_id,
                     idx=commit.idx,
                     repo=commit.repository,
                     alias=commit.repository.alias,
                     message=commit.message,
                     author=commit.author,
                     date=commit.date)
                 node = FileNode(safe_bytes(f_path), b'', commit=commit)
         else:
             commit = EmptyCommit(
                 repo=self.rhodecode_vcs_repo,
                 alias=self.rhodecode_vcs_repo.alias)
             node = FileNode(safe_bytes(f_path), b'', commit=commit)
         return node
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator(
         'repository.read', 'repository.write', 'repository.admin')
     def repo_files_diff(self):
         c = self.load_default_context()
         f_path = self._get_f_path(self.request.matchdict)
         diff1 = self.request.GET.get('diff1', '')
         diff2 = self.request.GET.get('diff2', '')
 
         path1, diff1 = parse_path_ref(diff1, default_path=f_path)
 
         ignore_whitespace = str2bool(self.request.GET.get('ignorews'))
         line_context = self.request.GET.get('context', 3)
 
         if not any((diff1, diff2)):
             h.flash(
                 'Need query parameter "diff1" or "diff2" to generate a diff.',
                 category='error')
             raise HTTPBadRequest()
 
         c.action = self.request.GET.get('diff')
         if c.action not in ['download', 'raw']:
             compare_url = h.route_path(
                 'repo_compare',
                 repo_name=self.db_repo_name,
                 source_ref_type='rev',
                 source_ref=diff1,
                 target_repo=self.db_repo_name,
                 target_ref_type='rev',
                 target_ref=diff2,
                 _query=dict(f_path=f_path))
             # redirect to new view if we render diff
             raise HTTPFound(compare_url)
 
         try:
             node1 = self._get_file_node(diff1, path1)
             node2 = self._get_file_node(diff2, f_path)
         except (RepositoryError, NodeError):
             log.exception("Exception while trying to get node from repository")
             raise HTTPFound(
                 h.route_path('repo_files', repo_name=self.db_repo_name,
                              commit_id='tip', f_path=f_path))
 
         if all(isinstance(node.commit, EmptyCommit)
                for node in (node1, node2)):
             raise HTTPNotFound()
 
         c.commit_1 = node1.commit
         c.commit_2 = node2.commit
 
         if c.action == 'download':
             _diff = diffs.get_gitdiff(node1, node2,
                                       ignore_whitespace=ignore_whitespace,
                                       context=line_context)
             # NOTE: this was using diff_format='gitdiff'
             diff = diffs.DiffProcessor(_diff, diff_format='newdiff')
 
             response = Response(self.path_filter.get_raw_patch(diff))
             response.content_type = 'text/plain'
             response.content_disposition = (
                 f'attachment; filename={f_path}_{diff1}_vs_{diff2}.diff'
             )
             charset = self._get_default_encoding(c)
             if charset:
                 response.charset = charset
             return response
 
         elif c.action == 'raw':
             _diff = diffs.get_gitdiff(node1, node2,
                                       ignore_whitespace=ignore_whitespace,
                                       context=line_context)
             # NOTE: this was using diff_format='gitdiff'
             diff = diffs.DiffProcessor(_diff, diff_format='newdiff')
 
             response = Response(self.path_filter.get_raw_patch(diff))
             response.content_type = 'text/plain'
             charset = self._get_default_encoding(c)
             if charset:
                 response.charset = charset
             return response
 
         # in case we ever end up here
         raise HTTPNotFound()
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator(
         'repository.read', 'repository.write', 'repository.admin')
     def repo_files_diff_2way_redirect(self):
         """
         Kept only to make OLD links work
         """
         f_path = self._get_f_path_unchecked(self.request.matchdict)
         diff1 = self.request.GET.get('diff1', '')
         diff2 = self.request.GET.get('diff2', '')
 
         if not any((diff1, diff2)):
             h.flash(
                 'Need query parameter "diff1" or "diff2" to generate a diff.',
                 category='error')
             raise HTTPBadRequest()
 
         compare_url = h.route_path(
             'repo_compare',
             repo_name=self.db_repo_name,
             source_ref_type='rev',
             source_ref=diff1,
             target_ref_type='rev',
             target_ref=diff2,
             _query=dict(f_path=f_path, diffmode='sideside',
                         target_repo=self.db_repo_name,))
         raise HTTPFound(compare_url)
 
     @LoginRequired()
     def repo_files_default_commit_redirect(self):
         """
         Special page that redirects to the landing page of files based on the default
         commit for repository
         """
         c = self.load_default_context()
         ref_name = c.rhodecode_db_repo.landing_ref_name
         landing_url = h.repo_files_by_ref_url(
             c.rhodecode_db_repo.repo_name,
             c.rhodecode_db_repo.repo_type,
             f_path='',
             ref_name=ref_name,
             commit_id='tip',
             query=dict(at=ref_name)
         )
 
         raise HTTPFound(landing_url)
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator(
         'repository.read', 'repository.write', 'repository.admin')
     def repo_files(self):
         c = self.load_default_context()
 
         view_name = getattr(self.request.matched_route, 'name', None)
 
         c.annotate = view_name == 'repo_files:annotated'
         # default is false, but .rst/.md files later are auto rendered, we can
         # overwrite auto rendering by setting this GET flag
         c.renderer = view_name == 'repo_files:rendered' or not self.request.GET.get('no-render', False)
 
         commit_id, f_path = self._get_commit_and_path()
 
         c.commit = self._get_commit_or_redirect(commit_id)
         c.branch = self.request.GET.get('branch', None)
         c.f_path = f_path
         at_rev = self.request.GET.get('at')
 
         # files or dirs
         try:
             c.file = c.commit.get_node(f_path, pre_load=['is_binary', 'size', 'data'])
 
             c.file_author = True
             c.file_tree = ''
 
             # prev link
             try:
                 prev_commit = c.commit.prev(c.branch)
                 c.prev_commit = prev_commit
                 c.url_prev = h.route_path(
                     'repo_files', repo_name=self.db_repo_name,
                     commit_id=prev_commit.raw_id, f_path=f_path)
                 if c.branch:
                     c.url_prev += '?branch=%s' % c.branch
             except (CommitDoesNotExistError, VCSError):
                 c.url_prev = '#'
                 c.prev_commit = EmptyCommit()
 
             # next link
             try:
                 next_commit = c.commit.next(c.branch)
                 c.next_commit = next_commit
                 c.url_next = h.route_path(
                     'repo_files', repo_name=self.db_repo_name,
                     commit_id=next_commit.raw_id, f_path=f_path)
                 if c.branch:
                     c.url_next += '?branch=%s' % c.branch
             except (CommitDoesNotExistError, VCSError):
                 c.url_next = '#'
                 c.next_commit = EmptyCommit()
 
             # load file content
             if c.file.is_file():
 
                 c.lf_node = {}
 
                 has_lf_enabled = self._is_lf_enabled(self.db_repo)
                 if has_lf_enabled:
                     c.lf_node = c.file.get_largefile_node()
 
                 c.file_source_page = 'true'
                 c.file_last_commit = c.file.last_commit
 
                 c.file_size_too_big = c.file.size > c.visual.cut_off_limit_file
 
                 if not (c.file_size_too_big or c.file.is_binary):
                     if c.annotate:  # annotation has precedence over renderer
                         c.annotated_lines = filenode_as_annotated_lines_tokens(
                             c.file
                         )
                     else:
                         c.renderer = (
                             c.renderer and h.renderer_from_filename(c.file.path)
                         )
                         if not c.renderer:
                             c.lines = filenode_as_lines_tokens(c.file)
 
                 _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)
                 c.on_branch_head = is_head
 
                 branch = c.commit.branch if (
                     c.commit.branch and '/' not in c.commit.branch) else None
                 c.branch_or_raw_id = branch or c.commit.raw_id
                 c.branch_name = c.commit.branch or h.short_id(c.commit.raw_id)
 
                 author = c.file_last_commit.author
                 c.authors = [[
                     h.email(author),
                     h.person(author, 'username_or_name_or_email'),
                     1
                 ]]
 
             else:  # load tree content at path
                 c.file_source_page = 'false'
                 c.authors = []
                 # this loads a simple tree without metadata to speed things up
                 # later via ajax we call repo_nodetree_full and fetch whole
                 c.file_tree = self._get_tree_at_commit(c, c.commit.raw_id, f_path, at_rev=at_rev)
 
                 c.readme_data, c.readme_file = \
                     self._get_readme_data(self.db_repo, c.visual.default_renderer,
                                           c.commit.raw_id, f_path)
 
         except RepositoryError as e:
             h.flash(h.escape(safe_str(e)), category='error')
             raise HTTPNotFound()
 
         if self.request.environ.get('HTTP_X_PJAX'):
             html = render('rhodecode:templates/files/files_pjax.mako',
                           self._get_template_context(c), self.request)
         else:
             html = render('rhodecode:templates/files/files.mako',
                           self._get_template_context(c), self.request)
         return Response(html)
 
     @HasRepoPermissionAnyDecorator(
         'repository.read', 'repository.write', 'repository.admin')
     def repo_files_annotated_previous(self):
         self.load_default_context()
 
         commit_id, f_path = self._get_commit_and_path()
         commit = self._get_commit_or_redirect(commit_id)
         prev_commit_id = commit.raw_id
         line_anchor = self.request.GET.get('line_anchor')
         is_file = False
         try:
             _file = commit.get_node(f_path)
             is_file = _file.is_file()
         except (NodeDoesNotExistError, CommitDoesNotExistError, VCSError):
             pass
 
         if is_file:
             history = commit.get_path_history(f_path)
             prev_commit_id = history[1].raw_id \
                 if len(history) > 1 else prev_commit_id
         prev_url = h.route_path(
             'repo_files:annotated', repo_name=self.db_repo_name,
             commit_id=prev_commit_id, f_path=f_path,
             _anchor=f'L{line_anchor}')
 
         raise HTTPFound(prev_url)
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator(
         'repository.read', 'repository.write', 'repository.admin')
     def repo_nodetree_full(self):
         """
         Returns rendered html of file tree that contains commit date,
         author, commit_id for the specified combination of
         repo, commit_id and file path
         """
         c = self.load_default_context()
 
         commit_id, f_path = self._get_commit_and_path()
         commit = self._get_commit_or_redirect(commit_id)
         try:
             dir_node = commit.get_node(f_path)
         except RepositoryError as e:
             return Response(f'error: {h.escape(safe_str(e))}')
 
         if dir_node.is_file():
             return Response('')
 
         c.file = dir_node
         c.commit = commit
         at_rev = self.request.GET.get('at')
 
         html = self._get_tree_at_commit(
             c, commit.raw_id, dir_node.path, full_load=True, at_rev=at_rev)
 
         return Response(html)
 
     def _get_attachement_headers(self, f_path):
         f_name = safe_str(f_path.split(Repository.NAME_SEP)[-1])
         safe_path = f_name.replace('"', '\\"')
         encoded_path = urllib.parse.quote(f_name)
 
         headers = "attachment; " \
                   "filename=\"{}\"; " \
                   "filename*=UTF-8\'\'{}".format(safe_path, encoded_path)
 
         return safe_bytes(headers).decode('latin-1', errors='replace')
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator(
         'repository.read', 'repository.write', 'repository.admin')
     def repo_file_raw(self):
         """
         Action for show as raw, some mimetypes are "rendered",
         those include images, icons.
         """
         c = self.load_default_context()
 
         commit_id, f_path = self._get_commit_and_path()
         commit = self._get_commit_or_redirect(commit_id)
         file_node = self._get_filenode_or_redirect(commit, f_path)
 
         raw_mimetype_mapping = {
             # map original mimetype to a mimetype used for "show as raw"
             # you can also provide a content-disposition to override the
             # default "attachment" disposition.
             # orig_type: (new_type, new_dispo)
 
             # show images inline:
             # Do not re-add SVG: it is unsafe and permits XSS attacks. One can
             # for example render an SVG with javascript inside or even render
             # HTML.
             'image/x-icon': ('image/x-icon', 'inline'),
             'image/png': ('image/png', 'inline'),
             'image/gif': ('image/gif', 'inline'),
             'image/jpeg': ('image/jpeg', 'inline'),
             'application/pdf': ('application/pdf', 'inline'),
         }
 
         mimetype = file_node.mimetype
         try:
             mimetype, disposition = raw_mimetype_mapping[mimetype]
         except KeyError:
             # we don't know anything special about this, handle it safely
             if file_node.is_binary:
                 # do same as download raw for binary files
                 mimetype, disposition = 'application/octet-stream', 'attachment'
             else:
                 # do not just use the original mimetype, but force text/plain,
                 # otherwise it would serve text/html and that might be unsafe.
                 # Note: underlying vcs library fakes text/plain mimetype if the
                 # mimetype can not be determined and it thinks it is not
                 # binary.This might lead to erroneous text display in some
                 # cases, but helps in other cases, like with text files
                 # without extension.
                 mimetype, disposition = 'text/plain', 'inline'
 
         if disposition == 'attachment':
             disposition = self._get_attachement_headers(f_path)
 
         stream_content = file_node.stream_bytes()
 
         response = Response(app_iter=stream_content)
         response.content_disposition = disposition
         response.content_type = mimetype
 
         charset = self._get_default_encoding(c)
         if charset:
             response.charset = charset
 
         return response
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator(
         'repository.read', 'repository.write', 'repository.admin')
     def repo_file_download(self):
         c = self.load_default_context()
 
         commit_id, f_path = self._get_commit_and_path()
         commit = self._get_commit_or_redirect(commit_id)
         file_node = self._get_filenode_or_redirect(commit, f_path)
 
         if self.request.GET.get('lf'):
             # only if lf get flag is passed, we download this file
             # as LFS/Largefile
             lf_node = file_node.get_largefile_node()
             if lf_node:
                 # overwrite our pointer with the REAL large-file
                 file_node = lf_node
 
         disposition = self._get_attachement_headers(f_path)
 
         stream_content = file_node.stream_bytes()
 
         response = Response(app_iter=stream_content)
         response.content_disposition = disposition
         response.content_type = file_node.mimetype
 
         charset = self._get_default_encoding(c)
         if charset:
             response.charset = charset
 
         return response
 
     def _get_nodelist_at_commit(self, repo_name, repo_id, commit_id, f_path):
 
         cache_seconds = safe_int(
             rhodecode.CONFIG.get('rc_cache.cache_repo.expiration_time'))
         cache_on = cache_seconds > 0
         log.debug(
             'Computing FILE SEARCH for repo_id %s commit_id `%s` and path `%s`'
             'with caching: %s[TTL: %ss]' % (
                 repo_id, commit_id, f_path, cache_on, cache_seconds or 0))
 
         cache_namespace_uid = f'repo.{repo_id}'
         region = rc_cache.get_or_create_region('cache_repo', cache_namespace_uid)
 
         @region.conditional_cache_on_arguments(namespace=cache_namespace_uid, condition=cache_on)
         def compute_file_search(_name_hash, _repo_id, _commit_id, _f_path):
             log.debug('Generating cached nodelist for repo_id:%s, %s, %s',
                       _repo_id, commit_id, f_path)
             try:
                 _d, _f = ScmModel().get_quick_filter_nodes(repo_name, _commit_id, _f_path)
             except (RepositoryError, CommitDoesNotExistError, Exception) as e:
                 log.exception(safe_str(e))
                 h.flash(h.escape(safe_str(e)), category='error')
                 raise HTTPFound(h.route_path(
                     'repo_files', repo_name=self.db_repo_name,
                     commit_id='tip', f_path='/'))
 
             return _d + _f
 
         result = compute_file_search(self.db_repo.repo_name_hash, self.db_repo.repo_id,
                                      commit_id, f_path)
         return filter(lambda n: self.path_filter.path_access_allowed(n['name']), result)
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator(
         'repository.read', 'repository.write', 'repository.admin')
     def repo_nodelist(self):
         self.load_default_context()
 
         commit_id, f_path = self._get_commit_and_path()
         commit = self._get_commit_or_redirect(commit_id)
 
         metadata = self._get_nodelist_at_commit(
             self.db_repo_name, self.db_repo.repo_id, commit.raw_id, f_path)
         return {'nodes': [x for x in metadata]}
 
     def _create_references(self, branches_or_tags, symbolic_reference, f_path, ref_type):
         items = []
         for name, commit_id in branches_or_tags.items():
             sym_ref = symbolic_reference(commit_id, name, f_path, ref_type)
             items.append((sym_ref, name, ref_type))
         return items
 
     def _symbolic_reference(self, commit_id, name, f_path, ref_type):
         return commit_id
 
     def _symbolic_reference_svn(self, commit_id, name, f_path, ref_type):
         return commit_id
 
         # NOTE(dan): old code we used in "diff" mode compare
         new_f_path = vcspath.join(name, f_path)
         return f'{new_f_path}@{commit_id}'
 
     def _get_node_history(self, commit_obj, f_path, commits=None):
         """
         get commit history for given node
 
         :param commit_obj: commit to calculate history
         :param f_path: path for node to calculate history for
         :param commits: if passed don't calculate history and take
             commits defined in this list
         """
         _ = self.request.translate
 
         # calculate history based on tip
         tip = self.rhodecode_vcs_repo.get_commit()
         if commits is None:
             pre_load = ["author", "branch"]
             try:
                 commits = tip.get_path_history(f_path, pre_load=pre_load)
             except (NodeDoesNotExistError, CommitError):
                 # this node is not present at tip!
                 commits = commit_obj.get_path_history(f_path, pre_load=pre_load)
 
         history = []
         commits_group = ([], _("Changesets"))
         for commit in commits:
             branch = ' (%s)' % commit.branch if commit.branch else ''
             n_desc = f'r{commit.idx}:{commit.short_id}{branch}'
             commits_group[0].append((commit.raw_id, n_desc, 'sha'))
         history.append(commits_group)
 
         symbolic_reference = self._symbolic_reference
 
         if self.rhodecode_vcs_repo.alias == 'svn':
             adjusted_f_path = RepoFilesView.adjust_file_path_for_svn(
                 f_path, self.rhodecode_vcs_repo)
             if adjusted_f_path != f_path:
                 log.debug(
                     'Recognized svn tag or branch in file "%s", using svn '
                     'specific symbolic references', f_path)
                 f_path = adjusted_f_path
                 symbolic_reference = self._symbolic_reference_svn
 
         branches = self._create_references(
             self.rhodecode_vcs_repo.branches, symbolic_reference, f_path, 'branch')
         branches_group = (branches, _("Branches"))
 
         tags = self._create_references(
             self.rhodecode_vcs_repo.tags, symbolic_reference, f_path, 'tag')
         tags_group = (tags, _("Tags"))
 
         history.append(branches_group)
         history.append(tags_group)
 
         return history, commits
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator(
         'repository.read', 'repository.write', 'repository.admin')
     def repo_file_history(self):
         self.load_default_context()
 
         commit_id, f_path = self._get_commit_and_path()
         commit = self._get_commit_or_redirect(commit_id)
         file_node = self._get_filenode_or_redirect(commit, f_path)
 
         if file_node.is_file():
             file_history, _hist = self._get_node_history(commit, f_path)
 
             res = []
             for section_items, section in file_history:
                 items = []
                 for obj_id, obj_text, obj_type in section_items:
                     at_rev = ''
                     if obj_type in ['branch', 'bookmark', 'tag']:
                         at_rev = obj_text
                     entry = {
                         'id': obj_id,
                         'text': obj_text,
                         'type': obj_type,
                         'at_rev': at_rev
                     }
 
                     items.append(entry)
 
                 res.append({
                     'text': section,
                     'children': items
                 })
 
             data = {
                 'more': False,
                 'results': res
             }
             return data
 
         log.warning('Cannot fetch history for directory')
         raise HTTPBadRequest()
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator(
         'repository.read', 'repository.write', 'repository.admin')
     def repo_file_authors(self):
         c = self.load_default_context()
 
         commit_id, f_path = self._get_commit_and_path()
         commit = self._get_commit_or_redirect(commit_id)
         file_node = self._get_filenode_or_redirect(commit, f_path)
 
         if not file_node.is_file():
             raise HTTPBadRequest()
 
         c.file_last_commit = file_node.last_commit
         if self.request.GET.get('annotate') == '1':
             # use _hist from annotation if annotation mode is on
             commit_ids = {x[1] for x in file_node.annotate}
             _hist = (
                 self.rhodecode_vcs_repo.get_commit(commit_id)
                 for commit_id in commit_ids)
         else:
             _f_history, _hist = self._get_node_history(commit, f_path)
         c.file_author = False
 
         unique = collections.OrderedDict()
         for commit in _hist:
             author = commit.author
             if author not in unique:
                 unique[commit.author] = [
                     h.email(author),
                     h.person(author, 'username_or_name_or_email'),
                     1  # counter
                 ]
 
             else:
                 # increase counter
                 unique[commit.author][2] += 1
 
         c.authors = [val for val in unique.values()]
 
         return self._get_template_context(c)
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
     def repo_files_check_head(self):
         self.load_default_context()
 
         commit_id, f_path = self._get_commit_and_path()
         _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)
 
         new_path = self.request.POST.get('path')
         operation = self.request.POST.get('operation')
         path_exist = ''
 
         if new_path and operation in ['create', 'upload']:
             new_f_path = os.path.join(f_path.lstrip('/'), new_path)
             try:
                 commit_obj = self.rhodecode_vcs_repo.get_commit(commit_id)
                 # NOTE(dan): construct whole path without leading /
                 file_node = commit_obj.get_node(new_f_path)
                 if file_node is not None:
                     path_exist = new_f_path
             except EmptyRepositoryError:
                 pass
             except Exception:
                 pass
 
         return {
             'branch': _branch_name,
             'sha': _sha_commit_id,
             'is_head': is_head,
             'path_exists': path_exist
         }
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
     def repo_files_remove_file(self):
         _ = self.request.translate
         c = self.load_default_context()
         commit_id, f_path = self._get_commit_and_path()
 
         self._ensure_not_locked()
         _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)
 
         self.forbid_non_head(is_head, f_path)
         self.check_branch_permission(_branch_name)
 
         c.commit = self._get_commit_or_redirect(commit_id)
         c.file = self._get_filenode_or_redirect(c.commit, f_path)
 
         c.default_message = _(
             'Deleted file {} via RhodeCode Enterprise').format(f_path)
         c.f_path = f_path
 
         return self._get_template_context(c)
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
     @CSRFRequired()
     def repo_files_delete_file(self):
         _ = self.request.translate
 
         c = self.load_default_context()
         commit_id, f_path = self._get_commit_and_path()
 
         self._ensure_not_locked()
         _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)
 
         self.forbid_non_head(is_head, f_path)
         self.check_branch_permission(_branch_name)
 
         c.commit = self._get_commit_or_redirect(commit_id)
         c.file = self._get_filenode_or_redirect(c.commit, f_path)
 
         c.default_message = _(
             'Deleted file {} via RhodeCode Enterprise').format(f_path)
         c.f_path = f_path
         node_path = f_path
         author = self._rhodecode_db_user.full_contact
         message = self.request.POST.get('message') or c.default_message
         try:
             nodes = {
                 safe_bytes(node_path): {
                     'content': b''
                 }
             }
             ScmModel().delete_nodes(
                 user=self._rhodecode_db_user.user_id, repo=self.db_repo,
                 message=message,
                 nodes=nodes,
                 parent_commit=c.commit,
                 author=author,
             )
 
             h.flash(
                 _('Successfully deleted file `{}`').format(
                     h.escape(f_path)), category='success')
         except Exception:
             log.exception('Error during commit operation')
             h.flash(_('Error occurred during commit'), category='error')
         raise HTTPFound(
             h.route_path('repo_commit', repo_name=self.db_repo_name,
                          commit_id='tip'))
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
     def repo_files_edit_file(self):
         _ = self.request.translate
         c = self.load_default_context()
         commit_id, f_path = self._get_commit_and_path()
 
         self._ensure_not_locked()
         _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)
 
         self.forbid_non_head(is_head, f_path, commit_id=commit_id)
         self.check_branch_permission(_branch_name, commit_id=commit_id)
 
         c.commit = self._get_commit_or_redirect(commit_id)
         c.file = self._get_filenode_or_redirect(c.commit, f_path)
 
         if c.file.is_binary:
             files_url = h.route_path(
                 'repo_files',
                 repo_name=self.db_repo_name,
                 commit_id=c.commit.raw_id, f_path=f_path)
             raise HTTPFound(files_url)
 
         c.default_message = _('Edited file {} via RhodeCode Enterprise').format(f_path)
         c.f_path = f_path
 
         return self._get_template_context(c)
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
     @CSRFRequired()
     def repo_files_update_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)
         c.file = self._get_filenode_or_redirect(c.commit, f_path)
 
         if c.file.is_binary:
             raise HTTPFound(h.route_path('repo_files', repo_name=self.db_repo_name,
                                          commit_id=c.commit.raw_id, f_path=f_path))
 
         _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)
 
         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 = _('Edited file {} via RhodeCode Enterprise').format(f_path)
         c.f_path = f_path
 
         old_content = c.file.str_content
         sl = old_content.splitlines(1)
         first_line = sl[0] if sl else ''
 
         r_post = self.request.POST
         # line endings:  0 - Unix, 1 - Mac, 2 - DOS
         line_ending_mode = detect_mode(first_line, 0)
         content = convert_line_endings(r_post.get('content', ''), line_ending_mode)
 
         message = r_post.get('message') or c.default_message
 
         org_node_path = c.file.str_path
         filename = r_post['filename']
 
         root_path = c.file.dir_path
         pure_path = self.create_pure_path(root_path, filename)
         node_path = pure_path.as_posix()
 
         default_redirect_url = h.route_path('repo_commit', repo_name=self.db_repo_name,
                                             commit_id=commit_id)
         if content == old_content and node_path == org_node_path:
             h.flash(_('No changes detected on {}').format(h.escape(org_node_path)),
                     category='warning')
             raise HTTPFound(default_redirect_url)
 
         try:
             mapping = {
                 c.file.bytes_path: {
                     'org_filename': org_node_path,
                     'filename': safe_bytes(node_path),
                     'content': safe_bytes(content),
                     'lexer': '',
                     'op': 'mod',
                     'mode': c.file.mode
                 }
             }
 
             commit = ScmModel().update_nodes(
                 user=self._rhodecode_db_user.user_id,
                 repo=self.db_repo,
                 message=message,
                 nodes=mapping,
                 parent_commit=c.commit,
             )
 
             h.flash(_('Successfully committed changes to file `{}`').format(
                     h.escape(f_path)), category='success')
             default_redirect_url = h.route_path(
                 'repo_commit', repo_name=self.db_repo_name, commit_id=commit.raw_id)
 
         except Exception:
             log.exception('Error occurred during commit')
             h.flash(_('Error occurred during commit'), category='error')
 
         raise HTTPFound(default_redirect_url)
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
     def repo_files_add_file(self):
         _ = self.request.translate
         c = self.load_default_context()
         commit_id, f_path = self._get_commit_and_path()
 
         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)
 
         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)
 
         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)
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
     @CSRFRequired()
     def repo_files_create_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)
 
         # calculate redirect URL
         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)
 
         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.f_path = f_path
 
         r_post = self.request.POST
         message = r_post.get('message') or c.default_message
         filename = r_post.get('filename')
         unix_mode = 0
 
         if not filename:
             # If there's no commit, redirect to repo summary
             if type(c.commit) is EmptyCommit:
                 redirect_url = h.route_path(
                     'repo_summary', repo_name=self.db_repo_name)
             else:
                 redirect_url = default_redirect_url
             h.flash(_('No filename specified'), category='warning')
             raise HTTPFound(redirect_url)
 
         root_path = f_path
         pure_path = self.create_pure_path(root_path, filename)
         node_path = pure_path.as_posix().lstrip('/')
 
         author = self._rhodecode_db_user.full_contact
         content = convert_line_endings(r_post.get('content', ''), unix_mode)
         nodes = {
             safe_bytes(node_path): {
                 'content': safe_bytes(content)
             }
         }
 
         try:
 
             commit = ScmModel().create_nodes(
                 user=self._rhodecode_db_user.user_id,
                 repo=self.db_repo,
                 message=message,
                 nodes=nodes,
                 parent_commit=c.commit,
                 author=author,
             )
 
             h.flash(_('Successfully committed new file `{}`').format(
                     h.escape(node_path)), category='success')
 
             default_redirect_url = h.route_path(
                 'repo_commit', repo_name=self.db_repo_name, commit_id=commit.raw_id)
 
         except NonRelativePathError:
             log.exception('Non Relative path found')
             h.flash(_('The location specified must be a relative path and must not '
                       'contain .. in the path'), category='warning')
             raise HTTPFound(default_redirect_url)
         except (NodeError, NodeAlreadyExistsError) as e:
             h.flash(h.escape(safe_str(e)), category='error')
         except Exception:
             log.exception('Error occurred during commit')
             h.flash(_('Error occurred during commit'), category='error')
 
         raise HTTPFound(default_redirect_url)
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
     @CSRFRequired()
     def repo_files_upload_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)
 
         # calculate redirect URL
         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 = (_('Added file via RhodeCode Enterprise'))
         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
 
         nodes = {}
 
         for file_obj in r_post.getall('files_upload') or []:
             content = file_obj.file
             filename = file_obj.filename
 
             root_path = f_path
             pure_path = self.create_pure_path(root_path, filename)
             node_path = pure_path.as_posix().lstrip('/')
 
             nodes[safe_bytes(node_path)] = {
                 'content': content
             }
 
         if not nodes:
             error = 'missing files'
             return {
                 'error': error,
                 'redirect_url': default_redirect_url
             }
 
         author = self._rhodecode_db_user.full_contact
 
         try:
             commit = ScmModel().create_nodes(
                 user=self._rhodecode_db_user.user_id,
                 repo=self.db_repo,
                 message=message,
                 nodes=nodes,
                 parent_commit=c.commit,
                 author=author,
             )
             if len(nodes) == 1:
                 flash_message = _('Successfully committed {} new files').format(len(nodes))
             else:
                 flash_message = _('Successfully committed 1 new file')
 
             h.flash(flash_message, category='success')
 
             default_redirect_url = h.route_path(
                 'repo_commit', repo_name=self.db_repo_name, commit_id=commit.raw_id)
 
         except NonRelativePathError:
             log.exception('Non Relative path found')
             error = _('The location specified must be a relative path and must not '
                       'contain .. in the path')
             h.flash(error, category='warning')
 
             return {
                 'error': error,
                 'redirect_url': default_redirect_url
             }
         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
         }
+
+    @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

@@ -1,1044 +1,1063 b''
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 Scm model for RhodeCode
 """
 
 import os.path
 import traceback
 import logging
 import io
 
 from sqlalchemy import func
 from zope.cachedescriptors.property import Lazy as LazyProperty
 
 import rhodecode
 from rhodecode.lib.str_utils import safe_bytes
 from rhodecode.lib.vcs import get_backend
 from rhodecode.lib.vcs.exceptions import RepositoryError, NodeNotChangedError
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.lib.vcs.backends.base import EmptyCommit
 from rhodecode.lib import helpers as h, rc_cache
 from rhodecode.lib.auth import (
     HasRepoPermissionAny, HasRepoGroupPermissionAny,
     HasUserGroupPermissionAny)
 from rhodecode.lib.exceptions import NonRelativePathError, IMCCommitError
 from rhodecode.lib import hooks_utils
 from rhodecode.lib.utils import (
     get_filesystem_repos, make_db_config)
 from rhodecode.lib.str_utils import safe_str
 from rhodecode.lib.system_info import get_system_info
 from rhodecode.model import BaseModel
 from rhodecode.model.db import (
     or_, false, null,
     Repository, CacheKey, UserFollowing, UserLog, User, RepoGroup,
     PullRequest, FileStore)
 from rhodecode.model.settings import VcsSettingsModel
 from rhodecode.model.validation_schema.validators import url_validator, InvalidCloneUrl
 
 log = logging.getLogger(__name__)
 
 
 class UserTemp(object):
     def __init__(self, user_id):
         self.user_id = user_id
 
     def __repr__(self):
         return "<{}('id:{}')>".format(self.__class__.__name__, self.user_id)
 
 
 class RepoTemp(object):
     def __init__(self, repo_id):
         self.repo_id = repo_id
 
     def __repr__(self):
         return "<{}('id:{}')>".format(self.__class__.__name__, self.repo_id)
 
 
 class SimpleCachedRepoList(object):
     """
     Lighter version of of iteration of repos without the scm initialisation,
     and with cache usage
     """
     def __init__(self, db_repo_list, repos_path, order_by=None, perm_set=None):
         self.db_repo_list = db_repo_list
         self.repos_path = repos_path
         self.order_by = order_by
         self.reversed = (order_by or '').startswith('-')
         if not perm_set:
             perm_set = ['repository.read', 'repository.write',
                         'repository.admin']
         self.perm_set = perm_set
 
     def __len__(self):
         return len(self.db_repo_list)
 
     def __repr__(self):
         return '<{} ({})>'.format(self.__class__.__name__, self.__len__())
 
     def __iter__(self):
         for dbr in self.db_repo_list:
             # check permission at this level
             has_perm = HasRepoPermissionAny(*self.perm_set)(
                 dbr.repo_name, 'SimpleCachedRepoList check')
             if not has_perm:
                 continue
 
             tmp_d = {
                 'name': dbr.repo_name,
                 'dbrepo': dbr.get_dict(),
                 'dbrepo_fork': dbr.fork.get_dict() if dbr.fork else {}
             }
             yield tmp_d
 
 
 class _PermCheckIterator(object):
 
     def __init__(
             self, obj_list, obj_attr, perm_set, perm_checker,
             extra_kwargs=None):
         """
         Creates iterator from given list of objects, additionally
         checking permission for them from perm_set var
 
         :param obj_list: list of db objects
         :param obj_attr: attribute of object to pass into perm_checker
         :param perm_set: list of permissions to check
         :param perm_checker: callable to check permissions against
         """
         self.obj_list = obj_list
         self.obj_attr = obj_attr
         self.perm_set = perm_set
         self.perm_checker = perm_checker(*self.perm_set)
         self.extra_kwargs = extra_kwargs or {}
 
     def __len__(self):
         return len(self.obj_list)
 
     def __repr__(self):
         return '<{} ({})>'.format(self.__class__.__name__, self.__len__())
 
     def __iter__(self):
         for db_obj in self.obj_list:
             # check permission at this level
             # NOTE(marcink): the __dict__.get() is ~4x faster then getattr()
             name = db_obj.__dict__.get(self.obj_attr, None)
             if not self.perm_checker(name, self.__class__.__name__, **self.extra_kwargs):
                 continue
 
             yield db_obj
 
 
 class RepoList(_PermCheckIterator):
 
     def __init__(self, db_repo_list, perm_set=None, extra_kwargs=None):
         if not perm_set:
             perm_set = ['repository.read', 'repository.write', 'repository.admin']
 
         super().__init__(
             obj_list=db_repo_list,
             obj_attr='_repo_name', perm_set=perm_set,
             perm_checker=HasRepoPermissionAny,
             extra_kwargs=extra_kwargs)
 
 
 class RepoGroupList(_PermCheckIterator):
 
     def __init__(self, db_repo_group_list, perm_set=None, extra_kwargs=None):
         if not perm_set:
             perm_set = ['group.read', 'group.write', 'group.admin']
 
         super().__init__(
             obj_list=db_repo_group_list,
             obj_attr='_group_name', perm_set=perm_set,
             perm_checker=HasRepoGroupPermissionAny,
             extra_kwargs=extra_kwargs)
 
 
 class UserGroupList(_PermCheckIterator):
 
     def __init__(self, db_user_group_list, perm_set=None, extra_kwargs=None):
         if not perm_set:
             perm_set = ['usergroup.read', 'usergroup.write', 'usergroup.admin']
 
         super().__init__(
             obj_list=db_user_group_list,
             obj_attr='users_group_name', perm_set=perm_set,
             perm_checker=HasUserGroupPermissionAny,
             extra_kwargs=extra_kwargs)
 
 
 class ScmModel(BaseModel):
     """
     Generic Scm Model
     """
 
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
 
         settings_model = VcsSettingsModel(sa=self.sa)
         return settings_model.get_repos_location()
 
     def repo_scan(self, repos_path=None):
         """
         Listing of repositories in given path. This path should not be a
         repository itself. Return a dictionary of repository objects
 
         :param repos_path: path to directory containing repositories
         """
 
         if repos_path is None:
             repos_path = self.repos_path
 
         log.info('scanning for repositories in %s', repos_path)
 
         config = make_db_config()
         config.set('extensions', 'largefiles', '')
         repos = {}
 
         for name, path in get_filesystem_repos(repos_path, recursive=True):
             # name need to be decomposed and put back together using the /
             # since this is internal storage separator for rhodecode
             name = Repository.normalize_repo_name(name)
 
             try:
                 if name in repos:
                     raise RepositoryError('Duplicate repository name %s '
                                           'found in %s' % (name, path))
                 elif path[0] in rhodecode.BACKENDS:
                     backend = get_backend(path[0])
                     repos[name] = backend(path[1], config=config,
                                           with_wire={"cache": False})
             except OSError:
                 continue
             except RepositoryError:
                 log.exception('Failed to create a repo')
                 continue
 
         log.debug('found %s paths with repositories', len(repos))
         return repos
 
     def get_repos(self, all_repos=None, sort_key=None):
         """
         Get all repositories from db and for each repo create it's
         backend instance and fill that backed with information from database
 
         :param all_repos: list of repository names as strings
             give specific repositories list, good for filtering
 
         :param sort_key: initial sorting of repositories
         """
         if all_repos is None:
             all_repos = self.sa.query(Repository)\
                 .filter(Repository.group_id == null())\
                 .order_by(func.lower(Repository.repo_name)).all()
         repo_iter = SimpleCachedRepoList(
             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()\
                 .filter(RepoGroup.group_parent_id == null()).all()
         return [x for x in RepoGroupList(all_groups)]
 
     def mark_for_invalidation(self, repo_name, delete=False):
         """
         Mark caches of this repo invalid in the database. `delete` flag
         removes the cache entries
 
         :param repo_name: the repo_name for which caches should be marked
             invalid, or deleted
         :param delete: delete the entry keys instead of setting bool
             flag on them, and also purge caches used by the dogpile
         """
         repo = Repository.get_by_repo_name(repo_name)
 
         if repo:
             invalidation_namespace = CacheKey.REPO_INVALIDATION_NAMESPACE.format(
                 repo_id=repo.repo_id)
             CacheKey.set_invalidate(invalidation_namespace, delete=delete)
 
             repo_id = repo.repo_id
             config = repo._config
             config.set('extensions', 'largefiles', '')
             repo.update_commit_cache(config=config, cs_cache=None)
             if delete:
                 cache_namespace_uid = f'cache_repo.{repo_id}'
                 rc_cache.clear_cache_namespace('cache_repo', cache_namespace_uid, method=rc_cache.CLEAR_INVALIDATE)
 
     def toggle_following_repo(self, follow_repo_id, user_id):
 
         f = self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_repo_id == follow_repo_id)\
             .filter(UserFollowing.user_id == user_id).scalar()
 
         if f is not None:
             try:
                 self.sa.delete(f)
                 return
             except Exception:
                 log.error(traceback.format_exc())
                 raise
 
         try:
             f = UserFollowing()
             f.user_id = user_id
             f.follows_repo_id = follow_repo_id
             self.sa.add(f)
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def toggle_following_user(self, follow_user_id, user_id):
         f = self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_user_id == follow_user_id)\
             .filter(UserFollowing.user_id == user_id).scalar()
 
         if f is not None:
             try:
                 self.sa.delete(f)
                 return
             except Exception:
                 log.error(traceback.format_exc())
                 raise
 
         try:
             f = UserFollowing()
             f.user_id = user_id
             f.follows_user_id = follow_user_id
             self.sa.add(f)
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def is_following_repo(self, repo_name, user_id, cache=False):
         r = self.sa.query(Repository)\
             .filter(Repository.repo_name == repo_name).scalar()
 
         f = self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_repository == r)\
             .filter(UserFollowing.user_id == user_id).scalar()
 
         return f is not None
 
     def is_following_user(self, username, user_id, cache=False):
         u = User.get_by_username(username)
 
         f = self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_user == u)\
             .filter(UserFollowing.user_id == user_id).scalar()
 
         return f is not None
 
     def get_followers(self, repo):
         repo = self._get_repo(repo)
 
         return self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_repository == repo).count()
 
     def get_forks(self, repo):
         repo = self._get_repo(repo)
         return self.sa.query(Repository)\
             .filter(Repository.fork == repo).count()
 
     def get_pull_requests(self, repo):
         repo = self._get_repo(repo)
         return self.sa.query(PullRequest)\
             .filter(PullRequest.target_repo == repo)\
             .filter(PullRequest.status != PullRequest.STATUS_CLOSED).count()
 
     def get_artifacts(self, repo):
         repo = self._get_repo(repo)
         return self.sa.query(FileStore)\
             .filter(FileStore.repo == repo)\
             .filter(or_(FileStore.hidden == null(), FileStore.hidden == false())).count()
 
     def mark_as_fork(self, repo, fork, user):
         repo = self._get_repo(repo)
         fork = self._get_repo(fork)
         if fork and repo.repo_id == fork.repo_id:
             raise Exception("Cannot set repository as fork of itself")
 
         if fork and repo.repo_type != fork.repo_type:
             raise RepositoryError(
                 "Cannot set repository as fork of repository with other type")
 
         repo.fork = fork
         self.sa.add(repo)
         return repo
 
     def pull_changes(self, repo, username, remote_uri=None, validate_uri=True, **kwargs):
         dbrepo = self._get_repo(repo)
         remote_uri = remote_uri or dbrepo.clone_uri
         if not remote_uri:
             raise Exception("This repository doesn't have a clone uri")
 
         repo = dbrepo.scm_instance(cache=False)
         repo.config.clear_section('hooks')
 
         try:
             # NOTE(marcink): add extra validation so we skip invalid urls
             # this is due this tasks can be executed via scheduler without
             # proper validation of remote_uri
             if validate_uri:
                 config = make_db_config(clear_session=False)
                 url_validator(remote_uri, dbrepo.repo_type, config)
         except InvalidCloneUrl:
             raise
 
         repo_name = dbrepo.repo_name
         try:
             # TODO: we need to make sure those operations call proper hooks !
             repo.fetch(remote_uri, **kwargs)
 
             self.mark_for_invalidation(repo_name)
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def push_changes(self, repo, username, remote_uri=None, validate_uri=True, **kwargs):
         dbrepo = self._get_repo(repo)
         remote_uri = remote_uri or dbrepo.push_uri
         if not remote_uri:
             raise Exception("This repository doesn't have a clone uri")
 
         repo = dbrepo.scm_instance(cache=False)
         repo.config.clear_section('hooks')
 
         try:
             # NOTE(marcink): add extra validation so we skip invalid urls
             # this is due this tasks can be executed via scheduler without
             # proper validation of remote_uri
             if validate_uri:
                 config = make_db_config(clear_session=False)
                 url_validator(remote_uri, dbrepo.repo_type, config)
         except InvalidCloneUrl:
             raise
 
         try:
             repo.push(remote_uri, **kwargs)
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def commit_change(self, repo, repo_name, commit, user, author, message,
                       content: bytes, f_path: bytes, branch: str = None):
         """
         Commits changes
         """
         user = self._get_user(user)
 
         # message and author needs to be unicode
         # proper backend should then translate that into required type
         message = safe_str(message)
         author = safe_str(author)
         imc = repo.in_memory_commit
         imc.change(FileNode(f_path, content, mode=commit.get_file_mode(f_path)))
         try:
             # TODO: handle pre-push action !
             tip = imc.commit(
                 message=message, author=author, parents=[commit],
                 branch=branch or commit.branch)
         except Exception as e:
             log.error(traceback.format_exc())
             raise IMCCommitError(str(e))
         finally:
             # always clear caches, if commit fails we want fresh object also
             self.mark_for_invalidation(repo_name)
 
         # We trigger the post-push action
         hooks_utils.trigger_post_push_hook(
             username=user.username, action='push_local', hook_type='post_push',
             repo_name=repo_name, repo_type=repo.alias, commit_ids=[tip.raw_id])
         return tip
 
     def _sanitize_path(self, f_path: bytes):
         if f_path.startswith(b'/') or f_path.startswith(b'./') or b'../' in f_path:
             raise NonRelativePathError(b'%b is not an relative path' % f_path)
         if f_path:
             f_path = os.path.normpath(f_path)
         return f_path
 
     def get_dirnode_metadata(self, request, commit, dir_node):
         if not dir_node.is_dir():
             return []
 
         data = []
         for node in dir_node:
             if not node.is_file():
                 # we skip file-nodes
                 continue
 
             last_commit = node.last_commit
             last_commit_date = last_commit.date
             data.append({
                 'name': node.name,
                 'size': h.format_byte_size_binary(node.size),
                 'modified_at': h.format_date(last_commit_date),
                 'modified_ts': last_commit_date.isoformat(),
                 'revision': last_commit.revision,
                 'short_id': last_commit.short_id,
                 'message': h.escape(last_commit.message),
                 'author': h.escape(last_commit.author),
                 'user_profile': h.gravatar_with_user(
                     request, last_commit.author),
             })
 
         return data
 
     def get_nodes(self, repo_name, commit_id, root_path='/', flat=True,
                   extended_info=False, content=False, max_file_bytes=None):
         """
         recursive walk in root dir and return a set of all path in that dir
         based on repository walk function
 
         :param repo_name: name of repository
         :param commit_id: commit id for which to list nodes
         :param root_path: root path to list
         :param flat: return as a list, if False returns a dict with description
         :param extended_info: show additional info such as md5, binary, size etc
         :param content: add nodes content to the return data
         :param max_file_bytes: will not return file contents over this limit
 
         """
         _files = list()
         _dirs = list()
 
         try:
             _repo = self._get_repo(repo_name)
             commit = _repo.scm_instance().get_commit(commit_id=commit_id)
             root_path = root_path.lstrip('/')
 
             # get RootNode, inject pre-load options before walking
             top_node = commit.get_node(root_path)
             extended_info_pre_load = []
             if extended_info:
                 extended_info_pre_load += ['md5']
             top_node.default_pre_load = ['is_binary', 'size'] + extended_info_pre_load
 
             for __, dirs, files in commit.walk(top_node):
 
                 for f in files:
                     _content = None
                     _data = f_name = f.str_path
 
                     if not flat:
                         _data = {
                             "name": h.escape(f_name),
                             "type": "file",
                             }
                         if extended_info:
                             _data.update({
                                 "md5": f.md5,
                                 "binary": f.is_binary,
                                 "size": f.size,
                                 "extension": f.extension,
                                 "mimetype": f.mimetype,
                                 "lines": f.lines()[0]
                             })
 
                         if content:
                             over_size_limit = (max_file_bytes is not None
                                                and f.size > max_file_bytes)
                             full_content = None
                             if not f.is_binary and not over_size_limit:
                                 full_content = f.str_content
 
                             _data.update({
                                 "content": full_content,
                             })
                     _files.append(_data)
 
                 for d in dirs:
                     _data = d_name = d.str_path
                     if not flat:
                         _data = {
                             "name": h.escape(d_name),
                             "type": "dir",
                             }
                     if extended_info:
                         _data.update({
                             "md5": "",
                             "binary": False,
                             "size": 0,
                             "extension": "",
                         })
                     if content:
                         _data.update({
                             "content": None
                         })
                     _dirs.append(_data)
         except RepositoryError:
             log.exception("Exception in get_nodes")
             raise
 
         return _dirs, _files
 
     def get_quick_filter_nodes(self, repo_name, commit_id, root_path='/'):
         """
         Generate files for quick filter in files view
         """
 
         _files = list()
         _dirs = list()
         try:
             _repo = self._get_repo(repo_name)
             commit = _repo.scm_instance().get_commit(commit_id=commit_id)
             root_path = root_path.lstrip('/')
 
             top_node = commit.get_node(root_path)
             top_node.default_pre_load = []
 
             for __, dirs, files in commit.walk(top_node):
                 for f in files:
 
                     _data = {
                         "name": h.escape(f.str_path),
                         "type": "file",
                         }
 
                     _files.append(_data)
 
                 for d in dirs:
 
                     _data = {
                         "name": h.escape(d.str_path),
                         "type": "dir",
                         }
 
                     _dirs.append(_data)
         except RepositoryError:
             log.exception("Exception in get_quick_filter_nodes")
             raise
 
         return _dirs, _files
 
     def get_node(self, repo_name, commit_id, file_path,
                  extended_info=False, content=False, max_file_bytes=None, cache=True):
         """
         retrieve single node from commit
         """
 
         try:
 
             _repo = self._get_repo(repo_name)
             commit = _repo.scm_instance().get_commit(commit_id=commit_id)
 
             file_node = commit.get_node(file_path)
             if file_node.is_dir():
                 raise RepositoryError('The given path is a directory')
 
             _content = None
             f_name = file_node.str_path
 
             file_data = {
                 "name": h.escape(f_name),
                 "type": "file",
             }
 
             if extended_info:
                 file_data.update({
                     "extension": file_node.extension,
                     "mimetype": file_node.mimetype,
                 })
 
                 if cache:
                     md5 = file_node.md5
                     is_binary = file_node.is_binary
                     size = file_node.size
                 else:
                     is_binary, md5, size, _content = file_node.metadata_uncached()
 
                 file_data.update({
                     "md5": md5,
                     "binary": is_binary,
                     "size": size,
                 })
 
             if content and cache:
                 # get content + cache
                 size = file_node.size
                 over_size_limit = (max_file_bytes is not None and size > max_file_bytes)
                 full_content = None
                 all_lines = 0
                 if not file_node.is_binary and not over_size_limit:
                     full_content = safe_str(file_node.content)
                     all_lines, empty_lines = file_node.count_lines(full_content)
 
                 file_data.update({
                     "content": full_content,
                     "lines": all_lines
                 })
             elif content:
                 # get content *without* cache
                 if _content is None:
                     is_binary, md5, size, _content = file_node.metadata_uncached()
 
                 over_size_limit = (max_file_bytes is not None and size > max_file_bytes)
                 full_content = None
                 all_lines = 0
                 if not is_binary and not over_size_limit:
                     full_content = safe_str(_content)
                     all_lines, empty_lines = file_node.count_lines(full_content)
 
                 file_data.update({
                     "content": full_content,
                     "lines": all_lines
                 })
 
         except RepositoryError:
             log.exception("Exception in get_node")
             raise
 
         return file_data
 
     def get_fts_data(self, repo_name, commit_id, root_path='/'):
         """
         Fetch node tree for usage in full text search
         """
 
         tree_info = list()
 
         try:
             _repo = self._get_repo(repo_name)
             commit = _repo.scm_instance().get_commit(commit_id=commit_id)
             root_path = root_path.lstrip('/')
             top_node = commit.get_node(root_path)
             top_node.default_pre_load = []
 
             for __, dirs, files in commit.walk(top_node):
 
                 for f in files:
                     is_binary, md5, size, _content = f.metadata_uncached()
                     _data = {
                         "name": f.str_path,
                         "md5": md5,
                         "extension": f.extension,
                         "binary": is_binary,
                         "size": size
                     }
 
                     tree_info.append(_data)
 
         except RepositoryError:
             log.exception("Exception in get_nodes")
             raise
 
         return tree_info
 
     def create_nodes(self, user, repo, message, nodes, parent_commit=None,
                      author=None, trigger_push_hook=True):
         """
         Commits given multiple nodes into repo
 
         :param user: RhodeCode User object or user_id, the commiter
         :param repo: RhodeCode Repository object
         :param message: commit message
         :param nodes: mapping {filename:{'content':content},...}
         :param parent_commit: parent commit, can be empty than it's
            initial commit
         :param author: author of commit, cna be different that commiter
            only for git
         :param trigger_push_hook: trigger push hooks
 
         :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 = []
         for filename, content_dict in nodes.items():
             if not isinstance(filename, bytes):
                 raise ValueError(f'filename key in nodes needs to be bytes , or {upload_file_types}')
             content = content_dict['content']
             if not isinstance(content, upload_file_types + (bytes,)):
                 raise ValueError('content key value in nodes needs to be bytes')
 
         for f_path in nodes:
             f_path = self._sanitize_path(f_path)
             content = nodes[f_path]['content']
 
             # decoding here will force that we have proper encoded values
             # in any other case this will throw exceptions and deny commit
 
             if isinstance(content, bytes):
                 pass
             elif isinstance(content, upload_file_types):
                 content = content.read()
             else:
                 raise Exception(f'Content is of unrecognized type {type(content)}, expected {upload_file_types}')
             processed_nodes.append((f_path, content))
 
         # add multiple nodes
         for path, content in processed_nodes:
             imc.add(FileNode(path, content=content))
 
         # TODO: handle pre push scenario
         tip = imc.commit(message=message,
                          author=author,
                          parents=parents,
                          branch=parent_commit.branch)
 
         self.mark_for_invalidation(repo.repo_name)
         if trigger_push_hook:
             hooks_utils.trigger_post_push_hook(
                 username=user.username, action='push_local',
                 repo_name=repo.repo_name, repo_type=scm_instance.alias,
                 hook_type='post_push',
                 commit_ids=[tip.raw_id])
         return tip
 
     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():
             # new filename, can be renamed from the old one, also sanitaze
             # the path for any hack around relative paths like ../../ etc.
             filename = self._sanitize_path(data['filename'])
             old_filename = self._sanitize_path(_filename)
             content = data['content']
             file_mode = data.get('mode')
             filenode = FileNode(old_filename, content=content, mode=file_mode)
             op = data['op']
             if op == 'add':
                 imc.add(filenode)
             elif op == 'del':
                 imc.remove(filenode)
             elif op == 'mod':
                 if filename != old_filename:
                     # TODO: handle renames more efficient, needs vcs lib changes
                     imc.remove(filenode)
                     imc.add(FileNode(filename, content=content, mode=file_mode))
                 else:
                     imc.change(filenode)
 
         try:
             # TODO: handle pre push scenario commit changes
             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:
             # always clear caches, if commit fails we want fresh object also
             self.mark_for_invalidation(repo.repo_name)
 
         if trigger_push_hook:
             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 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):
         """
         Deletes given multiple nodes into `repo`
 
         :param user: RhodeCode User object or user_id, the committer
         :param repo: RhodeCode Repository object
         :param message: commit message
         :param nodes: mapping {filename:{'content':content},...}
         :param parent_commit: parent commit, can be empty than it's initial
            commit
         :param author: author of commit, cna be different that commiter only
            for git
         :param trigger_push_hook: trigger push hooks
 
         :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:
             f_path = self._sanitize_path(f_path)
             # content can be empty but for compatibility it allows same dicts
             # structure as add_nodes
             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))
 
         # TODO: handle pre push scenario
         tip = imc.commit(message=message,
                          author=author,
                          parents=parents,
                          branch=parent_commit.branch)
 
         self.mark_for_invalidation(repo.repo_name)
         if trigger_push_hook:
             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 strip(self, repo, commit_id, branch):
         scm_instance = repo.scm_instance(cache=False)
         scm_instance.config.clear_section('hooks')
         scm_instance.strip(commit_id, branch)
         self.mark_for_invalidation(repo.repo_name)
 
     def get_unread_journal(self):
         return self.sa.query(UserLog).count()
 
     @classmethod
     def backend_landing_ref(cls, repo_type):
         """
         Return a default landing ref based on a repository type.
         """
 
         landing_ref = {
             'hg': ('branch:default', 'default'),
             'git': ('branch:master', 'master'),
             'svn': ('rev:tip', 'latest tip'),
             'default': ('rev:tip', 'latest tip'),
         }
 
         return landing_ref.get(repo_type) or landing_ref['default']
 
     def get_repo_landing_revs(self, translator, repo=None):
         """
         Generates select option with tags branches and bookmarks (for hg only)
         grouped by type
 
         :param repo:
         """
         from rhodecode.lib.vcs.backends.git import GitRepository
 
         _ = translator
         repo = self._get_repo(repo)
 
         if repo:
             repo_type = repo.repo_type
         else:
             repo_type = 'default'
 
         default_landing_ref, landing_ref_lbl = self.backend_landing_ref(repo_type)
 
         default_ref_options = [
             [default_landing_ref, landing_ref_lbl]
         ]
         default_choices = [
             default_landing_ref
         ]
 
         if not repo:
             # presented at NEW repo creation
             return default_choices, default_ref_options
 
         repo = repo.scm_instance()
 
         ref_options = [(default_landing_ref, landing_ref_lbl)]
         choices = [default_landing_ref]
 
         # branches
         branch_group = [(f'branch:{safe_str(b)}', safe_str(b)) for b in repo.branches]
         if not branch_group:
             # new repo, or without maybe a branch?
             branch_group = default_ref_options
 
         branches_group = (branch_group, _("Branches"))
         ref_options.append(branches_group)
         choices.extend([x[0] for x in branches_group[0]])
 
         # bookmarks for HG
         if repo.alias == 'hg':
             bookmarks_group = (
                 [(f'book:{safe_str(b)}', safe_str(b))
                     for b in repo.bookmarks],
                 _("Bookmarks"))
             ref_options.append(bookmarks_group)
             choices.extend([x[0] for x in bookmarks_group[0]])
 
         # tags
         tags_group = (
             [(f'tag:{safe_str(t)}', safe_str(t))
                 for t in repo.tags],
             _("Tags"))
         ref_options.append(tags_group)
         choices.extend([x[0] for x in tags_group[0]])
 
         return choices, ref_options
 
     def get_server_info(self, environ=None):
         server_info = get_system_info(environ)
         return server_info

rhodecode/templates/files/files_source.mako

@@ -1,189 +1,189 b''
 <%namespace name="sourceblock" file="/codeblocks/source.mako"/>
 
 <%
     at_ref = request.GET.get('at')
     if at_ref:
         query={'at': at_ref}
         default_commit_id = at_ref or c.rhodecode_db_repo.landing_ref_name
     else:
         query=None
         default_commit_id = c.commit.raw_id
 %>
 
 <div id="codeblock" class="browserblock">
     <div class="browser-header">
         <div class="browser-nav">
             <div class="pull-left">
                 ## loads the history for a file
                 ${h.hidden('file_refs_filter')}
             </div>
 
             <div class="pull-right">
 
             ## Download
             % if c.lf_node:
               <a class="btn btn-default" href="${h.route_path('repo_file_download',repo_name=c.repo_name,commit_id=c.commit.raw_id,f_path=c.f_path, _query=dict(lf=1))}">
               ${_('Download largefile')}
               </a>
             % else:
               <a  class="btn btn-default" href="${h.route_path('repo_file_download',repo_name=c.repo_name,commit_id=c.commit.raw_id,f_path=c.f_path)}">
               ${_('Download file')}
               </a>
             % endif
 
             %if h.HasRepoPermissionAny('repository.write','repository.admin')(c.repo_name):
               ## on branch head, can edit files
               %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)}">
                         ${_('Edit on branch: ')}<code>${c.branch_name}</code>
                     </a>
 
                     <a class="btn btn-danger" href="${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)}">
                         ${_('Delete')}
                     </a>
                   % endif
               ## not on head, forbid all
               % else:
                ${h.link_to(_('Edit'), '#Edit', class_="btn btn-default disabled tooltip", title=_('Editing files allowed only when on branch head commit'))}
                ${h.link_to(_('Delete'), '#Delete', class_="btn btn-default btn-danger disabled tooltip", title=_('Deleting files allowed only when on branch head commit'))}
               % endif
             %endif
 
           </div>
         </div>
         <div id="file_history_container"></div>
 
         </div>
     </div>
 
     <div class="codeblock">
       <div class=" codeblock-header">
         <div class="file-filename">
             <i class="icon-file"></i> ${c.file.name}
         </div>
 
         <div class="file-stats">
 
           <div class="stats-info">
             <span class="stats-first-item">
                 % if c.file_size_too_big:
                     0 ${(_('lines'))}
                 % else:
                     ${c.file.lines()[0]} ${_ungettext('line', 'lines', c.file.lines()[0])}
                 % endif
             </span>
 
             <span> | ${h.format_byte_size_binary(c.file.size)}</span>
             % if c.lf_node:
             <span title="${_('This file is a pointer to large binary file')}"> | ${_('LargeFile')} ${h.format_byte_size_binary(c.lf_node.size)} </span>
             % endif
             <span>
                 | ${c.file.mimetype}
             </span>
 
             % if not c.file_size_too_big:
             <span> |
                 ${h.get_lexer_for_filenode(c.file).__class__.__name__}
             </span>
             % endif
 
           </div>
         </div>
       </div>
 
       <div class="path clear-fix">
           <div class="pull-left">
             ${h.files_breadcrumbs(c.repo_name, c.rhodecode_db_repo.repo_type, c.commit.raw_id, c.file.path, c.rhodecode_db_repo.landing_ref_name, request.GET.get('at'))}
           </div>
 
           <div class="pull-right stats">
                 <a id="file_history_overview" href="#loadHistory">
                     ${_('History')}
                 </a>
                  |
                 %if c.annotate:
                   ${h.link_to(_('Source'), h.route_path('repo_files', repo_name=c.repo_name,commit_id=c.commit.raw_id,f_path=c.f_path))}
                 %else:
                   ${h.link_to(_('Annotation'), h.route_path('repo_files:annotated',repo_name=c.repo_name,commit_id=c.commit.raw_id,f_path=c.f_path))}
                 %endif
                  | ${h.link_to(_('Raw'), h.route_path('repo_file_raw',repo_name=c.repo_name,commit_id=c.commit.raw_id,f_path=c.f_path))}
                  % if not c.file.is_binary:
                  |<a href="#copySource" onclick="return false;" class="no-grey clipboard-action" data-clipboard-text="${c.file.content}">${_('Copy content')}</a>
                  |<a href="#copyPermaLink" onclick="return false;" class="no-grey clipboard-action" data-clipboard-text="${h.route_url('repo_files', repo_name=c.repo_name,commit_id=c.commit.raw_id,f_path=c.f_path)}">${_('Copy permalink')}</a>
                  % endif
 
           </div>
           <div class="clear-fix"></div>
       </div>
 
       <div class="code-body clear-fix ">
         %if c.file.is_binary:
              <% rendered_binary = h.render_binary(c.repo_name, c.file)%>
              % if rendered_binary:
                  <div class="text-center">
                  ${rendered_binary}
                  </div>
              % else:
                  <div>
                   ${_('Binary file ({})').format(c.file.mimetype)}
                  </div>
              % endif
         %else:
             % if c.file_size_too_big:
                 ${_('File size {} is bigger then allowed limit {}. ').format(h.format_byte_size_binary(c.file.size), h.format_byte_size_binary(c.visual.cut_off_limit_file))} ${h.link_to(_('Show as raw'),
                 h.route_path('repo_file_raw',repo_name=c.repo_name,commit_id=c.commit.raw_id,f_path=c.f_path))}
             % else:
                 %if c.renderer and not c.annotate:
                     ## pick relative url based on renderer
                     <%
                         relative_urls = {
                             'raw': h.route_path('repo_file_raw',repo_name=c.repo_name,commit_id=c.commit.raw_id,f_path=c.f_path),
                             'standard': h.route_path('repo_files',repo_name=c.repo_name,commit_id=c.commit.raw_id,f_path=c.f_path),
                         }
                     %>
                     ${h.render(c.file.str_content, renderer=c.renderer, relative_urls=relative_urls)}
                 %else:
                     <table class="cb codehilite">
                     %if c.annotate:
                       <% color_hasher = h.color_hasher() %>
                       %for annotation, lines in c.annotated_lines:
                         ${sourceblock.render_annotation_lines(annotation, lines, color_hasher)}
                       %endfor
                     %else:
                       %for line_num, tokens in enumerate(c.lines, 1):
                         ${sourceblock.render_line(line_num, tokens)}
                       %endfor
                     %endif
                     </table>
                 %endif
             % endif
         %endif
     </div>
 
     </div>
 
 <script type="text/javascript">
 % if request.GET.get('mark'):
 
 $(function(){
   $(".codehilite").mark(
       "${request.GET.get('mark')}",
           {
               "className": 'match',
               "accuracy": "complementary",
               "ignorePunctuation": ":._(){}[]!'+=".split(""),
               "each": function(el) {
                   // and also highlight lines !
                   $($(el).closest('tr')).find('td.cb-lineno').addClass('cb-line-selected');
               }
           }
   );
 
 });
 % endif
 </script>

rhodecode/templates/files/files_upload.mako

@@ -1,211 +1,222 b''
 <%inherit file="/base/base.mako"/>
 
 <%def name="title()">
     ${_('{} Files Upload').format(c.repo_name)}
     %if c.rhodecode_name:
         &middot; ${h.branding(c.rhodecode_name)}
     %endif
 </%def>
 
 <%def name="menu_bar_nav()">
     ${self.menu_items(active='repositories')}
 </%def>
 
 <%def name="breadcrumbs_links()"></%def>
 
 <%def name="menu_bar_subnav()">
     ${self.repo_menu(active='files')}
 </%def>
 
 <%def name="main()">
 
 <div class="box">
     ## Template for uploads
     <div style="display: none" id="tpl-dropzone">
         <div class="dz-preview dz-file-preview">
             <div class="dz-details">
 
                 <div class="dz-filename">
                     <span data-dz-name></span>
                 </div>
                 <div class="dz-filename-size">
                     <span class="dz-size" data-dz-size></span>
 
                 </div>
 
                 <div class="dz-sending" style="display: none">${_('Uploading...')}</div>
                 <div class="dz-response" style="display: none">
                     ${_('Uploaded')} 100%
                 </div>
 
             </div>
             <div class="dz-progress">
                 <span class="dz-upload" data-dz-uploadprogress></span>
             </div>
 
             <div class="dz-error-message">
             </div>
         </div>
     </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}
         </span>
         % 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>
 
             <div class="dropzone-wrapper" id="file-uploader" style="border: none; padding: 40px 0">
                 <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>
 
             </div>
         </div>
 
     </div>
 
     <div class="upload-form edit-file-fieldset">
         <div class="fieldset">
             <div class="message">
                 <textarea id="commit" name="message"  placeholder="${c.default_message}"></textarea>
             </div>
         </div>
         <div class="pull-left">
             ${h.submit('commit_btn',_('Commit changes'), class_="btn btn-small btn-success")}
         </div>
     </div>
     ##${h.end_form()}
 
     <div class="file-upload-transaction-wrapper" style="display: none">
     <div class="file-upload-transaction">
         <h3>${_('Committing...')}</h3>
         <p>${_('Please wait while the files are being uploaded')}</p>
         <p class="error" style="display: none">
 
         </p>
         <i class="icon-spin animate-spin"></i>
         <p></p>
     </div>
     </div>
 
 </div>
 
 <script type="text/javascript">
 
     $(document).ready(function () {
 
         //see: https://www.dropzonejs.com/#configuration
         myDropzone = new Dropzone("div#file-uploader", {
             url: "${form_url}",
             headers: {"X-CSRF-Token": CSRF_TOKEN},
             paramName: function () {
                 return "files_upload"
             }, // The name that will be used to transfer the file
             parallelUploads: 20,
             maxFiles: 20,
             uploadMultiple: true,
             //chunking: true, // use chunking transfer, not supported at the moment
             //maxFilesize: 2, // in MBs
             autoProcessQueue: false, // if false queue will not be processed automatically.
             createImageThumbnails: false,
             previewTemplate: document.querySelector('#tpl-dropzone').innerHTML,
             accept: function (file, done) {
                 done();
             },
             init: function () {
                 this.on("addedfile", function (file) {
 
                 });
 
                 this.on("sending", function (file, xhr, formData) {
                     formData.append("message", $('#commit').val());
                     $(file.previewElement).find('.dz-sending').show();
                 });
 
                 this.on("success", function (file, response) {
                     $(file.previewElement).find('.dz-sending').hide();
                     $(file.previewElement).find('.dz-response').show();
 
                     if (response.error !== null) {
                         $('.file-upload-transaction-wrapper .error').html('ERROR: {0}'.format(response.error));
                         $('.file-upload-transaction-wrapper .error').show();
                         $('.file-upload-transaction-wrapper i').hide()
                     }
 
                     var redirect_url = response.redirect_url || '/';
                     window.location = redirect_url
 
                 });
 
                 this.on("error", function (file, errorMessage, xhr) {
                     var error = null;
 
                     if (xhr !== undefined){
                         var httpStatus = xhr.status + " " + xhr.statusText;
                         if (xhr !== undefined && xhr.status >= 500) {
                             error = httpStatus;
                         }
                     }
 
                     if (error === null) {
                         error = errorMessage.error || errorMessage || httpStatus;
                     }
 
                     $(file.previewElement).find('.dz-error-message').html('ERROR: {0}'.format(error));
                 });
             }
             });
 
         $('#commit_btn').on('click', function(e) {
             e.preventDefault();
             var button = $(this);
             if (button.hasClass('clicked')) {
                 button.attr('disabled', true);
             } else {
                 button.addClass('clicked');
             }
 
             var files = myDropzone.getQueuedFiles();
             if (files.length === 0) {
                 alert("Missing files");
                 e.preventDefault();
             }
 
             $('.upload-form').hide();
             $('.file-upload-transaction-wrapper').show();
             myDropzone.processQueue();
 
         });
 
     });
 
 </script>
 </%def>

rhodecode/tests/routes.py

@@ -1,319 +1,320 b''
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
 def get_url_defs():
     from rhodecode.apps._base import ADMIN_PREFIX
 
     return {
         "home": "/",
         "main_page_repos_data": "/_home_repos",
         "main_page_repo_groups_data": "/_home_repo_groups",
         "repo_group_home": "/{repo_group_name}",
         "user_autocomplete_data": "/_users",
         "user_group_autocomplete_data": "/_user_groups",
         "repo_list_data": "/_repos",
         "goto_switcher_data": "/_goto_data",
         "admin_home": ADMIN_PREFIX + "",
         "admin_audit_logs": ADMIN_PREFIX + "/audit_logs",
         "admin_defaults_repositories": ADMIN_PREFIX + "/defaults/repositories",
         "admin_defaults_repositories_update": ADMIN_PREFIX
         + "/defaults/repositories/update",
         "search": ADMIN_PREFIX + "/search",
         "search_repo": "/{repo_name}/search",
         "my_account_auth_tokens": ADMIN_PREFIX + "/my_account/auth_tokens",
         "my_account_auth_tokens_add": ADMIN_PREFIX + "/my_account/auth_tokens/new",
         "my_account_auth_tokens_delete": ADMIN_PREFIX
         + "/my_account/auth_tokens/delete",
         "repos": ADMIN_PREFIX + "/repos",
         "repos_data": ADMIN_PREFIX + "/repos_data",
         "repo_groups": ADMIN_PREFIX + "/repo_groups",
         "repo_groups_data": ADMIN_PREFIX + "/repo_groups_data",
         "user_groups": ADMIN_PREFIX + "/user_groups",
         "user_groups_data": ADMIN_PREFIX + "/user_groups_data",
         "user_profile": "/_profiles/{username}",
         "profile_user_group": "/_profile_user_group/{user_group_name}",
         "repo_summary": "/{repo_name}",
         "repo_creating_check": "/{repo_name}/repo_creating_check",
         "edit_repo": "/{repo_name}/settings",
         "edit_repo_vcs": "/{repo_name}/settings/vcs",
         "edit_repo_vcs_update": "/{repo_name}/settings/vcs/update",
         "edit_repo_vcs_svn_pattern_delete": "/{repo_name}/settings/vcs/svn_pattern/delete",
         "repo_archivefile": "/{repo_name}/archive/{fname}",
         "repo_files_diff": "/{repo_name}/diff/{f_path}",
         "repo_files_diff_2way_redirect": "/{repo_name}/diff-2way/{f_path}",
         "repo_files": "/{repo_name}/files/{commit_id}/{f_path}",
         "repo_files:default_path": "/{repo_name}/files/{commit_id}/",
         "repo_files:default_commit": "/{repo_name}/files",
         "repo_files:rendered": "/{repo_name}/render/{commit_id}/{f_path}",
         "repo_files:annotated": "/{repo_name}/annotate/{commit_id}/{f_path}",
         "repo_files:annotated_previous": "/{repo_name}/annotate-previous/{commit_id}/{f_path}",
         "repo_files_nodelist": "/{repo_name}/nodelist/{commit_id}/{f_path}",
         "repo_file_raw": "/{repo_name}/raw/{commit_id}/{f_path}",
         "repo_file_download": "/{repo_name}/download/{commit_id}/{f_path}",
         "repo_file_history": "/{repo_name}/history/{commit_id}/{f_path}",
         "repo_file_authors": "/{repo_name}/authors/{commit_id}/{f_path}",
         "repo_files_remove_file": "/{repo_name}/remove_file/{commit_id}/{f_path}",
         "repo_files_delete_file": "/{repo_name}/delete_file/{commit_id}/{f_path}",
         "repo_files_edit_file": "/{repo_name}/edit_file/{commit_id}/{f_path}",
         "repo_files_update_file": "/{repo_name}/update_file/{commit_id}/{f_path}",
         "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",
         "journal_rss": ADMIN_PREFIX + "/journal/rss",
         "journal_atom": ADMIN_PREFIX + "/journal/atom",
         "journal_public": ADMIN_PREFIX + "/public_journal",
         "journal_public_atom": ADMIN_PREFIX + "/public_journal/atom",
         "journal_public_atom_old": ADMIN_PREFIX + "/public_journal_atom",
         "journal_public_rss": ADMIN_PREFIX + "/public_journal/rss",
         "journal_public_rss_old": ADMIN_PREFIX + "/public_journal_rss",
         "toggle_following": ADMIN_PREFIX + "/toggle_following",
         "upload_file": "/_file_store/upload",
         "download_file": "/_file_store/download/{fid}",
         "download_file_by_token": "/_file_store/token-download/{_auth_token}/{fid}",
         "gists_show": ADMIN_PREFIX + "/gists",
         "gists_new": ADMIN_PREFIX + "/gists/new",
         "gists_create": ADMIN_PREFIX + "/gists/create",
         "gist_show": ADMIN_PREFIX + "/gists/{gist_id}",
         "gist_delete": ADMIN_PREFIX + "/gists/{gist_id}/delete",
         "gist_edit": ADMIN_PREFIX + "/gists/{gist_id}/edit",
         "gist_edit_check_revision": ADMIN_PREFIX
         + "/gists/{gist_id}/edit/check_revision",
         "gist_update": ADMIN_PREFIX + "/gists/{gist_id}/update",
         "gist_show_rev": ADMIN_PREFIX + "/gists/{gist_id}/rev/{revision}",
         "gist_show_formatted": ADMIN_PREFIX
         + "/gists/{gist_id}/rev/{revision}/{format}",
         "gist_show_formatted_path": ADMIN_PREFIX
         + "/gists/{gist_id}/rev/{revision}/{format}/{f_path}",
         "login": ADMIN_PREFIX + "/login",
         "logout": ADMIN_PREFIX + "/logout",
         "register": ADMIN_PREFIX + "/register",
         "reset_password": ADMIN_PREFIX + "/password_reset",
         "reset_password_confirmation": ADMIN_PREFIX + "/password_reset_confirmation",
         "admin_permissions_application": ADMIN_PREFIX + "/permissions/application",
         "admin_permissions_application_update": ADMIN_PREFIX
         + "/permissions/application/update",
         "repo_commit_raw": "/{repo_name}/changeset-diff/{commit_id}",
         "user_group_members_data": ADMIN_PREFIX
         + "/user_groups/{user_group_id}/members",
         "user_groups_new": ADMIN_PREFIX + "/user_groups/new",
         "user_groups_create": ADMIN_PREFIX + "/user_groups/create",
         "edit_user_group": ADMIN_PREFIX + "/user_groups/{user_group_id}/edit",
         "edit_user_group_advanced_sync": ADMIN_PREFIX
         + "/user_groups/{user_group_id}/edit/advanced/sync",
         "edit_user_group_global_perms_update": ADMIN_PREFIX
         + "/user_groups/{user_group_id}/edit/global_permissions/update",
         "user_groups_update": ADMIN_PREFIX + "/user_groups/{user_group_id}/update",
         "user_groups_delete": ADMIN_PREFIX + "/user_groups/{user_group_id}/delete",
         "edit_user_group_perms": ADMIN_PREFIX
         + "/user_groups/{user_group_id}/edit/permissions",
         "edit_user_group_perms_update": ADMIN_PREFIX
         + "/user_groups/{user_group_id}/edit/permissions/update",
         "edit_repo_group": "/{repo_group_name}/_edit",
         "edit_repo_group_perms": "/{repo_group_name:}/_settings/permissions",
         "edit_repo_group_perms_update": "/{repo_group_name}/_settings/permissions/update",
         "edit_repo_group_advanced": "/{repo_group_name}/_settings/advanced",
         "edit_repo_group_advanced_delete": "/{repo_group_name}/_settings/advanced/delete",
         "edit_user_ssh_keys": ADMIN_PREFIX + "/users/{user_id}/edit/ssh_keys",
         "edit_user_ssh_keys_generate_keypair": ADMIN_PREFIX
         + "/users/{user_id}/edit/ssh_keys/generate",
         "edit_user_ssh_keys_add": ADMIN_PREFIX + "/users/{user_id}/edit/ssh_keys/new",
         "edit_user_ssh_keys_delete": ADMIN_PREFIX
         + "/users/{user_id}/edit/ssh_keys/delete",
         "users": ADMIN_PREFIX + "/users",
         "users_data": ADMIN_PREFIX + "/users_data",
         "users_create": ADMIN_PREFIX + "/users/create",
         "users_new": ADMIN_PREFIX + "/users/new",
         "user_edit": ADMIN_PREFIX + "/users/{user_id}/edit",
         "user_edit_advanced": ADMIN_PREFIX + "/users/{user_id}/edit/advanced",
         "user_edit_global_perms": ADMIN_PREFIX
         + "/users/{user_id}/edit/global_permissions",
         "user_edit_global_perms_update": ADMIN_PREFIX
         + "/users/{user_id}/edit/global_permissions/update",
         "user_update": ADMIN_PREFIX + "/users/{user_id}/update",
         "user_delete": ADMIN_PREFIX + "/users/{user_id}/delete",
         "user_create_personal_repo_group": ADMIN_PREFIX
         + "/users/{user_id}/create_repo_group",
         "edit_user_auth_tokens": ADMIN_PREFIX + "/users/{user_id}/edit/auth_tokens",
         "edit_user_auth_tokens_add": ADMIN_PREFIX
         + "/users/{user_id}/edit/auth_tokens/new",
         "edit_user_auth_tokens_delete": ADMIN_PREFIX
         + "/users/{user_id}/edit/auth_tokens/delete",
         "edit_user_emails": ADMIN_PREFIX + "/users/{user_id}/edit/emails",
         "edit_user_emails_add": ADMIN_PREFIX + "/users/{user_id}/edit/emails/new",
         "edit_user_emails_delete": ADMIN_PREFIX + "/users/{user_id}/edit/emails/delete",
         "edit_user_ips": ADMIN_PREFIX + "/users/{user_id}/edit/ips",
         "edit_user_ips_add": ADMIN_PREFIX + "/users/{user_id}/edit/ips/new",
         "edit_user_ips_delete": ADMIN_PREFIX + "/users/{user_id}/edit/ips/delete",
         "edit_user_perms_summary": ADMIN_PREFIX
         + "/users/{user_id}/edit/permissions_summary",
         "edit_user_perms_summary_json": ADMIN_PREFIX
         + "/users/{user_id}/edit/permissions_summary/json",
         "edit_user_audit_logs": ADMIN_PREFIX + "/users/{user_id}/edit/audit",
         "edit_user_audit_logs_download": ADMIN_PREFIX
         + "/users/{user_id}/edit/audit/download",
         "admin_settings": ADMIN_PREFIX + "/settings",
         "admin_settings_update": ADMIN_PREFIX + "/settings/update",
         "admin_settings_global": ADMIN_PREFIX + "/settings/global",
         "admin_settings_global_update": ADMIN_PREFIX + "/settings/global/update",
         "admin_settings_vcs": ADMIN_PREFIX + "/settings/vcs",
         "admin_settings_vcs_update": ADMIN_PREFIX + "/settings/vcs/update",
         "admin_settings_vcs_svn_pattern_delete": ADMIN_PREFIX
         + "/settings/vcs/svn_pattern_delete",
         "admin_settings_mapping": ADMIN_PREFIX + "/settings/mapping",
         "admin_settings_mapping_update": ADMIN_PREFIX + "/settings/mapping/update",
         "admin_settings_visual": ADMIN_PREFIX + "/settings/visual",
         "admin_settings_visual_update": ADMIN_PREFIX + "/settings/visual/update",
         "admin_settings_issuetracker": ADMIN_PREFIX + "/settings/issue-tracker",
         "admin_settings_issuetracker_update": ADMIN_PREFIX
         + "/settings/issue-tracker/update",
         "admin_settings_issuetracker_test": ADMIN_PREFIX
         + "/settings/issue-tracker/test",
         "admin_settings_issuetracker_delete": ADMIN_PREFIX
         + "/settings/issue-tracker/delete",
         "admin_settings_email": ADMIN_PREFIX + "/settings/email",
         "admin_settings_email_update": ADMIN_PREFIX + "/settings/email/update",
         "admin_settings_hooks": ADMIN_PREFIX + "/settings/hooks",
         "admin_settings_hooks_update": ADMIN_PREFIX + "/settings/hooks/update",
         "admin_settings_hooks_delete": ADMIN_PREFIX + "/settings/hooks/delete",
         "admin_settings_search": ADMIN_PREFIX + "/settings/search",
         "admin_settings_labs": ADMIN_PREFIX + "/settings/labs",
         "admin_settings_labs_update": ADMIN_PREFIX + "/settings/labs/update",
         "admin_settings_sessions": ADMIN_PREFIX + "/settings/sessions",
         "admin_settings_sessions_cleanup": ADMIN_PREFIX + "/settings/sessions/cleanup",
         "admin_settings_system": ADMIN_PREFIX + "/settings/system",
         "admin_settings_system_update": ADMIN_PREFIX + "/settings/system/updates",
         "admin_settings_open_source": ADMIN_PREFIX + "/settings/open_source",
         "repo_group_new": ADMIN_PREFIX + "/repo_group/new",
         "repo_group_create": ADMIN_PREFIX + "/repo_group/create",
         "repo_new": ADMIN_PREFIX + "/repos/new",
         "repo_create": ADMIN_PREFIX + "/repos/create",
         "admin_permissions_global": ADMIN_PREFIX + "/permissions/global",
         "admin_permissions_global_update": ADMIN_PREFIX + "/permissions/global/update",
         "admin_permissions_object": ADMIN_PREFIX + "/permissions/object",
         "admin_permissions_object_update": ADMIN_PREFIX + "/permissions/object/update",
         "admin_permissions_ips": ADMIN_PREFIX + "/permissions/ips",
         "admin_permissions_overview": ADMIN_PREFIX + "/permissions/overview",
         "admin_permissions_ssh_keys": ADMIN_PREFIX + "/permissions/ssh_keys",
         "admin_permissions_ssh_keys_data": ADMIN_PREFIX + "/permissions/ssh_keys/data",
         "admin_permissions_ssh_keys_update": ADMIN_PREFIX
         + "/permissions/ssh_keys/update",
         "pullrequest_show": "/{repo_name}/pull-request/{pull_request_id}",
         "pull_requests_global": ADMIN_PREFIX + "/pull-request/{pull_request_id}",
         "pull_requests_global_0": ADMIN_PREFIX + "/pull_requests/{pull_request_id}",
         "pull_requests_global_1": ADMIN_PREFIX + "/pull-requests/{pull_request_id}",
         "notifications_show_all": ADMIN_PREFIX + "/notifications",
         "notifications_mark_all_read": ADMIN_PREFIX + "/notifications_mark_all_read",
         "notifications_show": ADMIN_PREFIX + "/notifications/{notification_id}",
         "notifications_update": ADMIN_PREFIX
         + "/notifications/{notification_id}/update",
         "notifications_delete": ADMIN_PREFIX
         + "/notifications/{notification_id}/delete",
         "my_account": ADMIN_PREFIX + "/my_account/profile",
         "my_account_edit": ADMIN_PREFIX + "/my_account/edit",
         "my_account_update": ADMIN_PREFIX + "/my_account/update",
         "my_account_pullrequests": ADMIN_PREFIX + "/my_account/pull_requests",
         "my_account_pullrequests_data": ADMIN_PREFIX + "/my_account/pull_requests/data",
         "my_account_emails": ADMIN_PREFIX + "/my_account/emails",
         "my_account_emails_add": ADMIN_PREFIX + "/my_account/emails/new",
         "my_account_emails_delete": ADMIN_PREFIX + "/my_account/emails/delete",
         "my_account_password": ADMIN_PREFIX + "/my_account/password",
         "my_account_password_update": ADMIN_PREFIX + "/my_account/password/update",
         "my_account_repos": ADMIN_PREFIX + "/my_account/repos",
         "my_account_watched": ADMIN_PREFIX + "/my_account/watched",
         "my_account_perms": ADMIN_PREFIX + "/my_account/perms",
         "my_account_notifications": ADMIN_PREFIX + "/my_account/notifications",
         "my_account_ssh_keys": ADMIN_PREFIX + "/my_account/ssh_keys",
         "my_account_ssh_keys_generate": ADMIN_PREFIX + "/my_account/ssh_keys/generate",
         "my_account_ssh_keys_add": ADMIN_PREFIX + "/my_account/ssh_keys/new",
         "my_account_ssh_keys_delete": ADMIN_PREFIX + "/my_account/ssh_keys/delete",
         "pullrequest_show_all": "/{repo_name}/pull-request",
         "pullrequest_show_all_data": "/{repo_name}/pull-request-data",
         "bookmarks_home": "/{repo_name}/bookmarks",
         "branches_home": "/{repo_name}/branches",
         "tags_home": "/{repo_name}/tags",
         "repo_changelog": "/{repo_name}/changelog",
         "repo_commits": "/{repo_name}/commits",
         "repo_commits_file": "/{repo_name}/commits/{commit_id}/{f_path}",
         "repo_commits_elements": "/{repo_name}/commits_elements",
         "repo_commit": "/{repo_name}/changeset/{commit_id}",
         "repo_commit_comment_create": "/{repo_name}/changeset/{commit_id}/comment/create",
         "repo_commit_comment_preview": "/{repo_name}/changeset/{commit_id}/comment/preview",
         "repo_commit_comment_delete": "/{repo_name}/changeset/{commit_id}/comment/{comment_id}/delete",
         "repo_commit_comment_edit": "/{repo_name}/changeset/{commit_id}/comment/{comment_id}/edit",
         "repo_commit_children": "/{repo_name}/changeset_children/{commit_id}",
         "repo_commit_parents": "/{repo_name}/changeset_parents/{commit_id}",
         "repo_commit_patch": "/{repo_name}/changeset-patch/{commit_id}",
         "repo_commit_download": "/{repo_name}/changeset-download/{commit_id}",
         "repo_commit_data": "/{repo_name}/changeset-data/{commit_id}",
         "repo_compare": "/{repo_name}/compare/{source_ref_type}@{source_ref}...{target_ref_type}@{target_ref}",
         "repo_compare_select": "/{repo_name}/compare",
         "rss_feed_home": "/{repo_name}/feed-rss",
         "atom_feed_home": "/{repo_name}/feed-atom",
         "rss_feed_home_old": "/{repo_name}/feed/rss",
         "atom_feed_home_old": "/{repo_name}/feed/atom",
         "repo_fork_new": "/{repo_name}/fork",
         "repo_fork_create": "/{repo_name}/fork/create",
         "repo_forks_show_all": "/{repo_name}/forks",
         "repo_forks_data": "/{repo_name}/forks/data",
         "edit_repo_issuetracker": "/{repo_name}/settings/issue_trackers",
         "edit_repo_issuetracker_test": "/{repo_name}/settings/issue_trackers/test",
         "edit_repo_issuetracker_delete": "/{repo_name}/settings/issue_trackers/delete",
         "edit_repo_issuetracker_update": "/{repo_name}/settings/issue_trackers/update",
         "edit_repo_maintenance": "/{repo_name}/settings/maintenance",
         "edit_repo_maintenance_execute": "/{repo_name}/settings/maintenance/execute",
         "repo_changelog_file": "/{repo_name}/changelog/{commit_id}/{f_path}",
         "pullrequest_repo_refs": "/{repo_name}/pull-request/refs/{target_repo_name:.*?[^/]}",
         "pullrequest_repo_targets": "/{repo_name}/pull-request/repo-destinations",
         "pullrequest_new": "/{repo_name}/pull-request/new",
         "pullrequest_create": "/{repo_name}/pull-request/create",
         "pullrequest_update": "/{repo_name}/pull-request/{pull_request_id}/update",
         "pullrequest_merge": "/{repo_name}/pull-request/{pull_request_id}/merge",
         "pullrequest_delete": "/{repo_name}/pull-request/{pull_request_id}/delete",
         "pullrequest_comment_create": "/{repo_name}/pull-request/{pull_request_id}/comment",
         "pullrequest_comment_delete": "/{repo_name}/pull-request/{pull_request_id}/comment/{comment_id}/delete",
         "pullrequest_comment_edit": "/{repo_name}/pull-request/{pull_request_id}/comment/{comment_id}/edit",
         "edit_repo_caches": "/{repo_name}/settings/caches",
         "edit_repo_perms": "/{repo_name}/settings/permissions",
         "edit_repo_fields": "/{repo_name}/settings/fields",
         "edit_repo_remote": "/{repo_name}/settings/remote",
         "edit_repo_statistics": "/{repo_name}/settings/statistics",
         "edit_repo_advanced": "/{repo_name}/settings/advanced",
         "edit_repo_advanced_delete": "/{repo_name}/settings/advanced/delete",
         "edit_repo_advanced_archive": "/{repo_name}/settings/advanced/archive",
         "edit_repo_advanced_fork": "/{repo_name}/settings/advanced/fork",
         "edit_repo_advanced_locking": "/{repo_name}/settings/advanced/locking",
         "edit_repo_advanced_journal": "/{repo_name}/settings/advanced/journal",
         "repo_stats": "/{repo_name}/repo_stats/{commit_id}",
         "repo_refs_data": "/{repo_name}/refs-data",
         "repo_refs_changelog_data": "/{repo_name}/refs-data-changelog",
         "repo_artifacts_stream_store": "/_file_store/stream-upload",
     }
 
 
 def route_path(name, params=None, **kwargs):
     import urllib.parse
 
     base_url = get_url_defs()[name].format(**kwargs)
 
     if params:
         base_url = f"{base_url}?{urllib.parse.urlencode(params)}"
     return base_url