Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, context, name, description):
PulpCliCommand.__init__(self, name, description, self.status)
self.context = context
self.create_option('--repo-id', _('identifies the repository'), required=True)
OPTION_NAME, OPTION_NOTES)
from pulp.client.extensions.extensions import (PulpCliCommand, PulpCliOption)
DESC_CREATE = _('creates a new consumer group')
DESC_DELETE = _('deletes a consumer group')
DESC_UPDATE = _('updates the metadata about the group itself (not its members)')
DESC_LIST = _('lists consumer groups on the Pulp server')
DESC_SEARCH = _('searches for consumer groups on the Pulp server')
# Defaults to pass to render_document_list when displaying groups
DEFAULT_FILTERS = ['id', 'display_name', 'description', 'consumer_ids', 'notes']
DEFAULT_ORDER = DEFAULT_FILTERS
class CreateConsumerGroupCommand(PulpCliCommand):
def __init__(self, context, name='create', description=DESC_CREATE, method=None):
self.context = context
self.prompt = context.prompt
if method is None:
method = self.run
super(CreateConsumerGroupCommand, self).__init__(name, description, method)
self.add_option(OPTION_GROUP_ID)
self.add_option(OPTION_NAME)
self.add_option(OPTION_DESCRIPTION)
self.add_option(OPTION_NOTES)
def run(self, **kwargs):
def __init__(self, context):
PulpCliSection.__init__(self, 'user', 'manage users')
self.context = context
self.prompt = context.prompt # for easier access
# Common Options
login_option = PulpCliOption('--login', 'uniquely identifies the user; '
'only alphanumeric, -, ., and _ allowed',
required=True,
validate_func=validators.id_validator_allow_dots)
name_option = PulpCliOption('--name', 'user-readable full name of the user',
required=False)
# Create command
create_command = PulpCliCommand('create', 'creates a user', self.create)
create_command.add_option(login_option)
create_command.add_option(PulpCliOption('--password', 'password for the new user, '
'if you do not want to be prompted for one',
required=False))
create_command.add_option(name_option)
self.add_command(create_command)
# Update Command
update_command = PulpCliCommand('update', 'changes metadata of an existing user',
self.update)
update_command.add_option(PulpCliOption('--login', 'identifies the user to be updated',
required=True))
update_command.add_option(name_option)
update_command.add_option(PulpCliOption('--password', 'new password for the user, use -p '
'if you want to be prompted for the password',
required=False))
def __init__(self, context, name, description):
PulpCliCommand.__init__(self, name, description, self.unbind)
self.context = context
self.prompt = context.prompt
self.add_option(PulpCliOption('--repo-id', 'repository id', required=True))
if not consumer_id:
self.prompt.render_failure_message("This consumer is not registered to the Pulp server.")
return
delta = dict([(k, v) for k, v in kwargs.items() if v is not None])
if 'note' in delta.keys():
if delta['note']:
delta['notes'] = args_to_notes_dict(kwargs['note'], include_none=False)
delta.pop('note')
try:
self.context.server.consumer.update(consumer_id, delta)
self.prompt.render_success_message('Consumer [%s] successfully updated' % consumer_id)
except NotFoundException:
self.prompt.write('Consumer [%s] does not exist on the server' % consumer_id, tag='not-found')
class UnregisterCommand(PulpCliCommand):
def __init__(self, context, name, description):
PulpCliCommand.__init__(self, name, description, self.unregister)
self.context = context
self.prompt = context.prompt
d = 'if specified, the local consumer identification certificate will be '\
'removed even if the server cannot be contacted'
self.create_flag('--force', _(d))
def unregister(self, **kwargs):
consumer_id = load_consumer_id(self.context)
if not consumer_id:
self.context.prompt.render_failure_message("This consumer is not registered to the Pulp server.")
return
def __init__(self, context):
PulpCliSection.__init__(self, 'role', 'role lifecycle (list, create, update, etc.) commands')
self.context = context
self.prompt = context.prompt # for easier access
# Common Options
id_option = PulpCliOption('--role-id', 'uniquely identifies the role; only alphanumeric, -, and _ allowed', required=True)
# Create command
create_command = PulpCliCommand('create', 'creates a role', self.create)
create_command.add_option(id_option)
create_command.add_option(PulpCliOption('--display-name', 'user-friendly name for the role', required=False))
create_command.add_option(PulpCliOption('--description', 'user-friendly text describing the role', required=False))
self.add_command(create_command)
# Update command
update_command = PulpCliCommand('update', 'updates a role', self.update)
update_command.add_option(PulpCliOption('--role-id', 'identifies the role to be updated', required=True))
update_command.add_option(PulpCliOption('--display-name', 'user-friendly name for the role', required=False))
update_command.add_option(PulpCliOption('--description', 'user-friendly text describing the role', required=False))
self.add_command(update_command)
# Delete Command
delete_command = PulpCliCommand('delete', 'deletes a role', self.delete)
delete_command.add_option(PulpCliOption('--role-id', 'identifies the role to be deleted', required=True))
self.add_command(delete_command)
def __init__(self, context, name, description):
PulpCliCommand.__init__(self, name, description, self.list)
self.context = context
The content *catalog* section.
"""
NAME = 'catalog'
DESCRIPTION = _('manage the content catalog')
def __init__(self, context):
"""
:param context: The client context.
:type context: pulp.client.extensions.core.ClientContext
"""
super(CatalogSection, self).__init__(self.NAME, self.DESCRIPTION)
self.add_command(CatalogDeleteCommand(context))
class ListCommand(PulpCliCommand):
"""
List command.
:ivar context: The client context.
:type context: pulp.client.extensions.core.ClientContext
"""
NAME = 'list'
DESCRIPTION = _('list sources')
TITLE = _('Content Sources')
def __init__(self, context):
"""
:param context: The client context.
:type context: pulp.client.extensions.core.ClientContext
"""
super(ListCommand, self).__init__(self.NAME, self.DESCRIPTION, self._run)
if delta.get(OPTION_NAME.keyword, None) is not None:
delta['display_name'] = delta.pop(OPTION_NAME.keyword)
if delta.pop(OPTION_NOTES.keyword, None) is not None:
delta['notes'] = kwargs[OPTION_NOTES.keyword]
try:
self.context.server.repo_group.update(kwargs[OPTION_GROUP_ID.keyword], delta)
msg = 'Repo group [%(g)s] successfully updated'
self.prompt.render_success_message(msg % {'g': kwargs['group-id']})
except NotFoundException:
msg = 'Repo group [%(g)s] does not exist on the server'
self.prompt.write(msg % {'g': kwargs['group-id']}, tag='not-found')
class ListRepositoryGroupsCommand(PulpCliCommand):
"""
Lists repository groups in the Pulp server.
"""
def __init__(self, context, name='list', description=DESC_LIST, method=None):
self.context = context
self.prompt = context.prompt
if method is None:
method = self.run
super(ListRepositoryGroupsCommand, self).__init__(name, description, method)
self.add_option(PulpCliFlag('--details', _('if specified, all the repo group information '
'is displayed')))
self.add_option(PulpCliOption('--fields', _('comma-separated list of repo group fields; '
def __init__(self, context, strategy, name=NAME_DELETE, description=DESC_DELETE):
PulpCliCommand.__init__(self, name, description, self.run)
self.context = context
self.strategy = strategy
self.add_option(OPT_SCHEDULE_ID)
def run(self, **kwargs):
schedule_id = kwargs[OPT_SCHEDULE_ID.keyword]
self.strategy.delete_schedule(schedule_id, kwargs)
self.context.prompt.render_success_message(_('Schedule successfully deleted'))
class UpdateScheduleCommand(PulpCliCommand):
"""
Provides options for manipulating the standard schedule configuration such
as the schedule itself or whether or not it is enabled. Subclasses or'
instances should add any extra options that are required. The corresponding
strategy method will be invoked with all of the user specified arguments.
"""
def __init__(self, context, strategy, name=NAME_UPDATE, description=DESC_UPDATE):
PulpCliCommand.__init__(self, name, description, self.run)
self.context = context
self.strategy = strategy
self.add_option(OPT_SCHEDULE_ID)
self.add_option(OPT_FAILURE_THRESHOLD)
self.add_option(OPT_ENABLED)