Source code for gerritssh.gerritsite

r'''
Representations of a Gerrit site, and the abstract base class for all
executable commands.

All sessions with a Gerrit instance begin by creating a `Site` object and
establishing a connection::

    import gerritssh as gssh
    mysite = gssh.Site('gerrit.exmaple.com').connect()

Commands are executed by creating an instance of a `SiteCommand` concrete
class, and then executing them against the `Site` object. The following
snippet will connect to a site, and then print out a list of all open
reviews by project::

    import gerritssh as gssh
    mysite = gssh.Site('gerrit.example.org').connect()
    lsprojects = gssh.ProjectList()
    lsprojects.execute_on(mysite)

    for p in lsprojects:
        openp = gssh.open_reviews(project=p).execute_on(mysite)
        if not openp:
            continue

        print('\n{0}\n{1}\n'.format(p, '=' * len(p)))
        for r in openp:
            print('{0}\t{1}\n'.format(r.ref, r.summary))


This example also shows both ways of iterating over the results from
executing a command. The line ``for p in lsprojects`` iterates
directly over the `ProjectList` object, while the line ``for r in openp:``
iterates over the list of results returned by calling `execute_on`

.. note::
    This module was original called simply 'site', but that clashed
    with the built in site module which is automatically imported during
    initialization. This lead to strange failures during runs of ``tox`` that
    took a little while to debug.

'''

import logging
import re
import json
import collections
import abc

import semantic_version as SV

from gerritssh import GerritsshException
from gerritssh.borrowed import ssh
from gerritssh.internal.cmdoptions import *  # noqa


_logger = logging.getLogger(__name__)


