Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def run(self):
# rev is a commit
# we use branch on purpose so we get the latest thing
# TODO: check if rev is HEAD on {branch}, warn then?
branch = nested_get(self.event, "msg", "commit", "branch")
# self.project is dist-git, we need to get upstream
dg = DistGit(self.config, self.package_config)
self.package_config.upstream_project_url = (
dg.get_project_url_from_distgit_spec()
)
if not self.package_config.upstream_project_url:
raise PackitException(
"URL in specfile is not set. We don't know where the upstream project lives."
)
n, r = get_namespace_and_repo_name(self.package_config.upstream_project_url)
up = self.upstream_service.get_project(repo=r, namespace=n)
lp = LocalProject(git_project=up)
def get_job_input_from_dist_git_commit(
self, event: dict
) -> Optional[Tuple[JobTriggerType, PackageConfig, GitProject]]:
""" this corresponds to dist-git event when someone pushes new commits """
topic = nested_get(event, "topic")
logger.debug(f"topic = {topic}")
if topic == NewDistGitCommit.topic:
repo_namespace = nested_get(event, "msg", "commit", "namespace")
repo_name = nested_get(event, "msg", "commit", "repo")
ref = nested_get(event, "msg", "commit", "branch")
if not (repo_namespace and repo_name):
logger.warning(
"We could not figure out the full name of the repository."
)
return None
if not ref:
logger.warning("Target branch for the new commits is not set.")
return None
logger.info(
f"New commits added to dist-git repo {repo_namespace}/{repo_name}, branch {ref}."
)
msg_id = nested_get(event, "msg_id")
logger.info(f"msg_id = {msg_id}")
dg_proj = self.pagure_service.get_project(
def get_job_input_from_github_pr(
self, event: dict
) -> Optional[Tuple[JobTriggerType, PackageConfig, GitProject]]:
""" look into the provided event and see if it's one for a new github pr """
action = nested_get(event, "action")
logger.debug(f"action = {action}")
pr_id = nested_get(event, "number")
is_pr = nested_get(event, "pull_request")
if not is_pr:
logger.info("Not a pull request event.")
return None
if action in ["opened", "reopened", "synchronize"] and pr_id:
# we can't use head repo here b/c the app is set up against the upstream repo
# and not the fork, on the other hand, we don't process packit.yaml from
# the PR but what's in the upstream
base_repo_namespace = nested_get(
event, "pull_request", "base", "repo", "owner", "login"
)
base_repo_name = nested_get(event, "pull_request", "base", "repo", "name")
if not (base_repo_name and base_repo_namespace):
logger.warning(
"We could not figure out the full name of the repository."
)
return None
base_ref = nested_get(event, "pull_request", "head", "sha")
if not base_ref:
logger.warning("Ref where the PR is coming from is not set.")
return None
target_repo = nested_get(event, "repository", "full_name")
logger.info(f"GitHub pull request {pr_id} event for repo {target_repo}.")
gh_proj = get_github_project(
return None
if action in ["opened", "reopened", "synchronize"] and pr_id:
# we can't use head repo here b/c the app is set up against the upstream repo
# and not the fork, on the other hand, we don't process packit.yaml from
# the PR but what's in the upstream
base_repo_namespace = nested_get(
event, "pull_request", "base", "repo", "owner", "login"
)
base_repo_name = nested_get(event, "pull_request", "base", "repo", "name")
if not (base_repo_name and base_repo_namespace):
logger.warning(
"We could not figure out the full name of the repository."
)
return None
base_ref = nested_get(event, "pull_request", "head", "sha")
if not base_ref:
logger.warning("Ref where the PR is coming from is not set.")
return None
target_repo = nested_get(event, "repository", "full_name")
logger.info(f"GitHub pull request {pr_id} event for repo {target_repo}.")
gh_proj = get_github_project(
self.config, repo=base_repo_name, namespace=base_repo_namespace
)
package_config = get_package_config_from_repo(gh_proj, base_ref)
https_url = event["repository"]["html_url"]
package_config.upstream_project_url = https_url
return JobTriggerType.pull_request, package_config, gh_proj
return None
"""
look into the provided event and see if it's one for a published github release;
if it is, process it and return input for the job handler
"""
action = nested_get(event, "action")
logger.debug(f"action = {action}")
release = nested_get(event, "release")
if action == "published" and release:
repo_namespace = nested_get(event, "repository", "owner", "login")
repo_name = nested_get(event, "repository", "name")
if not (repo_namespace and repo_name):
logger.warning(
"We could not figure out the full name of the repository."
)
return None
release_ref = nested_get(event, "release", "tag_name")
if not release_ref:
logger.warning("Release tag name is not set.")
return None
logger.info(
f"New release event {release_ref} for repo {repo_namespace}/{repo_name}."
)
gh_proj = get_github_project(
self.config, repo=repo_name, namespace=repo_namespace
)
package_config = get_package_config_from_repo(gh_proj, release_ref)
https_url = event["repository"]["html_url"]
package_config.upstream_project_url = https_url
return JobTriggerType.release, package_config, gh_proj
return None
def get_job_input_from_github_pr(
self, event: dict
) -> Optional[Tuple[JobTriggerType, PackageConfig, GitProject]]:
""" look into the provided event and see if it's one for a new github pr """
action = nested_get(event, "action")
logger.debug(f"action = {action}")
pr_id = nested_get(event, "number")
is_pr = nested_get(event, "pull_request")
if not is_pr:
logger.info("Not a pull request event.")
return None
if action in ["opened", "reopened", "synchronize"] and pr_id:
# we can't use head repo here b/c the app is set up against the upstream repo
# and not the fork, on the other hand, we don't process packit.yaml from
# the PR but what's in the upstream
base_repo_namespace = nested_get(
event, "pull_request", "base", "repo", "owner", "login"
)
base_repo_name = nested_get(event, "pull_request", "base", "repo", "name")
if not (base_repo_name and base_repo_namespace):
logger.warning(
"We could not figure out the full name of the repository."
)
if topic == NewDistGitCommit.topic:
repo_namespace = nested_get(event, "msg", "commit", "namespace")
repo_name = nested_get(event, "msg", "commit", "repo")
ref = nested_get(event, "msg", "commit", "branch")
if not (repo_namespace and repo_name):
logger.warning(
"We could not figure out the full name of the repository."
)
return None
if not ref:
logger.warning("Target branch for the new commits is not set.")
return None
logger.info(
f"New commits added to dist-git repo {repo_namespace}/{repo_name}, branch {ref}."
)
msg_id = nested_get(event, "msg_id")
logger.info(f"msg_id = {msg_id}")
dg_proj = self.pagure_service.get_project(
repo=repo_name, namespace=repo_namespace
)
package_config = get_package_config_from_repo(dg_proj, ref)
return JobTriggerType.commit, package_config, dg_proj
return None
def get_job_input_from_github_release(
self, event: dict
) -> Optional[Tuple[JobTriggerType, PackageConfig, GitProject]]:
"""
look into the provided event and see if it's one for a published github release;
if it is, process it and return input for the job handler
"""
action = nested_get(event, "action")
logger.debug(f"action = {action}")
release = nested_get(event, "release")
if action == "published" and release:
repo_namespace = nested_get(event, "repository", "owner", "login")
repo_name = nested_get(event, "repository", "name")
if not (repo_namespace and repo_name):
logger.warning(
"We could not figure out the full name of the repository."
)
return None
release_ref = nested_get(event, "release", "tag_name")
if not release_ref:
logger.warning("Release tag name is not set.")
return None
logger.info(
f"New release event {release_ref} for repo {repo_namespace}/{repo_name}."
)
gh_proj = get_github_project(
self.config, repo=repo_name, namespace=repo_namespace
)
def get_job_input_from_github_release(
self, event: dict
) -> Optional[Tuple[JobTriggerType, PackageConfig, GitProject]]:
"""
look into the provided event and see if it's one for a published github release;
if it is, process it and return input for the job handler
"""
action = nested_get(event, "action")
logger.debug(f"action = {action}")
release = nested_get(event, "release")
if action == "published" and release:
repo_namespace = nested_get(event, "repository", "owner", "login")
repo_name = nested_get(event, "repository", "name")
if not (repo_namespace and repo_name):
logger.warning(
"We could not figure out the full name of the repository."
)
return None
release_ref = nested_get(event, "release", "tag_name")
if not release_ref:
logger.warning("Release tag name is not set.")
return None
logger.info(
f"New release event {release_ref} for repo {repo_namespace}/{repo_name}."
)
gh_proj = get_github_project(
self.config, repo=repo_name, namespace=repo_namespace
)
package_config = get_package_config_from_repo(gh_proj, release_ref)
def get_job_input_from_github_pr(
self, event: dict
) -> Optional[Tuple[JobTriggerType, PackageConfig, GitProject]]:
""" look into the provided event and see if it's one for a new github pr """
action = nested_get(event, "action")
logger.debug(f"action = {action}")
pr_id = nested_get(event, "number")
is_pr = nested_get(event, "pull_request")
if not is_pr:
logger.info("Not a pull request event.")
return None
if action in ["opened", "reopened", "synchronize"] and pr_id:
# we can't use head repo here b/c the app is set up against the upstream repo
# and not the fork, on the other hand, we don't process packit.yaml from
# the PR but what's in the upstream
base_repo_namespace = nested_get(
event, "pull_request", "base", "repo", "owner", "login"
)
base_repo_name = nested_get(event, "pull_request", "base", "repo", "name")
if not (base_repo_name and base_repo_namespace):