#!/usr/bin/env perl

use strict;
use warnings;
use Test::More;
use FindBin qw($Bin);
use lib "$Bin/lib";
use Carp qw(croak);
use MemcachedTest;
use IO::Socket qw(AF_INET SOCK_STREAM);
use IO::Select;

if (!supports_proxy()) {
    plan skip_all => 'proxy not enabled';
    exit 0;
}

# Set up some server sockets.
sub mock_server {
    my $port = shift;
    my $srv = IO::Socket->new(
        Domain => AF_INET,
        Type => SOCK_STREAM,
        Proto => 'tcp',
        LocalHost => '127.0.0.1',
        LocalPort => $port,
        ReusePort => 1,
        Listen => 5) || die "IO::Socket: $@";
    return $srv;
}

# Accept and validate a new backend connection.
sub accept_backend {
    my $srv = shift;
    my $be = $srv->accept();
    $be->autoflush(1);
    ok(defined $be, "mock backend created");
    like(<$be>, qr/version/, "received version command");
    print $be "VERSION 1.0.0-mock\r\n";

    return $be;
}

note("Initialization:" . __LINE__);

my @mocksrvs = ();
#diag "making mock servers";
for my $port (11411, 11412, 11413) {
    my $srv = mock_server($port);
    ok(defined $srv, "mock server created");
    push(@mocksrvs, $srv);
}

my $p_srv = new_memcached('-o proxy_config=./t/proxyunits.lua -t 1');
my $ps = $p_srv->sock;
$ps->autoflush(1);

my $pss = IO::Select->new();
$pss->add($ps);

# set up server backend sockets.
my @mbe = ();
#diag "accepting mock backends";
for my $msrv (@mocksrvs) {
    my $be = accept_backend($msrv);
    push(@mbe, $be);
}

# Put a version command down the pipe to ensure the socket is clear.
# client version commands skip the proxy code
sub check_version {
    my $ps = shift;
    print $ps "version\r\n";
    like(<$ps>, qr/VERSION /, "version received");
}

# Send a touch command to all backends, and verify response.
# This makes sure socket buffers are clean between tests.
sub check_sanity {
    my $ps = shift;
    my $cmd = "touch /sanity/a 50\r\n";
    print $ps $cmd;
    foreach my $idx (keys @mbe) {
        my $be = $mbe[$idx];
        is(scalar <$be>, $cmd, "sanity check: touch cmd received for be " . $idx);
        print $be "TOUCHED\r\n";
    }
    is(scalar <$ps>, "TOUCHED\r\n", "sanity check: TOUCHED response received.");
}

# $ps_send : request to proxy
# $be_recv : ref to a hashmap from be index to an array of received requests for validation.
# $be_send : ref to a hashmap from be index to an array of responses to proxy.
# $ps_recv : ref to response returned by proxy
# backends in $be_recv and $be_send are vistied by looping through the @mbe.
sub proxy_test {
    my %args = @_;

    my $ps_send = $args{ps_send};
    my $be_recv = $args{be_recv} // {};
    my $be_send = $args{be_send} // {};
    my $ps_recv = $args{ps_recv};

    # sends request to proxy
    if ($ps_send) {
        print $ps $ps_send;
    }

    # verify all backends received request
    foreach my $idx (keys @mbe) {
        if (exists $be_recv->{$idx}) {
            my $be = $mbe[$idx];
            foreach my $recv (@{$be_recv->{$idx}}) {
                is(scalar <$be>, $recv, "be " . $idx . " received expected response");
            }
        }
    }

    # backends send responses
    foreach my $idx (keys @mbe) {
        if (exists $be_send->{$idx}) {
            my $be = $mbe[$idx];
            foreach my $send (@{$be_send->{$idx}}) {
                print $be $send;
            }
        }
    }

    if (defined $ps_recv) {
        if (scalar @{$ps_recv}) {
            foreach my $recv (@{$ps_recv}) {
                is(scalar <$ps>, $recv, "ps returned expected response.");
            }
            # makes sure nothing remains in $ps.
            check_version($ps);
        } else {
            # when $ps_recv is empty, make sure it is not readable.
            my @readable = $pss->can_read(0.1);
            is(scalar @readable, 0, "ps is expected to be non-readable");
        }
    }
}

{
    note("Bad syntax tests");
    # Write a request with bad syntax, and check the response.
    print $ps "set with the wrong number of tokens\n";
    is(scalar <$ps>, "CLIENT_ERROR parsing request\r\n", "got CLIENT_ERROR for bad syntax");

    for ('get', 'incr', 'decr', 'touch', 'gat', 'gats', 'mg', 'md', 'ma', 'ms') {
        print $ps "$_\r\n";
        is(scalar <$ps>, "CLIENT_ERROR parsing request\r\n", "$_ got CLIENT_ERROR for too few tokens");
    }

    my $space = ' ' x 200;
    print $ps "get$space key key\r\n";
    is(scalar <$ps>, "CLIENT_ERROR malformed request\r\n");
    is(scalar <$ps>, "CLIENT_ERROR malformed request\r\n");
    is(scalar <$ps>, "END\r\n"); # god damn multiget syntax.
}

{
    note("Test dead backend");
    my $start = int(time());
    print $ps "get /dead/foo\r\n";
    is(scalar <$ps>, "SERVER_ERROR backend failure\r\n", "Backend failed");
    my $end = int(time());
    cmp_ok($end - $start, '<', 3, "backend failed immediately");

    print $ps "get /deadrespcode/foo\r\n";
    is(scalar <$ps>, "ERROR code_correct\r\n", "Backend had correct response code on failure");
}

{
    note("millisecond timer");
    print $ps "mg /millis/key\r\n";
    my $res = <$ps>;
    if ($res =~ m/^HD t(\d+)/) {
        my $time = $1;
        my $now = int(time());
        cmp_ok($time, '>', $now*5, "mcp.time_millis is a reasonable value: $now*5 vs $time");
        cmp_ok($time, '!=', 0, 'mcp.time_millis is non zero');
    } else {
        fail("mcp.time_millis failure: $res");
    }
}

