view release on metacpan or  search on metacpan

t/streaming_callback.t  view on Meta::CPAN

    'and each one decodes to what the daemon actually sent';
};

# ---------------------------------------------------------------------------
subtest 'on_event: errorDetail croaks at the event that reports it' => sub {
  my $raw = $FIXTURES->child('images_build_error_stream.ndjson')->slurp_raw;
  my @lines = grep { /\S/ } split /(?<=\n)/, $raw;

  # at_end => 'die': if the transport carried on collecting after the failing
  # event it would run off the end of the script. The buffered path has to
  # read the whole stream before it can scan it; this one croaks on the spot.
  my $client = transport(chunked(\@lines), step => 9, at_end => 'die');

  my @got;
  eval {
    $client->post('/build', undef, on_event => sub { push @got, $_[0] });
  };
  my $err = $@;

  ok ref $err && $err->isa('API::Docker::Error::Stream'),
    'the same error class the buffered path raises';
  like "$err", qr/Docker API stream error \(POST \/v1\.41\/build\)/,
    'named by endpoint, as before';
  is_deeply $err->events, [ decode_json($lines[-1]) ],
    'the object carries the failing event alone: a callback stream keeps no '
    . 'history, the caller was handed all of it as it arrived';
  is scalar(@got), scalar(@lines) - 1,
    'the events before the failure were delivered; the failing one was not';
};

subtest 'on_event: croak_on_error => 0 hands the error event over as data' => sub {
  my $raw = $FIXTURES->child('images_build_error_stream.ndjson')->slurp_raw;
  my @lines = grep { /\S/ } split /(?<=\n)/, $raw;
  my $client = transport(chunked(\@lines, closed => 1), step => 9);

  my @got;
  my $summary = $client->get('/events',
    croak_on_error => 0,
    on_event       => sub { push @got, $_[0] },
  );

  is scalar(@got), scalar(@lines), 'nothing was withheld';
  ok $got[-1]{errorDetail}, 'the errorDetail event arrived as an ordinary one';
  is $summary->{stopped}, 0, 'and the stream ran to the end';
};

# ---------------------------------------------------------------------------
subtest 'on_frame: frames are reassembled across chunk boundaries' => sub {
  my $body = $FIXTURES->child('containers_logs_multiplexed.bin')->slurp_raw;
  # Split at 5 bytes: the first frame's 8-byte header straddles two chunks,
  # so a reader that expected a header to arrive whole gets nothing right.
  my @pieces = ($body =~ /(.{1,5})/gs);
  my $client = transport(chunked(\@pieces, closed => 1), step => 3);

  my @got;
  my $summary = $client->stream_frames('GET', '/containers/x/logs',
    on_frame => sub { push @got, $_[0] });

  is_deeply \@got, [
    { stream => 'stdout', data => "OUT\n" },
    { stream => 'stderr', data => "ERR\n" },
  ], 'demultiplexed exactly as the buffered path demultiplexes it';
  is_deeply $summary, { delivered => 2, stopped => 0 }, 'two frames, ran out';
};

subtest 'on_frame: the callback stops between frames' => sub {
  my $body = $FIXTURES->child('containers_logs_multiplexed.bin')->slurp_raw;
  my $client = transport(chunked([$body]), step => 4, at_end => 'die');

  my @got;
  my $summary = eval {
    $client->stream_frames('GET', '/containers/x/logs',
      on_frame => sub {
        my ($frame, $stop) = @_;
        push @got, $frame;
        $stop->();
      });
  };

  is $@, '', 'returned without reading for the chunk that never comes';
  is scalar(@got), 1, 'one frame, and the second was never delivered';
  is_deeply $summary, { delivered => 1, stopped => 1 }, 'summary says so';
};

subtest 'on_frame: a stream cut off mid-frame croaks' => sub {
  my $body = $FIXTURES->child('containers_logs_multiplexed.bin')->slurp_raw;
  # 12 bytes is the first frame whole; 22 leaves the second one's header
  # complete and its 4-byte payload two bytes short -- the case where a reader
  # knows exactly how much it is waiting for and never gets it.
  my $client = transport(
    chunked([ substr($body, 0, 22) ], closed => 1), step => 4);

  my @got;
  eval {
    $client->stream_frames('GET', '/containers/x/logs',
      on_frame => sub { push @got, $_[0] });
  };
  is scalar(@got), 1, 'the frame that did arrive whole was delivered';
  like $@, qr/closed mid-frame, leaving 10 bytes/,
    'the daemon closing in the middle of a frame is an error, not an empty '
    . 'tail: the buffered path can fall back to raw, this one cannot';
};

subtest 'on_frame: an unframed body croaks rather than inventing frames' => sub {
  my $body = $FIXTURES->child('containers_logs_tty.bin')->slurp_raw;
  my $client = transport(chunked([$body], closed => 1), step => 4);

  eval {
    $client->stream_frames('GET', '/containers/x/logs',
      on_frame => sub { });
  };
  like $@, qr/not a framed stream/,
    'framing cannot be sniffed without the whole body, so an undeclared one '
    . 'that is not framed is refused';
  like $@, qr/tty => 1/, 'and the message names the way to declare it';
};

subtest 'stream_frames: tty => 1 delivers raw frames per chunk' => sub {
  my $body = $FIXTURES->child('containers_logs_tty.bin')->slurp_raw;
  my @pieces = ($body =~ /(.{1,4})/gs);
  my $client = transport(chunked(\@pieces, closed => 1));