view release on metacpan or  search on metacpan

t/28_pass2_coverage.t  view on Meta::CPAN

# reset() called from a query-timeout error callback and from a cancel
# error callback both survive the surrounding teardown (the same
# connect_gen guard mechanism that protects the on_error path).

# 12a. reset() from inside a query-timeout callback.
{
    my $tries = 0;
    my $ok;
    my $ch; $ch = EV::ClickHouse->new(
        host => $host, port => $nport, protocol => 'native',
        on_connect => sub {
            return if $tries;
            $ch->query("select sleep(2)", { query_timeout => 0.1 }, sub {
                my (undef, $err) = @_;
                return if $tries++;
                # On timeout, reset and re-issue a real query — the new
                # socket must survive the outer cleanup_connection.
                $ch->reset;
                $ch->query("select 1", sub {
                    my ($r) = @_;
                    $ok = $r && @$r ? $r->[0][0] == 1 : 0;
                    EV::break;
                });
            });
        },
        on_error => sub { },
    );
    run_with_timeout(5);
    ok $ok, '$ch->reset from query-timeout cb survives outer teardown';
    $ch->finish if $ch->is_connected;
}

# 12b. reset() from inside a query callback that received a cancel error.
{
    my $tries = 0;
    my $ok;
    my $ch; $ch = EV::ClickHouse->new(
        host => $host, port => $nport, protocol => 'native',
        on_connect => sub {
            return if $tries;
            $ch->query("select sleep(2)", sub {
                my (undef, $err) = @_;
                return if $tries++;
                $ch->reset;
                $ch->query("select 2", sub {
                    my ($r) = @_;
                    $ok = $r && @$r ? $r->[0][0] == 2 : 0;
                    EV::break;
                });
            });
            EV::timer(0.1, 0, sub { $ch->cancel });
        },
        on_error => sub { },
    );
    run_with_timeout(5);
    ok $ok, '$ch->reset from cancel cb survives outer teardown';
    $ch->finish if $ch->is_connected;
}

# Regression: Pool::_pick prefers any idle member over a busy one
# (regression: the previous logic round-robined whenever ANY member was
# idle, which could route the next pick to a busy connection).
{
    my $pool = EV::ClickHouse::Pool->new(
        host => $host, port => $nport, protocol => 'native',
        size => 3,
    );
    my @c = $pool->conns;
    # Wait for all to connect.
    my $cn = 0;
    $_->ping(sub { $cn++ }) for @c;
    my $w = EV::timer(0, 0.05, sub { EV::break if $cn >= 3 });
    run_with_timeout(5);
    undef $w;
    # Force c[0] busy with a long sleep, leave c[1], c[2] idle.
    $c[0]->query("select sleep(2)", sub { });
    # Now _pick should NEVER return c[0] while it's busy.
    my %picks;
    $picks{ "$_" }++ for map { $pool->_pick } 1 .. 6;
    ok !exists $picks{"$c[0]"},
       'Pool::_pick avoids the busy member when others are idle';
    eval { $c[0]->cancel };
    eval { $pool->finish };
}

# Coverage: reconnect_jitter accepted, insert_streamer named-rows.

# 13. reconnect_jitter is accepted as a constructor option without croaking,
#     and a fully-disabled value (0) keeps backoff deterministic.
{
    my $ok;
    my $ch; $ch = EV::ClickHouse->new(
        host             => $host, port => $nport, protocol => 'native',
        reconnect_delay  => 0.1,
        reconnect_jitter => 0.5,
        on_connect       => sub { $ok = 1; EV::break },
    );
    run_with_timeout(5);
    ok $ok, 'reconnect_jitter accepted without breaking connect';
    $ch->finish;
}

# 14. insert_streamer columns => [...] reorders hash rows positionally.
{
    my $err;
    my $ch; $ch = EV::ClickHouse->new(
        host => $host, port => $nport, protocol => 'native',
        on_connect => sub {
            $ch->query("create temporary table _named28 (a UInt32, b String) engine = Memory", sub {
                my $s = $ch->insert_streamer('_named28',
                    columns    => [qw(a b)],
                    batch_size => 10,
                );
                # mix hashref + arrayref pushes; column order in hash is
                # deliberately not the table order.
                $s->push_row({ b => 'x', a => 1 });
                $s->push_row({ a => 2, b => 'y' });
                $s->push_row([3, 'z']);
                $s->finish(sub {
                    (undef, $err) = @_;
                    EV::break;
                });
            });
        },
        on_error => sub { $err = $_[0]; EV::break },
    );
    run_with_timeout(5);
    is $err, undef, 'insert_streamer columns => [...] reorders hash rows';

    # And verify the rows landed in the right columns.
    my $ch2; $ch2 = EV::ClickHouse->new(
        host => $host, port => $nport, protocol => 'native',
        on_connect => sub {
            $ch->query("select a, b from _named28 order by a format TabSeparated", sub {
                my ($rows) = @_;
                is_deeply $rows, [[1, 'x'], [2, 'y'], [3, 'z']],
                    '  rows landed in declared column order';
                EV::break;
            });
        },
    );