tests: fixed merge tests cases
This commit is contained in:
RhodeCode Admin
parent
02b7fd344c
commit
dd312df5a1
1 changed files with 243 additions and 155 deletions
|
|
@ -1,4 +1,3 @@
|
|||
|
||||
# Copyright (C) 2010-2023 RhodeCode GmbH
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
|
|
@ -25,7 +24,12 @@ import pytest
|
|||
|
||||
from rhodecode.lib.vcs import backends
|
||||
from rhodecode.lib.vcs.backends.base import (
|
||||
Config, BaseInMemoryCommit, Reference, MergeResponse, MergeFailureReason)
|
||||
Config,
|
||||
BaseInMemoryCommit,
|
||||
Reference,
|
||||
MergeResponse,
|
||||
MergeFailureReason,
|
||||
)
|
||||
from rhodecode.lib.vcs.exceptions import VCSError, RepositoryError
|
||||
from rhodecode.lib.vcs.nodes import FileNode
|
||||
from rhodecode.tests.vcs.conftest import BackendTestMixin
|
||||
|
|
@ -37,20 +41,20 @@ class TestRepositoryBase(BackendTestMixin):
|
|||
recreate_repo_per_test = False
|
||||
|
||||
def test_init_accepts_unicode_path(self, tmpdir):
|
||||
path = str(tmpdir.join(u'unicode ä'))
|
||||
path = str(tmpdir.join("unicode ä"))
|
||||
self.Backend(path, create=True)
|
||||
|
||||
def test_init_accepts_str_path(self, tmpdir):
|
||||
path = str(tmpdir.join('str ä'))
|
||||
path = str(tmpdir.join("str ä"))
|
||||
self.Backend(path, create=True)
|
||||
|
||||
def test_init_fails_if_path_does_not_exist(self, tmpdir):
|
||||
path = str(tmpdir.join('i-do-not-exist'))
|
||||
path = str(tmpdir.join("i-do-not-exist"))
|
||||
with pytest.raises(VCSError):
|
||||
self.Backend(path)
|
||||
|
||||
def test_init_fails_if_path_is_not_a_valid_repository(self, tmpdir):
|
||||
path = str(tmpdir.mkdir(u'unicode ä'))
|
||||
path = str(tmpdir.mkdir("unicode ä"))
|
||||
with pytest.raises(VCSError):
|
||||
self.Backend(path)
|
||||
|
||||
|
|
@ -58,7 +62,7 @@ class TestRepositoryBase(BackendTestMixin):
|
|||
assert self.repo.commit_ids
|
||||
|
||||
def test_name(self):
|
||||
assert self.repo.name.startswith('vcs-test')
|
||||
assert self.repo.name.startswith("vcs-test")
|
||||
|
||||
@pytest.mark.backends("hg", "git")
|
||||
def test_has_default_branch_name(self):
|
||||
|
|
@ -75,6 +79,7 @@ class TestRepositoryBase(BackendTestMixin):
|
|||
def test_empty_changeset_is_deprecated(self):
|
||||
def get_empty_changeset(repo):
|
||||
return repo.EMPTY_CHANGESET
|
||||
|
||||
pytest.deprecated_call(get_empty_changeset, self.repo)
|
||||
|
||||
def test_bookmarks(self):
|
||||
|
|
@ -87,9 +92,9 @@ class TestRepositoryBase(BackendTestMixin):
|
|||
def test_check_url_on_remote_url(self):
|
||||
config = Config()
|
||||
url = {
|
||||
'hg': 'https://code.rhodecode.com/rhodecode-vcsserver',
|
||||
'svn': 'https://code.rhodecode.com/svn-doc',
|
||||
'git': 'https://code.rhodecode.com/appenlight',
|
||||
"hg": "https://code.rhodecode.com/rhodecode-vcsserver",
|
||||
"svn": "https://code.rhodecode.com/svn-doc",
|
||||
"git": "https://code.rhodecode.com/appenlight",
|
||||
}[self.repo.alias]
|
||||
|
||||
assert self.Backend.check_url(url, config)
|
||||
|
|
@ -110,7 +115,8 @@ class TestRepositoryBase(BackendTestMixin):
|
|||
|
||||
def test_last_change(self, local_dt_to_utc):
|
||||
assert self.repo.last_change >= local_dt_to_utc(
|
||||
datetime.datetime(2010, 1, 1, 21, 0))
|
||||
datetime.datetime(2010, 1, 1, 21, 0)
|
||||
)
|
||||
|
||||
def test_last_change_in_empty_repository(self, vcsbackend, local_dt_to_utc):
|
||||
delta = datetime.timedelta(seconds=1)
|
||||
|
|
@ -126,13 +132,15 @@ class TestRepositoryBase(BackendTestMixin):
|
|||
|
||||
def test_repo_equality_broken_object(self):
|
||||
import copy
|
||||
|
||||
_repo = copy.copy(self.repo)
|
||||
delattr(_repo, 'path')
|
||||
delattr(_repo, "path")
|
||||
assert self.repo != _repo
|
||||
|
||||
def test_repo_equality_other_object(self):
|
||||
class dummy(object):
|
||||
path = self.repo.path
|
||||
|
||||
assert self.repo != dummy()
|
||||
|
||||
def test_get_commit_is_implemented(self):
|
||||
|
|
@ -154,7 +162,7 @@ class TestRepositoryBase(BackendTestMixin):
|
|||
|
||||
@pytest.mark.backends("hg")
|
||||
def test__get_url_unicode(self):
|
||||
url = u'/home/repos/malmö'
|
||||
url = "/home/repos/malmö"
|
||||
assert self.repo._get_url(url)
|
||||
|
||||
|
||||
|
|
@ -165,6 +173,7 @@ class TestDeprecatedRepositoryAPI(BackendTestMixin):
|
|||
def test_revisions_is_deprecated(self):
|
||||
def get_revisions(repo):
|
||||
return repo.revisions
|
||||
|
||||
pytest.deprecated_call(get_revisions, self.repo)
|
||||
|
||||
def test_get_changeset_is_deprecated(self):
|
||||
|
|
@ -176,63 +185,67 @@ class TestDeprecatedRepositoryAPI(BackendTestMixin):
|
|||
def test_in_memory_changeset_is_deprecated(self):
|
||||
def get_imc(repo):
|
||||
return repo.in_memory_changeset
|
||||
|
||||
pytest.deprecated_call(get_imc, self.repo)
|
||||
|
||||
|
||||
# TODO: these tests are incomplete, must check the resulting compare result for
|
||||
# correcteness
|
||||
class TestRepositoryCompare:
|
||||
|
||||
@pytest.mark.parametrize('merge', [True, False])
|
||||
@pytest.mark.parametrize("merge", [True, False])
|
||||
def test_compare_commits_of_same_repository(self, vcsbackend, merge):
|
||||
target_repo = vcsbackend.create_repo(number_of_commits=5)
|
||||
target_repo.compare(
|
||||
target_repo[1].raw_id, target_repo[3].raw_id, target_repo,
|
||||
merge=merge)
|
||||
target_repo[1].raw_id, target_repo[3].raw_id, target_repo, merge=merge
|
||||
)
|
||||
|
||||
@pytest.mark.xfail_backends('svn')
|
||||
@pytest.mark.parametrize('merge', [True, False])
|
||||
@pytest.mark.xfail_backends("svn")
|
||||
@pytest.mark.parametrize("merge", [True, False])
|
||||
def test_compare_cloned_repositories(self, vcsbackend, merge):
|
||||
target_repo = vcsbackend.create_repo(number_of_commits=5)
|
||||
source_repo = vcsbackend.clone_repo(target_repo)
|
||||
assert target_repo != source_repo
|
||||
|
||||
vcsbackend.add_file(source_repo, b'newfile', b'somecontent')
|
||||
vcsbackend.add_file(source_repo, b"newfile", b"somecontent")
|
||||
source_commit = source_repo.get_commit()
|
||||
|
||||
target_repo.compare(
|
||||
target_repo[1].raw_id, source_repo[3].raw_id, source_repo,
|
||||
merge=merge)
|
||||
target_repo[1].raw_id, source_repo[3].raw_id, source_repo, merge=merge
|
||||
)
|
||||
|
||||
@pytest.mark.xfail_backends('svn')
|
||||
@pytest.mark.parametrize('merge', [True, False])
|
||||
@pytest.mark.xfail_backends("svn")
|
||||
@pytest.mark.parametrize("merge", [True, False])
|
||||
def test_compare_unrelated_repositories(self, vcsbackend, merge):
|
||||
orig = vcsbackend.create_repo(number_of_commits=5)
|
||||
unrelated = vcsbackend.create_repo(number_of_commits=5)
|
||||
assert orig != unrelated
|
||||
|
||||
orig.compare(
|
||||
orig[1].raw_id, unrelated[3].raw_id, unrelated, merge=merge)
|
||||
orig.compare(orig[1].raw_id, unrelated[3].raw_id, unrelated, merge=merge)
|
||||
|
||||
|
||||
class TestRepositoryGetCommonAncestor:
|
||||
|
||||
def test_get_common_ancestor_from_same_repo_existing(self, vcsbackend):
|
||||
target_repo = vcsbackend.create_repo(number_of_commits=5)
|
||||
|
||||
expected_ancestor = target_repo[2].raw_id
|
||||
|
||||
assert target_repo.get_common_ancestor(
|
||||
commit_id1=target_repo[2].raw_id,
|
||||
commit_id2=target_repo[4].raw_id,
|
||||
repo2=target_repo
|
||||
) == expected_ancestor
|
||||
assert (
|
||||
target_repo.get_common_ancestor(
|
||||
commit_id1=target_repo[2].raw_id,
|
||||
commit_id2=target_repo[4].raw_id,
|
||||
repo2=target_repo,
|
||||
)
|
||||
== expected_ancestor
|
||||
)
|
||||
|
||||
assert target_repo.get_common_ancestor(
|
||||
commit_id1=target_repo[4].raw_id,
|
||||
commit_id2=target_repo[2].raw_id,
|
||||
repo2=target_repo
|
||||
) == expected_ancestor
|
||||
assert (
|
||||
target_repo.get_common_ancestor(
|
||||
commit_id1=target_repo[4].raw_id,
|
||||
commit_id2=target_repo[2].raw_id,
|
||||
repo2=target_repo,
|
||||
)
|
||||
== expected_ancestor
|
||||
)
|
||||
|
||||
@pytest.mark.xfail_backends("svn")
|
||||
def test_get_common_ancestor_from_cloned_repo_existing(self, vcsbackend):
|
||||
|
|
@ -240,22 +253,28 @@ class TestRepositoryGetCommonAncestor:
|
|||
source_repo = vcsbackend.clone_repo(target_repo)
|
||||
assert target_repo != source_repo
|
||||
|
||||
vcsbackend.add_file(source_repo, b'newfile', b'somecontent')
|
||||
vcsbackend.add_file(source_repo, b"newfile", b"somecontent")
|
||||
source_commit = source_repo.get_commit()
|
||||
|
||||
expected_ancestor = target_repo[4].raw_id
|
||||
|
||||
assert target_repo.get_common_ancestor(
|
||||
commit_id1=target_repo[4].raw_id,
|
||||
commit_id2=source_commit.raw_id,
|
||||
repo2=source_repo
|
||||
) == expected_ancestor
|
||||
assert (
|
||||
target_repo.get_common_ancestor(
|
||||
commit_id1=target_repo[4].raw_id,
|
||||
commit_id2=source_commit.raw_id,
|
||||
repo2=source_repo,
|
||||
)
|
||||
== expected_ancestor
|
||||
)
|
||||
|
||||
assert target_repo.get_common_ancestor(
|
||||
commit_id1=source_commit.raw_id,
|
||||
commit_id2=target_repo[4].raw_id,
|
||||
repo2=target_repo
|
||||
) == expected_ancestor
|
||||
assert (
|
||||
target_repo.get_common_ancestor(
|
||||
commit_id1=source_commit.raw_id,
|
||||
commit_id2=target_repo[4].raw_id,
|
||||
repo2=target_repo,
|
||||
)
|
||||
== expected_ancestor
|
||||
)
|
||||
|
||||
@pytest.mark.xfail_backends("svn")
|
||||
def test_get_common_ancestor_from_unrelated_repo_missing(self, vcsbackend):
|
||||
|
|
@ -263,17 +282,23 @@ class TestRepositoryGetCommonAncestor:
|
|||
unrelated = vcsbackend.create_repo(number_of_commits=5)
|
||||
assert original != unrelated
|
||||
|
||||
assert original.get_common_ancestor(
|
||||
commit_id1=original[0].raw_id,
|
||||
commit_id2=unrelated[0].raw_id,
|
||||
repo2=unrelated
|
||||
) is None
|
||||
assert (
|
||||
original.get_common_ancestor(
|
||||
commit_id1=original[0].raw_id,
|
||||
commit_id2=unrelated[0].raw_id,
|
||||
repo2=unrelated,
|
||||
)
|
||||
is None
|
||||
)
|
||||
|
||||
assert original.get_common_ancestor(
|
||||
commit_id1=original[-1].raw_id,
|
||||
commit_id2=unrelated[-1].raw_id,
|
||||
repo2=unrelated
|
||||
) is None
|
||||
assert (
|
||||
original.get_common_ancestor(
|
||||
commit_id1=original[-1].raw_id,
|
||||
commit_id2=unrelated[-1].raw_id,
|
||||
repo2=unrelated,
|
||||
)
|
||||
is None
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.backends("git", "hg")
|
||||
|
|
@ -281,51 +306,57 @@ class TestRepositoryMerge(object):
|
|||
def prepare_for_success(self, vcsbackend):
|
||||
self.target_repo = vcsbackend.create_repo(number_of_commits=1)
|
||||
self.source_repo = vcsbackend.clone_repo(self.target_repo)
|
||||
vcsbackend.add_file(self.target_repo, b'README_MERGE1', b'Version 1')
|
||||
vcsbackend.add_file(self.source_repo, b'README_MERGE2', b'Version 2')
|
||||
vcsbackend.add_file(self.target_repo, b"README_MERGE1", b"Version 1")
|
||||
vcsbackend.add_file(self.source_repo, b"README_MERGE2", b"Version 2")
|
||||
imc = self.source_repo.in_memory_commit
|
||||
imc.add(FileNode(b'file_x', content=self.source_repo.name))
|
||||
imc.add(FileNode(b"file_x", content=self.source_repo.name))
|
||||
imc.commit(
|
||||
message=u'Automatic commit from repo merge test',
|
||||
author=u'Automatic <automatic@rhodecode.com>')
|
||||
message="Automatic commit from repo merge test",
|
||||
author="Automatic <automatic@rhodecode.com>",
|
||||
)
|
||||
self.target_commit = self.target_repo.get_commit()
|
||||
self.source_commit = self.source_repo.get_commit()
|
||||
# This only works for Git and Mercurial
|
||||
default_branch = self.target_repo.DEFAULT_BRANCH_NAME
|
||||
self.target_ref = Reference('branch', default_branch, self.target_commit.raw_id)
|
||||
self.source_ref = Reference('branch', default_branch, self.source_commit.raw_id)
|
||||
self.workspace_id = 'test-merge-{}'.format(vcsbackend.alias)
|
||||
self.target_ref = Reference("branch", default_branch, self.target_commit.raw_id)
|
||||
self.source_ref = Reference("branch", default_branch, self.source_commit.raw_id)
|
||||
self.workspace_id = "test-merge-{}".format(vcsbackend.alias)
|
||||
self.repo_id = repo_id_generator(self.target_repo.path)
|
||||
|
||||
def prepare_for_conflict(self, vcsbackend):
|
||||
self.target_repo = vcsbackend.create_repo(number_of_commits=1)
|
||||
self.source_repo = vcsbackend.clone_repo(self.target_repo)
|
||||
vcsbackend.add_file(self.target_repo, b'README_MERGE', b'Version 1')
|
||||
vcsbackend.add_file(self.source_repo, b'README_MERGE', b'Version 2')
|
||||
vcsbackend.add_file(self.target_repo, b"README_MERGE", b"Version 1")
|
||||
vcsbackend.add_file(self.source_repo, b"README_MERGE", b"Version 2")
|
||||
self.target_commit = self.target_repo.get_commit()
|
||||
self.source_commit = self.source_repo.get_commit()
|
||||
# This only works for Git and Mercurial
|
||||
default_branch = self.target_repo.DEFAULT_BRANCH_NAME
|
||||
self.target_ref = Reference('branch', default_branch, self.target_commit.raw_id)
|
||||
self.source_ref = Reference('branch', default_branch, self.source_commit.raw_id)
|
||||
self.workspace_id = 'test-merge-{}'.format(vcsbackend.alias)
|
||||
self.target_ref = Reference("branch", default_branch, self.target_commit.raw_id)
|
||||
self.source_ref = Reference("branch", default_branch, self.source_commit.raw_id)
|
||||
self.workspace_id = "test-merge-{}".format(vcsbackend.alias)
|
||||
self.repo_id = repo_id_generator(self.target_repo.path)
|
||||
|
||||
def test_merge_success(self, vcsbackend):
|
||||
self.prepare_for_success(vcsbackend)
|
||||
|
||||
merge_response = self.target_repo.merge(
|
||||
self.repo_id, self.workspace_id, self.target_ref, self.source_repo,
|
||||
self.repo_id,
|
||||
self.workspace_id,
|
||||
self.target_ref,
|
||||
self.source_repo,
|
||||
self.source_ref,
|
||||
'test user', 'test@rhodecode.com', 'merge message 1',
|
||||
dry_run=False)
|
||||
"test user",
|
||||
"test@rhodecode.com",
|
||||
"merge message 1",
|
||||
dry_run=False,
|
||||
)
|
||||
expected_merge_response = MergeResponse(
|
||||
True, True, merge_response.merge_ref,
|
||||
MergeFailureReason.NONE)
|
||||
True, True, merge_response.merge_ref, MergeFailureReason.NONE
|
||||
)
|
||||
assert merge_response == expected_merge_response
|
||||
|
||||
target_repo = backends.get_backend(vcsbackend.alias)(
|
||||
self.target_repo.path)
|
||||
target_repo = backends.get_backend(vcsbackend.alias)(self.target_repo.path)
|
||||
target_commits = list(target_repo.get_commits())
|
||||
commit_ids = [c.raw_id for c in target_commits[:-1]]
|
||||
assert self.source_ref.commit_id in commit_ids
|
||||
|
|
@ -333,49 +364,67 @@ class TestRepositoryMerge(object):
|
|||
|
||||
merge_commit = target_commits[-1]
|
||||
assert merge_commit.raw_id == merge_response.merge_ref.commit_id
|
||||
assert merge_commit.message.strip() == 'merge message 1'
|
||||
assert merge_commit.author == 'test user <test@rhodecode.com>'
|
||||
assert merge_commit.message.strip() == "merge message 1"
|
||||
assert merge_commit.author == "test user <test@rhodecode.com>"
|
||||
|
||||
# We call it twice so to make sure we can handle updates
|
||||
target_ref = Reference(
|
||||
self.target_ref.type, self.target_ref.name,
|
||||
merge_response.merge_ref.commit_id)
|
||||
self.target_ref.type,
|
||||
self.target_ref.name,
|
||||
merge_response.merge_ref.commit_id,
|
||||
)
|
||||
|
||||
merge_response = target_repo.merge(
|
||||
self.repo_id, self.workspace_id, target_ref, self.source_repo, self.source_ref,
|
||||
'test user', 'test@rhodecode.com', 'merge message 2',
|
||||
dry_run=False)
|
||||
self.repo_id,
|
||||
self.workspace_id,
|
||||
target_ref,
|
||||
self.source_repo,
|
||||
self.source_ref,
|
||||
"test user",
|
||||
"test@rhodecode.com",
|
||||
"merge message 2",
|
||||
dry_run=False,
|
||||
)
|
||||
expected_merge_response = MergeResponse(
|
||||
True, True, merge_response.merge_ref,
|
||||
MergeFailureReason.NONE)
|
||||
True, True, merge_response.merge_ref, MergeFailureReason.NONE
|
||||
)
|
||||
assert merge_response == expected_merge_response
|
||||
|
||||
target_repo = backends.get_backend(
|
||||
vcsbackend.alias)(self.target_repo.path)
|
||||
merge_commit = target_repo.get_commit(
|
||||
merge_response.merge_ref.commit_id)
|
||||
assert merge_commit.message.strip() == 'merge message 1'
|
||||
assert merge_commit.author == 'test user <test@rhodecode.com>'
|
||||
target_repo = backends.get_backend(vcsbackend.alias)(self.target_repo.path)
|
||||
merge_commit = target_repo.get_commit(merge_response.merge_ref.commit_id)
|
||||
assert merge_commit.message.strip() == "merge message 1"
|
||||
assert merge_commit.author == "test user <test@rhodecode.com>"
|
||||
|
||||
def test_merge_success_dry_run(self, vcsbackend):
|
||||
self.prepare_for_success(vcsbackend)
|
||||
|
||||
merge_response = self.target_repo.merge(
|
||||
self.repo_id, self.workspace_id, self.target_ref, self.source_repo,
|
||||
self.source_ref, dry_run=True)
|
||||
self.repo_id,
|
||||
self.workspace_id,
|
||||
self.target_ref,
|
||||
self.source_repo,
|
||||
self.source_ref,
|
||||
dry_run=True,
|
||||
)
|
||||
|
||||
# We call it twice so to make sure we can handle updates
|
||||
merge_response_update = self.target_repo.merge(
|
||||
self.repo_id, self.workspace_id, self.target_ref, self.source_repo,
|
||||
self.source_ref, dry_run=True)
|
||||
self.repo_id,
|
||||
self.workspace_id,
|
||||
self.target_ref,
|
||||
self.source_repo,
|
||||
self.source_ref,
|
||||
dry_run=True,
|
||||
)
|
||||
|
||||
assert merge_response.merge_ref
|
||||
assert merge_response_update.merge_ref
|
||||
|
||||
# Multiple merges may differ in their commit id. Therefore, we set the
|
||||
# commit id to `None` before comparing the merge responses.
|
||||
new_merge_ref = merge_response.merge_ref.commit_id = None
|
||||
merge_response.merge_ref = new_merge_ref
|
||||
merge_response.merge_ref.commit_id = 'abcdeabcde'
|
||||
|
||||
new_update_merge_ref = merge_response_update.merge_ref.commit_id = None
|
||||
merge_response_update.merge_ref = new_update_merge_ref
|
||||
merge_response_update.merge_ref.commit_id = 'abcdeabcde'
|
||||
|
||||
assert merge_response == merge_response_update
|
||||
assert merge_response.possible is True
|
||||
|
|
@ -383,36 +432,59 @@ class TestRepositoryMerge(object):
|
|||
assert merge_response.merge_ref
|
||||
assert merge_response.failure_reason is MergeFailureReason.NONE
|
||||
|
||||
@pytest.mark.parametrize('dry_run', [True, False])
|
||||
@pytest.mark.parametrize("dry_run", [True, False])
|
||||
def test_merge_conflict(self, vcsbackend, dry_run):
|
||||
self.prepare_for_conflict(vcsbackend)
|
||||
|
||||
expected_merge_response = MergeResponse(
|
||||
False, False, None, MergeFailureReason.MERGE_FAILED)
|
||||
False, False, None, MergeFailureReason.MERGE_FAILED
|
||||
)
|
||||
|
||||
merge_response = self.target_repo.merge(
|
||||
self.repo_id, self.workspace_id, self.target_ref,
|
||||
self.source_repo, self.source_ref,
|
||||
'test_user', 'test@rhodecode.com', 'test message', dry_run=dry_run)
|
||||
self.repo_id,
|
||||
self.workspace_id,
|
||||
self.target_ref,
|
||||
self.source_repo,
|
||||
self.source_ref,
|
||||
"test_user",
|
||||
"test@rhodecode.com",
|
||||
"test message",
|
||||
dry_run=dry_run,
|
||||
)
|
||||
assert merge_response == expected_merge_response
|
||||
|
||||
# We call it twice so to make sure we can handle updates
|
||||
merge_response = self.target_repo.merge(
|
||||
self.repo_id, self.workspace_id, self.target_ref, self.source_repo,
|
||||
self.repo_id,
|
||||
self.workspace_id,
|
||||
self.target_ref,
|
||||
self.source_repo,
|
||||
self.source_ref,
|
||||
'test_user', 'test@rhodecode.com', 'test message', dry_run=dry_run)
|
||||
"test_user",
|
||||
"test@rhodecode.com",
|
||||
"test message",
|
||||
dry_run=dry_run,
|
||||
)
|
||||
assert merge_response == expected_merge_response
|
||||
|
||||
def test_merge_target_is_not_head(self, vcsbackend):
|
||||
self.prepare_for_success(vcsbackend)
|
||||
target_ref = Reference(
|
||||
self.target_ref.type, self.target_ref.name, '0' * 40)
|
||||
target_ref = Reference(self.target_ref.type, self.target_ref.name, "0" * 40)
|
||||
expected_merge_response = MergeResponse(
|
||||
False, False, None, MergeFailureReason.TARGET_IS_NOT_HEAD,
|
||||
metadata={'target_ref': target_ref})
|
||||
False,
|
||||
False,
|
||||
None,
|
||||
MergeFailureReason.TARGET_IS_NOT_HEAD,
|
||||
metadata={"target_ref": target_ref},
|
||||
)
|
||||
merge_response = self.target_repo.merge(
|
||||
self.repo_id, self.workspace_id, target_ref, self.source_repo,
|
||||
self.source_ref, dry_run=True)
|
||||
self.repo_id,
|
||||
self.workspace_id,
|
||||
target_ref,
|
||||
self.source_repo,
|
||||
self.source_ref,
|
||||
dry_run=True,
|
||||
)
|
||||
|
||||
assert merge_response == expected_merge_response
|
||||
|
||||
|
|
@ -420,59 +492,76 @@ class TestRepositoryMerge(object):
|
|||
self.prepare_for_success(vcsbackend)
|
||||
|
||||
source_ref = Reference(
|
||||
self.source_ref.type, 'not_existing', self.source_ref.commit_id)
|
||||
self.source_ref.type, "not_existing", self.source_ref.commit_id
|
||||
)
|
||||
expected_merge_response = MergeResponse(
|
||||
False, False, None, MergeFailureReason.MISSING_SOURCE_REF,
|
||||
metadata={'source_ref': source_ref})
|
||||
False,
|
||||
False,
|
||||
None,
|
||||
MergeFailureReason.MISSING_SOURCE_REF,
|
||||
metadata={"source_ref": source_ref},
|
||||
)
|
||||
|
||||
merge_response = self.target_repo.merge(
|
||||
self.repo_id, self.workspace_id, self.target_ref,
|
||||
self.source_repo, source_ref,
|
||||
dry_run=True)
|
||||
self.repo_id,
|
||||
self.workspace_id,
|
||||
self.target_ref,
|
||||
self.source_repo,
|
||||
source_ref,
|
||||
dry_run=True,
|
||||
)
|
||||
|
||||
assert merge_response == expected_merge_response
|
||||
|
||||
def test_merge_raises_exception(self, vcsbackend):
|
||||
self.prepare_for_success(vcsbackend)
|
||||
expected_merge_response = MergeResponse(
|
||||
False, False, None, MergeFailureReason.UNKNOWN,
|
||||
metadata={'exception': 'ErrorForTest'})
|
||||
False,
|
||||
False,
|
||||
None,
|
||||
MergeFailureReason.UNKNOWN,
|
||||
metadata={"exception": "ErrorForTest"},
|
||||
)
|
||||
|
||||
with mock.patch.object(self.target_repo, '_merge_repo',
|
||||
side_effect=RepositoryError()):
|
||||
with mock.patch.object(
|
||||
self.target_repo, "_merge_repo", side_effect=RepositoryError()
|
||||
):
|
||||
merge_response = self.target_repo.merge(
|
||||
self.repo_id, self.workspace_id, self.target_ref,
|
||||
self.source_repo, self.source_ref,
|
||||
dry_run=True)
|
||||
self.repo_id,
|
||||
self.workspace_id,
|
||||
self.target_ref,
|
||||
self.source_repo,
|
||||
self.source_ref,
|
||||
dry_run=True,
|
||||
)
|
||||
|
||||
assert merge_response == expected_merge_response
|
||||
|
||||
def test_merge_invalid_user_name(self, vcsbackend):
|
||||
repo = vcsbackend.create_repo(number_of_commits=1)
|
||||
ref = Reference('branch', 'master', 'not_used')
|
||||
workspace_id = 'test-errors-in-merge'
|
||||
ref = Reference("branch", "master", "not_used")
|
||||
workspace_id = "test-errors-in-merge"
|
||||
repo_id = repo_id_generator(workspace_id)
|
||||
with pytest.raises(ValueError):
|
||||
repo.merge(repo_id, workspace_id, ref, self, ref)
|
||||
repo.merge(repo_id, workspace_id, ref, self, ref)
|
||||
|
||||
def test_merge_invalid_user_email(self, vcsbackend):
|
||||
repo = vcsbackend.create_repo(number_of_commits=1)
|
||||
ref = Reference('branch', 'master', 'not_used')
|
||||
workspace_id = 'test-errors-in-merge'
|
||||
ref = Reference("branch", "master", "not_used")
|
||||
workspace_id = "test-errors-in-merge"
|
||||
repo_id = repo_id_generator(workspace_id)
|
||||
with pytest.raises(ValueError):
|
||||
repo.merge(
|
||||
repo_id, workspace_id, ref, self, ref, 'user name')
|
||||
repo.merge(repo_id, workspace_id, ref, self, ref, "user name")
|
||||
|
||||
def test_merge_invalid_message(self, vcsbackend):
|
||||
repo = vcsbackend.create_repo(number_of_commits=1)
|
||||
ref = Reference('branch', 'master', 'not_used')
|
||||
workspace_id = 'test-errors-in-merge'
|
||||
ref = Reference("branch", "master", "not_used")
|
||||
workspace_id = "test-errors-in-merge"
|
||||
repo_id = repo_id_generator(workspace_id)
|
||||
with pytest.raises(ValueError):
|
||||
repo.merge(
|
||||
repo_id, workspace_id, ref, self, ref,
|
||||
'user name', 'user@email.com')
|
||||
repo_id, workspace_id, ref, self, ref, "user name", "user@email.com"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vcs_repository_support")
|
||||
|
|
@ -483,24 +572,24 @@ class TestRepositoryStrip(BackendTestMixin):
|
|||
def _get_commits(cls):
|
||||
commits = [
|
||||
{
|
||||
'message': 'Initial commit',
|
||||
'author': 'Joe Doe <joe.doe@example.com>',
|
||||
'date': datetime.datetime(2010, 1, 1, 20),
|
||||
'branch': 'master',
|
||||
'added': [
|
||||
FileNode(b'foobar', content='foobar'),
|
||||
FileNode(b'foobar2', content='foobar2'),
|
||||
"message": "Initial commit",
|
||||
"author": "Joe Doe <joe.doe@example.com>",
|
||||
"date": datetime.datetime(2010, 1, 1, 20),
|
||||
"branch": "master",
|
||||
"added": [
|
||||
FileNode(b"foobar", content="foobar"),
|
||||
FileNode(b"foobar2", content="foobar2"),
|
||||
],
|
||||
},
|
||||
]
|
||||
for x in range(10):
|
||||
commit_data = {
|
||||
'message': 'Changed foobar - commit%s' % x,
|
||||
'author': 'Jane Doe <jane.doe@example.com>',
|
||||
'date': datetime.datetime(2010, 1, 1, 21, x),
|
||||
'branch': 'master',
|
||||
'changed': [
|
||||
FileNode(b'foobar', 'FOOBAR - %s' % x),
|
||||
"message": "Changed foobar - commit%s" % x,
|
||||
"author": "Jane Doe <jane.doe@example.com>",
|
||||
"date": datetime.datetime(2010, 1, 1, 21, x),
|
||||
"branch": "master",
|
||||
"changed": [
|
||||
FileNode(b"foobar", "FOOBAR - %s" % x),
|
||||
],
|
||||
}
|
||||
commits.append(commit_data)
|
||||
|
|
@ -527,9 +616,8 @@ class TestRepositoryStrip(BackendTestMixin):
|
|||
assert tip.idx == 4
|
||||
|
||||
|
||||
@pytest.mark.backends('hg', 'git')
|
||||
@pytest.mark.backends("hg", "git")
|
||||
class TestRepositoryPull(object):
|
||||
|
||||
def test_pull(self, vcsbackend):
|
||||
source_repo = vcsbackend.repo
|
||||
target_repo = vcsbackend.create_repo()
|
||||
|
|
@ -550,8 +638,8 @@ class TestRepositoryPull(object):
|
|||
target_repo = vcsbackend.create_repo()
|
||||
|
||||
second_commit = source_repo[1].raw_id
|
||||
if vcsbackend.alias == 'git':
|
||||
second_commit_ref = 'refs/test-refs/a'
|
||||
if vcsbackend.alias == "git":
|
||||
second_commit_ref = "refs/test-refs/a"
|
||||
source_repo.set_refs(second_commit_ref, second_commit)
|
||||
|
||||
target_repo.pull(source_repo.path, commit_ids=[second_commit])
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue