view release on metacpan or  search on metacpan

t/request-body-stream.t  view on Meta::CPAN

use strict;
use warnings;
use Test2::V0;
use Future::AsyncAwait;
use File::Temp qw(tempfile);

use PAGI::Request::BodyStream;

# Mock receive function
sub mock_receive {
    my @chunks = @_;
    my $i = 0;
    return sub {
        my $chunk = $chunks[$i++];
        return Future->done($chunk);
    };
}

subtest 'basic streaming' => sub {
    my $receive = mock_receive(
        { type => 'http.request', body => 'Hello', more => 1 },
        { type => 'http.request', body => ' World', more => 0 },
    );

    my $stream = PAGI::Request::BodyStream->new(receive => $receive);

    my $chunk1 = $stream->next_chunk->get;
    is $chunk1, 'Hello', 'first chunk';
    ok !$stream->is_done, 'not done yet';

    my $chunk2 = $stream->next_chunk->get;
    is $chunk2, ' World', 'second chunk';
    ok $stream->is_done, 'done after last chunk';

    is $stream->bytes_read, 11, 'bytes_read correct';
};

subtest 'max_bytes limit' => sub {
    my $receive = mock_receive(
        { type => 'http.request', body => 'Hello World', more => 0 },
    );

    my $stream = PAGI::Request::BodyStream->new(
        receive => $receive,
        max_bytes => 5,
    );

    like dies { $stream->next_chunk->get }, qr/max_bytes exceeded/, 'throws on limit exceeded';
};

subtest 'UTF-8 decoding' => sub {
    my $receive = mock_receive(
        { type => 'http.request', body => "caf\xc3\xa9", more => 0 },
    );

    my $stream = PAGI::Request::BodyStream->new(
        receive => $receive,
        decode => 'UTF-8',
    );

    my $chunk = $stream->next_chunk->get;
    is $chunk, "café", 'UTF-8 decoded';
};

subtest 'disconnect handling' => sub {
    my $receive = mock_receive(
        { type => 'http.request', body => 'Hello', more => 1 },
        { type => 'http.disconnect' },
    );

    my $stream = PAGI::Request::BodyStream->new(receive => $receive);

    my $chunk1 = $stream->next_chunk->get;
    is $chunk1, 'Hello', 'first chunk';

    my $chunk2 = $stream->next_chunk->get;
    is $chunk2, undef, 'undef on disconnect';
    ok $stream->is_done, 'done after disconnect';
};

t/request-body-stream.t  view on Meta::CPAN

subtest 'bytes_read tracking' => sub {
    my $receive = mock_receive(
        { type => 'http.request', body => '12345', more => 1 },
        { type => 'http.request', body => '67890', more => 0 },
    );

    my $stream = PAGI::Request::BodyStream->new(receive => $receive);

    is $stream->bytes_read, 0, 'zero before reading';

    $stream->next_chunk->get;
    is $stream->bytes_read, 5, 'five after first chunk';

    $stream->next_chunk->get;
    is $stream->bytes_read, 10, 'ten after second chunk';
};

subtest 'stream_to_file rejects decode option' => sub {
    my $receive = mock_receive(
        { type => 'http.request', body => 'test', more => 0 },
    );

    my $stream = PAGI::Request::BodyStream->new(
        receive => $receive,
        decode => 'UTF-8',
    );

    like dies { $stream->stream_to_file('/tmp/test.bin')->get },
        qr/cannot be used with decode/,
        'stream_to_file throws with decode option';
};

subtest 'PAGI::Request body_stream' => sub {
    use PAGI::Request;

    my $chunks = [
        { type => 'http.request', body => 'Hello', more => 1 },
        { type => 'http.request', body => ' World', more => 0 },
    ];
    my $i = 0;
    my $receive = sub { Future->done($chunks->[$i++]) };

    my $scope = { method => 'POST', path => '/', headers => [] };
    my $req = PAGI::Request->new($scope, $receive);

    my $stream = $req->body_stream;
    isa_ok $stream, 'PAGI::Request::BodyStream';

    my $chunk1 = $stream->next_chunk->get;
    is $chunk1, 'Hello', 'first chunk via Request';
};

subtest 'body_stream mutual exclusivity' => sub {
    use PAGI::Request;

    # Streaming then buffered should fail
    my $scope1 = { method => 'POST', path => '/', headers => [] };
    my $receive1 = sub { Future->done({ type => 'http.request', body => 'test', more => 0 }) };
    my $req1 = PAGI::Request->new($scope1, $receive1);
    $req1->body_stream;
    like dies { $req1->body->get }, qr/streaming/, 'body after stream fails';

    # Buffered then streaming should fail
    my $scope2 = { method => 'POST', path => '/', headers => [] };
    my $receive2 = sub { Future->done({ type => 'http.request', body => 'x', more => 0 }) };
    my $req2 = PAGI::Request->new($scope2, $receive2);
    $req2->body->get;
    like dies { $req2->body_stream }, qr/consumed|read/, 'stream after body fails';
};

done_testing;