'''
Command line behavior for centpkg
'''
#
# Author(s):
# Jesse Keating <jkeating@redhat.com>
# Pat Riehecky <riehecky@fnal.gov>
# Brian Stinson <bstinson@ksu.edu>
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation; either version 2 of the License, or (at your
# option) any later version. See http://www.gnu.org/copyleft/gpl.html for
# the full text of the license.
from __future__ import print_function
import argparse
import textwrap
from centpkg.utils import config_get_safely, do_add_remote, do_fork
import centpkg.utils
from pyrpkg.cli import cliClient
from pyrpkg import rpkgError
from urllib.parse import urlparse
from configparser import ConfigParser
import gitlab
import json
import os
import sys
_DEFAULT_API_BASE_URL = 'https://gitlab.com'
_DEFAULT_PARENT_NAMESPACE = {
'centpkg': 'redhat/centos-stream',
'centpkg-sig': 'CentOS',
}
class centpkgClient(cliClient):
def __init__(self, config, name='centpkg'):
self.DEFAULT_CLI_NAME = name
# Save the config for utilities such as get_repo_name()
centpkg.utils.dist_git_config = config
centpkg.utils.dist_git_config.add_section('__default')
centpkg.utils.dist_git_config.set('__default', 'cli_name', name)
super(centpkgClient, self).__init__(config, name)
self.setup_centos_subparsers()
def setup_centos_subparsers(self):
self.register_do_fork()
self.register_request_gated_side_tag()
self.register_current_state()
def register_do_fork(self):
help_msg = 'Create a new fork of the current repository'
distgit_section = '{0}.distgit'.format(self.name)
# uses default dist-git url in case section is not present
try:
distgit_api_base_url = config_get_safely(self.config, distgit_section, "apibaseurl")
except rpkgError:
distgit_api_base_url = _DEFAULT_API_BASE_URL
description = textwrap.dedent('''
Create a new fork of the current repository
Before the operation, you need to generate an API token at
https://{1}/-/user_settings/personal_access_tokens, select the "api"
scope and save it in your local user configuration located
at ~/.config/rpkg/{0}.conf. For example:
[{0}.distgit]
token = <api_key_here>
Below is a basic example of the command to fork a current repository:
{0} fork
'''.format(self.name, urlparse(distgit_api_base_url).netloc))
fork_parser = self.subparsers.add_parser(
'fork',
formatter_class=argparse.RawDescriptionHelpFormatter,
help=help_msg,
description=description)
fork_parser.set_defaults(command=self.do_distgit_fork)
def do_distgit_fork(self):
"""create fork of the distgit repository
That includes creating fork itself using API call and then adding
remote tracked repository
"""
distgit_section = '{0}.distgit'.format(self.name)
distgit_api_base_url = config_get_safely(self.config, distgit_section, "apibaseurl")
distgit_token = config_get_safely(self.config, distgit_section, 'token')
ret, repo_path = do_fork(
logger=self.log,
base_url=distgit_api_base_url,
token=distgit_token,
repo_name=self.cmd.repo_name,
namespace=self.cmd.ns,
cli_name=self.name,
)
# assemble url of the repo in web browser
fork_url = '{0}/{1}'.format(
distgit_api_base_url.rstrip('/'),
repo_path,
)
if ret:
msg = "Fork of the repository has been created: '{0}'"
else:
msg = "Repo '{0}' already exists."
self.log.info(msg.format(fork_url))
distgit_remote_base_url = self.config.get(
'{0}'.format(self.name),
"gitbaseurl",
vars={'repo': '{0}/{1}'.format(self.cmd.ns, self.cmd.repo_name)},
)
remote_name = repo_path.split('/')[0]
ret = do_add_remote(
base_url=distgit_api_base_url,
remote_base_url=distgit_remote_base_url,
repo=self.cmd.repo,
repo_path=repo_path,
remote_name=remote_name,
)
if ret:
msg = "Adding as remote '{0}'."
else:
msg = "Remote with name '{0}' already exists."
self.log.info(msg.format(remote_name))
def register_current_state(self):
"""Register command line parser for subcommand current-state"""
parser = self.subparsers.add_parser(
"current-state",
help="Show the phase and state of current dist-git branch - Internal Only",
)
parser.set_defaults(command=self.request_current_state)
# FIXME: This is alot of duplication.
# Get it working now. De-duplicate later
def request_current_state(self):
# Only run if we have internal configuraions, or scratch
internal_config_file = "/etc/rpkg/centpkg_internal.conf"
if os.path.exists(internal_config_file):
# Get our internal only variables
cfg = ConfigParser()
cfg.read(internal_config_file)
pp_api_url = config_get_safely(cfg, "centpkg.internal", 'pp_api_url')
gitbz_query_url = config_get_safely(cfg, "centpkg.internal", 'gitbz_query_url')
rhel_dist_git = config_get_safely(cfg, "centpkg.internal", 'rhel_dist_git')
# Find out divergent branch and stabalization
self.log.info("Checking current state:")
stream_version = self.cmd.target.split('-')[0]
rhel_version = centpkg.utils.stream_mapping(stream_version)
try:
x_version, active_y, is_beta, in_stabilization = centpkg.utils.determine_active_y_version(rhel_version, pp_api_url)
except AssertionError as e:
self.log.error(" Error: centpkg cannot determine the development phase.")
self.log.error(" Please file an issue at https://git.centos.org/centos/centpkg")
self.log.error(" Workaround: Use the --rhel-target option")
self.log.error("Exiting")
raise SystemExit(1)
if is_beta:
# Special case: for X.0 betas, there will never be a prior branch
# In this case, always work on the active branch.
divergent_branch = True
else:
divergent_branch = centpkg.utils.does_divergent_branch_exist(
self.cmd.repo_name,
x_version,
active_y,
rhel_dist_git,
"rpms")
# Good to know
if in_stabilization :
self.log.info(" we are in stabilization mode.")
else:
self.log.info(" we are not in stabilization mode.")
if divergent_branch and not is_beta:
self.log.info(" a divergent branch was found.")
elif divergent_branch and is_beta:
self.log.info(" we are working on a beta release.")
else:
self.log.info(" a divergent branch was not found.")
else:
self.log.error("NO INTERNAL CONFIGURATION")
self.log.error("Hint: If you are internal, install the rhel-packager package from https://download.devel.redhat.com/rel-eng/RCMTOOLS/latest-RCMTOOLS-2-RHEL-9/compose/BaseOS/x86_64/os/Packages/")
self.log.error("Exiting")
raise SystemExit(1)
# Overloaded build
def register_build(self):
# Do all the work from the super class
super(centpkgClient, self).register_build()
build_parser = self.subparsers.choices['build']
build_parser.formatter_class = argparse.RawDescriptionHelpFormatter
build_parser.description = textwrap.dedent('''
{0}
centpkg now sets the rhel metadata with --rhel-target.
* exception - This will build for the current in-development Y-stream release.
It is equivalent to passing latest when not in the Blocker and Exception Phase.
* zstream - If pre-GA of a y-stream release, this will build for 0day.
If post-GA of a Y-stream release, this will build for the Z-stream of that release.
* latest - This will always build for the next Y-stream release
* none - This will not build in brew. No rhel metadata is set.
'''.format('\n'.join(textwrap.wrap(build_parser.description))))
# Now add our additional option
build_parser.add_argument(
'--rhel-target',
choices=['exception', 'zstream', 'latest', 'none'],
help='Set the rhel-target metadata')
# Overloaded _build
def _build(self, sets=None):
# Only do rhel-target if we are centpkg, not centpkg-sig, or if we are doing scratch
if not os.path.basename(sys.argv[0]).endswith('-sig') and not self.args.scratch:
# Only run if we have internal configuraions
internal_config_file = "/etc/rpkg/centpkg_internal.conf"
if os.path.exists(internal_config_file):
# If rhel-target is set, no need to check, just use it
if hasattr(self.args, 'rhel_target') and self.args.rhel_target:
# If rhel-target set to none, do nothing with metadata
if self.args.rhel_target.lower() != "none":
# If custom-user-metadata set, add onto it
if hasattr(self.args, 'custom_user_metadata') and self.args.custom_user_metadata:
try:
temp_custom_user_metadata = json.loads(self.args.custom_user_metadata)
# Use ValueError instead of json.JSONDecodeError for Python 2 and 3 compatibility
except ValueError as e:
self.parser.error("--custom-user-metadata is not valid JSON: %s" % e)
if not isinstance(temp_custom_user_metadata, dict):
self.parser.error("--custom-user-metadata must be a JSON object")
temp_custom_user_metadata["rhel-target"] = self.args.rhel_target
else:
temp_custom_user_metadata = {"rhel-target": self.args.rhel_target}
self.args.custom_user_metadata = json.dumps(temp_custom_user_metadata)
else:
# Get our internal only variables
cfg = ConfigParser()
cfg.read(internal_config_file)
pp_api_url = config_get_safely(cfg, "centpkg.internal", 'pp_api_url')
gitbz_query_url = config_get_safely(cfg, "centpkg.internal", 'gitbz_query_url')
rhel_dist_git = config_get_safely(cfg, "centpkg.internal", 'rhel_dist_git')
# Find out divergent branch and stabalization
self.log.info("Checking rhel-target information:")
stream_version = self.cmd.target.split('-')[0]
rhel_version = centpkg.utils.stream_mapping(stream_version)
try:
x_version, active_y, is_beta, in_stabilization = centpkg.utils.determine_active_y_version(rhel_version, pp_api_url)
except AssertionError as e:
self.log.error(" Error: centpkg cannot determine the development phase.")
self.log.error(" Please file an issue at https://git.centos.org/centos/centpkg")
self.log.error(" Workaround: Use the --rhel-target option")
self.log.error("Exiting")
raise SystemExit(1)
if is_beta:
# Special case: for X.0 betas, there will never be a prior branch
# In this case, always work on the active branch.
divergent_branch = True
else:
divergent_branch = centpkg.utils.does_divergent_branch_exist(
self.cmd.repo_name,
x_version,
active_y,
rhel_dist_git,
"rpms")
# Good to know
if divergent_branch and not is_beta:
self.log.info(" a divergent branch was found.")
elif divergent_branch and is_beta:
self.log.info(" we are working on a beta release.")
else:
self.log.info(" a divergent branch was not found.")
if in_stabilization :
self.log.info(" we are in stabilization mode.")
else:
self.log.info(" we are not in stabilization mode.")
# Update args.custom_user_metadata
if hasattr(self.args, 'custom_user_metadata') and self.args.custom_user_metadata:
try:
temp_custom_user_metadata = json.loads(self.args.custom_user_metadata)
# Use ValueError instead of json.JSONDecodeError for Python 2 and 3 compatibility
except ValueError as e:
self.parser.error("--custom-user-metadata is not valid JSON: %s" % e)
if not isinstance(temp_custom_user_metadata, dict):
self.parser.error("--custom-user-metadata must be a JSON object")
if divergent_branch:
temp_custom_user_metadata["rhel-target"] = "latest"
elif not divergent_branch and not in_stabilization :
temp_custom_user_metadata["rhel-target"] = "zstream"
else:
self.log.error("We are currently in Stabalization mode")
self.log.error("You must either set the rhel-target (--rhel-target)")
self.log.error("or branch for the previous version.")
self.log.error("Exiting")
raise SystemExit(1)
self.args.custom_user_metadata = json.dumps(temp_custom_user_metadata)
else:
if divergent_branch:
self.args.custom_user_metadata = '{"rhel-target": "latest"}'
elif not divergent_branch and not in_stabilization :
self.args.custom_user_metadata = '{"rhel-target": "zstream"}'
else:
self.log.error("We are currently in Stabalization mode")
self.log.error("You must either set the rhel-target (--rhel-target)")
self.log.error("or branch for the previous version.")
self.log.error("Exiting")
raise SystemExit(1)
# Good to know
self.log.info(" rhel-target: " + json.loads(self.args.custom_user_metadata)['rhel-target'])
else:
self.log.error("NO INTERNAL CONFIGURATION")
self.log.error("Only scratch builds are allowed without internal configurations")
self.log.error("Hint: Install the rhel-packager package from https://download.devel.redhat.com/rel-eng/RCMTOOLS/latest-RCMTOOLS-2-RHEL-9/compose/BaseOS/x86_64/os/Packages/ if you want to build the package via internal (Red Hat) configuration.")
self.log.error("Exiting")
raise SystemExit(1)
# Proceed with build
return super(centpkgClient, self)._build(sets)
def register_request_gated_side_tag(self):
"""Register command line parser for subcommand request-gated-side-tag"""
parser = self.subparsers.add_parser(
"request-gated-side-tag",
help="Create a new dynamic gated side tag",
)
parser.add_argument("--base-tag", help="name of base tag")
parser.set_defaults(command=self.request_gated_side_tag)
def request_gated_side_tag(self):
self.args.suffix = "stack-gate"
super(centpkgClient, self).request_side_tag()
def clone(self):
# Since gitlab allows git repos to be checked out with incorrect capitilization
# we need to check the spelling when cloning
gl = gitlab.Gitlab(_DEFAULT_API_BASE_URL)
parent_namespace = _DEFAULT_PARENT_NAMESPACE[self.DEFAULT_CLI_NAME]
if "/" in self.args.repo[0]:
project_name_with_namespace = parent_namespace+"/"+self.args.repo[0]
short_name=self.args.repo[0].split("/")[-1]
else:
project_name_with_namespace = parent_namespace+"/rpms/"+self.args.repo[0]
short_name=self.args.repo[0]
try:
project = gl.projects.get(project_name_with_namespace)
except gitlab.exceptions.GitlabGetError as e:
self.log.error("There is no repository with the given name. Did you spell it correctly?")
self.log.error("Fatal: ")
self.log.error(e)
raise SystemExit(1)
if not project.name == short_name:
self.log.error("Fatal: ")
self.log.error("Project name is wrong: " + short_name + " it should be: " + project.name)
raise SystemExit(1)
super(centpkgClient, self).clone()