1467

pending completion

Build # 1467

Build Type

Pull #624

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/PulpQE/pulp-smash/commit/12fc82457">Re-enable sync &amp; publish Docker tests

The tests in `pulp_smash/tests/docker/` talk directly to the pulp_docker
plugin. This isn&#39;t a good idea. It is assumed that clients such as
Docker will talk to the API exposed by Crane, and as a result,
pulp_docker changes more frequently and in more significant ways than is
typical for other plugins. In addition, testing pulp_docker directly
means that Pulp Smash isn&#39;t verifying the behaviour that real clients
see.

Re-write two test cases (resulting in `SyncPublishV1TestCase` and
`SyncPublishV1TestCase`) so that they communicate with the interface
exposed by Crane, not the interface exposed by pulp_docker.

See: </a><a class="double-link" href="https://github.com/PulpQE/pulp-smash/commit/b470bfb59">b470bfb59</a>

Pull Request Pull Request #624: DO NOT MERGE: Re-enable sync & publish Docker tests

Run Details

19 of 19 new or added lines in 1 file covered. (100.0%)

389 of 628 relevant lines covered (61.94%)

1.24 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

42.58

/pulp_smash/cli.py

# coding=utf-8
"""A client for working with Pulp systems via their CLI."""
import contextlib
import os
import socket
from urllib.parse import urlparse

import plumbum

from pulp_smash import exceptions


# A dict mapping hostnames to *nix service managers.
#
# For example: {'old.example.com': 'sysv', 'new.example.com', 'systemd'}
_SERVICE_MANAGERS = {}

# A dict mapping hostnames to *nix package managers.
#
# For example: {'old.example.com': 'yum', 'new.example.com', 'yum'}
_PACKAGE_MANAGERS = {}


def _get_hostname(urlstring):
    """Get the hostname from a URL string.

    ``urlparse`` follows RFC 1808 and requires that network locations be
    prefixed with "//". This function is a thin wrapper. It treats the leading
    "//" as optional::

    >>> urls = ('ftp://localhost', '//localhost', 'localhost', 'localhost:123')
    >>> for url in urls:
    ...     print((_get_hostname(url), urlparse(url).hostname))
    ('localhost', 'localhost')
    ('localhost', 'localhost')
    ('localhost', None)
    ('localhost', None)

    Usage::

        if server_config is None:
            server_config = get_config()
        hostname = _get_hostname(server_config.base_url)

    :param urlstring: A string such as "localhost:3000", "pulp.example.com" or
        "https://pulp.example.com".
    :returns: A hostname, such as "localhost" or "pulp.example.com".

    """
    parts = urlparse(urlstring)
    if parts.hostname is None:
        return _get_hostname('//' + parts.path)
    return parts.hostname


def _is_root(cfg):
    """Tell if we are root on the target system.

    :param pulp_smash.config.ServerConfig cfg: Information about the target
        system.
    :returns: Either ``True`` or ``False``.
    """
    if Client(cfg).run(('id', '-u')).stdout.strip() == '0':
        return True
    return False


def echo_handler(completed_proc):
    """Immediately return ``completed_proc``."""
    return completed_proc


