runtimes/google/ipc/client_test.go

package ipc_test

import (
	"fmt"
	"io"
	"os"
	"path/filepath"
	"reflect"
	"runtime"
	"testing"
	"time"

	"veyron.io/veyron/veyron2"
	"veyron.io/veyron/veyron2/ipc"
	"veyron.io/veyron/veyron2/naming"
	"veyron.io/veyron/veyron2/rt"
	old_verror "veyron.io/veyron/veyron2/verror"
	verror "veyron.io/veyron/veyron2/verror2"
	"veyron.io/veyron/veyron2/vlog"

	"veyron.io/veyron/veyron/lib/expect"
	"veyron.io/veyron/veyron/lib/flags/consts"
	"veyron.io/veyron/veyron/lib/modules"
	"veyron.io/veyron/veyron/lib/modules/core"
	"veyron.io/veyron/veyron/profiles"
)

var r veyron2.Runtime

func init() {
	modules.RegisterChild("ping", "<name>", childPing)
	var err error
	if r, err = rt.New(); err != nil {
		panic(err)
	}

	r.Namespace().CacheCtl(naming.DisableCache(true))
}

func testArgs(args ...string) []string {
	var targs = []string{"--", "--veyron.tcp.address=127.0.0.1:0"}
	return append(targs, args...)
}

func runMountTable(t *testing.T, r veyron2.Runtime) (*modules.Shell, func()) {
	sh, err := modules.NewShell(r.Principal())
	if err != nil {
		t.Fatalf("unexpected error: %s", err)
	}
	root, err := sh.Start(core.RootMTCommand, nil, testArgs()...)
	if err != nil {
		t.Fatalf("unexpected error for root mt: %s", err)
	}
	sh.Forget(root)

	rootSession := expect.NewSession(t, root.Stdout(), time.Minute)
	rootName := rootSession.ExpectVar("MT_NAME")
	if t.Failed() {
		t.Fatalf("%s", rootSession.Error())
	}
	sh.SetVar(consts.NamespaceRootPrefix, rootName)
	if err = r.Namespace().SetRoots(rootName); err != nil {
		t.Fatalf("unexpected error setting namespace roots: %s", err)
	}

	deferFn := func() {
		if testing.Verbose() {
			vlog.Infof("------ shell cleanup ------")
			sh.Cleanup(os.Stderr, os.Stderr)
			vlog.Infof("------ root shutdown ------")
			root.Shutdown(os.Stderr, os.Stderr)
		} else {
			sh.Cleanup(nil, nil)
			root.Shutdown(nil, nil)
		}
	}
	return sh, deferFn
}

func runClient(t *testing.T, sh *modules.Shell) error {
	clt, err := sh.Start(core.EchoClientCommand, nil, "echoServer", "a message")
	if err != nil {
		t.Fatalf("unexpected error: %s", err)
	}
	s := expect.NewSession(t, clt.Stdout(), 30*time.Second)
	s.Expect("echoServer: a message")
	if s.Failed() {
		return s.Error()
	}
	return nil
}

func numServers(t *testing.T, name string) int {
	servers, err := r.Namespace().Resolve(r.NewContext(), name)
	if err != nil {
		return 0
	}
	return len(servers)
}