[docs]class SSHConnectionError(GerritsshException): ''' Raised when a Site object fails to connect via SSH, or when a method is called on an unconnected site which requires a connection. ''' pass
[docs]class InvalidCommandError(GerritsshException): ''' Raised when an attempt is made to execute a BaseCommand object. ''' pass
[docs]class Site(object): ''' An individual Gerrit site. An object of this class manages all access, execution of commands, etc. :param str sitename: The top level URL for the site, e.g. 'gerrit.example.com' :param username: The optional user to log in as :param port: The optional port to connect on :param keyfile: The optional file containing the SSH key to use :raises: TypeError if sitename is not a string Usage:: import gerritssh try: mysite = gerritssh.Site('gerrit.example.com').connect() msg = 'Connected to {}, running Gerrit version {}' print(msg.format(mysite.site, mysite.version) except gerritssh.SSHConnectionError: print('Failed to connect to site '+mysite.site) ''' def __init__(self, sitename, username=None, port=None, keyfile=None): if not isinstance(sitename, str): raise TypeError('sitename must be a string') self.__init_args = (sitename, username, port, keyfile) self.__site = sitename self.__ssh_prefix = 'gerrit' self.__version = SV.Version('0.0.0') self.__keyfile = keyfile self.__ssh = ssh.GerritSSHClient(sitename, username, port, keyfile) def __repr__(self): ''' String representation of the instance ''' return ('<gerritssh.gerritsite.Site(args=%s, connected=%s)>' % (self.__init_args, self.connected))
[docs] def copy(self): ''' Construct an unconnected copy of this Site. This can be used to create additional Site instances from one which has already been initialized. An obvious use would be to open multiple connections to the same site, from a point in the code which is not aware of the initial values used to identify the Gerrit instance. for example:: # In the command line parsing module site = gerritssh.Site(sitename, username, port).connect() ... # In another module: def new_connection(asite): return asite.copy().connect() This method is aliased by __copy__ and __deepcopy__ and allows Site objects to be safely used with copy.copy() and copy.deepCopy() ''' _logger.debug('copy<%s>' % self) return Site(*self.__init_args) # Alias the magic methods used by the copy module
__copy__ = copy def __deepcopy__(self, memo): ''' Deep copies are potentially dangerous. In this case we simply return a new, shallow copy ''' _logger.debug('deepcopy<%s, %s>' % (self, memo)) return self.copy() def __extract_version(self, vstr): results = re.search(r'gerrit version (\d+\.\d+\.\d+).*$', vstr) ver = results.groups()[0] if results else '0.0.0' return ver def __do_command(self, command, args=''): ''' Private method to actually execute a command :returns [str]: The output from the command as a list of strings :raises: :exc: `SSHException` if the command fails ''' cmdline = '{0} {1} {2}'.format(self.__ssh_prefix, command, args) _logger.debug('Site Executing: %s' % cmdline) result = self.__ssh.execute(cmdline) _logger.debug('Command Response:%s' % repr(result)) # return result if isinstance(result, str) else result.decode('utf-8') retval = [l for s in result.stdout.readlines() for l in s.splitlines()] retval = [(l if isinstance(l, str) else str(l.decode('utf-8'))) for l in retval] _logger.debug('Returning:{0}'.format(retval)) return retval
[docs] def connect(self): ''' Establish an SSH connection to the site :returns: self to allow chaining :raises: `SSHConnectionError` if it is not possible to connect to the site ''' _logger.debug('Attempting to connect to site: {0}'.format(self.site)) if self.connected: _logger.debug('Already connected') return try: resp = self.__do_command('version') except ssh.SSHException as e: _logger.debug('Failed to connect: ' + str(e.args)) raise SSHConnectionError('Failed to connect to ' + self.site) _logger.debug('Connected OK: resp: {0}'.format(resp[0].strip())) self.__version = SV.Version(self.__extract_version(resp[0])) return self
[docs] def disconnect(self): ''' Terminate the connection to the site :returns: self to allow chaining ''' _logger.debug('Disconnecting from ' + self.site) self.__ssh.disconnect() return self
[docs] def execute(self, cmd): ''' Execute a command and return the results :param cmd: The command to execute as either a `SiteComand` object, or a string. If a SiteCommand object is passed in, double-dispatch is used to evaluate the command. Otherwise, the string is treated as a valid command string and executed directly. :returns [str]: A list of stripped strings containing the output of the command. :raises: `InvalidCommandError` if the cmd object does not report a valid command :raises: `SSHConnectionError` if there is no current connection to the site :raises: `CalledProcessError` if the command returns an error ''' if not self.connected: _logger.debug('Attempted to execute command without a connection') raise SSHConnectionError('No connection') if isinstance(cmd, SiteCommand): return cmd.execute_on(self) elif cmd and not isinstance(cmd, str): _logger.debug('Invalid argument to cmd. Got type ' + str(type(cmd))) raise InvalidCommandError(('Expected an instance of SiteCommand,' ' or a string. Got ') + str(type(cmd))) if not cmd: _logger.debug('No command found') raise InvalidCommandError('No command found') return self.__do_command(cmd)
@property
[docs] def site(self): ''' The original site name provided to the constructor This needs to be an immutable attribute of the instance once it is created,hence the definition of a 'read-only' property. ''' return self.__site
@property
[docs] def version(self): ''' After connection, provides the version of Gerrit running on the site. :returns: A semantic_version Version object containing the values extracted from the response to a 'gerrit version' command. Before connecting, or if a valid version number can not be found in the response from Gerrit, it has the value '0.0.0'. This needs to be an immutable attribute of the instance once it is created,hence the definition of a 'read-only' property. ''' return self.__version
[docs] def version_in(self, constraint): ''' Does the site's version match a constraint specifier. Client's are free to roll their own tests, but this method makes it unnecessary for them to actually import the semantic_version module directly. :param str constraint: A requirement specification conforming to the Semantic Version package's documentation at http://pythonhosted.org/semantic_version/#requirement-specification :returns: True if the site's version satisfies the requirement. :raises: SSHConnectionError if there is no active connection Usage:: s=Site('example.gerrit.com').connect() # Check that the site is running Gerrit 2.5 or later s.version_in('>=2.5') # Check that the site is running 2.6, 2.7, or 2.8 s.version_in('>=2.6,<2.9') ''' if not self.connected: _logger.debug('Attempt to get version of unconnected site') raise SSHConnectionError('Site is not connected') spec = SV.Spec(constraint) return self.version in spec
@property
[docs] def connected(self): ''' Indicates if there is a connection active. ''' return self.__ssh.connected # The unusual class definition is a version-agnostic means # of setting the metaclass attribute for the class. It creates, at runtime, # a temporary class 'newbase' with a meta-class of ABCMeta and base class of # object which is then used as a the base class for SiteCommand, allowing it # to inherit the metaclass. See the documentation of type() for details.
[docs]class SiteCommand(abc.ABCMeta('newbase', (object,), {})): ''' Base class for a command to be executed against a :class:`Site` This is not meant to be used directly by clients. Instead is allows for duck-typing of sub-classes representing the various Command Line tools supported by Gerrit. Clients can use this to support commands which are missing from the release version of gerritssh or to create macro commands. The key method to override is :meth:`execute_on` which uses the provided Site object to actually implement the command. The method returns its results, usually as an iterable. The parameters for the command are meant to be provided to the constructor of the concrete class. On completion, the `execute_on` method should store its results in self._results as an iterable object, to allow iteration over the object. The constructor creates a parser and parses the provided options string, storing the results in self._parsed_options. For example, a minimal class to represent the ls-projects command, with the response in JSON format (on a site which supports it) might declare the method as follows:: class JSONProjects(SiteCommand): def __init__(self): super(JSONProjects,self).__init__() def execute_on(self, site): self._results = site.execute('ls-projects') return self._results and be used thus:: site=Site('gerrit.example.com').connect() cmd=JSONProjects() projects = cmd.execute_on(site) or, the command object can be passed to the site (useful in building macro operations):: site=Site('gerrit.example.com').connect() cmd=JSONProjects() projects = site.execute(cmd) Either way, providing the SiteCommand class sets _results properly, the caller can then iterate over the results in two ways. By directly iterating over the returned value:: projects = site.execute(cmd) for p in projects: pass or, by iterating over the command object itself:: site.execute(cmd) # or cmd.execute_on(site) for p in cmd: pass :param cmd_supported_in: The semantic-version compliant version specification defining which Gerrit versions support the command. If not specified, or specified as None, defaults to '>=2.4' :param option_set: An OptionSet instance defining the options supported by the command If not specfied, or given as None, no options will be parsed. You can not provide an `option_str` argument without an `OptionSet`. :param option_str: The options to be parsed and passed to the Gerrit site. Defaults to '' if omitted or given as None. :raises: `TypeError` if `option_set` is not an OptionSet instance :raises: `ValueError` if `option_str` is specified without an `option_set` :raises: `SystemExit` if the option_str fails to parse ''' def __init__(self, cmd_supported_in, option_set, option_str): self._results = [] self._options = option_set self._option_str = option_str self._supported_in = cmd_supported_in or '>=2.4' if option_set: if not isinstance(option_set, OptionSet): _logger.debug('Invalid type for option_set argument:' + str(type(option_set))) raise TypeError('Invalid type for option_set argument') self._parser = CmdOptionParser(option_set) self._parsed_options = self._parser.parse(option_str or '') else: self._parser = None self._parsed_options = None if option_str: _logger.debug('Options string provided without OptionSet') raise ValueError('Options string provided without OptionSet') def __iter__(self): ''' Allows iteration through the results of the last command execution ''' return iter(self._results) @property
[docs] def results(self): ''' Results from the most recent execution ''' return self._results
[docs] def check_support_for(self, site): ''' Validate that the provided site supports the command and all provided options. :param site: A connected `Site` instance :raises: `NotImplementedError` If the command or one of the options are unsupported on the site. ''' not_supported = ( 'Gerrit version {0} does not support '.format(site.version) ) if not site.version_in(self._supported_in): _logger.debug('Command not supported') raise NotImplementedError(not_supported + 'this command') if self._parsed_options: if not self._parsed_options.supported_in(site.version): _logger.debug('Some options not supported in: ' + str(self._parsed_options)) raise NotImplementedError(not_supported + 'one or more options provided')
@abc.abstractmethod
[docs] def execute_on(self, the_site): ''' Execute the command on the given site This method must be overridden in concrete classes, and thus the base class implementation is guaranteed to raise an exception. :raises: TypeError '''
@staticmethod
[docs] def text_to_list(text_or_list, nonempty=False): r''' Split a single string containing embedded newlines into a list of trimmed strings Useful for cleaning up multi-line output from commands. Note that a list of strings (perhaps the output from multiple commands) will be flattened to a single list. :param str text_or_list: Either a string with embedded newlines, or a list or tuple of strings with embedded newlines. :param bool nonempty: If true, all empty lines will be removed from the output. :returns [str]: List of stripped strings, one string per embedded line. :raises: :exc:`TypeError` if `text_or_list` contains anything other than strings. Usage:: >>> SiteCommand.text_to_list('a\n \nb') ['a', '', 'b'] >>> SiteCommand.text_to_list('a\n \nb\n', True) ['a', 'b'] >>> SiteCommand.text_to_list(['a\nb','c\nd']) ['a','b','c','d'] ''' if isinstance(text_or_list, str): text_or_list = [text_or_list] if not (isinstance(text_or_list, collections.Iterable) and all([isinstance(x, str) for x in text_or_list])): raise TypeError('Argument must be a string or list of strings') stripempty = lambda x: x if nonempty else True stripped_list = [l.strip() for s in text_or_list for l in s.splitlines()] return [l for l in stripped_list if stripempty(l)]
@staticmethod
[docs] def text_to_json(text_or_list): ''' Convert one or more JSON strings to a list of dictionaries. Every string is split and stripped (via :meth:`text_to_list`) before decoding. All empty strings (or substrings) are ignored. :param text_or_list: Either a single string, or a list of strings to be interpreted as JSON. :returns [dict]: A list of dictionaries, one per string, produced by interpreting each string JSON. :raises: :exc:`TypeError` if text_or_list` is not one or more strings or if one of the strings can't be decoded as valid JSON. ''' if isinstance(text_or_list, str): text_or_list = SiteCommand.text_to_list(text_or_list, nonempty=True) if not (isinstance(text_or_list, collections.Iterable) and all([isinstance(x, str) for x in text_or_list])): raise TypeError('Argument must be one or more strings') jstrings = [s for l in text_or_list for s in SiteCommand.text_to_list(l, nonempty=True)] return [json.loads(s) for s in [_f for _f in jstrings if _f]]
__all__ = ['Site', 'SSHConnectionError', 'InvalidCommandError', 'SiteCommand']