def 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: :meth:`pulp_smash.cli.CompletedProcess.check_returncode`.
    """
    completed_proc.check_returncode()
    return completed_proc


class CompletedProcess(object):
    # pylint:disable=too-few-public-methods
    """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.
    * :meth:`check_returncode` returns a custom exception, not
      ``subprocess.CalledProcessError``.

    All constructor arguments are stored as instance attributes.

    :param args: A string or a sequence. The arguments passed to
        :meth:`pulp_smash.cli.Client.run`.
    :param returncode: The integer exit code of the executed process. Negative
        for signals.
    :param stdout: The standard output of the executed process.
    :param stderr: The standard error of the executed process.
    """

    def __init__(self, args, returncode, stdout, stderr):
        """Initialize a new object."""
        self.args = args
        self.returncode = returncode
        self.stdout = stdout
        self.stderr = stderr

    def __repr__(self):
        """Provide an ``eval``-compatible string representation."""
        str_kwargs = ', '.join([
            'args={!r}'.format(self.args),
            'returncode={!r}'.format(self.returncode),
            'stdout={!r}'.format(self.stdout),
            'stderr={!r}'.format(self.stderr),
        ])
        return '{}({})'.format(type(self).__name__, str_kwargs)

    def check_returncode(self):
        """Raise an exception if ``returncode`` is non-zero.

        Raise :class:`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.
        """
        if self.returncode != 0:
            raise exceptions.CalledProcessError(
                self.args,
                self.returncode,
                self.stdout,
                self.stderr,
            )


class Client(object):  # pylint:disable=too-few-public-methods
    """A convenience object for working with a CLI.

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

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

    You can customize how commands are executed and how responses are handled
    by fiddling with the two public instance attributes:

    ``machine``
        A `Plumbum`_ machine. :meth:`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 ``server_config.cli_transport`` is ``'local'`` or ``'ssh``, set
    ``machine`` so that commands run locally or over SSH, respectively. If
    ``server_config.cli_transport`` is ``None``, guess how to set ``machine``
    by comparing the hostname embedded in ``server_config.base_url`` against
    the current system's hostname. If they match, set ``machine`` to execute
    commands locally; and vice versa.

    :param pulp_smash.config.ServerConfig server_config: Information about the
        system on which commands will be executed.
    :param response_handler: A callback function. Defaults to
        :func:`pulp_smash.cli.code_handler`.

    .. _Plumbum: http://plumbum.readthedocs.io/en/latest/index.html
    """

    def __init__(self, server_config, response_handler=None):
        """Initialize this object with needed instance attributes."""
        # How do we make requests?
        hostname = _get_hostname(server_config.base_url)
        if server_config.cli_transport is None:
            transport = 'local' if hostname == socket.getfqdn() else 'ssh'
        else:
            transport = server_config.cli_transport
        if transport == 'local':
            self.machine = plumbum.machines.local
        else:  # transport == 'ssh'
            # The SshMachine is a wrapper around the system's "ssh" binary.
            # Thus, it uses ~/.ssh/config, ~/.ssh/known_hosts, etc.
            self.machine = (
                plumbum.machines.SshMachine(hostname)
            )

        # How do we handle responses?
        if response_handler is None:
            self.response_handler = code_handler
        else:
            self.response_handler = response_handler

    def run(self, args, **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 :class:`pulp_smash.cli.Client` for a usage example.

        .. _BaseCommand.run:
           http://plumbum.readthedocs.io/en/latest/api/commands.html#plumbum.commands.base.BaseCommand.run
        .. _subprocess.Popen:
           https://docs.python.org/3/library/subprocess.html#subprocess.Popen
        """
        # Let self.response_handler check return codes. See:
        # https://plumbum.readthedocs.io/en/latest/api/commands.html#plumbum.commands.base.BaseCommand.run
        kwargs.setdefault('retcode')

        code, stdout, stderr = self.machine[args[0]].run(args[1:], **kwargs)
        completed_process = CompletedProcess(args, code, stdout, stderr)
        return self.response_handler(completed_process)