// TODO(cnicolaou): figure out how to test and see what the internals
// of tryCall are doing - e.g. using stats counters.
func TestMultipleEndpoints(t *testing.T) {
	sh, fn := runMountTable(t, r)
	defer fn()
	srv, err := sh.Start(core.EchoServerCommand, nil, testArgs("echoServer", "echoServer")...)
	if err != nil {
		t.Fatalf("unexpected error: %s", err)
	}
	s := expect.NewSession(t, srv.Stdout(), time.Minute)
	s.ExpectVar("NAME")

	runClient(t, sh)

	// Create a fake set of 100 entries in the mount table
	ctx := r.NewContext()
	for i := 0; i < 100; i++ {
		// 203.0.113.0 is TEST-NET-3 from RFC5737
		ep := naming.FormatEndpoint("tcp", fmt.Sprintf("203.0.113.%d:443", i))
		n := naming.JoinAddressName(ep, "")
		if err := r.Namespace().Mount(ctx, "echoServer", n, time.Hour); err != nil {
			t.Fatalf("unexpected error: %s", err)
		}
	}

	// Verify that there are 102 entries for echoServer in the mount table.
	if got, want := numServers(t, "echoServer"), 102; got != want {
		t.Fatalf("got: %d, want: %d", got, want)
	}

	// TODO(cnicolaou): ok, so it works, but I'm not sure how
	// long it should take or if the parallel connection code
	// really works. Use counters to inspect it for example.
	if err := runClient(t, sh); err != nil {
		t.Fatalf("unexpected error: %s", err)
	}

	srv.CloseStdin()
	srv.Shutdown(nil, nil)

	// Verify that there are 100 entries for echoServer in the mount table.
	if got, want := numServers(t, "echoServer"), 100; got != want {
		t.Fatalf("got: %d, want: %d", got, want)
	}
}

func TestTimeoutCall(t *testing.T) {
	client := r.Client()
	ctx, _ := r.NewContext().WithTimeout(100 * time.Millisecond)
	name := naming.JoinAddressName(naming.FormatEndpoint("tcp", "203.0.113.10:443"), "")
	_, err := client.StartCall(ctx, name, "echo", []interface{}{"args don't matter"})
	if !verror.Is(err, verror.Timeout.ID) {
		t.Fatalf("wrong error: %s", err)
	}
}

type sleeper struct {
	done <-chan struct{}
}

func (s *sleeper) Sleep(call ipc.ServerContext) error {
	select {
	case <-s.done:
	case <-time.After(time.Hour):
	}
	return nil
}

func (s *sleeper) Ping(call ipc.ServerContext) (string, error) {
	return "pong", nil
}

func (s *sleeper) Source(call ipc.ServerCall, start int) error {
	i := start
	backoff := 25 * time.Millisecond
	for {
		select {
		case <-s.done:
			return nil
		case <-time.After(backoff):
			call.Send(i)
			i++
		}
		backoff *= 2
	}
}

func (s *sleeper) Sink(call ipc.ServerCall) (int, error) {
	i := 0
	for {
		if err := call.Recv(&i); err != nil {
			return i, err
		}
	}
}

func childPing(stdin io.Reader, stdout, stderr io.Writer, env map[string]string, args ...string) error {
	name := args[1]
	call, err := r.Client().StartCall(r.NewContext(), name, "Ping", nil)
	if err != nil {
		fmt.Errorf("unexpected error: %s", err)
	}
	got := ""
	verr := call.Finish(&got, &err)
	if verr != nil {
		fmt.Errorf("unexpected error: %s", verr)
	}
	if err != nil {
		fmt.Errorf("unexpected error: %s", err)
	}
	fmt.Fprintf(stdout, "RESULT=%s\n", got)
	return nil
}

func initServer(t *testing.T, r veyron2.Runtime) (string, ipc.Server, func()) {
	server, err := r.NewServer()
	if err != nil {
		t.Fatalf("unexpected error: %s", err)
	}
	done := make(chan struct{})
	deferFn := func() { close(done); server.Stop() }

	ep, err := server.Listen(profiles.LocalListenSpec)
	if err != nil {
		t.Fatalf("unexpected error: %s", err)
	}
	server.Serve("", &sleeper{done}, nil)
	name := naming.JoinAddressName(ep.String(), "")
	return name, server, deferFn
}

func testForVerror(t *testing.T, err error, verr ...verror.IDAction) {
	_, file, line, _ := runtime.Caller(1)
	loc := fmt.Sprintf("%s:%d", filepath.Base(file), line)
	found := false
	for _, v := range verr {
		if verror.Is(err, v.ID) {
			found = true
			break
		}
	}
	if !found {
		if _, ok := err.(verror.E); !ok {
			t.Fatalf("%s: err %v not a verror", loc, err)
		}
		stack := ""
		if err != nil {
			stack = err.(verror.E).Stack().String()
		}
		t.Fatalf("%s: expecting one of: %v, got: %v: stack: %s", loc, verr, err, stack)
	}
}

