''' Command line behavior for centpkg ''' # # Author(s): # Jesse Keating # Pat Riehecky # Brian Stinson # # 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 six.moves.urllib_parse import urlparse import six.moves.configparser as ConfigParser import json import koji import os import sys _DEFAULT_API_BASE_URL = 'https://gitlab.com' 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() 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}/-/profile/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 = 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)) # 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 '''.format('\n'.join(textwrap.wrap(build_parser.description)))) # Now add our additional option build_parser.add_argument( '--rhel-target', choices=['exception', 'zstream', 'latest'], help='Set the rhel-target metadata') # Overloaded _build def _build(self, sets=None): # Only do rhel-target if we are centpkg, not centpkg-sig if not os.path.basename(sys.argv[0]).endswith('-sig') and not self.args.scratch: # Only run if we have internal configuraions, or scratch 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 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.SafeConfigParser() 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: active_y, in_stabilization = centpkg.utils.determine_active_y_version(rhel_version, pp_api_url) except AssertionError as e: self.log.info(" Error: centpkg cannot determine the development phase.") self.log.info(" Please file an issue at https://git.centos.org/centos/centpkg") self.log.info(" Workaround: Use the --rhel-target option") self.log.info("Exiting") raise SystemExit divergent_branch = centpkg.utils.does_divergent_branch_exist( self.cmd.repo_name, rhel_version, rhel_dist_git, pp_api_url, "rpms") # Good to know if divergent_branch : self.log.info(" a divergent branch was found.") 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.info("We are currently in Stabalization mode") self.log.info("You must either set the rhel-target (--rhel-target)") self.log.info("or branch for the previous version.") self.log.info("Exiting") raise SystemExit 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.info("We are currently in Stabalization mode") self.log.info("You must either set the rhel-target (--rhel-target)") self.log.info("or branch for the previous version.") self.log.info("Exiting") raise SystemExit # Good to know self.log.info(" rhel-target: " + json.loads(self.args.custom_user_metadata)['rhel-target']) else: self.log.info("NO INTERNAL CONFIGURATION") self.log.info("Only scratch builds are allowed without internal configurations") self.log.info("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.info("Exiting") raise SystemExit # 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()