Gerrit-Client

 view release on metacpan or  search on metacpan

lib/Gerrit/Client.pm  view on Meta::CPAN

##
## You should have received a copy of the GNU Lesser General Public
## License along with this library; if not, write to the Free Software
## Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
##
##
#############################################################################

=head1 NAME

Gerrit::Client - interact with Gerrit code review tool

=head1 SYNOPSIS

  use AnyEvent;
  use Gerrit::Client qw(stream_events);

  # alert me when new patch sets arrive in
  # ssh://gerrit.example.com:29418/myproject
  my $stream = stream_events(
    url => 'ssh://gerrit.example.com:29418',
    on_event => sub {
      my ($event) = @_;
      if ($event->{type} eq 'patchset-added'
          && $event->{change}{project} eq 'myproject') {
        system("xmessage", "New patch set arrived!");
      }
    }
  );

  AE::cv()->recv(); # must run an event loop for callbacks to be activated

This module provides some utility functions for interacting with the Gerrit code
review tool.

This module is an L<AnyEvent> user and may be used with any event loop supported
by AnyEvent.

=cut

package Gerrit::Client;
use strict;
use warnings;

use AnyEvent::HTTP;
use AnyEvent::Handle;
use AnyEvent::Util;
use AnyEvent;
use Capture::Tiny qw(capture);
use Carp;
use Data::Alias;
use Data::Dumper;
use Digest::MD5 qw(md5_hex);
use English qw(-no_match_vars);
use File::Path;
use File::Spec::Functions;
use File::Temp;
use File::chdir;
use JSON;
use Params::Validate qw(:all);
use Scalar::Util qw(weaken);
use URI;
use URI::Escape;

use base 'Exporter';
our @EXPORT_OK = qw(
  for_each_patchset
  stream_events
  git_environment
  next_change_id
  random_change_id
  review
  query
  quote
);

our @GIT             = ('git');
our @SSH             = ('ssh');
our $VERSION         = 20140611;
our $DEBUG           = !!$ENV{GERRIT_CLIENT_DEBUG};
our $MAX_CONNECTIONS = 2;
our $MAX_FORKS       = 4;

sub _debug_print {
  return unless $DEBUG;
  print STDERR __PACKAGE__ . ': ', @_, "\n";
}

# parses a gerrit URL and returns a hashref with following keys:
#   cmd => arrayref, base ssh command for interacting with gerrit
#   project => the gerrit project name (e.g. "my/project")
sub _gerrit_parse_url {
  my ($url) = @_;

  if ( !ref($url) || !$url->isa('URI') ) {
    $url = URI->new($url);
  }

  if ( $url->scheme() ne 'ssh' ) {
    croak "gerrit URL $url is not supported; only ssh URLs are supported\n";
  }

  my $project = $url->path();
  $url->path(undef);

  # remove useless leading/trailing components
  $project =~ s{\A/+}{};
  $project =~ s{\.git\z}{}i;

  return {
    cmd => [
      @SSH,
      '-oBatchMode=yes',    # never do interactive prompts
      '-oServerAliveInterval=30'
      ,    # try to avoid the server silently dropping connection
      ( $url->port() ? ( '-p', $url->port() ) : () ),
      ( $url->user() ? ( $url->user() . '@' ) : q{} ) . $url->host(),
      'gerrit',
    ],
    project => $project,
    gerrit => $url->as_string(),

lib/Gerrit/Client.pm  view on Meta::CPAN

        $handle_error->( $handle, $error );
      }
    );
    $handle->{r_h_stderr} = AnyEvent::Handle->new( fh => $r2, );
    $handle->{r_h_stderr}->on_error( sub { } );
    $handle->{warn_on_stderr} = 1;

    # run stream-events with stdout connected to pipe ...
    $handle->{cv} = run_cmd(
      \@ssh,
      '>'  => $w,
      '2>' => $w2,
      '$$' => \$handle->{pid},
    );
    $handle->{cv}->cb(
      sub {
        my ($status) = shift->recv();
        $handle_error->( $handle, "ssh exited with status $status" );
      }
    );

    my %read_req;
    %read_req = (

      # read one json item at a time
      json => sub {
        my ( $h, $data ) = @_;

        # every successful read resets sleep period
        $sleep = $INIT_SLEEP;

        $h->push_read(%read_req);
        $on_event->($data);
      }
    );
    $handle->{r_h}->push_read(%read_req);

    my %err_read_req;
    %err_read_req = (
      line => sub {
        my ( $h, $line ) = @_;

        if ( $handle->{warn_on_stderr} ) {
          warn __PACKAGE__ . ': ssh stderr: ' . $line;
        }
        $h->push_read(%err_read_req);
      }
    );
    $handle->{r_h_stderr}->push_read(%err_read_req);
  };

  my $stash = {};
  $restart->($stash);

  my $out = { stash => $stash };
  if ( defined wantarray ) {
    $out->{guard} = guard {
      $cleanup->($stash);
    };
    $out_weak = $out;
    weaken($out_weak);
  }
  else {
    $out_weak = $out;
  }
  return $out;
}