# Basic test with a backend; write a request to the client socket, read it
# from a backend socket, and write a response to the backend socket.
#
# The array @mbe holds references to our sockets for the backends listening on
# the above mocked servers. In most tests we're only routing to the first
# backend in the list ($mbe[0])
#
# In this case the client will receive an error and the backend gets closed,
# so we have to re-establish it.
{
    note("Test missing END:" . __LINE__);

    # Test a fix for passing through partial read data if END ends up missing.
    my $be = $mbe[0];
    my $w = $p_srv->new_sock;
    print $w "watch proxyevents\n";
    is(<$w>, "OK\r\n", "watcher enabled");

    # write a request to proxy.
    print $ps "get /b/a\r\n";

    # verify request is received by backend.
    is(scalar <$be>, "get /b/a\r\n", "get passthrough");

    # write a response with partial data.
    print $be "VALUE /b/a 0 2\r\nhi\r\nEN";

    # verify the error response from proxy
    is(scalar <$ps>, "SERVER_ERROR backend failure\r\n", "backend failure error");

    # verify a particular proxy event logline is received
    like(<$w>, qr/ts=(\S+) gid=\d+ type=proxy_backend error=timeout name=127.0.0.1 port=\d+ label=b\d+ depth=1 rbuf=EN/, "got backend error log line");

    # backend is disconnected due to the error, so we have to re-establish it.
    $mbe[0] = accept_backend($mocksrvs[0]);
}

# This test is similar to the above one, except we also establish a watcher to
# check for appropriate log entries.
{
    note("Test trailingdata:" . __LINE__);

    # Test a log line with detailed data from backend failures.
    my $be = $mbe[0];
    my $w = $p_srv->new_sock;
    print $w "watch proxyevents\n";
    is(<$w>, "OK\r\n", "watcher enabled");

    print $ps "get /b/c\r\n";
    is(scalar <$be>, "get /b/c\r\n", "get passthrough");
    # Set off a "trailing data" error
    print $be "VALUE /b/c 0 2\r\nok\r\nEND\r\ngarbage";

    is(scalar <$ps>, "VALUE /b/c 0 2\r\n", "got value back");
    is(scalar <$ps>, "ok\r\n", "got data back");
    is(scalar <$ps>, "END\r\n", "got end string");

    like(<$w>, qr/ts=(\S+) gid=\d+ type=proxy_backend error=trailingdata name=127.0.0.1 port=\d+ label=b\d+ depth=0 rbuf=garbage/, "got backend error log line");

    $mbe[0] = accept_backend($mocksrvs[0]);
}

{
    note("Test proxyevents for a backend without a label");

    my $msrv_nolabel = mock_server(11414);
    ok(defined $msrv_nolabel, "mock server with no label created");
    my $mbe_nolabel = accept_backend($msrv_nolabel);

    # Trigger a proxyevent error by adding trailing data ("garbage") to the
    # backend response.
    my $w = $p_srv->new_sock;
    print $w "watch proxyevents\n";
    is(<$w>, "OK\r\n", "watcher enabled");

    print $ps "get /nolabel/c\r\n";
    is(scalar <$mbe_nolabel>, "get /nolabel/c\r\n", "get passthrough");
    # Set off a "trailing data" error
    print $mbe_nolabel "VALUE /nolabel/c 0 2\r\nok\r\nEND\r\ngarbage";

    is(scalar <$ps>, "VALUE /nolabel/c 0 2\r\n", "got value back");
    is(scalar <$ps>, "ok\r\n", "got data back");
    is(scalar <$ps>, "END\r\n", "got end string");

    # Verify a proxy event logline is received with empty label
    like(<$w>, qr/ts=(\S+) gid=\d+ type=proxy_backend error=trailingdata name=127.0.0.1 port=11414 label= depth=0 rbuf=garbage/, "got backend error log line");
}

# This is an example of a test which will only pass before a bugfix is issued.
# It's good practice where possible to write a failing test, then check it
# against a code fix. We then leave the test in the file for reference.
# Though noting when it was fixed is probably better than what I did here :)
SKIP: {
    note("Test bugfix for missingend:" . __LINE__);
    skip "Remove this skip line to demonstrate pre-patch bug", 1;
    # Test issue with finding response complete when read lands between value
    # size and value + response line in size.
    my $be = $mbe[0];
    my $w = $p_srv->new_sock;
    print $w "watch proxyevents\n";
    is(<$w>, "OK\r\n", "watcher enabled");

    print $ps "get /b/c\r\n";
    is(scalar <$be>, "get /b/c\r\n", "get passthrough");

    # Set off a "missingend" error.
    # The server will wake up several times, thinking it has read the
    # full size of response but it only read enough for the value portion.
    print $be "VALUE /b/c 0 5\r\nhe";
    sleep 0.1;
    print $be "llo";
    sleep 0.1;
    print $be "\r\nEND\r\n";

    is(scalar <$ps>, "SERVER_ERROR backend failure\r\n");

    like(<$w>, qr/ts=(\S+) gid=\d+ type=proxy_backend error=missingend name=127.0.0.1 port=\d+ label=b\d+ depth=1 rbuf=/, "got missingend error log line");

    $mbe[0] = accept_backend($mocksrvs[0]);
}

{
    # Test issue with finding response complete when read lands between value
    # size and value + response line in size.
    my $be = $mbe[0];

    print $ps "get /b/c\r\n";
    is(scalar <$be>, "get /b/c\r\n", "get passthrough");

    # Set off a "missingend" error.
    # The server will wake up several times, thinking it has read the
    # full size of response but it only read enough for the value portion.
    print $be "VALUE /b/c 0 5\r\nhe";
    sleep 0.1;
    print $be "llo";
    sleep 0.1;
    print $be "\r\nEND\r\n";

    is(scalar <$ps>, "VALUE /b/c 0 5\r\n", "got value back");
    is(scalar <$ps>, "hello\r\n", "got data back");
    is(scalar <$ps>, "END\r\n", "got end string");
}

#diag "ready for main tests";
# Target a single backend, validating basic syntax.
# Should test all command types.
# uses /b/ path for "basic"
{
    note("Test all commands to a single backend:" . __LINE__);

    # Test invalid route.
    print $ps "set /invalid/a 0 0 2\r\nhi\r\n";
    is(scalar <$ps>, "SERVER_ERROR no set route\r\n");

    # Testing against just one backend. Results should make sense despite our
    # invalid request above.
    my $be = $mbe[0];
    my $cmd;

    # TODO: add more tests for the varying response codes.

    # Basic set.
    $cmd = "set /b/a 0 0 2";
    print $ps "$cmd\r\nhi\r\n";
    is(scalar <$be>, "$cmd\r\n", "set passthrough");
    is(scalar <$be>, "hi\r\n", "set value");
    print $be "STORED\r\n";

    is(scalar <$ps>, "STORED\r\n", "got STORED from set");

    # set (large value)
    my $datasize = 256000;
    my $data = 'x' x $datasize;
    $cmd = "set /b/a 0 0 $datasize";
    print $ps "$cmd\r\n$data\r\n";
    is(scalar <$be>, "$cmd\r\n", "set passthrough (large value)");
    is(scalar <$be>, "$data\r\n", "set value (large value)");
    print $be "STORED\r\n";

    is(scalar <$ps>, "STORED\r\n", "got STORED from set (large value)");

    # set (pipelined)
    my $num_repetitions = 5;
    $cmd = "set /b/a 0 0 2";
    $data = "ab";
    my $req_ps = "$cmd\r\n$data\r\n";
    my $resp_be = "STORED\r\n";

    my $repeated_req_ps = "";
    my $repeated_resp_be = "";
    for (1..$num_repetitions) {
        $repeated_req_ps = $repeated_req_ps . $req_ps;
        $repeated_resp_be = $repeated_resp_be . $resp_be;
    }

    print $ps $repeated_req_ps;

    for (1..$num_repetitions) {
        is(scalar <$be>, "$cmd\r\n", "set passthrough (repeated)");
        is(scalar <$be>, "$data\r\n", "set value (repeated)");
    }

    print $be $repeated_resp_be;

    for (1..$num_repetitions) {
        is(scalar <$ps>, "STORED\r\n", "got STORED from set (repeated)");
    }

    # Basic get
    $cmd = "get /b/a\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "get passthrough");
    print $be "VALUE /b/a 0 2\r\nhi\r\nEND\r\n";

    is(scalar <$ps>, "VALUE /b/a 0 2\r\n", "get rline");
    is(scalar <$ps>, "hi\r\n", "get data");
    is(scalar <$ps>, "END\r\n", "get end");

    # get (large value)
    $datasize = 256000;
    $data = 'x' x $datasize;
    $cmd = "get /b/a\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "get passthrough (large value)");
    print $be "VALUE /b/a 0 $datasize\r\n$data\r\nEND\r\n";

    is(scalar <$ps>, "VALUE /b/a 0 $datasize\r\n", "get rline (large value)");
    is(scalar <$ps>, "$data\r\n", "get data (large value)");
    is(scalar <$ps>, "END\r\n", "get end (large value)");

    # touch
    $cmd = "touch /b/a 50\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "touch passthrough");
    print $be "TOUCHED\r\n";

    is(scalar <$ps>, "TOUCHED\r\n", "got touch response");

    # gets
    $cmd = "gets /b/a\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "gets passthrough");
    print $be "VALUE /b/a 0 2 2\r\nhi\r\nEND\r\n";

    is(scalar <$ps>, "VALUE /b/a 0 2 2\r\n", "gets rline");
    is(scalar <$ps>, "hi\r\n", "gets data");
    is(scalar <$ps>, "END\r\n", "gets end");

    # gat
    $cmd = "gat 10 /b/a\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "gat passthrough");
    print $be "VALUE /b/a 0 2\r\nhi\r\nEND\r\n";

    is(scalar <$ps>, "VALUE /b/a 0 2\r\n", "gat rline");
    is(scalar <$ps>, "hi\r\n", "gat data");
    is(scalar <$ps>, "END\r\n", "gat end");

    # gat (cache miss)
    $cmd = "gat 10 /b/a\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "gat passthrough");
    print $be "END\r\n";

    is(scalar <$ps>, "END\r\n", "gat end, cache miss");

    # gats
    $cmd = "gats 11 /b/a\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "gats passthrough");
    print $be "VALUE /b/a 0 2 1\r\nhi\r\nEND\r\n";

    is(scalar <$ps>, "VALUE /b/a 0 2 1\r\n", "gats rline");
    is(scalar <$ps>, "hi\r\n", "gats data");
    is(scalar <$ps>, "END\r\n", "gats end");

    # cas
    $cmd = "cas /b/a 0 0 2 5";
    print $ps "$cmd\r\nhi\r\n";
    is(scalar <$be>, "$cmd\r\n", "cas passthrough");
    is(scalar <$be>, "hi\r\n", "cas value");
    print $be "STORED\r\n";

    is(scalar <$ps>, "STORED\r\n", "got STORED from cas");

    # cas (already exists failure)
    $cmd = "cas /b/a 0 0 3 6";
    print $ps "$cmd\r\nabc\r\n";
    is(scalar <$be>, "$cmd\r\n", "cas passthrough");
    is(scalar <$be>, "abc\r\n", "cas value");
    print $be "EXISTS\r\n";

    is(scalar <$ps>, "EXISTS\r\n", "got EXISTS from cas");

    # add
    $cmd = "add /b/a 0 0 2";
    print $ps "$cmd\r\nhi\r\n";
    is(scalar <$be>, "$cmd\r\n", "add passthrough");
    is(scalar <$be>, "hi\r\n", "add value");
    print $be "STORED\r\n";

    is(scalar <$ps>, "STORED\r\n", "got STORED from add");

    # add (re-add failure)
    $cmd = "add /b/a 0 0 4";
    print $ps "$cmd\r\nabcd\r\n";
    is(scalar <$be>, "$cmd\r\n", "add passthrough");
    is(scalar <$be>, "abcd\r\n", "add value");
    print $be "NOT_STORED\r\n";

    is(scalar <$ps>, "NOT_STORED\r\n", "got STORED from add");

    # delete
    $cmd = "delete /b/a\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "delete passthrough");
    print $be "DELETED\r\n";

    is(scalar <$ps>, "DELETED\r\n", "got delete response");

    # incr
    $cmd = "incr /b/a 1\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "incr passthrough");
    print $be "2\r\n";

    is(scalar <$ps>, "2\r\n", "got incr response");

    # decr
    $cmd = "decr /b/a 1\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "decr passthrough");
    print $be "10\r\n";

    is(scalar <$ps>, "10\r\n", "got decr response");

    # append
    $cmd = "append /b/a 0 0 2";
    print $ps "$cmd\r\nhi\r\n";
    is(scalar <$be>, "$cmd\r\n", "append passthrough");
    is(scalar <$be>, "hi\r\n", "append value");
    print $be "STORED\r\n";

    is(scalar <$ps>, "STORED\r\n", "got STORED from append");

    # prepend
    $cmd = "prepend /b/a 0 0 2";
    print $ps "$cmd\r\nhi\r\n";
    is(scalar <$be>, "$cmd\r\n", "prepend passthrough");
    is(scalar <$be>, "hi\r\n", "prepend value");
    print $be "STORED\r\n";

    is(scalar <$ps>, "STORED\r\n", "got STORED from prepend");

    # [meta commands]
    # testing the bare meta commands.
    # TODO: add more tests for tokens and changing response codes.
    # mg
    $cmd = "mg /b/a\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "mg passthrough");
    print $be "HD\r\n";

    is(scalar <$ps>, "HD\r\n", "got mg response");
    # ms
    $cmd = "ms /b/a 2";
    print $ps "$cmd\r\nhi\r\n";
    is(scalar <$be>, "$cmd\r\n", "ms passthrough");
    is(scalar <$be>, "hi\r\n", "ms value");
    print $be "HD\r\n";

    is(scalar <$ps>, "HD\r\n", "got HD from ms");

    # md
    $cmd = "md /b/a\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "md passthrough");
    print $be "HD\r\n";

    is(scalar <$ps>, "HD\r\n", "got HD from md");
    # ma
    $cmd = "ma /b/a\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "ma passthrough");
    print $be "HD\r\n";

    is(scalar <$ps>, "HD\r\n", "got HD from ma");
    # mn?
    # me?
}

# run a cleanser check between each set of tests.
# This ensures nothing was left in the client pipeline.
check_sanity($ps);

{
    note("Test multiget:" . __LINE__);

    # multiget syntax
    # - gets broken into individual gets on backend
    my $be = $mbe[0];
    my $cmd = "get /b/a /b/b /b/c\r\n";
    print $ps $cmd;
    is(scalar <$be>, "get /b/a\r\n", "multiget breakdown a");
    is(scalar <$be>, "get /b/b\r\n", "multiget breakdown b");
    is(scalar <$be>, "get /b/c\r\n", "multiget breakdown c");

    print $be "VALUE /b/a 0 1\r\na\r\n",
              "END\r\n",
              "VALUE /b/b 0 1\r\nb\r\n",
              "END\r\n",
              "VALUE /b/c 0 1\r\nc\r\n",
              "END\r\n";

    for my $key ('a', 'b', 'c') {
        is(scalar <$ps>, "VALUE /b/$key 0 1\r\n", "multiget res $key");
        is(scalar <$ps>, "$key\r\n", "multiget value $key");
    }
    is(scalar <$ps>, "END\r\n", "final END from multiget");

    # Test multiget workaround with misses (known bug)
    print $ps $cmd;
    is(scalar <$be>, "get /b/a\r\n", "multiget breakdown a");
    is(scalar <$be>, "get /b/b\r\n", "multiget breakdown b");
    is(scalar <$be>, "get /b/c\r\n", "multiget breakdown c");

    print $be "END\r\nEND\r\nEND\r\n";
    is(scalar <$ps>, "END\r\n", "final END from multiget");

    # If bugged, the backend will have closed.
    print $ps "get /b/a\r\n";
    is(scalar <$be>, "get /b/a\r\n", "get works after empty multiget");
    print $be "END\r\n";
    is(scalar <$ps>, "END\r\n", "end after empty multiget");
}

check_sanity($ps);

{
    note("Test noreply:" . __LINE__);

    # noreply tests.
    # - backend should receive with noreply/q stripped or mangled
    # - backend should reply as normal
    # - frontend should get nothing; to test issue another command and ensure
    # it only gets that response.
    my $be = $mbe[0];
    my $cmd = "set /b/a 0 0 2 noreply\r\nhi\r\n";
    print $ps $cmd;
    is(scalar <$be>, "set /b/a 0 0 2 noreplY\r\n", "set received with broken noreply");
    is(scalar <$be>, "hi\r\n", "set payload received");

    print $be "STORED\r\n";

    # To ensure success, make another req and ensure res isn't STORED
    $cmd = "touch /b/a 50\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "canary touch received");
    print $be "TOUCHED\r\n";

    is(scalar <$ps>, "TOUCHED\r\n", "got TOUCHED instread of STORED");
}

check_sanity($ps);