func TestTimeoutResponse(t *testing.T) {
	name, _, fn := initServer(t, r)
	defer fn()
	ctx, _ := r.NewContext().WithTimeout(100 * time.Millisecond)
	call, err := r.Client().StartCall(ctx, name, "Sleep", nil)
	if err != nil {
		testForVerror(t, err, verror.Timeout)
		return
	}
	verr := call.Finish(&err)
	// TODO(cnicolaou): this should be Timeout only.
	testForVerror(t, verr, verror.Timeout, verror.BadProtocol)
}

func TestArgsAndResponses(t *testing.T) {
	name, _, fn := initServer(t, r)
	defer fn()

	call, err := r.Client().StartCall(r.NewContext(), name, "Sleep", []interface{}{"too many args"})
	if err != nil {
		t.Fatalf("unexpected error: %s", err)
	}
	verr := call.Finish(&err)
	testForVerror(t, verr, verror.BadProtocol)

	call, err = r.Client().StartCall(r.NewContext(), name, "Ping", nil)
	if err != nil {
		t.Fatalf("unexpected error: %s", err)
	}
	pong := ""
	dummy := ""
	verr = call.Finish(&pong, &dummy, &err)
	testForVerror(t, verr, verror.BadProtocol)
}

func TestAccessDenied(t *testing.T) {
	r1, _ := rt.New()
	r2, _ := rt.New()

	// The server and client use different runtimes and hence different
	// principals and without any shared blessings the server will deny
	// access to the client
	name, _, fn := initServer(t, r1)
	defer fn()

	client := r2.Client()
	call, err := client.StartCall(r2.NewContext(), name, "Sleep", nil)
	if err != nil {
		t.Fatalf("unexpected error: %s", err)
	}
	verr := call.Finish(&err)
	testForVerror(t, verr, verror.NoAccess)
}

func TestCancelledBeforeFinish(t *testing.T) {
	name, _, fn := initServer(t, r)
	defer fn()

	ctx, cancel := r.NewContext().WithCancel()
	call, err := r.Client().StartCall(ctx, name, "Sleep", nil)
	if err != nil {
		t.Fatalf("unexpected error: %s", err)
	}
	// Cancel before we call finish.
	cancel()
	verr := call.Finish(&err)
	// TOO(cnicolaou): this should be Cancelled only.
	testForVerror(t, verr, verror.Cancelled, verror.BadProtocol)
}

func TestCancelledDuringFinish(t *testing.T) {
	name, _, fn := initServer(t, r)
	defer fn()

	ctx, cancel := r.NewContext().WithCancel()
	call, err := r.Client().StartCall(ctx, name, "Sleep", nil)
	if err != nil {
		t.Fatalf("unexpected error: %s", err)
	}
	// Cancel whilst the RPC is running.
	go func() {
		time.Sleep(100 * time.Millisecond)
		cancel()
	}()
	verr := call.Finish(&err)
	// TOO(cnicolaou): this should be Cancelled only.
	testForVerror(t, verr, verror.Cancelled, verror.BadProtocol)
}

func TestRendezvous(t *testing.T) {
	sh, fn := runMountTable(t, r)
	defer fn()

	name := "echoServer"

	// We start the client before we start the server, StartCall will reresolve
	// the name until it finds an entry or timesout after an exponential
	// backoff of some minutes.
	startServer := func() {
		time.Sleep(10 * time.Millisecond)
		srv, _ := sh.Start(core.EchoServerCommand, nil, testArgs("message", name)...)
		s := expect.NewSession(t, srv.Stdout(), time.Minute)
		s.ExpectVar("NAME")
	}
	go startServer()

	call, err := r.Client().StartCall(r.NewContext(), name, "Echo", []interface{}{"hello"})
	if err != nil {
		t.Fatalf("unexpected error: %s", err)
	}

	response := ""
	verr := call.Finish(&response, &err)
	if verr != nil {
		testForVerror(t, verr, verror.Cancelled)
		return
	}
	if got, want := response, "message: hello\n"; got != want {
		t.Errorf("got %q, want %q", got, want)
	}
}