class ServiceManager(object):
    """A service manager on a system.

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

    from pulp_smash import cli, config
    >>> svc_mgr = cli.ServiceManager(config.get_config())
    >>> completed_process = svc_manager.stop(['httpd'])
    >>> completed_process = svc_manager.start(['httpd'])

    In the example above, the ``svc_mgr`` object represents the service manager
    on the host referenced by :func:`pulp_smash.config.get_config`.

    Upon instantiation, a :class:`ServiceManager` object talks to its target
    system and uses simple heuristics to determine which service manager is
    available. As a result, it's possible to manage services on heterogeneous
    systems with homogeneous commands.

    Upon instantiation, a :class:`ServiceManager` object also talks to its
    target system 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.

    :param pulp_smash.config.ServerConfig cfg: Information about the target
        system.
    :raises pulp_smash.exceptions.NoKnownServiceManagerError: If unable to find
        any service manager on the target system.
    """

    def __init__(self, cfg):
        """Initialize a new object."""
        self._client = Client(cfg)
        self._sudo = () if _is_root(cfg) else ('sudo',)
        self._svc_mgr = self._get_service_manager(cfg)
        self._on_jenkins = 'JENKINS_HOME' in os.environ

    @staticmethod
    def _get_service_manager(server_config):
        """Talk to the target system and determine the type of service manager.

        Return "systemd" or "sysv" if the service manager appears to be one of
        those. Raise an exception otherwise.
        """
        hostname = _get_hostname(server_config.base_url)
        try:
            return _SERVICE_MANAGERS[hostname]
        except KeyError:
            pass

        client = Client(server_config, echo_handler)
        commands_managers = (
            ('which systemctl', 'systemd'),
            ('which service', 'sysv'),
            ('test -x /sbin/service', 'sysv'),
        )
        for command, service_manager in commands_managers:
            if client.run(command.split()).returncode == 0:
                _SERVICE_MANAGERS[hostname] = service_manager
                return service_manager
        raise exceptions.NoKnownServiceManagerError(
            'Unable to determine the service manager used by {}. It does not '
            'appear to be any of {}.'
            .format(hostname, {manager for _, manager in commands_managers})
        )

    @contextlib.contextmanager
    def _disable_selinux(self):
        """Context manager to temporary disable SELinux."""
        if self._on_jenkins:
            self._client.run(self._sudo + ('setenforce', '0'))
        yield
        if self._on_jenkins:
            self._client.run(self._sudo + ('setenforce', '1'))

    def start(self, services):
        """Start the given services.

        :param services: An iterable of service names.
        :return: An iterable of :class:`pulp_smash.cli.CompletedProcess`
            objects.
        """
        if self._svc_mgr == 'sysv':
            with self._disable_selinux():
                return self._start_sysv(services)
        elif self._svc_mgr == 'systemd':
            return self._start_systemd(services)
        else:
            raise NotImplementedError(
                'Service manager not supported: {}'.format(self._svc_mgr)
            )

    def _start_sysv(self, services):
        return tuple((
            self._client.run(self._sudo + ('service', service, 'start'))
            for service in services
        ))

    def _start_systemd(self, services):
        cmd = self._sudo + ('systemctl', 'start') + tuple(services)
        return (self._client.run(cmd),)

    def stop(self, services):
        """Stop the given services.

        :param services: An iterable of service names.
        :return: An iterable of :class:`pulp_smash.cli.CompletedProcess`
            objects.
        """
        if self._svc_mgr == 'sysv':
            with self._disable_selinux():
                return self._stop_sysv(services)
        elif self._svc_mgr == 'systemd':
            return self._stop_systemd(services)
        else:
            raise NotImplementedError(
                'Service manager not supported: {}'.format(self._svc_mgr)
            )

    def _stop_sysv(self, services):
        return tuple((
            self._client.run(self._sudo + ('service', service, 'stop'))
            for service in services
        ))

    def _stop_systemd(self, services):
        cmd = self._sudo + ('systemctl', 'stop') + tuple(services)
        return (self._client.run(cmd),)

    def restart(self, services):
        """Restart the given services.

        :param services: An iterable of service names.
        :return: An iterable of :class:`pulp_smash.cli.CompletedProcess`
            objects.
        """
        if self._svc_mgr == 'sysv':
            with self._disable_selinux():
                return self._restart_sysv(services)
        elif self._svc_mgr == 'systemd':
            return self._restart_systemd(services)
        else:
            raise NotImplementedError(
                'Service manager not supported: {}'.format(self._svc_mgr)
            )

    def _restart_sysv(self, services):
        return tuple((
            self._client.run(self._sudo + ('service', service, 'restart'))
            for service in services
        ))

    def _restart_systemd(self, services):
        cmd = self._sudo + ('systemctl', 'restart') + tuple(services)
        return (self._client.run(cmd),)