=item B<< for_each_patchset(ssh_url => $ssh_url, workdir => $workdir, ...) >>

Set up a high-level event watcher to invoke a custom callback or
command for each existing or incoming patch set on Gerrit. This method
is suitable for performing automated testing or sanity checks on
incoming patches.

For each patch set, a git repository is set up with the working tree
and HEAD set to the patch. The callback is invoked with the current
working directory set to the top level of this git repository.

Returns a guard object. Event processing terminates when the object is
destroyed.

Options:

=over

=item B<ssh_url>

The Gerrit ssh URL, e.g. C<ssh://user@gerrit.example.com:29418/>.
May also be specified as 'url' for backwards compatibility.
Mandatory.

=item B<http_url>

=item B<< http_auth_cb => $sub->($response_headers, $request_headers) >>

=item B<http_username>

=item B<http_password>

These arguments have the same meaning as for the L<review> function.
Provide them if you want to post reviews via REST.

=item B<workdir>

The top-level working directory under which git repositories and other data
should be stored. Mandatory. Will be created if it does not exist.

The working directory is persistent across runs. Removing the
directory may cause the processing of patch sets which have already
been processed.

=item B<< on_patchset => $sub->($change, $patchset) >>

=item B<< on_patchset_fork => $sub->($change, $patchset) >>

=item B<< on_patchset_cmd => $sub->($change, $patchset) | $cmd_ref >>

Callbacks invoked for each patchset. Only one of the above callback
forms may be used.

lib/Gerrit/Client.pm  view on Meta::CPAN

Defaults to 1.

=item B<< query => $query | 0 >>

The Gerrit query used to find the initial set of patches to be
processed.  The query is executed when the loop begins and whenever
the connection to Gerrit is interrupted, to avoid missed patchsets.

Defaults to "status:open", meaning every open patch will be processed.

Note that the query is not applied to incoming patchsets observed via
stream-events. The B<wanted> parameter may be used for that case.

If a false value is passed, querying is disabled altogether. This
means only patchsets arriving while the loop is running will be
processed.

=back

=cut

sub for_each_patchset {
  my (%args) = @_;

  $args{ssh_url} ||= $args{url};
  if ($args{http_username} && $args{http_password}) {
    $args{http_auth_cb} ||= http_digest_auth($args{http_username}, $args{http_password});
  }

  $args{ssh_url} || croak 'missing ssh_url argument';
       $args{on_patchset}
    || $args{on_patchset_cmd}
    || $args{on_patchset_fork}
    || croak 'missing on_patchset{_cmd,_fork} argument';
  $args{workdir} || croak 'missing workdir argument';
  $args{on_error} ||= sub { warn __PACKAGE__, ': ', @_ };

  if ( !exists( $args{git_work_tree} ) ) {
    $args{git_work_tree} = 1;
  }

  if ( !exists( $args{query} ) ) {
    $args{query} = 'status:open';
  }

  if ( !-d $args{workdir} ) {
    mkpath( $args{workdir} );
  }

  # drop the path section of the URL to get base gerrit URL
  my $url = URI->new($args{ssh_url});
  $url->path( undef );
  $args{ssh_url} = $url->as_string();

  require "Gerrit/Client/ForEach.pm";
  my $self = bless {}, 'Gerrit::Client::ForEach';
  $self->{args} = \%args;

  my $weakself = $self;
  weaken($weakself);

  # stream_events takes care of incoming changes, perform a query to find
  # existing changes
  my $do_query = sub {
    return unless $args{query};

    query(
      $args{query},
      ssh_url           => $args{ssh_url},
      current_patch_set => 1,
      on_error          => sub { $args{on_error}->(@_) },
      on_success        => sub {
        return unless $weakself;
        my (@results) = @_;
        foreach my $change (@results) {

          # simulate patch set creation
          my ($event) = {
            type     => 'patchset-created',
            change   => $change,
            patchSet => delete $change->{currentPatchSet},
          };
          $weakself->_handle_for_each_event($event);
        }
      },
    );
  };

  # Unfortunately, we have no idea how long it takes between starting the
  # stream-events command and when the streaming of events begins, so if
  # we query straight away, we could miss some changes which arrive while
  # stream-events is e.g. still in ssh negotiation.
  # Therefore, introduce this arbitrary delay between when we start
  # stream-events and when we'll perform a query.
  my $query_timer;
  my $do_query_soon = sub {
    $query_timer = AE::timer( 4, 0, $do_query );
  };

  $self->{stream} = Gerrit::Client::stream_events(
    ssh_url  => $args{ssh_url},
    on_event => sub {
      $weakself->_handle_for_each_event(@_);
    },
    on_error => sub {
      my ($error) = @_;

      $args{on_error}->("connection lost: $error, attempting to recover\n");

      # after a few seconds to allow reconnect, perform the base query again
      $do_query_soon->();

      return 1;
    },
  );

  $do_query_soon->();

  return $self;
}