func TestCallback(t *testing.T) {
	sh, fn := runMountTable(t, r)
	defer fn()

	name, _, fn := initServer(t, r)
	defer fn()

	srv, err := sh.Start("ping", nil, name)
	if err != nil {
		t.Fatalf("unexpected error: %s", err)
	}
	s := expect.NewSession(t, srv.Stdout(), time.Minute)
	if got, want := s.ExpectVar("RESULT"), "pong"; got != want {
		t.Errorf("got %q, want %q", got, want)
	}
}

func TestStreamTimeout(t *testing.T) {
	name, _, fn := initServer(t, r)
	defer fn()

	want := 10
	ctx, _ := r.NewContext().WithTimeout(300 * time.Millisecond)
	call, err := r.Client().StartCall(ctx, name, "Source", []interface{}{want})
	if err != nil {
		if !verror.Is(err, verror.Timeout.ID) && !verror.Is(err, verror.BadProtocol.ID) {
			t.Fatalf("verror should be a timeout or badprotocol, not %s: stack %s",
				err, err.(verror.E).Stack())
		}
		return
	}

	for {
		got := 0
		err := call.Recv(&got)
		if err == nil {
			if got != want {
				t.Fatalf("got %d, want %d", got, want)
			}
			want++
			continue
		}
		// TOO(cnicolaou): this should be Timeout only.
		testForVerror(t, err, verror.Timeout, verror.BadProtocol)
		break
	}
	verr := call.Finish(&err)
	testForVerror(t, verr, verror.Timeout, verror.BadProtocol)
}

func TestStreamAbort(t *testing.T) {
	name, _, fn := initServer(t, r)
	defer fn()

	ctx := r.NewContext()
	call, err := r.Client().StartCall(ctx, name, "Sink", nil)
	if err != nil {
		t.Fatalf("unexpected error: %s", err)
	}

	want := 10
	for i := 0; i <= want; i++ {
		if err := call.Send(i); err != nil {
			t.Fatalf("unexpected error: %s", err)
		}
	}
	call.CloseSend()
	verr := call.Send(100)
	testForVerror(t, verr, verror.Aborted)

	result := 0
	verr = call.Finish(&result, &err)
	if verr != nil {
		t.Fatalf("unexpected error: %s", verr)
	}
	if got, want := err, (old_verror.Standard{Msg: "EOF"}); !reflect.DeepEqual(got, want) {
		t.Fatalf("got %v, want %v", got, want)
	}
	if got := result; got != want {
		t.Errorf("got %d, want %d", got, want)
	}
}

func TestNoServersAvailable(t *testing.T) {
	_, fn := runMountTable(t, r)
	defer fn()
	name := "noservers"
	ctx, _ := r.NewContext().WithTimeout(300 * time.Millisecond)
	call, verr := r.Client().StartCall(ctx, name, "Sleep", nil)
	if verr != nil {
		testForVerror(t, verr, verror.Timeout, verror.BadProtocol, verror.NoExist)
		return
	}
	// The local namespace client may return the 'current' name when it encounters
	// a timeout or other networking error, which means we may end up invoking an
	// RPC on that entry - which in our case means we end up invoking:
	// <mount table endpoint>/noservers.Sleep(). This RPC will fail immediately
	// since we've already reached our timeout, but we can't see that error
	// until we call Finish.
	err := call.Finish(&verr)
	testForVerror(t, err, verror.Timeout, verror.BadProtocol)
}

func TestNoMountTable(t *testing.T) {
	r.Namespace().SetRoots()
	name := "a_mount_table_entry"

	// If there is no mount table, then we'll get a NoServers error message.
	ctx, _ := r.NewContext().WithTimeout(300 * time.Millisecond)
	_, verr := r.Client().StartCall(ctx, name, "Sleep", nil)
	testForVerror(t, verr, verror.NoServers)
}

// TODO(cnicolaou:) tests for:
// -- Test for bad discharges error and correct invalidation, client.go:870..880