class PackageManager(object):
    """A package manager on a system.

    Each instance of this class represents the package manager on a system. 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 :func:`pulp_smash.config.get_config`.

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

    Upon instantiation, a :class:`PackageManager` object also talks to its
    target system 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.

    :param pulp_smash.config.ServerConfig cfg: Information about the target
        system.
    :raises pulp_smash.exceptions.NoKnownPackageManagerError: If unable to find
        any package manager on the target system.
    """

    def __init__(self, cfg):
        """Initialize a new object."""
        self._client = Client(cfg)
        self._sudo = () if _is_root(cfg) else ('sudo',)
        self._pkg_mgr = self._get_package_manager(cfg)

    @staticmethod
    def _get_package_manager(cfg):
        """Talk to the target system and determine the package manager.

        Return "dnf" or "yum" if the package manager appears to be one of
        those. Raise an exception otherwise.
        """
        hostname = _get_hostname(cfg.base_url)
        try:
            return _PACKAGE_MANAGERS[hostname]
        except KeyError:
            pass

        client = Client(cfg, echo_handler)
        commands_managers = (
            (('which', 'dnf'), 'dnf'),
            (('which', 'yum'), 'yum'),
        )
        for cmd, pkg_mgr in commands_managers:
            if client.run(cmd).returncode == 0:
                _PACKAGE_MANAGERS[hostname] = pkg_mgr
                return pkg_mgr
        raise exceptions.NoKnownPackageManagerError(
            'Unable to determine the package manager used by {}. It does not '
            'appear to be any of {}.'
            .format(hostname, {pkg_mgr for _, pkg_mgr in commands_managers})
        )

    def install(self, *args):
        """Install the named packages.

        :rtype: pulp_smash.cli.CompletedProcess
        """
        if self._pkg_mgr == 'dnf':
            cmd = self._sudo + ('dnf', '-y', 'install') + tuple(args)
        elif self._pkg_mgr == 'yum':
            cmd = self._sudo + ('yum', '-y', 'install') + tuple(args)
        else:
            cmd = None
        assert cmd is not None
        return self._client.run(cmd)

    def uninstall(self, *args):
        """Uninstall the named packages.

        :rtype: pulp_smash.cli.CompletedProcess
        """
        if self._pkg_mgr == 'dnf':
            cmd = self._sudo + ('dnf', '-y', 'remove') + tuple(args)
        elif self._pkg_mgr == 'yum':
            cmd = self._sudo + ('yum', '-y', 'remove') + tuple(args)
        else:
            cmd = None
        assert cmd is not None
        return self._client.run(cmd)

    def upgrade(self, *args):
        """Upgrade the named packages.

        :rtype: pulp_smash.cli.CompletedProcess
        """
        if self._pkg_mgr == 'dnf':
            cmd = self._sudo + ('dnf', '-y', 'upgrade') + tuple(args)
        elif self._pkg_mgr == 'yum':
            cmd = self._sudo + ('yum', '-y', 'update') + tuple(args)
        else:
            cmd = None
        assert cmd is not None
        return self._client.run(cmd)

