pulp_smash.cli

Location: Pulp SmashAPI Documentationpulp_smash.cli

A client for working with Pulp hosts via their CLI.

class pulp_smash.cli.BaseServiceManager

Bases: object

A base service manager.

Each subclass must implement the abstract methods to provide the service management on a single or multiple hosts.

Subclasses should take advantage of the helper methods offered by this class in order to manage services and check the proper service manager software available on a host.

This base class also offers a context manager to temporary disable SELinux. It is useful when managing services on hosts running RHEL 6 and earlier, which has SELinux issues when running on Jenkins.

Make sure to call this class __init__ method on the subclass __init__ method to ensure the helper methods functionality.

is_active(services)

Check whether given services are active.

Parameters:services – A list or tuple of services to check.
Returns:boolean
restart(services)

Restart the given services.

Parameters:services – A list or tuple of services to be restarted.
start(services)

Start the given services.

Parameters:services – A list or tuple of services to be started.
stop(services)

Stop the given services.

Parameters:services – A list or tuple of services to be stopped.
class pulp_smash.cli.Client(cfg, response_handler=None, pulp_host=None)

Bases: object

A convenience object for working with a CLI.

This class provides the ability to execute shell commands on either the local host or a remote host. Here is a typical usage example:

>>> from pulp_smash import cli, config
>>> client = cli.Client(config.PulpSmashConfig.load())
>>> response = client.run(('echo', '-n', 'foo'))
>>> response.returncode == 0
True
>>> response.stdout == 'foo'
True
>>> response.stderr == ''
True

Smartly chosen defaults make this example concise, but it’s also quite flexible. For example, if a single Pulp application is deployed across several hosts, one can choose on which host commands are executed:

>>> from pulp_smash import cli, config
>>> cfg = config.PulpSmashConfig.load()
>>> client = cli.Client(cfg, pulp_host=cfg.get_hosts('shell')[0])
>>> response = client.run(('echo', '-n', 'foo'))

You can customize how Client objects execute commands and handle responses by fiddling with the two public instance attributes:

machine
A Plumbum machine. run() delegates all command execution responsibilities to this object.
response_handler
A callback function. Each time machine executes a command, the result is handed to this callback, and the callback’s return value is handed to the user.

If pulp_host.roles['shell']['transport'] is 'local' or 'ssh, machine will be set so that commands run locally or over SSH, respectively. If pulp_host.roles['shell']['transport'] is None, the constructor will guess how to set machine by comparing the hostname embedded in pulp_host.hostname against the current host’s hostname. If they match, machine is set to execute commands locally; and vice versa.

Parameters:
  • cfg (pulp_smash.config.PulpSmashConfig) – Information about the host on which commands will be executed.
  • response_handler – A callback function. Defaults to pulp_smash.cli.code_handler().
  • pulp_host (pulp_smash.config.PulpHost) – A specific host to target. Defaults to the first host with the pulp cli role when targeting Pulp 2, and the first host with the shell role when targeting Pulp 3. If Pulp 3 gets a CLI, this latter default may change.
is_superuser

Check if the current client is root.

If the current client is in root mode it stores the status as a cache to avoid it to be called again.

This property is named is_supersuser to avoid conflict with existing is_root function.

machine

Initialize the plumbum machine lazily.

run(args, sudo=False, **kwargs)

Run a command and return self.response_handler(result).

This method is a thin wrapper around Plumbum’s BaseCommand.run method, which is itself a thin wrapper around the standard library’s subprocess.Popen class. See their documentation for detailed usage instructions. See pulp_smash.cli.Client for a usage example.

Parameters:
  • args – Any arguments to be passed to the process (a tuple).
  • sudo – If the command should run as superuser (a boolean).
  • kwargs – Extra named arguments passed to plumbumBaseCommand.run.
class pulp_smash.cli.CompletedProcess(args, returncode, stdout, stderr)

Bases: object

A process that has finished running.

This class is similar to the subprocess.CompletedProcess class available in Python 3.5 and above. Significant differences include the following:

  • All constructor arguments are required.
  • check_returncode() returns a custom exception, not subprocess.CalledProcessError.

All constructor arguments are stored as instance attributes.

Parameters:
  • args – A string or a sequence. The arguments passed to pulp_smash.cli.Client.run().
  • returncode – The integer exit code of the executed process. Negative for signals.
  • stdout – The standard output of the executed process.
  • stderr – The standard error of the executed process.
check_returncode()

Raise an exception if returncode is non-zero.

Raise pulp_smash.exceptions.CalledProcessError if returncode is non-zero.

Why not raise subprocess.CalledProcessError? Because stdout and stderr are not included when str() is called on a CalledProcessError object. A typical message is:

"Command '('ls', 'foo')' returned non-zero exit status 2"

This information is valuable. One could still make subprocess.CalledProcessError work by overloading args:

>>> if isinstance(args, (str, bytes)):
...     custom_args = (args, stdout, stderr)
... else:
...     custom_args = tuple(args) + (stdout, stderr)
>>> subprocess.CalledProcessError(args, returncode)

But this seems like a hack.

In addition, it’s generally good for an application to raise expected exceptions from its own namespace, so as to better abstract away dependencies.

class pulp_smash.cli.GlobalServiceManager(cfg)

Bases: pulp_smash.cli.BaseServiceManager

A service manager that manages services on all Pulp hosts.

Each instance of this class manages a single service. When a method like start() is executed, it will start a service on all hosts that are declared as running that service. For example, imagine that the following is executed:

>>> from pulp_smash import cli, config
>>> cfg = config.get_config()
>>> svc_mgr = cli.GlobalServiceManager(cfg)
>>> svc_mgr.start(['httpd'])

In this case, the service manager will iterate over all hosts in cfg. For each host that is declared as fulfilling the api role, Apache (httpd) will be restarted.

When asked to perform an action, this object may talk to each target host and determines whether it is running as root. If not root, all commands are prefixed with “sudo”. Please ensure that Pulp Smash can either execute commands as root or can successfully execute “sudo”. You may need to edit your ~/.ssh/config file.

For conceptual information on why both a pulp_smash.cli.ServiceManager and a pulp_smash.cli.GlobalServiceManager are necessary, see pulp_smash.config.PulpSmashConfig.

Parameters:cfg (pulp_smash.config.PulpSmashConfig) – Information about the Pulp deployment.
Raises:pulp_smash.exceptions.NoKnownServiceManagerError – If unable to find any service manager on one of the target hosts.
get_client(pulp_host, **kwargs)

Get an already instantiated client from cache.

is_active(services)

Check whether given services are active.

Parameters:services – A list or tuple of services to check.
Returns:boolean
restart(services)

Restart the services on every host that has the services.

Parameters:services – An iterable of service names.
Returns:A dict mapping the affected hosts’ hostnames with a list of pulp_smash.cli.CompletedProcess objects.
start(services)

Start the services on every host that has the services.

Parameters:services – An iterable of service names.
Returns:A dict mapping the affected hosts’ hostnames with a list of pulp_smash.cli.CompletedProcess objects.
stop(services)

Stop the services on every host that has the services.

Parameters:services – An iterable of service names.
Returns:A dict mapping the affected hosts’ hostnames with a list of pulp_smash.cli.CompletedProcess objects.
class pulp_smash.cli.PackageManager(cfg, raise_if_unsupported=None)

Bases: object

A package manager on a host.

Each instance of this class represents the package manager on a host. An example may help to clarify this idea:

>>> from pulp_smash import cli, config
>>> pkg_mgr = cli.PackageManager(config.get_config())
>>> completed_process = pkg_mgr.install('vim')
>>> completed_process = pkg_mgr.uninstall('vim')

In the example above, the pkg_mgr object represents the package manager on the host referenced by pulp_smash.config.get_config().

Upon instantiation, a PackageManager object talks to its target host and uses simple heuristics to determine which package manager is used. As a result, it’s possible to manage packages on heterogeneous host with homogeneous commands.

Upon instantiation, this object talks to the target host and determines whether it is running as root. If not root, all commands are prefixed with “sudo”. Please ensure that Pulp Smash can either execute commands as root or can successfully execute “sudo”. You may need to edit your ~/.ssh/config file.

Parameters:
  • cfg (pulp_smash.config.PulpSmashConfig) – Information about the target host.
  • raise_if_unsupported (tuple) –

    a tuple of Exception and optional string message to force raise_if_unsupported on initialization:

    pm = PackageManager(cfg, (unittest.SkipTest, 'Test requires yum'))
    # will raise and skip if unsupported package manager
    

    The optional is calling pm.raise_if_unsupported explicitly.

apply_erratum(erratum)

Dispatch to proper _{self.name}_apply_erratum.

install(*args)

Install the named packages.

Return type:pulp_smash.cli.CompletedProcess
name

Return the name of the Package Manager.

raise_if_unsupported(exc, message='Unsupported package manager')

Check if the package manager is supported else raise exc.

Use case:

pm = PackageManager(cfg)
pm.raise_if_unsupported(unittest.SkipTest, 'Test requires yum/dnf')
# will raise and skip if not yum or dnf
pm.install('foobar')
uninstall(*args)

Uninstall the named packages.

Return type:pulp_smash.cli.CompletedProcess
upgrade(*args)

Upgrade the named packages.