{
    subtest 'quiet flag: HD response' => sub {
        # be_recv must receive a response with quiet flag replaced by a space.
        # ps_recv must not receoved HD response.
        proxy_test(
            ps_send => "ms /b/a 2 q\r\nhi\r\n",
            be_recv => {0 => ["ms /b/a 2  \r\n", "hi\r\n"]},
            be_send => {0 => ["HD\r\n"]},
            ps_recv => [],
        );
    };

    subtest 'quiet flag: EX response' => sub {
        # be_recv must receive a response with quiet flag replaced by a space.
        # ps_recv must return EX response from the backend.
        proxy_test(
            ps_send => "ms /b/a 2 q\r\nhi\r\n",
            be_recv => {0 => ["ms /b/a 2  \r\n", "hi\r\n"]},
            be_send => {0 => ["EX\r\n"]},
            ps_recv => ["EX\r\n"],
        );
    };

    subtest 'quiet flag: backend failure' => sub {
        # be_recv must receive a response with quiet flag replaced by a space.
        # ps_recv must return backend failure response from the backend.
        proxy_test(
            ps_send => "ms /b/a 2 q\r\nhi\r\n",
            be_recv => {0 => ["ms /b/a 2  \r\n", "hi\r\n"]},
            be_send => {0 => ["garbage\r\n"]},
            ps_recv => ["SERVER_ERROR backend failure\r\n"],
        );
        $mbe[0] = accept_backend($mocksrvs[0]);
    };
}

check_sanity($ps);

# Test Lua request API
{
    note("Test Lua request APIs:" . __LINE__);

    my $be = $mbe[0];

    # fetching the key.
    print $ps "get /getkey/testkey\r\n";
    # look for the key to be slightly different to ensure we hit lua.
    is(scalar <$ps>, "VALUE |/getkey/testkey 0 2\r\n", "request:key()");
    is(scalar <$ps>, "ts\r\n", "request:key() value");
    is(scalar <$ps>, "END\r\n", "request:key() END");

    # rtrimkey
    # this overwrites part of the key with spaces, which should be skipped by
    # a valid protocol parser.
    print $ps "get /rtrimkey/onehalf\r\n";
    is(scalar <$be>, "get /rtrimkey/one    \r\n", "request:rtrimkey()");
    print $be "END\r\n";
    is(scalar <$ps>, "END\r\n", "rtrimkey END");

    # ltrimkey
    print $ps "get /ltrimkey/test\r\n";
    is(scalar <$be>, "get           test\r\n", "request:ltrimkey()");
    print $be "END\r\n";
    is(scalar <$ps>, "END\r\n", "ltrimkey END");

    subtest 'request:ntokens()' => sub {
        # ps_recv must return value that matches the number of tokens.
        proxy_test(
            ps_send => "mg /ntokens/test c v\r\n",
            ps_recv => ["VA 1 C123 v\r\n", "4\r\n"],
        );
    };

    subtest 'request:token() replacement' => sub {
        # be_recv must received a response with replaced CAS token.
        proxy_test(
            ps_send => "ms /token/replacement 2 C123\r\nhi\r\n",
            be_recv => {0 => ["ms /token/replacement 2 C456\r\n", "hi\r\n"]},
            be_send => {0 => ["NF\r\n"]},
            ps_recv => ["NF\r\n"],
        );
    };

    subtest 'request:token() remove' => sub {
        # be_recv must received a response with CAS token removed.
        proxy_test(
            ps_send => "ms /token/removal 2 C123\r\nhi\r\n",
            be_recv => {0 => ["ms /token/removal 2\r\n", "hi\r\n"]},
            be_send => {0 => ["NF\r\n"]},
            ps_recv => ["NF\r\n"],
        );
    };

    subtest 'request:token() fetch' => sub {
        # be_recv must received the key token in the P flag.
        proxy_test(
            ps_send => "ms /token/fetch 2 C123 P\r\nhi\r\n",
            be_recv => {0 => ["ms /token/fetch 2 C123 P/token/fetch\r\n", "hi\r\n"]},
            be_send => {0 => ["HD\r\n"]},
            ps_recv => ["HD\r\n"],
        );
    };

    # # command() integer

    subtest 'request:has_flag() meta positive 1' => sub {
        # ps_recv must receive HD C123 for a successful hash_flag call.
        proxy_test(
            ps_send => "mg /hasflag/test c\r\n",
            ps_recv => ["HD C123\r\n"],
        );
    };

    subtest 'request:has_flag() meta positive 2' => sub {
        # ps_recv must receive HD Oabc for a successful hash_flag call.
        proxy_test(
            ps_send => "mg /hasflag/test Oabc T999\r\n",
            ps_recv => ["HD Oabc\r\n"],
        );
    };

    subtest 'request:has_flag() meta negative' => sub {
        # ps_recv must receive NF when has_flag returns false.
        proxy_test(
            ps_send => "mg /hasflag/test T999\r\n",
            ps_recv => ["NF\r\n"],
        );
    };

    subtest 'request:has_flag() none-meta ' => sub {
        # ps_recv must receive END for a successful hash_flag call.
        proxy_test(
            ps_send => "get /hasflag/test\r\n",
            ps_recv => ["END\r\n"],
        );
    };

    subtest 'request:flag_token()' => sub {
        # be_recv must receive expected flags after a series of flag_token() calls.
        proxy_test(
            ps_send => "mg /flagtoken/a N10 k c R10\r\n",
            ps_recv => ["HD\r\n"],
        );
    };


    subtest 'request edit' => sub {
        # be_recv must receive the edited request.
        proxy_test(
            ps_send => "ms /request/edit 2\r\nhi\r\n",
            be_recv => {0 => ["ms /request/edit 2\r\n", "ab\r\n"]},
            be_send => {0 => ["HD\r\n"]},
            ps_recv => ["HD\r\n"],
        );
    };

    subtest 'request new' => sub {
        # be_recv must receive the new request.
        proxy_test(
            ps_send => "mg /request/old\r\n",
            be_recv => {0 => ["mg /request/new c\r\n"]},
            be_send => {0 => ["HD C123\r\n"]},
            ps_recv => ["HD C123\r\n"],
        );
    };

    subtest 'request clone response' => sub {
        # be must receive cloned meta-set from the previous meta-get.
        my $be = $mbe[0];
        my $be2 = $mbe[1];
        print $ps "mg /request/clone v\r\n";
        is(scalar <$be>, "mg /request/clone v\r\n", "get passthrough");
        print $be "VA 1 v\r\n4\r\n";
        is(scalar <$be2>, "ms /request/a 1\r\n", "received cloned meta-set");
        is(scalar <$be2>, "4\r\n", "received cloned meta-set value");
        print $be2 "HD\r\n";
        is(scalar <$ps>, "HD\r\n", "received HD");
    };
}

check_sanity($ps);

# Test Lua response API
{
    subtest 'response:elapsed() >100000micros' => sub {
        # ps_recv must not receive an error
        my $be = $mbe[0];
        my $cmd = "mg /response/elapsed\r\n";
        print $ps $cmd;
        is(scalar <$be>, $cmd, "be received request.");
        sleep 0.1;
        print $be "HD\r\n";
        is(scalar <$ps>, "HD\r\n", "proxy received HD");
    };


    subtest 'response:ok()' => sub {
        # ps_recv must not receive an error
        proxy_test(
            ps_send => "mg /response/ok\r\n",
            be_recv => {0 => ["mg /response/ok\r\n"]},
            be_send => {0 => ["HD\r\n"]},
            ps_recv => ["HD\r\n"],
        );
    };

    subtest 'response:ok() false 1' => sub {
        # ps_recv must receive an error
        proxy_test(
            ps_send => "mg /response/not_ok\r\n",
            be_recv => {0 => ["mg /response/not_ok\r\n"]},
            be_send => {0 => ["SERVER_ERROR\r\n"]},
            ps_recv => ["SERVER_ERROR\r\n"],
        );
    };

    subtest 'response:ok() false 2' => sub {
        # ps_recv must receive an error
        proxy_test(
            ps_send => "mg /response/not_ok\r\n",
            be_recv => {0 => ["mg /response/not_ok\r\n"]},
            be_send => {0 => ["GARBAGE\r\n"]},
            ps_recv => ["SERVER_ERROR\r\n"],
        );

        # test not_ok when backend is disconnected.
        # ps_recv must receive an error
        proxy_test(
            ps_send => "mg /response/not_ok\r\n",
            ps_recv => ["SERVER_ERROR\r\n"],
        );

        $mbe[0] = accept_backend($mocksrvs[0]);
        $mbe[0] = accept_backend($mocksrvs[0]);
    };

    subtest 'response:ok() false 3' => sub {
        # ps_recv must receive an error
        proxy_test(
            ps_send => "mg /response/not_ok\r\n",
            be_recv => {0 => ["mg /response/not_ok\r\n"]},
            be_send => {0 => ["HD\r"]},
            ps_recv => ["SERVER_ERROR\r\n"],
        );
        $mbe[0] = accept_backend($mocksrvs[0]);
    };

    subtest 'response:hit() mg' => sub {
        # ps_recv must not receive an error
        proxy_test(
            ps_send => "mg /response/hit\r\n",
            be_recv => {0 => ["mg /response/hit\r\n"]},
            be_send => {0 => ["HD\r\n"]},
            ps_recv => ["HD\r\n"],
        );
    };

    subtest 'response:hit() get' => sub {
        # ps_recv must not receive an error
        my $key = "/response/hit";
        proxy_test(
            ps_send => "get $key\r\n",
            be_recv => {0 => ["get $key\r\n"]},
            be_send => {0 => ["VALUE $key 0 1\r\na\r\nEND\r\n"]},
            ps_recv => ["VALUE $key 0 1\r\n", "a\r\n", "END\r\n"],
        );
    };

    subtest 'response:hit() false 1' => sub {
        # ps_recv must receive an error
        proxy_test(
            ps_send => "mg /response/not_hit\r\n",
            be_recv => {0 => ["mg /response/not_hit\r\n"]},
            be_send => {0 => ["EN\r\n"]},
            ps_recv => ["SERVER_ERROR\r\n"],
        );
    };

    subtest 'response:hit() false 2' => sub {
        # ps_recv must receive an error
        proxy_test(
            ps_send => "get /response/not_hit\r\n",
            be_recv => {0 => ["get /response/not_hit\r\n"]},
            be_send => {0 => ["END\r\n"]},
            ps_recv => ["SERVER_ERROR\r\n"],
        );
    };

    subtest 'response:hit() false 3' => sub {
        # ps_recv must receive an error
        proxy_test(
            ps_send => "mg /response/not_hit\r\n",
            be_recv => {0 => ["mg /response/not_hit\r\n"]},
            be_send => {0 => ["SERVER_ERROR\r\n"]},
            ps_recv => ["SERVER_ERROR\r\n"],
        );
    };

    subtest 'response:vlen()' => sub {
        # ps_recv must not receive an error
        proxy_test(
            ps_send => "mg /response/vlen v\r\n",
            be_recv => {0 => ["mg /response/vlen v\r\n"]},
            be_send => {0 => ["VA 1 v\r\n", "4\r\n"]},
            ps_recv => ["VA 1 v\r\n", "4\r\n"],
        );
    };

    subtest 'response:code() MCMC_CODE_OK' => sub {
        # ps_recv must not receive an error
        my $cmd = "mg /response/code_ok v\r\n";
        proxy_test(
            ps_send => $cmd,
            be_recv => {0 => [$cmd]},
            be_send => {0 => ["VA 1 v\r\n", "4\r\n"]},
            ps_recv => ["VA 1 v\r\n", "4\r\n"],
        );

        proxy_test(
            ps_send => "ms /response/code_ok 1\r\na\r\n",
            be_recv => {0 => ["ms /response/code_ok 1\r\n", "a\r\n"]},
            be_send => {0 => ["HD\r\n"]},
            ps_recv => ["HD\r\n"],
        );
    };

    subtest 'response:code() MCMC_CODE_MISS' => sub {
        # ps_recv must not receive an error
        my $cmd = "mg /response/code_miss v\r\n";
        proxy_test(
            ps_send => $cmd,
            be_recv => {0 => [$cmd]},
            be_send => {0 => ["EN\r\n"]},
            ps_recv => ["EN\r\n"],
        );
    };

    subtest 'response:code() MCMC_CODE_STORED' => sub {
        # ps_recv must not receive an error
        proxy_test(
            ps_send => "set /response/code_stored 0 0 1\r\na\r\n",
            be_recv => {0 => ["set /response/code_stored 0 0 1\r\n", "a\r\n"]},
            be_send => {0 => ["STORED\r\n"]},
            ps_recv => ["STORED\r\n"],
        );
    };

    subtest 'response:code() MCMC_CODE_EXISTS' => sub {
        # ps_recv must not receive an error
        proxy_test(
            ps_send => "set /response/code_exists 0 0 1\r\na\r\n",
            be_recv => {0 => ["set /response/code_exists 0 0 1\r\n", "a\r\n"]},
            be_send => {0 => ["EXISTS\r\n"]},
            ps_recv => ["EXISTS\r\n"],
        );
    };

    subtest 'response:code() MCMC_CODE_NOT_STORED' => sub {
        # ps_recv must not receive an error
        proxy_test(
            ps_send => "set /response/code_not_stored 0 0 1\r\na\r\n",
            be_recv => {0 => ["set /response/code_not_stored 0 0 1\r\n", "a\r\n"]},
            be_send => {0 => ["NOT_STORED\r\n"]},
            ps_recv => ["NOT_STORED\r\n"],
        );
    };

    subtest 'response:code() MCMC_CODE_NOT_FOUND' => sub {
        # ps_recv must not receive an error
        proxy_test(
            ps_send => "set /response/code_not_found 0 0 1\r\na\r\n",
            be_recv => {0 => ["set /response/code_not_found 0 0 1\r\n", "a\r\n"]},
            be_send => {0 => ["NOT_FOUND\r\n"]},
            ps_recv => ["NOT_FOUND\r\n"],
        );
    };

    subtest 'response:code() MCMC_CODE_TOUCHED' => sub {
        # ps_recv must not receive an error
        proxy_test(
            ps_send => "touch /response/code_touched 50\r\n",
            be_recv => {0 => ["touch /response/code_touched 50\r\n"]},
            be_send => {0 => ["TOUCHED\r\n"]},
            ps_recv => ["TOUCHED\r\n"],
        );
    };

    subtest 'response:code() MCMC_CODE_DELETED' => sub {
        # ps_recv must not receive an error
        proxy_test(
            ps_send => "delete /response/code_deleted\r\n",
            be_recv => {0 => ["delete /response/code_deleted\r\n"]},
            be_send => {0 => ["DELETED\r\n"]},
            ps_recv => ["DELETED\r\n"],
        );
    };

    subtest 'response:line()' => sub {
        # ps_recv must not receive an error
        my $cmd = "mg /response/line v\r\n";
        proxy_test(
            ps_send => $cmd,
            be_recv => {0 => [$cmd]},
            be_send => {0 => ["VA 1 v c123\r\n", "a\r\n"]},
            ps_recv => ["VA 1 v c123\r\n", "a\r\n"],
        );

        # ps_recv must not receive an error
        proxy_test(
            ps_send => "ms /response/line 2\r\nab\r\n",
            be_recv => {0 => ["ms /response/line 2\r\n", "ab\r\n"]},
            be_send => {0 => ["HD O123 C123\r\n"]},
            ps_recv => ["HD O123 C123\r\n"],
        );
    };

    subtest 'response:flag_blank()' => sub {
        proxy_test(
            ps_send => "mg /response/blank f Ofoo t\r\n",
            be_recv => {0 => ["mg /response/blank f Ofoo t\r\n"]},
            be_send => {0 => ["HD f1234 Ofoo t999\r\n"]},
            ps_recv => ["HD f1234      t999\r\n"],
        );
    };
}


# Test requests land in proper backend in basic scenarios
{
    note("Test routing by zone:" . __LINE__);

    # TODO: maybe should send values to ensure the right response?
    # I don't think this test is very useful though; probably better to try
    # harder when testing error conditions.
    for my $tu (['a', $mbe[0]], ['b', $mbe[1]], ['c', $mbe[2]]) {
        my $be = $tu->[1];
        my $cmd = "get /zonetest/" . $tu->[0] . "\r\n";
        print $ps $cmd;
        is(scalar <$be>, $cmd, "routed proper zone: " . $tu->[0]);
        print $be "END\r\n";
        is(scalar <$ps>, "END\r\n", "end from zone fetch");
    }
    my $cmd = "get /zonetest/invalid\r\n";
    print $ps $cmd;
    is(scalar <$ps>, "END\r\n", "END from invalid route");
}

check_sanity($ps);
# Test re-requests in lua.
# - fetch zones.z1() then fetch zones.z2()
# - return z1 or z2 or netiher
# - fetch all three zones
# - hit the same zone multiple times

# Test delayed read (timeout)

# Test Lua logging (see t/watcher.t)
{
    note("Test Lua logging:" . __LINE__);

    my $be = $mbe[0];
    my $watcher = $p_srv->new_sock;
    print $watcher "watch proxyuser proxyreqs\n";
    is(<$watcher>, "OK\r\n", "watcher enabled");

    # log(msg)
    print $ps "get /logtest/a\r\n";
    like(<$watcher>, qr/ts=(\S+) gid=\d+ type=proxy_user msg=testing manual log messages/,
        "log a manual message");
    is(scalar <$ps>, "END\r\n", "logtest END");

    # log_req(r, res)
    my $cmd = "get /logreqtest/a\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "got passthru for log");
    print $be "END\r\n";
    is(scalar <$ps>, "END\r\n", "got END from log test");
    like(<$watcher>, qr/ts=(\S+) gid=\d+ type=proxy_req elapsed=\d+ type=105 code=17 status=0 res=ok cfd=\d+ be=127.0.0.1:11411 detail=logreqtest req=get \/logreqtest\/a/, "found request log entry");

    # test log_req with nil res (should be 0's in places)
    # log_reqsample()
    $cmd = "get /logreqstest/a\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "got passthru for logsamp");
    print $be "END\r\n";
    is(scalar <$ps>, "END\r\n", "got END from log test");

    $cmd = "get /logreqstest/b\r\n";
    print $ps $cmd;
    is(scalar <$be>, $cmd, "got passthru for logsamp");

    # cause the sampler time limit to trigger.
    sleep 0.3;
    print $be "END\r\n";
    is(scalar <$ps>, "END\r\n", "got END from log test");
    like(<$watcher>, qr/ts=(\S+) gid=\d+ type=proxy_req elapsed=\d+ type=105 code=17 status=0 res=ok cfd=\d+ be=127.0.0.1:11411 detail=logsampletest req=get \/logreqstest\/b/, "only got b request from log sample");
}

# Test out of spec commands from client
# - wrong # of tokens
# - bad key size
# - etc

# Test errors/garbage from server
# - certain errors pass through to the client, most close the backend.
# - should be able to retrieve the error message
{
    note("Test error/garbage from backend:" . __LINE__);

    my $be = $mbe[0];
    print $ps "set /b/foo 0 0 2\r\nhi\r\n";
    is(scalar <$be>, "set /b/foo 0 0 2\r\n", "received set cmd");
    is(scalar <$be>, "hi\r\n", "received set data");
    # Send a classic back up the pipe.
    my $msg = "SERVER_ERROR object too large for cache\r\n";
    print $be $msg;
    is(scalar <$ps>, $msg, "client received error message");

    print $ps "get /b/foo\r\n";
    is(scalar <$be>, "get /b/foo\r\n", "backend still works");
    print $be "END\r\n";
    is(scalar <$ps>, "END\r\n", "got end back");

    # ERROR and CLIENT_ERROR should both break the backend.
    print $ps "get /b/moo\r\n";
    is(scalar <$be>, "get /b/moo\r\n", "received get command");
    $msg = "CLIENT_ERROR bad command line format\r\n";
    my $data;
    print $be $msg;
    is(scalar <$ps>, $msg, "client received error message");
    my $read = $be->read($data, 1);
    is($read, 0, "backend disconnected");

    $be = $mbe[0] = accept_backend($mocksrvs[0]);

    print $ps "get /b/too\r\n";
    is(scalar <$be>, "get /b/too\r\n", "received get command");
    $msg = "ERROR unhappy\r\n";
    print $be $msg;
    is(scalar <$ps>, $msg, "client received error message");
    $read = $be->read($data, 1);
    is($read, 0, "backend disconnected");

    $be = $mbe[0] = accept_backend($mocksrvs[0]);

    # Sometimes blank ERRORS can be sent.
    print $ps "get /b/zoo\r\n";
    is(scalar <$be>, "get /b/zoo\r\n", "received get command");
    $msg = "ERROR\r\n";
    print $be $msg;
    is(scalar <$ps>, $msg, "client received error message");
    $read = $be->read($data, 1);
    is($read, 0, "backend disconnected");

    $be = $mbe[0] = accept_backend($mocksrvs[0]);

    # Ensure garbage doesn't surface to client.
    print $ps "get /b/doo\r\n";
    is(scalar <$be>, "get /b/doo\r\n", "received get command");
    print $be "garbage\r\n"; # don't need the \r\n but it makes tests easier
    is(scalar <$ps>, "SERVER_ERROR backend failure\r\n", "generic backend error");

    $be = $mbe[0] = accept_backend($mocksrvs[0]);

    # Check errors from pipelined commands past a CLIENT_ERROR
    print $ps "get /b/quu\r\nget /b/muu\r\n";
    is(scalar <$be>, "get /b/quu\r\n", "received get command");
    is(scalar <$be>, "get /b/muu\r\n", "received next get command");
    print $be "CLIENT_ERROR bad protocol\r\nEND\r\n";
    is(scalar <$ps>, "CLIENT_ERROR bad protocol\r\n", "backend error");
    is(scalar <$ps>, "SERVER_ERROR backend failure\r\n", "backend error");

    $be = $mbe[0] = accept_backend($mocksrvs[0]);

    # Check that lua handles errors properly.
    print $ps "get /errcheck/a\r\n";
    is(scalar <$be>, "get /errcheck/a\r\n", "received get command");
    print $be "ERROR test1\r\n";
    is(scalar <$ps>, "ERROR\r\n", "lua saw correct error code");

    $be = $mbe[0] = accept_backend($mocksrvs[0]);

    print $ps "get /errcheck/b\r\n";
    is(scalar <$be>, "get /errcheck/b\r\n", "received get command");
    print $be "CLIENT_ERROR test2\r\n";
    is(scalar <$ps>, "CLIENT_ERROR\r\n", "lua saw correct error code");

    $be = $mbe[0] = accept_backend($mocksrvs[0]);

    print $ps "get /errcheck/c\r\n";
    is(scalar <$be>, "get /errcheck/c\r\n", "received get command");
    print $be "SERVER_ERROR test3\r\n";
    is(scalar <$ps>, "SERVER_ERROR\r\n", "lua saw correct error code");
}

check_sanity($ps);
done_testing();