1	# coding=utf-8
2	"""A client for working with Pulp systems via their CLI."""	2✔
3	import contextlib	2✔
4	import os	2✔
5	import socket	2✔
6	from urllib.parse import urlparse	2✔
7
8	import plumbum	2✔
9
10	from pulp_smash import exceptions	2✔
11
12
13	# A dict mapping hostnames to *nix service managers.
14	#
15	# For example: {'old.example.com': 'sysv', 'new.example.com', 'systemd'}
16	_SERVICE_MANAGERS = {}	2✔
17
18	# A dict mapping hostnames to *nix package managers.
19	#
20	# For example: {'old.example.com': 'yum', 'new.example.com', 'yum'}
21	_PACKAGE_MANAGERS = {}	2✔
22
23
24	def _get_hostname(urlstring):	2✔
25	"""Get the hostname from a URL string.
26
27	``urlparse`` follows RFC 1808 and requires that network locations be
28	prefixed with "//". This function is a thin wrapper. It treats the leading
29	"//" as optional::
30
31	>>> urls = ('ftp://localhost', '//localhost', 'localhost', 'localhost:123')
32	>>> for url in urls:
33	... print((_get_hostname(url), urlparse(url).hostname))
34	('localhost', 'localhost')
35	('localhost', 'localhost')
36	('localhost', None)
37	('localhost', None)
38
39	Usage::
40
41	if server_config is None:
42	server_config = get_config()
43	hostname = _get_hostname(server_config.base_url)
44
45	:param urlstring: A string such as "localhost:3000", "pulp.example.com" or
46	"https://pulp.example.com".
47	:returns: A hostname, such as "localhost" or "pulp.example.com".
48
49	"""
50	parts = urlparse(urlstring)	2✔
51	if parts.hostname is None:	2✔
52	return _get_hostname('//' + parts.path)	2✔
53	return parts.hostname	2✔
54
55
56	def _is_root(cfg):	2✔
57	"""Tell if we are root on the target system.
58
59	:param pulp_smash.config.ServerConfig cfg: Information about the target
60	system.
61	:returns: Either ``True`` or ``False``.
62	"""
63	if Client(cfg).run(('id', '-u')).stdout.strip() == '0':	2✔
64	return True	2✔
65	return False	2✔
66
67
68	def echo_handler(completed_proc):	2✔
69	"""Immediately return ``completed_proc``."""
70	return completed_proc	2✔
71
72
73	def code_handler(completed_proc):	2✔
74	"""Check the process for a non-zero return code. Return the process.
75
76	Check the return code by calling ``completed_proc.check_returncode()``.
77	See: :meth:`pulp_smash.cli.CompletedProcess.check_returncode`.
78	"""
79	completed_proc.check_returncode()	2✔
80	return completed_proc	2✔
81
82
83	class CompletedProcess(object):	2✔
84	# pylint:disable=too-few-public-methods
85	"""A process that has finished running.
86
87	This class is similar to the ``subprocess.CompletedProcess`` class
88	available in Python 3.5 and above. Significant differences include the
89	following:
90
91	* All constructor arguments are required.
92	* :meth:`check_returncode` returns a custom exception, not
93	``subprocess.CalledProcessError``.
94
95	All constructor arguments are stored as instance attributes.
96
97	:param args: A string or a sequence. The arguments passed to
98	:meth:`pulp_smash.cli.Client.run`.
99	:param returncode: The integer exit code of the executed process. Negative
100	for signals.
101	:param stdout: The standard output of the executed process.
102	:param stderr: The standard error of the executed process.
103	"""
104
105	def __init__(self, args, returncode, stdout, stderr):	2✔
106	"""Initialize a new object."""
107	self.args = args	2✔
108	self.returncode = returncode	2✔
109	self.stdout = stdout	2✔
110	self.stderr = stderr	2✔
111
112	def __repr__(self):	2✔
113	"""Provide an ``eval``-compatible string representation."""
114	str_kwargs = ', '.join([	2✔
115	'args={!r}'.format(self.args),
116	'returncode={!r}'.format(self.returncode),
117	'stdout={!r}'.format(self.stdout),
118	'stderr={!r}'.format(self.stderr),
119	])
120	return '{}({})'.format(type(self).__name__, str_kwargs)	2✔
121
122	def check_returncode(self):	2✔
123	"""Raise an exception if ``returncode`` is non-zero.
124
125	Raise :class:`pulp_smash.exceptions.CalledProcessError` if
126	``returncode`` is non-zero.
127
128	Why not raise ``subprocess.CalledProcessError``? Because stdout and
129	stderr are not included when str() is called on a CalledProcessError
130	object. A typical message is::
131
132	"Command '('ls', 'foo')' returned non-zero exit status 2"
133
134	This information is valuable. One could still make
135	``subprocess.CalledProcessError`` work by overloading ``args``:
136
137	>>> if isinstance(args, (str, bytes)):
138	... custom_args = (args, stdout, stderr)
139	... else:
140	... custom_args = tuple(args) + (stdout, stderr)
141	>>> subprocess.CalledProcessError(args, returncode)
142
143	But this seems like a hack.
144
145	In addition, it's generally good for an application to raise expected
146	exceptions from its own namespace, so as to better abstract away
147	dependencies.
148	"""
149	if self.returncode != 0:	2✔
150	raise exceptions.CalledProcessError(	2✔
151	self.args,
152	self.returncode,
153	self.stdout,
154	self.stderr,
155	)
156
157
158	class Client(object): # pylint:disable=too-few-public-methods	2✔
159	"""A convenience object for working with a CLI.
160
161	This class provides the ability to execute shell commands on either the
162	local system or a remote system. Here is a simple usage example:
163
164	>>> from pulp_smash import cli, config
165	>>> server_config = config.ServerConfig('localhost')
166	>>> client = cli.Client(server_config)
167	>>> response = client.run(('echo', '-n', 'foo'))
168	>>> response.returncode == 0
169	True
170	>>> response.stdout == 'foo'
171	True
172	>>> response.stderr == ''
173	True
174
175	You can customize how commands are executed and how responses are handled
176	by fiddling with the two public instance attributes:
177
178	``machine``
179	A `Plumbum`_ machine. :meth:`run` delegates all command execution
180	responsibilities to this object.
181	``response_handler``
182	A callback function. Each time ``machine`` executes a command, the
183	result is handed to this callback, and the callback's return value is
184	handed to the user.
185
186	If ``server_config.cli_transport`` is ``'local'`` or ``'ssh``, set
187	``machine`` so that commands run locally or over SSH, respectively. If
188	``server_config.cli_transport`` is ``None``, guess how to set ``machine``
189	by comparing the hostname embedded in ``server_config.base_url`` against
190	the current system's hostname. If they match, set ``machine`` to execute
191	commands locally; and vice versa.
192
193	:param pulp_smash.config.ServerConfig server_config: Information about the
194	system on which commands will be executed.
195	:param response_handler: A callback function. Defaults to
196	:func:`pulp_smash.cli.code_handler`.
197
198	.. _Plumbum: http://plumbum.readthedocs.io/en/latest/index.html
199	"""
200
201	def __init__(self, server_config, response_handler=None):	2✔
202	"""Initialize this object with needed instance attributes."""
203	# How do we make requests?
204	hostname = _get_hostname(server_config.base_url)	2✔
205	if server_config.cli_transport is None:	2✔
206	transport = 'local' if hostname == socket.getfqdn() else 'ssh'	2✔
207	else:
208	transport = server_config.cli_transport	2✔
209	if transport == 'local':	2✔
210	self.machine = plumbum.machines.local	2✔
211	else: # transport == 'ssh'
212	# The SshMachine is a wrapper around the system's "ssh" binary.
213	# Thus, it uses ~/.ssh/config, ~/.ssh/known_hosts, etc.
214	self.machine = (	×
215	plumbum.machines.SshMachine(hostname)
216	)
217
218	# How do we handle responses?
219	if response_handler is None:	2✔
220	self.response_handler = code_handler	2✔
221	else:
222	self.response_handler = response_handler	2✔
223
224	def run(self, args, **kwargs):	2✔
225	"""Run a command and ``return self.response_handler(result)``.
226
227	This method is a thin wrapper around Plumbum's `BaseCommand.run`_
228	method, which is itself a thin wrapper around the standard library's
229	`subprocess.Popen`_ class. See their documentation for detailed usage
230	instructions. See :class:`pulp_smash.cli.Client` for a usage example.
231
232	.. _BaseCommand.run:
233	http://plumbum.readthedocs.io/en/latest/api/commands.html#plumbum.commands.base.BaseCommand.run
234	.. _subprocess.Popen:
235	https://docs.python.org/3/library/subprocess.html#subprocess.Popen
236	"""
237	# Let self.response_handler check return codes. See:
238	# https://plumbum.readthedocs.io/en/latest/api/commands.html#plumbum.commands.base.BaseCommand.run
239	kwargs.setdefault('retcode')	×
240
241	code, stdout, stderr = self.machine[args[0]].run(args[1:], **kwargs)	×
242	completed_process = CompletedProcess(args, code, stdout, stderr)	×
243	return self.response_handler(completed_process)	×
244
245
246	class ServiceManager(object):	2✔
247	"""A service manager on a system.
248
249	Each instance of this class represents the service manager on a system. An
250	example may help to clarify this idea:
251
252	from pulp_smash import cli, config
253	>>> svc_mgr = cli.ServiceManager(config.get_config())
254	>>> completed_process = svc_manager.stop(['httpd'])
255	>>> completed_process = svc_manager.start(['httpd'])
256
257	In the example above, the ``svc_mgr`` object represents the service manager
258	on the host referenced by :func:`pulp_smash.config.get_config`.
259
260	Upon instantiation, a :class:`ServiceManager` object talks to its target
261	system and uses simple heuristics to determine which service manager is
262	available. As a result, it's possible to manage services on heterogeneous
263	systems with homogeneous commands.
264
265	Upon instantiation, a :class:`ServiceManager` object also talks to its
266	target system and determines whether it is running as root. If not root,
267	all commands are prefixed with "sudo". Please ensure that Pulp Smash can
268	either execute commands as root or can successfully execute ``sudo``. You
269	may need to edit your ``~/.ssh/config`` file.
270
271	:param pulp_smash.config.ServerConfig cfg: Information about the target
272	system.
273	:raises pulp_smash.exceptions.NoKnownServiceManagerError: If unable to find
274	any service manager on the target system.
275	"""
276
277	def __init__(self, cfg):	2✔
278	"""Initialize a new object."""
279	self._client = Client(cfg)	×
280	self._sudo = () if _is_root(cfg) else ('sudo',)	×
281	self._svc_mgr = self._get_service_manager(cfg)	×
282	self._on_jenkins = 'JENKINS_HOME' in os.environ	×
283
284	@staticmethod	2✔
285	def _get_service_manager(server_config):
286	"""Talk to the target system and determine the type of service manager.
287
288	Return "systemd" or "sysv" if the service manager appears to be one of
289	those. Raise an exception otherwise.
290	"""
291	hostname = _get_hostname(server_config.base_url)	×
292	try:	×
293	return _SERVICE_MANAGERS[hostname]	×
294	except KeyError:	×
295	pass	×
296
297	client = Client(server_config, echo_handler)	×
298	commands_managers = (	×
299	('which systemctl', 'systemd'),
300	('which service', 'sysv'),
301	('test -x /sbin/service', 'sysv'),
302	)
303	for command, service_manager in commands_managers:	×
304	if client.run(command.split()).returncode == 0:	×
305	_SERVICE_MANAGERS[hostname] = service_manager	×
306	return service_manager	×
307	raise exceptions.NoKnownServiceManagerError(	×
308	'Unable to determine the service manager used by {}. It does not '
309	'appear to be any of {}.'
310	.format(hostname, {manager for _, manager in commands_managers})
311	)
312
313	@contextlib.contextmanager	2✔
314	def _disable_selinux(self):
315	"""Context manager to temporary disable SELinux."""
316	if self._on_jenkins:	×
317	self._client.run(self._sudo + ('setenforce', '0'))	×
318	yield	×
319	if self._on_jenkins:	×
320	self._client.run(self._sudo + ('setenforce', '1'))	×
321
322	def start(self, services):	2✔
323	"""Start the given services.
324
325	:param services: An iterable of service names.
326	:return: An iterable of :class:`pulp_smash.cli.CompletedProcess`
327	objects.
328	"""
329	if self._svc_mgr == 'sysv':	×
330	with self._disable_selinux():	×
331	return self._start_sysv(services)	×
332	elif self._svc_mgr == 'systemd':	×
333	return self._start_systemd(services)	×
334	else:
335	raise NotImplementedError(	×
336	'Service manager not supported: {}'.format(self._svc_mgr)
337	)
338
339	def _start_sysv(self, services):	2✔
340	return tuple((	×
341	self._client.run(self._sudo + ('service', service, 'start'))
342	for service in services
343	))
344
345	def _start_systemd(self, services):	2✔
346	cmd = self._sudo + ('systemctl', 'start') + tuple(services)	×
347	return (self._client.run(cmd),)	×
348
349	def stop(self, services):	2✔
350	"""Stop the given services.
351
352	:param services: An iterable of service names.
353	:return: An iterable of :class:`pulp_smash.cli.CompletedProcess`
354	objects.
355	"""
356	if self._svc_mgr == 'sysv':	×
357	with self._disable_selinux():	×
358	return self._stop_sysv(services)	×
359	elif self._svc_mgr == 'systemd':	×
360	return self._stop_systemd(services)	×
361	else:
362	raise NotImplementedError(	×
363	'Service manager not supported: {}'.format(self._svc_mgr)
364	)
365
366	def _stop_sysv(self, services):	2✔
367	return tuple((	×
368	self._client.run(self._sudo + ('service', service, 'stop'))
369	for service in services
370	))
371
372	def _stop_systemd(self, services):	2✔
373	cmd = self._sudo + ('systemctl', 'stop') + tuple(services)	×
374	return (self._client.run(cmd),)	×
375
376	def restart(self, services):	2✔
377	"""Restart the given services.
378
379	:param services: An iterable of service names.
380	:return: An iterable of :class:`pulp_smash.cli.CompletedProcess`
381	objects.
382	"""
383	if self._svc_mgr == 'sysv':	×
384	with self._disable_selinux():	×
385	return self._restart_sysv(services)	×
386	elif self._svc_mgr == 'systemd':	×
387	return self._restart_systemd(services)	×
388	else:
389	raise NotImplementedError(	×
390	'Service manager not supported: {}'.format(self._svc_mgr)
391	)
392
393	def _restart_sysv(self, services):	2✔
394	return tuple((	×
395	self._client.run(self._sudo + ('service', service, 'restart'))
396	for service in services
397	))
398
399	def _restart_systemd(self, services):	2✔
400	cmd = self._sudo + ('systemctl', 'restart') + tuple(services)	×
401	return (self._client.run(cmd),)	×
402
403
404	class PackageManager(object):	2✔
405	"""A package manager on a system.
406
407	Each instance of this class represents the package manager on a system. An
408	example may help to clarify this idea:
409
410	>>> from pulp_smash import cli, config
411	>>> pkg_mgr = cli.PackageManager(config.get_config())
412	>>> completed_process = pkg_mgr.install('vim')
413	>>> completed_process = pkg_mgr.uninstall('vim')
414
415	In the example above, the ``pkg_mgr`` object represents the package manager
416	on the host referenced by :func:`pulp_smash.config.get_config`.
417
418	Upon instantiation, a :class:`PackageManager` object talks to its target
419	system and uses simple heuristics to determine which package manager is
420	used. As a result, it's possible to manage packages on heterogeneous
421	systems with homogeneous commands.
422
423	Upon instantiation, a :class:`PackageManager` object also talks to its
424	target system and determines whether it is running as root. If not root,
425	all commands are prefixed with "sudo". Please ensure that Pulp Smash can
426	either execute commands as root or can successfully execute ``sudo``. You
427	may need to edit your ``~/.ssh/config`` file.
428
429	:param pulp_smash.config.ServerConfig cfg: Information about the target
430	system.
431	:raises pulp_smash.exceptions.NoKnownPackageManagerError: If unable to find
432	any package manager on the target system.
433	"""
434
435	def __init__(self, cfg):	2✔
436	"""Initialize a new object."""
437	self._client = Client(cfg)	×
438	self._sudo = () if _is_root(cfg) else ('sudo',)	×
439	self._pkg_mgr = self._get_package_manager(cfg)	×
440
441	@staticmethod	2✔
442	def _get_package_manager(cfg):
443	"""Talk to the target system and determine the package manager.
444
445	Return "dnf" or "yum" if the package manager appears to be one of
446	those. Raise an exception otherwise.
447	"""
448	hostname = _get_hostname(cfg.base_url)	×
449	try:	×
450	return _PACKAGE_MANAGERS[hostname]	×
451	except KeyError:	×
452	pass	×
453
454	client = Client(cfg, echo_handler)	×
455	commands_managers = (	×
456	(('which', 'dnf'), 'dnf'),
457	(('which', 'yum'), 'yum'),
458	)
459	for cmd, pkg_mgr in commands_managers:	×
460	if client.run(cmd).returncode == 0:	×
461	_PACKAGE_MANAGERS[hostname] = pkg_mgr	×
462	return pkg_mgr	×
463	raise exceptions.NoKnownPackageManagerError(	×
464	'Unable to determine the package manager used by {}. It does not '
465	'appear to be any of {}.'
466	.format(hostname, {pkg_mgr for _, pkg_mgr in commands_managers})
467	)
468
469	def install(self, *args):	2✔
470	"""Install the named packages.
471
472	:rtype: pulp_smash.cli.CompletedProcess
473	"""
474	if self._pkg_mgr == 'dnf':	×
475	cmd = self._sudo + ('dnf', '-y', 'install') + tuple(args)	×
476	elif self._pkg_mgr == 'yum':	×
477	cmd = self._sudo + ('yum', '-y', 'install') + tuple(args)	×
478	else:
479	cmd = None	×
480	assert cmd is not None	×
481	return self._client.run(cmd)	×
482
483	def uninstall(self, *args):	2✔
484	"""Uninstall the named packages.
485
486	:rtype: pulp_smash.cli.CompletedProcess
487	"""
488	if self._pkg_mgr == 'dnf':	×
489	cmd = self._sudo + ('dnf', '-y', 'remove') + tuple(args)	×
490	elif self._pkg_mgr == 'yum':	×
491	cmd = self._sudo + ('yum', '-y', 'remove') + tuple(args)	×
492	else:
493	cmd = None	×
494	assert cmd is not None	×
495	return self._client.run(cmd)	×
496
497	def upgrade(self, *args):	2✔
498	"""Upgrade the named packages.
499
500	:rtype: pulp_smash.cli.CompletedProcess
501	"""
502	if self._pkg_mgr == 'dnf':	×
503	cmd = self._sudo + ('dnf', '-y', 'upgrade') + tuple(args)	×
504	elif self._pkg_mgr == 'yum':	×
505	cmd = self._sudo + ('yum', '-y', 'update') + tuple(args)	×
506	else:
507	cmd = None	×
508	assert cmd is not None	×
509	return self._client.run(cmd)	×

PulpQE / pulp-smash / 1467

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous