##// END OF EJS Templates
fix(caching): fixed problems with Cache query for users....
fix(caching): fixed problems with Cache query for users. The old way of querying caused the user get query to be always cached, and returning old results even in 2fa forms. The new limited query doesn't cache the user object resolving issues

File last commit:

r5198:919dd05c default
r5365:ae8a165b default
Show More
test_repo_compare.py
656 lines | 26.4 KiB | text/x-python | PythonLexer
# 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 mock
import pytest
import lxml.html
from rhodecode.lib.vcs.exceptions import RepositoryRequirementError
from rhodecode.tests import assert_session_flash
from rhodecode.tests.utils import AssertResponse, commit_change
from rhodecode.tests.routes import route_path
@pytest.mark.usefixtures("autologin_user", "app")
class TestCompareView(object):
def test_compare_index_is_reached_at_least_once(self, backend):
repo = backend.repo
self.app.get(
route_path('repo_compare_select', repo_name=repo.repo_name))
@pytest.mark.xfail_backends("svn", reason="Requires pull")
def test_compare_remote_with_different_commit_indexes(self, backend):
# Preparing the following repository structure:
#
# Origin repository has two commits:
#
# 0 1
# A -- D
#
# The fork of it has a few more commits and "D" has a commit index
# which does not exist in origin.
#
# 0 1 2 3 4
# A -- -- -- D -- E
# \- B -- C
#
fork = backend.create_repo()
origin = backend.create_repo()
# prepare fork
commit0 = commit_change(
fork.repo_name, filename=b'file1', content=b'A',
message='A - Initial Commit', vcs_type=backend.alias, parent=None, newfile=True)
commit1 = commit_change(
fork.repo_name, filename=b'file1', content=b'B',
message='B, child of A', vcs_type=backend.alias, parent=commit0)
commit_change( # commit 2
fork.repo_name, filename=b'file1', content=b'C',
message='C, child of B', vcs_type=backend.alias, parent=commit1)
commit3 = commit_change(
fork.repo_name, filename=b'file1', content=b'D',
message='D, child of A', vcs_type=backend.alias, parent=commit0, branch='feature')
commit4 = commit_change(
fork.repo_name, filename=b'file1', content=b'E',
message='E, child of D', vcs_type=backend.alias, parent=commit3, branch='feature')
# prepare origin repository, taking just the history up to D
origin_repo = origin.scm_instance(cache=False)
origin_repo.config.clear_section('hooks')
origin_repo.pull(fork.repo_full_path, commit_ids=[commit3.raw_id])
origin_repo = origin.scm_instance(cache=False) # cache rebuild
# Verify test fixture setup
# This does not work for git
if backend.alias != 'git':
assert 5 == len(fork.scm_instance(cache=False).commit_ids)
assert 2 == len(origin_repo.commit_ids)
# Comparing the revisions
response = self.app.get(
route_path('repo_compare',
repo_name=origin.repo_name,
source_ref_type="rev", source_ref=commit3.raw_id,
target_ref_type="rev", target_ref=commit4.raw_id,
params=dict(merge='1', target_repo=fork.repo_name)
),
status=200)
compare_page = ComparePage(response)
compare_page.contains_commits([commit4])
@pytest.mark.xfail_backends("svn", reason="Depends on branch support")
def test_compare_forks_on_branch_extra_commits(self, backend):
repo1 = backend.create_repo()
# commit something !
commit0 = commit_change(
repo1.repo_name, filename=b'file1', content=b'line1\n',
message='commit1', vcs_type=backend.alias, parent=None,
newfile=True)
# fork this repo
repo2 = backend.create_fork()
# add two extra commit into fork
commit1 = commit_change(
repo2.repo_name, filename=b'file1', content=b'line1\nline2\n',
message='commit2', vcs_type=backend.alias, parent=commit0)
commit2 = commit_change(
repo2.repo_name, filename=b'file1', content=b'line1\nline2\nline3\n',
message='commit3', vcs_type=backend.alias, parent=commit1)
commit_id1 = repo1.scm_instance().DEFAULT_BRANCH_NAME
commit_id2 = repo2.scm_instance().DEFAULT_BRANCH_NAME
response = self.app.get(
route_path('repo_compare',
repo_name=repo1.repo_name,
source_ref_type="branch", source_ref=commit_id2,
target_ref_type="branch", target_ref=commit_id1,
params=dict(merge='1', target_repo=repo2.repo_name)
))
response.mustcontain('%s@%s' % (repo1.repo_name, commit_id2))
response.mustcontain('%s@%s' % (repo2.repo_name, commit_id1))
compare_page = ComparePage(response)
compare_page.contains_change_summary(1, 2, 0)
compare_page.contains_commits([commit1, commit2])
anchor = 'a_c-{}-826e8142e6ba'.format(commit0.short_id)
compare_page.contains_file_links_and_anchors([('file1', anchor), ])
# Swap is removed when comparing branches since it's a PR feature and
# it is then a preview mode
compare_page.swap_is_hidden()
compare_page.target_source_are_disabled()
@pytest.mark.xfail_backends("svn", reason="Depends on branch support")
def test_compare_forks_on_branch_extra_commits_origin_has_incoming(self, backend):
repo1 = backend.create_repo()
# commit something !
commit0 = commit_change(
repo1.repo_name, filename=b'file1', content=b'line1\n',
message='commit1', vcs_type=backend.alias, parent=None,
newfile=True)
# fork this repo
repo2 = backend.create_fork()
# now commit something to origin repo
commit_change(
repo1.repo_name, filename=b'file2', content=b'line1file2\n',
message='commit2', vcs_type=backend.alias, parent=commit0,
newfile=True)
# add two extra commit into fork
commit1 = commit_change(
repo2.repo_name, filename=b'file1', content=b'line1\nline2\n',
message='commit2', vcs_type=backend.alias, parent=commit0)
commit2 = commit_change(
repo2.repo_name, filename=b'file1', content=b'line1\nline2\nline3\n',
message='commit3', vcs_type=backend.alias, parent=commit1)
commit_id1 = repo1.scm_instance().DEFAULT_BRANCH_NAME
commit_id2 = repo2.scm_instance().DEFAULT_BRANCH_NAME
response = self.app.get(
route_path('repo_compare',
repo_name=repo1.repo_name,
source_ref_type="branch", source_ref=commit_id2,
target_ref_type="branch", target_ref=commit_id1,
params=dict(merge='1', target_repo=repo2.repo_name),
))
response.mustcontain(f'{repo1.repo_name}@{commit_id2}')
response.mustcontain(f'{repo2.repo_name}@{commit_id1}')
compare_page = ComparePage(response)
compare_page.contains_change_summary(1, 2, 0)
compare_page.contains_commits([commit1, commit2])
anchor = f'a_c-{commit0.short_id}-826e8142e6ba'
compare_page.contains_file_links_and_anchors([('file1', anchor), ])
# Swap is removed when comparing branches since it's a PR feature and
# it is then a preview mode
compare_page.swap_is_hidden()
compare_page.target_source_are_disabled()
@pytest.mark.xfail_backends("svn")
# TODO(marcink): no svn support for compare two seperate repos
def test_compare_of_unrelated_forks(self, backend):
orig = backend.create_repo(number_of_commits=1)
fork = backend.create_repo(number_of_commits=1)
response = self.app.get(
route_path('repo_compare',
repo_name=orig.repo_name,
source_ref_type="rev", source_ref="tip",
target_ref_type="rev", target_ref="tip",
params=dict(merge='1', target_repo=fork.repo_name),
),
status=302)
response = response.follow()
response.mustcontain("Repositories unrelated.")
@pytest.mark.xfail_backends("svn")
def test_compare_cherry_pick_commits_from_bottom(self, backend):
# repo1:
# commit0:
# commit1:
# repo1-fork- in which we will cherry pick bottom commits
# commit0:
# commit1:
# commit2: x
# commit3: x
# commit4: x
# commit5:
# make repo1, and commit1+commit2
repo1 = backend.create_repo()
# commit something !
commit0 = commit_change(
repo1.repo_name, filename=b'file1', content=b'line1\n',
message='commit1', vcs_type=backend.alias, parent=None,
newfile=True)
commit1 = commit_change(
repo1.repo_name, filename=b'file1', content=b'line1\nline2\n',
message='commit2', vcs_type=backend.alias, parent=commit0)
# fork this repo
repo2 = backend.create_fork()
# now make commit3-6
commit2 = commit_change(
repo1.repo_name, filename=b'file1', content=b'line1\nline2\nline3\n',
message='commit3', vcs_type=backend.alias, parent=commit1)
commit3 = commit_change(
repo1.repo_name, filename=b'file1',content=b'line1\nline2\nline3\nline4\n',
message='commit4', vcs_type=backend.alias, parent=commit2)
commit4 = commit_change(
repo1.repo_name, filename=b'file1', content=b'line1\nline2\nline3\nline4\nline5\n',
message='commit5', vcs_type=backend.alias, parent=commit3)
commit_change( # commit 5
repo1.repo_name, filename=b'file1', content=b'line1\nline2\nline3\nline4\nline5\nline6\n',
message='commit6', vcs_type=backend.alias, parent=commit4)
response = self.app.get(
route_path('repo_compare',
repo_name=repo2.repo_name,
# parent of commit2, in target repo2
source_ref_type="rev", source_ref=commit1.raw_id,
target_ref_type="rev", target_ref=commit4.raw_id,
params=dict(merge='1', target_repo=repo1.repo_name),
))
response.mustcontain('%s@%s' % (repo2.repo_name, commit1.short_id))
response.mustcontain('%s@%s' % (repo1.repo_name, commit4.short_id))
# files
compare_page = ComparePage(response)
compare_page.contains_change_summary(1, 3, 0)
compare_page.contains_commits([commit2, commit3, commit4])
anchor = 'a_c-{}-826e8142e6ba'.format(commit1.short_id)
compare_page.contains_file_links_and_anchors([('file1', anchor),])
@pytest.mark.xfail_backends("svn")
def test_compare_cherry_pick_commits_from_top(self, backend):
# repo1:
# commit0:
# commit1:
# repo1-fork- in which we will cherry pick bottom commits
# commit0:
# commit1:
# commit2:
# commit3: x
# commit4: x
# commit5: x
# make repo1, and commit1+commit2
repo1 = backend.create_repo()
# commit something !
commit0 = commit_change(
repo1.repo_name, filename=b'file1', content=b'line1\n',
message='commit1', vcs_type=backend.alias, parent=None,
newfile=True)
commit1 = commit_change(
repo1.repo_name, filename=b'file1', content=b'line1\nline2\n',
message='commit2', vcs_type=backend.alias, parent=commit0)
# fork this repo
backend.create_fork()
# now make commit3-6
commit2 = commit_change(
repo1.repo_name, filename=b'file1', content=b'line1\nline2\nline3\n',
message='commit3', vcs_type=backend.alias, parent=commit1)
commit3 = commit_change(
repo1.repo_name, filename=b'file1',
content=b'line1\nline2\nline3\nline4\n', message='commit4',
vcs_type=backend.alias, parent=commit2)
commit4 = commit_change(
repo1.repo_name, filename=b'file1',
content=b'line1\nline2\nline3\nline4\nline5\n', message='commit5',
vcs_type=backend.alias, parent=commit3)
commit5 = commit_change(
repo1.repo_name, filename=b'file1',
content=b'line1\nline2\nline3\nline4\nline5\nline6\n',
message='commit6', vcs_type=backend.alias, parent=commit4)
response = self.app.get(
route_path('repo_compare',
repo_name=repo1.repo_name,
# parent of commit3, not in source repo2
source_ref_type="rev", source_ref=commit2.raw_id,
target_ref_type="rev", target_ref=commit5.raw_id,
params=dict(merge='1'),))
response.mustcontain('%s@%s' % (repo1.repo_name, commit2.short_id))
response.mustcontain('%s@%s' % (repo1.repo_name, commit5.short_id))
compare_page = ComparePage(response)
compare_page.contains_change_summary(1, 3, 0)
compare_page.contains_commits([commit3, commit4, commit5])
# files
anchor = 'a_c-{}-826e8142e6ba'.format(commit2.short_id)
compare_page.contains_file_links_and_anchors([('file1', anchor),])
@pytest.mark.xfail_backends("svn")
def test_compare_remote_branches(self, backend):
repo1 = backend.repo
repo2 = backend.create_fork()
commit_id1 = repo1.get_commit(commit_idx=3).raw_id
commit_id1_short = repo1.get_commit(commit_idx=3).short_id
commit_id2 = repo1.get_commit(commit_idx=6).raw_id
commit_id2_short = repo1.get_commit(commit_idx=6).short_id
response = self.app.get(
route_path('repo_compare',
repo_name=repo1.repo_name,
source_ref_type="rev", source_ref=commit_id1,
target_ref_type="rev", target_ref=commit_id2,
params=dict(merge='1', target_repo=repo2.repo_name),
))
response.mustcontain('%s@%s' % (repo1.repo_name, commit_id1))
response.mustcontain('%s@%s' % (repo2.repo_name, commit_id2))
compare_page = ComparePage(response)
# outgoing commits between those commits
compare_page.contains_commits(
[repo2.get_commit(commit_idx=x) for x in [4, 5, 6]])
# files
compare_page.contains_file_links_and_anchors([
('vcs/backends/hg.py', 'a_c-{}-9c390eb52cd6'.format(commit_id2_short)),
('vcs/backends/__init__.py', 'a_c-{}-41b41c1f2796'.format(commit_id1_short)),
('vcs/backends/base.py', 'a_c-{}-2f574d260608'.format(commit_id1_short)),
])
@pytest.mark.xfail_backends("svn")
def test_source_repo_new_commits_after_forking_simple_diff(self, backend):
repo1 = backend.create_repo()
r1_name = repo1.repo_name
commit0 = commit_change(
repo=r1_name, filename=b'file1',
content=b'line1', message='commit1', vcs_type=backend.alias,
newfile=True)
assert repo1.scm_instance().commit_ids == [commit0.raw_id]
# fork the repo1
repo2 = backend.create_fork()
assert repo2.scm_instance().commit_ids == [commit0.raw_id]
self.r2_id = repo2.repo_id
r2_name = repo2.repo_name
commit1 = commit_change(
repo=r2_name, filename=b'file1-fork',
content=b'file1-line1-from-fork', message='commit1-fork',
vcs_type=backend.alias, parent=repo2.scm_instance()[-1],
newfile=True)
commit2 = commit_change(
repo=r2_name, filename=b'file2-fork',
content=b'file2-line1-from-fork', message='commit2-fork',
vcs_type=backend.alias, parent=commit1,
newfile=True)
commit_change( # commit 3
repo=r2_name, filename=b'file3-fork',
content=b'file3-line1-from-fork', message='commit3-fork',
vcs_type=backend.alias, parent=commit2, newfile=True)
# compare !
commit_id1 = repo1.scm_instance().DEFAULT_BRANCH_NAME
commit_id2 = repo2.scm_instance().DEFAULT_BRANCH_NAME
response = self.app.get(
route_path('repo_compare',
repo_name=r2_name,
source_ref_type="branch", source_ref=commit_id1,
target_ref_type="branch", target_ref=commit_id2,
params=dict(merge='1', target_repo=r1_name),
))
response.mustcontain('%s@%s' % (r2_name, commit_id1))
response.mustcontain('%s@%s' % (r1_name, commit_id2))
response.mustcontain('No files')
response.mustcontain('No commits in this compare')
commit0 = commit_change(
repo=r1_name, filename=b'file2',
content=b'line1-added-after-fork', message='commit2-parent',
vcs_type=backend.alias, parent=None, newfile=True)
# compare !
response = self.app.get(
route_path('repo_compare',
repo_name=r2_name,
source_ref_type="branch", source_ref=commit_id1,
target_ref_type="branch", target_ref=commit_id2,
params=dict(merge='1', target_repo=r1_name),
))
response.mustcontain('%s@%s' % (r2_name, commit_id1))
response.mustcontain('%s@%s' % (r1_name, commit_id2))
response.mustcontain("""commit2-parent""")
response.mustcontain("""line1-added-after-fork""")
compare_page = ComparePage(response)
compare_page.contains_change_summary(1, 1, 0)
@pytest.mark.xfail_backends("svn")
def test_compare_commits(self, backend, xhr_header):
commit0 = backend.repo.get_commit(commit_idx=0)
commit1 = backend.repo.get_commit(commit_idx=1)
response = self.app.get(
route_path('repo_compare',
repo_name=backend.repo_name,
source_ref_type="rev", source_ref=commit0.raw_id,
target_ref_type="rev", target_ref=commit1.raw_id,
params=dict(merge='1')
),
extra_environ=xhr_header, )
# outgoing commits between those commits
compare_page = ComparePage(response)
compare_page.contains_commits(commits=[commit1])
def test_errors_when_comparing_unknown_source_repo(self, backend):
repo = backend.repo
self.app.get(
route_path('repo_compare',
repo_name='badrepo',
source_ref_type="rev", source_ref='tip',
target_ref_type="rev", target_ref='tip',
params=dict(merge='1', target_repo=repo.repo_name)
),
status=404)
def test_errors_when_comparing_unknown_target_repo(self, backend):
repo = backend.repo
badrepo = 'badrepo'
response = self.app.get(
route_path('repo_compare',
repo_name=repo.repo_name,
source_ref_type="rev", source_ref='tip',
target_ref_type="rev", target_ref='tip',
params=dict(merge='1', target_repo=badrepo),
),
status=302)
redirected = response.follow()
redirected.mustcontain(
'Could not find the target repo: `{}`'.format(badrepo))
def test_compare_not_in_preview_mode(self, backend_stub):
commit0 = backend_stub.repo.get_commit(commit_idx=0)
commit1 = backend_stub.repo.get_commit(commit_idx=1)
response = self.app.get(
route_path('repo_compare',
repo_name=backend_stub.repo_name,
source_ref_type="rev", source_ref=commit0.raw_id,
target_ref_type="rev", target_ref=commit1.raw_id,
))
# outgoing commits between those commits
compare_page = ComparePage(response)
compare_page.swap_is_visible()
compare_page.target_source_are_enabled()
def test_compare_of_fork_with_largefiles(self, backend_hg, settings_util):
orig = backend_hg.create_repo(number_of_commits=1)
fork = backend_hg.create_fork()
settings_util.create_repo_rhodecode_ui(
orig, 'extensions', value='', key='largefiles', active=False)
settings_util.create_repo_rhodecode_ui(
fork, 'extensions', value='', key='largefiles', active=True)
compare_module = ('rhodecode.lib.vcs.backends.hg.repository.'
'MercurialRepository.compare')
with mock.patch(compare_module) as compare_mock:
compare_mock.side_effect = RepositoryRequirementError()
response = self.app.get(
route_path('repo_compare',
repo_name=orig.repo_name,
source_ref_type="rev", source_ref="tip",
target_ref_type="rev", target_ref="tip",
params=dict(merge='1', target_repo=fork.repo_name),
),
status=302)
assert_session_flash(
response,
'Could not compare repos with different large file settings')
@pytest.mark.usefixtures("autologin_user")
class TestCompareControllerSvn(object):
def test_supports_references_with_path(self, app, backend_svn):
repo = backend_svn['svn-simple-layout']
commit_id = repo.get_commit(commit_idx=-1).raw_id
response = app.get(
route_path('repo_compare',
repo_name=repo.repo_name,
source_ref_type="tag",
source_ref="%s@%s" % ('tags/v0.1', commit_id),
target_ref_type="tag",
target_ref="%s@%s" % ('tags/v0.2', commit_id),
params=dict(merge='1'),
),
status=200)
# Expecting no commits, since both paths are at the same revision
response.mustcontain('No commits in this compare')
# Should find only one file changed when comparing those two tags
response.mustcontain('example.py')
compare_page = ComparePage(response)
compare_page.contains_change_summary(1, 5, 1)
def test_shows_commits_if_different_ids(self, app, backend_svn):
repo = backend_svn['svn-simple-layout']
source_id = repo.get_commit(commit_idx=-6).raw_id
target_id = repo.get_commit(commit_idx=-1).raw_id
response = app.get(
route_path('repo_compare',
repo_name=repo.repo_name,
source_ref_type="tag",
source_ref="%s@%s" % ('tags/v0.1', source_id),
target_ref_type="tag",
target_ref="%s@%s" % ('tags/v0.2', target_id),
params=dict(merge='1')
),
status=200)
# It should show commits
assert 'No commits in this compare' not in response.text
# Should find only one file changed when comparing those two tags
response.mustcontain('example.py')
compare_page = ComparePage(response)
compare_page.contains_change_summary(1, 5, 1)
class ComparePage(AssertResponse):
"""
Abstracts the page template from the tests
"""
def contains_file_links_and_anchors(self, files):
doc = lxml.html.fromstring(self.response.body)
for filename, file_id in files:
self.contains_one_anchor(file_id)
diffblock = doc.cssselect('[data-f-path="%s"]' % filename)
assert len(diffblock) == 2
for lnk in diffblock[0].cssselect('a'):
if 'permalink' in lnk.text:
assert '#{}'.format(file_id) in lnk.attrib['href']
break
else:
pytest.fail('Unable to find permalink')
def contains_change_summary(self, files_changed, inserted, deleted):
template = (
'{files_changed} file{plural} changed: '
'<span class="op-added">{inserted} inserted</span>, <span class="op-deleted">{deleted} deleted</span>')
self.response.mustcontain(template.format(
files_changed=files_changed,
plural="s" if files_changed > 1 else "",
inserted=inserted,
deleted=deleted))
def contains_commits(self, commits, ancestors=None):
response = self.response
for commit in commits:
# Expecting to see the commit message in an element which
# has the ID "c-{commit.raw_id}"
self.element_contains('#c-' + commit.raw_id, commit.message)
self.contains_one_link(
'r%s:%s' % (commit.idx, commit.short_id),
self._commit_url(commit))
if ancestors:
response.mustcontain('Ancestor')
for ancestor in ancestors:
self.contains_one_link(
ancestor.short_id, self._commit_url(ancestor))
def _commit_url(self, commit):
return '/%s/changeset/%s' % (commit.repository.name, commit.raw_id)
def swap_is_hidden(self):
assert '<a id="btn-swap"' not in self.response.text
def swap_is_visible(self):
assert '<a id="btn-swap"' in self.response.text
def target_source_are_disabled(self):
response = self.response
response.mustcontain("var enable_fields = false;")
response.mustcontain('.select2("enable", enable_fields)')
def target_source_are_enabled(self):
response = self.response
response.mustcontain("var enable_fields = true;")