Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_mergeable_request_pull_conflicts(self, send_email):
""" Test the mergeable_request_pull endpoint when the changes cannot
be merged due to conflicts.
"""
send_email.return_value = True
# Create a git repo to play with
origgitrepo = os.path.join(self.path, "repos", "test.git")
self.assertFalse(os.path.exists(origgitrepo))
os.makedirs(origgitrepo)
orig_repo = pygit2.init_repository(origgitrepo, bare=True)
os.makedirs(os.path.join(self.path, "repos_tmp"))
gitrepo = os.path.join(self.path, "repos_tmp", "test.git")
repo = pygit2.clone_repository(origgitrepo, gitrepo)
# Create a file in that git repo
with open(os.path.join(gitrepo, "sources"), "w") as stream:
stream.write("foo\n bar")
repo.index.add("sources")
repo.index.write()
# Commits the files added
tree = repo.index.write_tree()
author = pygit2.Signature("Alice Author", "alice@authors.tld")
committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld")
repo.create_commit(
"refs/heads/master", # the name of the reference to update
author,
committer,
"Add sources file for testing",
def _clone_and_top_commits(folder, branch, branch_ref=False):
""" Clone the repository, checkout the specified branch and return
the top commit of that branch if there is one.
Returns the repo, the path to the clone and the top commit(s) in a tuple
or the repo, the path to the clone and the reference to the branch
object if branch_ref is True.
"""
if not os.path.exists(folder):
os.makedirs(folder)
brepo = pygit2.init_repository(folder, bare=True)
newfolder = tempfile.mkdtemp(prefix="pagure-tests")
repo = pygit2.clone_repository(folder, newfolder)
branch_ref_obj = None
if "origin/%s" % branch in repo.listall_branches(pygit2.GIT_BRANCH_ALL):
branch_ref_obj = pagure.lib.git.get_branch_ref(repo, branch)
repo.checkout(branch_ref_obj)
if branch_ref:
return (repo, newfolder, branch_ref_obj)
parents = []
commit = None
try:
if branch_ref_obj:
commit = repo[branch_ref_obj.peel().hex]
else:
commit = repo.revparse_single("HEAD")
pagure.lib.git.update_git(issue, repo, repopath)
repo = pygit2.Repository(self.gitrepo)
cnt = len([commit
for commit in repo.walk(
repo.head.target, pygit2.GIT_SORT_TOPOLOGICAL)])
self.assertEqual(cnt, 9)
last_commit = repo.revparse_single('HEAD')
patch = pagure.lib.git.commit_to_patch(repo, last_commit)
for line in patch.split('\n'):
if line.startswith('--- a/'):
fileid = line.split('--- a/')[1]
break
newpath = tempfile.mkdtemp(prefix='pagure-dump-load')
clone_repo = pygit2.clone_repository(self.gitrepo, newpath)
self.assertEqual(len(os.listdir(newpath)), 4)
ticket_json = os.path.join(self.path, 'test_ticket.json')
self.assertFalse(os.path.exists(ticket_json))
shutil.copyfile(os.path.join(newpath, fileid), ticket_json)
self.assertTrue(os.path.exists(ticket_json))
jsondata = None
with open(ticket_json) as stream:
jsondata = json.load(stream)
self.assertNotEqual(jsondata, None)
shutil.rmtree(newpath)
# Test reloading the JSON
self.tearDown()
# list of binary strings representing parents of the new commit
[first_commit.oid.hex],
)
refname = "refs/heads/master:refs/heads/master"
ori_remote = clone_repo.remotes[0]
PagureRepo.push(ori_remote, refname)
# Set the second repo
new_gitrepo = repopath
if new_project:
# Create a new git repo to play with
new_gitrepo = os.path.join(self.newpath, new_project.fullname)
if not os.path.exists(new_gitrepo):
os.makedirs(new_gitrepo)
new_repo = pygit2.clone_repository(gitrepo, new_gitrepo)
repo = pygit2.Repository(new_gitrepo)
# Edit the sources file again
with open(os.path.join(new_gitrepo, "sources"), "w") as stream:
stream.write("foo\n bar\nbaz\n boose")
repo.index.add("sources")
repo.index.write()
# Commits the files added
tree = repo.index.write_tree()
author = _make_signature("Alice Author", "alice@authors.tld")
committer = _make_signature("Cecil Committer", "cecil@committers.tld")
repo.create_commit(
"refs/heads/%s" % branch_from,
author,
# list of binary strings representing parents of the new commit
[first_commit.oid.hex]
)
refname = 'refs/heads/master:refs/heads/master'
ori_remote = clone_repo.remotes[0]
PagureRepo.push(ori_remote, refname)
# Set the second repo
new_gitrepo = repopath
if new_project:
# Create a new git repo to play with
new_gitrepo = os.path.join(newpath, new_project.fullname)
if not os.path.exists(new_gitrepo):
os.makedirs(new_gitrepo)
new_repo = pygit2.clone_repository(gitrepo, new_gitrepo)
repo = pygit2.Repository(new_gitrepo)
if mtype != 'nochanges':
# Edit the sources file again
with open(os.path.join(new_gitrepo, 'sources'), 'w') as stream:
stream.write('foo\n bar\nbaz\n boose')
repo.index.add('sources')
repo.index.write()
# Commits the files added
tree = repo.index.write_tree()
author = pygit2.Signature(
'Alice Author', 'alice@authors.tld')
committer = pygit2.Signature(
'Cecil Committer', 'cecil@committers.tld')
# list of binary strings representing parents of the new commit
[first_commit.oid.hex]
)
refname = 'refs/heads/master:refs/heads/master'
ori_remote = clone_repo.remotes[0]
PagureRepo.push(ori_remote, refname)
# Set the second repo
new_gitrepo = repopath
if new_project:
# Create a new git repo to play with
new_gitrepo = os.path.join(newpath, new_project.fullname)
if not os.path.exists(new_gitrepo):
os.makedirs(new_gitrepo)
new_repo = pygit2.clone_repository(gitrepo, new_gitrepo)
repo = pygit2.Repository(new_gitrepo)
if mtype != 'nochanges':
# Edit the sources file again
with open(os.path.join(new_gitrepo, 'sources'), 'w') as stream:
stream.write('foo\n bar\nbaz\n boose')
repo.index.add('sources')
repo.index.write()
# Commits the files added
tree = repo.index.write_tree()
author = pygit2.Signature(
'Alice Author', 'alice@authors.tld')
committer = pygit2.Signature(
'Cecil Committer', 'cecil@committers.tld')
repo_path = os.path.join(self._temp_dir, "clone-into")
url = pathname2url(os.path.realpath(src_repo_relpath))
if url.startswith('///'):
url = 'file:' + url
else:
url = 'file://' + url
def create_repository(path, bare):
return init_repository(path, bare)
# here we override the name
def create_remote(repo, name, url):
return repo.remotes.create("custom_remote", url)
repo = clone_repository(url, repo_path, repository=create_repository, remote=create_remote)
assert not repo.is_empty
assert 'refs/remotes/custom_remote/master' in repo.listall_references()
assert repo.remotes["custom_remote"] is not None
callback -- an instance of pygit2.RemoteCallbacks to handle cedentials and
push_update_reference (default: None)
"""
if not callback:
callback = QuitRemoteCallbacks()
self.callback = callback
try:
self._repository = pygit2.Repository(path)
except (KeyError, GitError):
if not create:
raise RepositoryNotFound(
'Repository "%s" does not exist' % path)
if origin:
self._repository = pygit2.clone_repository(
url=origin, path=path, bare=False, callbacks=self.callback
)
else:
self._repository = pygit2.init_repository(path)
if garbageCollection:
self.init_garbageCollection(path)
self.path = path
base_dir = config['cloning']['temporary_clone_path']
clone_dir = os.path.join(base_dir, owner, name)
if os.path.exists(clone_dir):
shutil.rmtree(clone_dir)
callbacks = None
if 'https://' not in url:
keypair = pygit2.Keypair(username=config['cloning']['ssh_user'],
pubkey=config['cloning']['public_key'],
privkey=config['cloning']['private_key'],
passphrase=config['cloning']['ssh_password'])
callbacks = pygit2.RemoteCallbacks(credentials=keypair)
os.makedirs(clone_dir)
repo = clone_repository(url, clone_dir, bare=True, callbacks=callbacks)
repo = Repository(clone_dir)
return repo
continue
if exclude:
found = False
for e in exclude:
if fnmatch(r['name'], e):
found = True
break
if found:
continue
repo_path = os.path.join(clone_dir, r['name'])
repos.append(repo_path)
if not os.path.exists(repo_path):
log.debug("Cloning repo: %s/%s" % (organization, r['name']))
repo = pygit2.clone_repository(
r['url'], repo_path, callbacks=callbacks)
else:
repo = pygit2.Repository(repo_path)
if repo.status():
log.warning('repo %s not clean skipping update')
continue
log.debug("Syncing repo: %s/%s" % (organization, r['name']))
pull(repo, callbacks)
return repos