Return type:pulp_smash.cli.CompletedProcess
class pulp_smash.cli.RegistryClient(cfg, raise_if_unsupported=None, pulp_host=None)

Bases: object

A container registry client on test runner machine.

Each instance of this class represents the registry client on a host. An example may help to clarify this idea:

>>> from pulp_smash import cli, config
>>> registry = cli.RegistryClient(config.get_config())
>>> image = registry.pull('image_name')

In the example above, the registry object represents the client on the host where pulp-smash is running the test cases.

Upon instantiation, a RegistryClient object talks to its target host and uses simple heuristics to determine which registry client is used.

Upon instantiation, this object determines whether it is running as root. If not root, all commands are prefixed with “sudo”. Please ensure that Pulp Smash can either execute commands as root or can successfully execute “sudo” on the localhost.

Note

When running against a non-https registry the client config insecure-registries must be enabled.

For docker it is located in /etc/docker/daemon.json and content is:

{"insecure-registries": ["pulp_host:8080"]}

For podman it is located in /etc/containers/registries.conf with:

[registries.insecure]
registries = ['pulp_host:8080']
Parameters:
  • cfg (pulp_smash.config.PulpSmashConfig) – Information about the target host.
  • raise_if_unsupported (tuple) –

    a tuple of Exception and optional string message to force raise_if_unsupported on initialization:

    rc = RegistryClient(cfg, (unittest.SkipTest, 'Test requires podman'))
    # will raise and skip if unsupported package manager
    

    The optional is calling rc.raise_if_unsupported explicitly.

  • pulp_host – The host where the Registry Client will run, by default it is set to None and then the same machine where tests are executed will be assumed.
images(*args)

List all pulled images.

import_(*args)

Import a container as a file in to the registry.

inspect(*args)

Inspect metadata for pulled image.

login(*args)

Authenticate to a registry.

logout(*args)

Logs out of a registry.

name

Return the name of the Registry Client.

pull(*args)

Pulls image from registry.

raise_if_unsupported(exc, message='Unsupported registry client')

Check if the registry client is supported else raise exc.

Use case:

rc = RegistryClient(cfg)
rc.raise_if_unsupported(unittest.SkipTest, 'Test requires podman')
# will raise and skip if not podman or docker
rc.pull('busybox')
rmi(*args)

removes pulled image.

class pulp_smash.cli.ServiceManager(cfg, pulp_host)

Bases: pulp_smash.cli.BaseServiceManager

A service manager on a host.

Each instance of this class represents the service manager on a host. An example may help to clarify this idea:

>>> from pulp_smash import cli, config
>>> cfg = config.get_config()
>>> pulp_host = cfg.get_services(('api',))[0]
>>> svc_mgr = cli.ServiceManager(cfg, pulp_host)
>>> completed_process_list = svc_mgr.stop(['httpd'])
>>> completed_process_list = svc_mgr.start(['httpd'])

In the example above, svc_mgr represents the service manager (such as SysV or systemd) on a host. Upon instantiation, a ServiceManager object talks to its target host and uses simple heuristics to determine which service manager is available. As a result, it’s possible to manage services on heterogeneous hosts with homogeneous commands.

Upon instantiation, this object talks to the target host and determines whether it is running as root. If not root, all commands are prefixed with “sudo”. Please ensure that Pulp Smash can either execute commands as root or can successfully execute “sudo”. You may need to edit your ~/.ssh/config file.

For conceptual information on why both a pulp_smash.cli.ServiceManager and a pulp_smash.cli.GlobalServiceManager are necessary, see pulp_smash.config.PulpSmashConfig.

Parameters:
Raises:

pulp_smash.exceptions.NoKnownServiceManagerError – If unable to find any service manager on the target host.

is_active(services)

Check whether given services are active.

Parameters:services – A list or tuple of services to check.
Returns:boolean
restart(services)

Restart the given services.

Parameters:services – An iterable of service names.
Returns:An iterable of pulp_smash.cli.CompletedProcess objects.
start(services)

Start the given services.

Parameters:services – An iterable of service names.
Returns:An iterable of pulp_smash.cli.CompletedProcess objects.
stop(services)

Stop the given services.

Parameters:services – An iterable of service names.
Returns:An iterable of pulp_smash.cli.CompletedProcess objects.
pulp_smash.cli.code_handler(completed_proc)

Check the process for a non-zero return code. Return the process.

Check the return code by calling completed_proc.check_returncode(). See: pulp_smash.cli.CompletedProcess.check_returncode().

pulp_smash.cli.echo_handler(completed_proc)

Immediately return completed_proc.

pulp_smash.cli.is_root(cfg, pulp_host=None)

Tell if we are root on the target host.

Parameters:
Returns:

Either True or False.