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_flag_commit_with_uid(self):
""" Test flagging a commit with provided uid. """
repo_obj = pygit2.Repository(self.git_path)
commit = repo_obj.revparse_single('HEAD')
headers = {'Authorization': 'token aaabbbcccddd'}
data = {
'username': 'Jenkins',
'percent': 100,
'comment': 'Tests passed',
'url': 'http://jenkins.cloud.fedoraproject.org/',
'uid': 'jenkins_build_pagure_100+seed',
'status': 'success',
}
output = self.app.post(
'/api/0/test/c/%s/flag' % commit.oid.hex,
headers=headers, data=data)
self.assertEqual(output.status_code, 200)
data = json.loads(output.get_data(as_text=True))
def addfile(self):
"""Create a repository and add a file to the git index."""
# Write to file
self.file.write(b'First Line\n')
self.file.read()
# Add file to index
repo = Repository(self.dir.name)
index = repo.index
index.read()
index.add(self.filename)
index.write()
"""Obtain a :class:`JsonGitRepository`. Either a path to a repo or an
existing repo must be provided.
:param path:
The path to a repository If it is a path that does not exist, a new
bare git repository will be initialized there. If it is a path that
does exist, then the directory will be used as a bare git repository.
:type path: string
:param repo: An existing repository object.
:type repo: :class:`pygit2.Repository`
"""
if repo and path:
raise TypeError("Cannot define repo and path")
if path:
if os.path.isdir(path):
repo = pygit2.Repository(path)
else:
repo = pygit2.init_repository(path, True) # bare repo
if not repo:
raise TypeError("Missing repo or path")
dumps = kwargs.pop('dumps', import_json().dumps)
loads = kwargs.pop('loads', import_json().loads)
return JsonGitRepository(repo, dumps, loads)
# issues are enabled
flask.g.issues_enabled = True
else:
# If the namespace isn't in the list of namespaces
# issues are disabled
flask.g.issues_enabled = False
flask.g.issues_project_disabled = False
if not flask.g.repo.settings.get("issue_tracker", True):
# If the project specifically disabled its issue tracker,
# disable issues
flask.g.issues_project_disabled = True
flask.g.issues_enabled = False
flask.g.reponame = get_repo_path(flask.g.repo)
flask.g.repo_obj = pygit2.Repository(flask.g.reponame)
flask.g.repo_admin = pagure.utils.is_repo_admin(flask.g.repo)
flask.g.repo_committer = pagure.utils.is_repo_committer(flask.g.repo)
flask.g.repo_user = pagure.utils.is_repo_user(flask.g.repo)
flask.g.branches = sorted(flask.g.repo_obj.listall_branches())
repouser = flask.g.repo.user.user if flask.g.repo.is_fork else None
fas_user = flask.g.fas_user if pagure.utils.authenticated() else None
flask.g.repo_watch_levels = pagure.lib.query.get_watch_level_on_repo(
flask.g.session,
fas_user,
flask.g.repo.name,
repouser=repouser,
namespace=namespace,
)
items_per_page = pagure_config["ITEM_PER_PAGE"]
def _external_repo(self, name):
path = self._repo.config.get_multivar("automerge-%s.path" % name)
if not path:
raise ValueError(name)
return pygit2.Repository(path[0])
def update_repo(reponame):
""" For a given path to a repo, pull/rebase the last changes if
it can, add/remove/commit the new changes and push them to the
remote repo if any.
:kwarg reponame, full path to a git repo.
"""
LOG.info('Processing %s' % reponame)
if not os.path.exists(reponame):
raise GitSyncError(
'The indicated working directory does not exists: %s' %
reponame)
try:
repo = Repository(reponame)
except Exception as err:
print(err)
raise GitSyncError(
'The indicated working directory is not a valid git '
'repository: %s' % reponame)
index = repo.index
dopush = False
origin = None
index = repo.index
## Add or remove to staging the files according to their status
if repo.status:
status = repo.status()
for filepath, flag in status.items():
if flag == GIT_STATUS_WT_DELETED:
out = subprocess.check_output(command, stderr=subprocess.STDOUT)
_log.info(" command ran successfully")
_log.debug("Output: %s" % out)
except subprocess.CalledProcessError as err:
_log.debug(
"Command FAILED: {cmd} returned code {code} with the "
"following output: {output}".format(
cmd=err.cmd, code=err.returncode, output=err.output
)
)
raise pagure.exceptions.PagureException(
"Did not manage to rebase this pull-request"
)
class PagureRepo(pygit2.Repository):
""" An utility class allowing to go around pygit2's inability to be
stable.
"""
@staticmethod
def clone(path_from, path_to, checkout_branch=None, bare=False):
""" Clone the git repo at the specified path to the specified location.
This method is meant to replace pygit2.clone_repository which for us
leaks file descriptors on large project leading to "Too many open files
error" which then prevent some tasks from completing.
:arg path_from: the path or url of the git repository to clone
:type path_from: str
:arg path_to: the path where the git repository should be cloned
def do_status(self, directory, path):
if path:
try:
status = git.Repository(directory).status_file(path)
except (KeyError, ValueError):
return None
if status == git.GIT_STATUS_CURRENT:
return None
else:
if status & git.GIT_STATUS_WT_NEW:
return '??'
if status & git.GIT_STATUS_IGNORED:
return '!!'
if status & git.GIT_STATUS_INDEX_NEW:
index_status = 'A'
elif status & git.GIT_STATUS_INDEX_DELETED:
index_status = 'D'
elif status & git.GIT_STATUS_INDEX_MODIFIED:
"USER": "pagure",
"REPOBRIDGE_CONFIG": ":environment:",
"REPOBRIDGE_BASEURL": regioninfo["url"],
"REPOBRIDGE_CA": regioninfo["ca"],
"REPOBRIDGE_CERT": regioninfo["push_cert"]["cert"],
"REPOBRIDGE_KEY": regioninfo["push_cert"]["key"],
}
)
with open(os.devnull, "w") as devnull:
subprocess.check_call(
command,
stdout=devnull,
stderr=subprocess.STDOUT,
env=environ,
)
self.repo = pygit2.Repository(self.repopath)
self._origrepo = self.repo
# Make sure that all remote refs are mapped to local ones.
headname = None
if not self.repo.is_empty and not self.repo.head_is_unborn:
headname = self.repo.head.shorthand
# Sync up all the references, branches and PR heads
for ref in self._origrepo.listall_references():
if ref.startswith("refs/heads/"):
localname = ref.replace("refs/heads/", "")
if localname in (headname, "HEAD"):
# This gets checked out by default
continue
branch = self.repo.branches.remote.get("origin/%s" % localname)
self.repo.branches.local.create(localname, branch.peel())
def _get_repo(repo_key):
path = safe_join(current_app.config['RESTFULGIT_REPO_BASE_PATH'], repo_key)
try:
return Repository(path)
except KeyError:
raise NotFound("repository not found")