ref: Rename package ipc to rpc.
We are doing this because rpc better describes what we are offering.
Also now vrpc and rpc are more obviously related.
This has the potential downside of being confused with the go rpc package
but we believe the two packages will ralrely be used together..
MultiPart: 6/11
Change-Id: Ia5c35cb7938ac907be418b495f485aede3621e56
diff --git a/profiles/internal/rpc/test/client_test.go b/profiles/internal/rpc/test/client_test.go
new file mode 100644
index 0000000..7f22784
--- /dev/null
+++ b/profiles/internal/rpc/test/client_test.go
@@ -0,0 +1,512 @@
+package test
+
+import (
+ "fmt"
+ "io"
+ "os"
+ "path/filepath"
+ "runtime"
+ "strings"
+ "testing"
+ "time"
+
+ "v.io/v23"
+ "v.io/v23/context"
+ "v.io/v23/naming"
+ "v.io/v23/options"
+ "v.io/v23/rpc"
+ "v.io/v23/verror"
+
+ "v.io/x/ref/lib/flags/consts"
+ _ "v.io/x/ref/profiles"
+ inaming "v.io/x/ref/profiles/internal/naming"
+ "v.io/x/ref/test"
+ "v.io/x/ref/test/expect"
+ "v.io/x/ref/test/modules"
+ "v.io/x/ref/test/modules/core"
+ tsecurity "v.io/x/ref/test/security"
+)
+
+//go:generate v23 test generate .
+
+func newCtx() (*context.T, v23.Shutdown) {
+ ctx, shutdown := test.InitForTest()
+ v23.GetNamespace(ctx).CacheCtl(naming.DisableCache(true))
+ return ctx, shutdown
+}
+
+func runMountTable(t *testing.T, ctx *context.T) (*modules.Shell, func()) {
+ sh, err := modules.NewShell(ctx, nil, testing.Verbose(), t)
+ if err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ root, err := sh.Start(core.RootMTCommand, nil)
+ if err != nil {
+ t.Fatalf("unexpected error for root mt: %s", err)
+ }
+ sh.Forget(root)
+
+ root.ExpectVar("PID")
+ rootName := root.ExpectVar("MT_NAME")
+
+ deferFn := func() {
+ sh.Cleanup(os.Stderr, os.Stderr)
+ root.Shutdown(os.Stderr, os.Stderr)
+ }
+
+ if t.Failed() {
+ deferFn()
+ t.Fatalf("%s", root.Error())
+ }
+ sh.SetVar(consts.NamespaceRootPrefix, rootName)
+ if err = v23.GetNamespace(ctx).SetRoots(rootName); err != nil {
+ t.Fatalf("unexpected error setting namespace roots: %s", err)
+ }
+
+ 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, ctx *context.T, name string) int {
+ me, err := v23.GetNamespace(ctx).Resolve(ctx, name)
+ if err != nil {
+ return 0
+ }
+ return len(me.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) {
+ ctx, shutdown := newCtx()
+ defer shutdown()
+
+ sh, fn := runMountTable(t, ctx)
+ defer fn()
+ srv, err := sh.Start(core.EchoServerCommand, nil, "echoServer", "echoServer")
+ if err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ s := expect.NewSession(t, srv.Stdout(), time.Minute)
+ s.ExpectVar("PID")
+ s.ExpectVar("NAME")
+
+ // Verify that there are 1 entries for echoServer in the mount table.
+ if got, want := numServers(t, ctx, "echoServer"), 1; got != want {
+ t.Fatalf("got: %d, want: %d", got, want)
+ }
+
+ runClient(t, sh)
+
+ // Create a fake set of 100 entries in the mount table
+ 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 := v23.GetNamespace(ctx).Mount(ctx, "echoServer", n, time.Hour); err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ }
+
+ // Verify that there are 101 entries for echoServer in the mount table.
+ if got, want := numServers(t, ctx, "echoServer"), 101; got != want {
+ t.Fatalf("got: %q, want: %q", 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, ctx, "echoServer"), 100; got != want {
+ t.Fatalf("got: %d, want: %d", got, want)
+ }
+}
+
+func TestTimeoutCall(t *testing.T) {
+ ctx, shutdown := newCtx()
+ defer shutdown()
+ ctx, _ = context.WithTimeout(ctx, 100*time.Millisecond)
+ name := naming.JoinAddressName(naming.FormatEndpoint("tcp", "203.0.113.10:443"), "")
+ client := v23.GetClient(ctx)
+ _, err := client.StartCall(ctx, name, "echo", []interface{}{"args don't matter"})
+ if !verror.Is(err, verror.ErrTimeout.ID) {
+ t.Fatalf("wrong error: %s", err)
+ }
+}
+
+func childPing(stdin io.Reader, stdout, stderr io.Writer, env map[string]string, args ...string) error {
+ ctx, shutdown := test.InitForTest()
+ defer shutdown()
+ v23.GetNamespace(ctx).CacheCtl(naming.DisableCache(true))
+
+ name := args[0]
+ call, err := v23.GetClient(ctx).StartCall(ctx, name, "Ping", nil)
+ if err != nil {
+ fmt.Errorf("unexpected error: %s", err)
+ }
+ got := ""
+ if err := call.Finish(&got); err != nil {
+ fmt.Errorf("unexpected error: %s", err)
+ }
+ fmt.Fprintf(stdout, "RESULT=%s\n", got)
+ return nil
+}
+
+func initServer(t *testing.T, ctx *context.T) (string, func()) {
+ server, err := v23.NewServer(ctx)
+ if err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ done := make(chan struct{})
+ deferFn := func() { close(done); server.Stop() }
+
+ eps, err := server.Listen(v23.GetListenSpec(ctx))
+ if err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ server.Serve("", &simple{done}, nil)
+ return eps[0].Name(), 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)
+ if !verror.Is(err, verr.ID) {
+ if _, ok := err.(verror.E); !ok {
+ t.Fatalf("%s: err %v not a verror", loc, err)
+ }
+ stack := ""
+ if err != nil {
+ stack = verror.Stack(err).String()
+ }
+ t.Fatalf("%s: expecting one of: %v, got: %v: stack: %s", loc, verr, err, stack)
+ }
+}
+
+func TestTimeoutResponse(t *testing.T) {
+ ctx, shutdown := newCtx()
+ defer shutdown()
+ name, fn := initServer(t, ctx)
+ defer fn()
+ ctx, _ = context.WithTimeout(ctx, time.Millisecond)
+ call, err := v23.GetClient(ctx).StartCall(ctx, name, "Sleep", nil)
+ if err != nil {
+ testForVerror(t, err, verror.ErrTimeout)
+ return
+ }
+ err = call.Finish()
+ testForVerror(t, err, verror.ErrTimeout)
+}
+
+func TestArgsAndResponses(t *testing.T) {
+ ctx, shutdown := newCtx()
+ defer shutdown()
+ name, fn := initServer(t, ctx)
+ defer fn()
+
+ call, err := v23.GetClient(ctx).StartCall(ctx, name, "Sleep", []interface{}{"too many args"})
+ if err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ err = call.Finish()
+ testForVerror(t, err, verror.ErrBadProtocol)
+
+ call, err = v23.GetClient(ctx).StartCall(ctx, name, "Ping", nil)
+ if err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ pong := ""
+ dummy := ""
+ err = call.Finish(&pong, &dummy)
+ testForVerror(t, err, verror.ErrBadProtocol)
+}
+
+func TestAccessDenied(t *testing.T) {
+ ctx, shutdown := test.InitForTest()
+ defer shutdown()
+
+ name, fn := initServer(t, ctx)
+ defer fn()
+
+ ctx1, err := v23.SetPrincipal(ctx, tsecurity.NewPrincipal("test-blessing"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ call, err := v23.GetClient(ctx1).StartCall(ctx1, name, "Sleep", nil)
+ if err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ err = call.Finish()
+ testForVerror(t, err, verror.ErrNoAccess)
+}
+
+func TestCanceledBeforeFinish(t *testing.T) {
+ ctx, shutdown := newCtx()
+ defer shutdown()
+ name, fn := initServer(t, ctx)
+ defer fn()
+
+ ctx, cancel := context.WithCancel(ctx)
+ call, err := v23.GetClient(ctx).StartCall(ctx, name, "Sleep", nil)
+ if err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ // Cancel before we call finish.
+ cancel()
+ err = call.Finish()
+ // TOO(cnicolaou): this should be Canceled only.
+ testForVerror(t, err, verror.ErrCanceled)
+}
+
+func TestCanceledDuringFinish(t *testing.T) {
+ ctx, shutdown := newCtx()
+ defer shutdown()
+ name, fn := initServer(t, ctx)
+ defer fn()
+
+ ctx, cancel := context.WithCancel(ctx)
+ call, err := v23.GetClient(ctx).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()
+ }()
+ err = call.Finish()
+ testForVerror(t, err, verror.ErrCanceled)
+}
+
+func TestRendezvous(t *testing.T) {
+ ctx, shutdown := newCtx()
+ defer shutdown()
+ sh, fn := runMountTable(t, ctx)
+ 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, "message", name)
+ s := expect.NewSession(t, srv.Stdout(), time.Minute)
+ s.ExpectVar("PID")
+ s.ExpectVar("NAME")
+ }
+ go startServer()
+
+ call, err := v23.GetClient(ctx).StartCall(ctx, name, "Echo", []interface{}{"hello"})
+ if err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+
+ response := ""
+ if err := call.Finish(&response); err != nil {
+ testForVerror(t, err, verror.ErrCanceled)
+ return
+ }
+ if got, want := response, "message: hello\n"; got != want {
+ t.Errorf("got %q, want %q", got, want)
+ }
+}
+
+func TestCallback(t *testing.T) {
+ ctx, shutdown := newCtx()
+ defer shutdown()
+ sh, fn := runMountTable(t, ctx)
+ defer fn()
+
+ name, fn := initServer(t, ctx)
+ defer fn()
+
+ srv, err := sh.Start("childPing", 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) {
+ ctx, shutdown := newCtx()
+ defer shutdown()
+ name, fn := initServer(t, ctx)
+ defer fn()
+
+ want := 10
+ ctx, _ = context.WithTimeout(ctx, 300*time.Millisecond)
+ call, err := v23.GetClient(ctx).StartCall(ctx, name, "Source", []interface{}{want})
+ if err != nil {
+ if !verror.Is(err, verror.ErrTimeout.ID) {
+ t.Fatalf("verror should be a timeout not %s: stack %s",
+ err, verror.Stack(err))
+ }
+ 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.ErrTimeout)
+ break
+ }
+ err = call.Finish()
+ testForVerror(t, err, verror.ErrTimeout)
+}
+
+func TestStreamAbort(t *testing.T) {
+ ctx, shutdown := newCtx()
+ defer shutdown()
+ name, fn := initServer(t, ctx)
+ defer fn()
+
+ call, err := v23.GetClient(ctx).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()
+ err = call.Send(100)
+ testForVerror(t, err, verror.ErrAborted)
+
+ result := 0
+ err = call.Finish(&result)
+ if err != nil {
+ t.Errorf("unexpected error: %#v", err)
+ }
+ if got := result; got != want {
+ t.Errorf("got %d, want %d", got, want)
+ }
+}
+
+func TestNoServersAvailable(t *testing.T) {
+ ctx, shutdown := newCtx()
+ defer shutdown()
+ _, fn := runMountTable(t, ctx)
+ defer fn()
+ name := "noservers"
+ ctx, _ = context.WithTimeout(ctx, 1000*time.Millisecond)
+ call, err := v23.GetClient(ctx).StartCall(ctx, name, "Sleep", nil)
+ if err != nil {
+ testForVerror(t, err, verror.ErrNoServers)
+ return
+ }
+ err = call.Finish()
+ testForVerror(t, err, verror.ErrNoServers)
+}
+
+func TestNoMountTable(t *testing.T) {
+ ctx, shutdown := newCtx()
+ defer shutdown()
+ v23.GetNamespace(ctx).SetRoots()
+ name := "a_mount_table_entry"
+
+ // If there is no mount table, then we'll get a NoServers error message.
+ ctx, _ = context.WithTimeout(ctx, 300*time.Millisecond)
+ _, err := v23.GetClient(ctx).StartCall(ctx, name, "Sleep", nil)
+ testForVerror(t, err, verror.ErrNoServers)
+}
+
+// TestReconnect verifies that the client transparently re-establishes the
+// connection to the server if the server dies and comes back (on the same
+// endpoint).
+func TestReconnect(t *testing.T) {
+ t.Skip()
+ ctx, shutdown := test.InitForTest()
+ defer shutdown()
+
+ sh, err := modules.NewShell(ctx, v23.GetPrincipal(ctx), testing.Verbose(), t)
+ if err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ defer sh.Cleanup(os.Stderr, os.Stderr)
+ server, err := sh.Start(core.EchoServerCommand, nil, "--veyron.tcp.address=127.0.0.1:0", "mymessage", "")
+ if err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ server.ReadLine()
+ serverName := server.ExpectVar("NAME")
+ serverEP, _ := naming.SplitAddressName(serverName)
+ ep, _ := inaming.NewEndpoint(serverEP)
+
+ makeCall := func(ctx *context.T, opts ...rpc.CallOpt) (string, error) {
+ ctx, _ = context.WithDeadline(ctx, time.Now().Add(10*time.Second))
+ call, err := v23.GetClient(ctx).StartCall(ctx, serverName, "Echo", []interface{}{"bratman"}, opts...)
+ if err != nil {
+ return "", fmt.Errorf("START: %s", err)
+ }
+ var result string
+ if err := call.Finish(&result); err != nil {
+ return "", err
+ }
+ return result, nil
+ }
+
+ expected := "mymessage: bratman\n"
+ if result, err := makeCall(ctx); err != nil || result != expected {
+ t.Errorf("Got (%q, %v) want (%q, nil)", result, err, expected)
+ }
+ // Kill the server, verify client can't talk to it anymore.
+ if err := server.Shutdown(os.Stderr, os.Stderr); err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+
+ if _, err := makeCall(ctx, options.NoRetry{}); err == nil || (!strings.HasPrefix(err.Error(), "START") && !strings.Contains(err.Error(), "EOF")) {
+ t.Fatalf(`Got (%v) want ("START: <err>" or "EOF") as server is down`, err)
+ }
+
+ // Resurrect the server with the same address, verify client
+ // re-establishes the connection. This is racy if another
+ // process grabs the port.
+ server, err = sh.Start(core.EchoServerCommand, nil, "--veyron.tcp.address="+ep.Address, "mymessage again", "")
+ if err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ defer server.Shutdown(os.Stderr, os.Stderr)
+ expected = "mymessage again: bratman\n"
+ if result, err := makeCall(ctx); err != nil || result != expected {
+ t.Errorf("Got (%q, %v) want (%q, nil)", result, err, expected)
+ }
+
+}
+
+// TODO(cnicolaou:) tests for:
+// -- Test for bad discharges error and correct invalidation, client.go:870..880
diff --git a/profiles/internal/rpc/test/doc.go b/profiles/internal/rpc/test/doc.go
new file mode 100644
index 0000000..b7b20cb
--- /dev/null
+++ b/profiles/internal/rpc/test/doc.go
@@ -0,0 +1,2 @@
+// package test contains test for rpc code that do not rely on unexposed rpc declarations.
+package test
diff --git a/profiles/internal/rpc/test/glob_test.go b/profiles/internal/rpc/test/glob_test.go
new file mode 100644
index 0000000..6fc55de
--- /dev/null
+++ b/profiles/internal/rpc/test/glob_test.go
@@ -0,0 +1,384 @@
+package test
+
+import (
+ "errors"
+ "fmt"
+ "reflect"
+ "strings"
+ "testing"
+
+ "v.io/v23"
+ "v.io/v23/context"
+ "v.io/v23/i18n"
+ "v.io/v23/naming"
+ "v.io/v23/rpc"
+ "v.io/v23/rpc/reserved"
+ "v.io/v23/security"
+ "v.io/v23/verror"
+
+ "v.io/x/ref/lib/glob"
+ _ "v.io/x/ref/profiles"
+ "v.io/x/ref/test"
+ "v.io/x/ref/test/testutil"
+)
+
+func startGlobServer(ctx *context.T, tree *node) (string, func(), error) {
+ server, err := v23.NewServer(ctx)
+ if err != nil {
+ return "", nil, fmt.Errorf("failed to start debug server: %v", err)
+ }
+ endpoints, err := server.Listen(v23.GetListenSpec(ctx))
+ if err != nil {
+ return "", nil, fmt.Errorf("failed to listen: %v", err)
+ }
+ if err := server.ServeDispatcher("", &disp{tree}); err != nil {
+ return "", nil, err
+ }
+ ep := endpoints[0].String()
+ return ep, func() { server.Stop() }, nil
+}
+
+func TestGlob(t *testing.T) {
+ ctx, shutdown := test.InitForTest()
+ defer shutdown()
+
+ namespace := []string{
+ "a/b/c1/d1",
+ "a/b/c1/d2",
+ "a/b/c2/d1",
+ "a/b/c2/d2",
+ "a/x/y/z",
+ "leaf",
+ }
+ tree := newNode()
+ for _, p := range namespace {
+ tree.find(strings.Split(p, "/"), true)
+ }
+
+ ep, stop, err := startGlobServer(ctx, tree)
+ if err != nil {
+ t.Fatalf("startGlobServer: %v", err)
+ }
+ defer stop()
+
+ var (
+ noExist = verror.New(verror.ErrNoExist, ctx, "")
+ notImplemented = reserved.NewErrGlobNotImplemented(ctx)
+ maxRecursion = reserved.NewErrGlobMaxRecursionReached(ctx)
+ )
+
+ testcases := []struct {
+ name, pattern string
+ expected []string
+ errors []naming.GlobError
+ }{
+ {"", "...", []string{
+ "",
+ "a",
+ "a/b",
+ "a/b/c1",
+ "a/b/c1/d1",
+ "a/b/c1/d2",
+ "a/b/c2",
+ "a/b/c2/d1",
+ "a/b/c2/d2",
+ "a/x",
+ "a/x/y",
+ "a/x/y/z",
+ "leaf",
+ }, nil},
+ {"a", "...", []string{
+ "",
+ "b",
+ "b/c1",
+ "b/c1/d1",
+ "b/c1/d2",
+ "b/c2",
+ "b/c2/d1",
+ "b/c2/d2",
+ "x",
+ "x/y",
+ "x/y/z",
+ }, nil},
+ {"a/b", "...", []string{
+ "",
+ "c1",
+ "c1/d1",
+ "c1/d2",
+ "c2",
+ "c2/d1",
+ "c2/d2",
+ }, nil},
+ {"a/b/c1", "...", []string{
+ "",
+ "d1",
+ "d2",
+ }, nil},
+ {"a/b/c1/d1", "...", []string{
+ "",
+ }, nil},
+ {"a/x", "...", []string{
+ "",
+ "y",
+ "y/z",
+ }, nil},
+ {"a/x/y", "...", []string{
+ "",
+ "z",
+ }, nil},
+ {"a/x/y/z", "...", []string{
+ "",
+ }, nil},
+ {"", "", []string{""}, nil},
+ {"", "*", []string{"a", "leaf"}, nil},
+ {"a", "", []string{""}, nil},
+ {"a", "*", []string{"b", "x"}, nil},
+ {"a/b", "", []string{""}, nil},
+ {"a/b", "*", []string{"c1", "c2"}, nil},
+ {"a/b/c1", "", []string{""}, nil},
+ {"a/b/c1", "*", []string{"d1", "d2"}, nil},
+ {"a/b/c1/d1", "*", []string{}, nil},
+ {"a/b/c1/d1", "", []string{""}, nil},
+ {"a", "*/c?", []string{"b/c1", "b/c2"}, nil},
+ {"a", "*/*", []string{"b/c1", "b/c2", "x/y"}, nil},
+ {"a", "*/*/*", []string{"b/c1/d1", "b/c1/d2", "b/c2/d1", "b/c2/d2", "x/y/z"}, nil},
+ {"a/x", "*/*", []string{"y/z"}, nil},
+ {"bad", "", []string{}, []naming.GlobError{{Name: "", Error: noExist}}},
+ {"bad/foo", "", []string{}, []naming.GlobError{{Name: "", Error: noExist}}},
+ {"a/bad", "", []string{}, []naming.GlobError{{Name: "", Error: noExist}}},
+ {"a/b/bad", "", []string{}, []naming.GlobError{{Name: "", Error: noExist}}},
+ {"a/b/c1/bad", "", []string{}, []naming.GlobError{{Name: "", Error: noExist}}},
+ {"a/x/bad", "", []string{}, []naming.GlobError{{Name: "", Error: noExist}}},
+ {"a/x/y/bad", "", []string{}, []naming.GlobError{{Name: "", Error: noExist}}},
+ {"leaf", "", []string{""}, nil},
+ {"leaf", "*", []string{}, []naming.GlobError{{Name: "", Error: notImplemented}}},
+ {"leaf/foo", "", []string{}, []naming.GlobError{{Name: "", Error: noExist}}},
+ // muah is an infinite space to test rescursion limit.
+ {"muah", "*", []string{"ha"}, nil},
+ {"muah", "*/*", []string{"ha/ha"}, nil},
+ {"muah", "*/*/*/*/*/*/*/*/*/*/*/*", []string{"ha/ha/ha/ha/ha/ha/ha/ha/ha/ha/ha/ha"}, nil},
+ {"muah", "...", []string{
+ "",
+ "ha",
+ "ha/ha",
+ "ha/ha/ha",
+ "ha/ha/ha/ha",
+ "ha/ha/ha/ha/ha",
+ "ha/ha/ha/ha/ha/ha",
+ "ha/ha/ha/ha/ha/ha/ha",
+ "ha/ha/ha/ha/ha/ha/ha/ha",
+ "ha/ha/ha/ha/ha/ha/ha/ha/ha",
+ "ha/ha/ha/ha/ha/ha/ha/ha/ha/ha",
+ }, []naming.GlobError{{Name: "ha/ha/ha/ha/ha/ha/ha/ha/ha/ha/ha", Error: maxRecursion}}},
+ }
+ for _, tc := range testcases {
+ name := naming.JoinAddressName(ep, tc.name)
+ results, globErrors, err := testutil.GlobName(ctx, name, tc.pattern)
+ if err != nil {
+ t.Errorf("unexpected Glob error for (%q, %q): %v", tc.name, tc.pattern, err)
+ continue
+ }
+ if !reflect.DeepEqual(results, tc.expected) {
+ t.Errorf("unexpected result for (%q, %q). Got %q, want %q", tc.name, tc.pattern, results, tc.expected)
+ }
+ if len(globErrors) != len(tc.errors) {
+ t.Errorf("unexpected number of glob errors for (%q, %q): got %#v, expected %#v", tc.name, tc.pattern, globErrors, tc.errors)
+ }
+ for i, e := range globErrors {
+ if i >= len(tc.errors) {
+ t.Errorf("unexpected glob error for (%q, %q): %#v", tc.name, tc.pattern, e.Error)
+ continue
+ }
+ if e.Name != tc.errors[i].Name {
+ t.Errorf("unexpected glob error for (%q, %q): %v", tc.name, tc.pattern, e)
+ }
+ if got, expected := verror.ErrorID(e.Error), verror.ErrorID(tc.errors[i].Error); got != expected {
+ t.Errorf("unexpected error ID for (%q, %q): Got %v, expected %v", tc.name, tc.pattern, got, expected)
+ }
+ }
+ }
+}
+
+func TestGlobDeny(t *testing.T) {
+ ctx, shutdown := test.InitForTest()
+ defer shutdown()
+
+ tree := newNode()
+ tree.find([]string{"a", "b"}, true)
+ tree.find([]string{"a", "deny", "x"}, true)
+ ep, stop, err := startGlobServer(ctx, tree)
+ if err != nil {
+ t.Fatalf("startGlobServer: %v", err)
+ }
+ defer stop()
+
+ testcases := []struct {
+ name, pattern string
+ results []string
+ numErrors int
+ }{
+ {"", "*", []string{"a"}, 0},
+ {"", "*/*", []string{"a/b"}, 1},
+ {"a", "*", []string{"b"}, 1},
+ {"a/deny", "*", []string{}, 1},
+ {"a/deny/x", "*", []string{}, 1},
+ {"a/deny/y", "*", []string{}, 1},
+ }
+
+ // Ensure that we're getting the english error message.
+ ctx = i18n.ContextWithLangID(ctx, i18n.LangID("en-US"))
+
+ for _, tc := range testcases {
+ name := naming.JoinAddressName(ep, tc.name)
+ results, globErrors, err := testutil.GlobName(ctx, name, tc.pattern)
+ if err != nil {
+ t.Errorf("unexpected Glob error for (%q, %q): %v", tc.name, tc.pattern, err)
+ }
+ if !reflect.DeepEqual(results, tc.results) {
+ t.Errorf("unexpected results for (%q, %q). Got %v, expected %v", tc.name, tc.pattern, results, tc.results)
+ }
+ if len(globErrors) != tc.numErrors {
+ t.Errorf("unexpected number of errors for (%q, %q). Got %v, expected %v", tc.name, tc.pattern, len(globErrors), tc.numErrors)
+ }
+ for _, gerr := range globErrors {
+ if got, expected := verror.ErrorID(gerr.Error), reserved.ErrGlobMatchesOmitted.ID; got != expected {
+ t.Errorf("unexpected error for (%q, %q): Got %v, expected %v", tc.name, tc.pattern, got, expected)
+ }
+ // This error message is purposely vague to avoid leaking information that
+ // the user doesn't have access to.
+ // We check the actual error string to make sure that we don't start
+ // leaking new information by accident.
+ expectedStr := fmt.Sprintf(
+ `test.test:"%s".__Glob: some matches might have been omitted`,
+ tc.name)
+ if got := gerr.Error.Error(); got != expectedStr {
+ t.Errorf("unexpected error string: Got %q, expected %q", got, expectedStr)
+ }
+ }
+ }
+}
+
+type disp struct {
+ tree *node
+}
+
+func (d *disp) Lookup(suffix string) (interface{}, security.Authorizer, error) {
+ elems := strings.Split(suffix, "/")
+ var auth security.Authorizer
+ for _, e := range elems {
+ if e == "deny" {
+ auth = &denyAllAuthorizer{}
+ break
+ }
+ }
+ if len(elems) != 0 && elems[0] == "muah" {
+ // Infinite space. Each node has one child named "ha".
+ return rpc.ChildrenGlobberInvoker("ha"), auth, nil
+
+ }
+ if len(elems) != 0 && elems[len(elems)-1] == "leaf" {
+ return leafObject{}, auth, nil
+ }
+ if len(elems) < 2 || (elems[0] == "a" && elems[1] == "x") {
+ return &vChildrenObject{d.tree, elems}, auth, nil
+ }
+ return &globObject{d.tree, elems}, auth, nil
+}
+
+type denyAllAuthorizer struct{}
+
+func (denyAllAuthorizer) Authorize(*context.T) error {
+ return errors.New("no access")
+}
+
+type globObject struct {
+ n *node
+ suffix []string
+}
+
+func (o *globObject) Glob__(call rpc.ServerCall, pattern string) (<-chan naming.GlobReply, error) {
+ g, err := glob.Parse(pattern)
+ if err != nil {
+ return nil, err
+ }
+ n := o.n.find(o.suffix, false)
+ if n == nil {
+ return nil, verror.New(verror.ErrNoExist, call.Context(), o.suffix)
+ }
+ ch := make(chan naming.GlobReply)
+ go func() {
+ o.globLoop(ch, "", g, n)
+ close(ch)
+ }()
+ return ch, nil
+}
+
+func (o *globObject) globLoop(ch chan<- naming.GlobReply, name string, g *glob.Glob, n *node) {
+ if g.Len() == 0 {
+ ch <- naming.GlobReplyEntry{naming.MountEntry{Name: name}}
+ }
+ if g.Finished() {
+ return
+ }
+ for leaf, child := range n.children {
+ if ok, _, left := g.MatchInitialSegment(leaf); ok {
+ o.globLoop(ch, naming.Join(name, leaf), left, child)
+ }
+ }
+}
+
+type vChildrenObject struct {
+ n *node
+ suffix []string
+}
+
+func (o *vChildrenObject) GlobChildren__(call rpc.ServerCall) (<-chan string, error) {
+ n := o.n.find(o.suffix, false)
+ if n == nil {
+ return nil, verror.New(verror.ErrNoExist, call.Context(), o.suffix)
+ }
+ ch := make(chan string, len(n.children))
+ for child, _ := range n.children {
+ ch <- child
+ }
+ close(ch)
+ return ch, nil
+}
+
+type node struct {
+ children map[string]*node
+}
+
+func newNode() *node {
+ return &node{make(map[string]*node)}
+}
+
+func (n *node) find(names []string, create bool) *node {
+ if len(names) == 1 && names[0] == "" {
+ return n
+ }
+ for {
+ if len(names) == 0 {
+ return n
+ }
+ if next, ok := n.children[names[0]]; ok {
+ n = next
+ names = names[1:]
+ continue
+ }
+ if create {
+ nn := newNode()
+ n.children[names[0]] = nn
+ n = nn
+ names = names[1:]
+ continue
+ }
+ return nil
+ }
+}
+
+type leafObject struct{}
+
+func (l leafObject) Func(call rpc.ServerCall) error {
+ return nil
+}
diff --git a/profiles/internal/rpc/test/proxy_test.go b/profiles/internal/rpc/test/proxy_test.go
new file mode 100644
index 0000000..689a10f
--- /dev/null
+++ b/profiles/internal/rpc/test/proxy_test.go
@@ -0,0 +1,383 @@
+package test
+
+import (
+ "fmt"
+ "io"
+ "os"
+ "reflect"
+ "sort"
+ "strings"
+ "testing"
+ "time"
+
+ "v.io/v23"
+ "v.io/v23/context"
+ "v.io/v23/naming"
+ "v.io/v23/naming/ns"
+ "v.io/v23/options"
+ "v.io/v23/rpc"
+ "v.io/v23/security"
+ "v.io/v23/verror"
+ "v.io/v23/vtrace"
+
+ "v.io/x/ref/lib/flags"
+ _ "v.io/x/ref/profiles"
+ "v.io/x/ref/profiles/internal/lib/publisher"
+ inaming "v.io/x/ref/profiles/internal/naming"
+ irpc "v.io/x/ref/profiles/internal/rpc"
+ imanager "v.io/x/ref/profiles/internal/rpc/stream/manager"
+ "v.io/x/ref/profiles/internal/rpc/stream/proxy"
+ tnaming "v.io/x/ref/profiles/internal/testing/mocks/naming"
+ ivtrace "v.io/x/ref/profiles/internal/vtrace"
+ "v.io/x/ref/test/modules"
+ tsecurity "v.io/x/ref/test/security"
+)
+
+func testContext() (*context.T, func()) {
+ ctx, shutdown := v23.Init()
+ ctx, _ = context.WithTimeout(ctx, 20*time.Second)
+ var err error
+ if ctx, err = ivtrace.Init(ctx, flags.VtraceFlags{}); err != nil {
+ panic(err)
+ }
+ ctx, _ = vtrace.SetNewTrace(ctx)
+ return ctx, shutdown
+}
+
+func proxyServer(stdin io.Reader, stdout, stderr io.Writer, env map[string]string, args ...string) error {
+ ctx, shutdown := v23.Init()
+ defer shutdown()
+
+ expected := len(args)
+ listenSpec := v23.GetListenSpec(ctx)
+ protocol := listenSpec.Addrs[0].Protocol
+ addr := listenSpec.Addrs[0].Address
+ proxyShutdown, proxyEp, err := proxy.New(ctx, protocol, addr, "")
+ if err != nil {
+ return err
+ }
+ defer proxyShutdown()
+
+ fmt.Fprintf(stdout, "PID=%d\n", os.Getpid())
+ if expected > 0 {
+ pub := publisher.New(ctx, v23.GetNamespace(ctx), time.Minute)
+ defer pub.WaitForStop()
+ defer pub.Stop()
+ pub.AddServer(proxyEp.String(), false)
+ for _, name := range args {
+ if len(name) == 0 {
+ return fmt.Errorf("empty name specified on the command line")
+ }
+ pub.AddName(name)
+ }
+ // Wait for all the entries to be published.
+ for {
+ pubState := pub.Status()
+ if expected == len(pubState) {
+ break
+ }
+ fmt.Fprintf(stderr, "%s\n", pub.DebugString())
+ delay := time.Second
+ fmt.Fprintf(stderr, "Sleeping: %s\n", delay)
+ time.Sleep(delay)
+ }
+ }
+ fmt.Fprintf(stdout, "PROXY_NAME=%s\n", proxyEp.Name())
+ modules.WaitForEOF(stdin)
+ fmt.Fprintf(stdout, "DONE\n")
+ return nil
+}
+
+type testServer struct{}
+
+func (*testServer) Echo(call rpc.ServerCall, arg string) (string, error) {
+ return fmt.Sprintf("method:%q,suffix:%q,arg:%q", call.Method(), call.Suffix(), arg), nil
+}
+
+type testServerAuthorizer struct{}
+
+func (testServerAuthorizer) Authorize(*context.T) error {
+ return nil
+}
+
+type testServerDisp struct{ server interface{} }
+
+func (t testServerDisp) Lookup(suffix string) (interface{}, security.Authorizer, error) {
+ return t.server, testServerAuthorizer{}, nil
+}
+
+type proxyHandle struct {
+ ns ns.Namespace
+ sh *modules.Shell
+ proxy modules.Handle
+ name string
+}
+
+func (h *proxyHandle) Start(t *testing.T, ctx *context.T, args ...string) error {
+ sh, err := modules.NewShell(nil, nil, testing.Verbose(), t)
+ if err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ h.sh = sh
+ p, err := sh.Start("proxyServer", nil, args...)
+ if err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ h.proxy = p
+ p.ReadLine()
+ h.name = p.ExpectVar("PROXY_NAME")
+ if len(h.name) == 0 {
+ t.Fatalf("failed to get PROXY_NAME from proxyd")
+ }
+ return h.ns.Mount(ctx, "proxy", h.name, time.Hour)
+}
+
+func (h *proxyHandle) Stop(ctx *context.T) error {
+ defer h.sh.Cleanup(os.Stderr, os.Stderr)
+ if err := h.proxy.Shutdown(os.Stderr, os.Stderr); err != nil {
+ return err
+ }
+ if len(h.name) == 0 {
+ return nil
+ }
+ return h.ns.Unmount(ctx, "proxy", h.name)
+}
+
+func TestProxyOnly(t *testing.T) {
+ listenSpec := rpc.ListenSpec{Proxy: "proxy"}
+ testProxy(t, listenSpec)
+}
+
+func TestProxy(t *testing.T) {
+ proxyListenSpec := rpc.ListenSpec{Addrs: rpc.ListenAddrs{{"tcp", "127.0.0.1:0"}}}
+ proxyListenSpec.Proxy = "proxy"
+ testProxy(t, proxyListenSpec)
+}
+
+func TestWSProxy(t *testing.T) {
+ proxyListenSpec := rpc.ListenSpec{Addrs: rpc.ListenAddrs{{"tcp", "127.0.0.1:0"}}}
+ proxyListenSpec.Proxy = "proxy"
+ // The proxy uses websockets only, but the server is using tcp.
+ testProxy(t, proxyListenSpec, "--veyron.tcp.protocol=ws")
+}
+
+func testProxy(t *testing.T, spec rpc.ListenSpec, args ...string) {
+ ctx, shutdown := testContext()
+ defer shutdown()
+ var (
+ pserver = tsecurity.NewPrincipal("server")
+ serverKey = pserver.PublicKey()
+ // We use different stream managers for the client and server
+ // to prevent VIF re-use (in other words, we want to test VIF
+ // creation from both the client and server end).
+ smserver = imanager.InternalNew(naming.FixedRoutingID(0x555555555))
+ smclient = imanager.InternalNew(naming.FixedRoutingID(0x444444444))
+ ns = tnaming.NewSimpleNamespace()
+ )
+ defer smserver.Shutdown()
+ defer smclient.Shutdown()
+ client, err := irpc.InternalNewClient(smserver, ns)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer client.Close()
+ serverCtx, _ := v23.SetPrincipal(ctx, pserver)
+ server, err := irpc.InternalNewServer(serverCtx, smserver, ns, nil, pserver)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer server.Stop()
+
+ // If no address is specified then we'll only 'listen' via
+ // the proxy.
+ hasLocalListener := len(spec.Addrs) > 0 && len(spec.Addrs[0].Address) != 0
+
+ name := "mountpoint/server/suffix"
+ makeCall := func(opts ...rpc.CallOpt) (string, error) {
+ ctx, _ := context.WithDeadline(ctx, time.Now().Add(5*time.Second))
+ // Let's fail fast so that the tests don't take as long to run.
+ call, err := client.StartCall(ctx, name, "Echo", []interface{}{"batman"}, opts...)
+ if err != nil {
+ // proxy is down, we should return here/.... prepend
+ // the error with a well known string so that we can test for that.
+ return "", fmt.Errorf("RESOLVE: %s", err)
+ }
+ var result string
+ if err = call.Finish(&result); err != nil {
+ return "", err
+ }
+ return result, nil
+ }
+ proxy := &proxyHandle{ns: ns}
+ if err := proxy.Start(t, ctx, args...); err != nil {
+ t.Fatal(err)
+ }
+ defer proxy.Stop(ctx)
+ addrs := verifyMount(t, ctx, ns, spec.Proxy)
+ if len(addrs) != 1 {
+ t.Fatalf("failed to lookup proxy")
+ }
+
+ eps, err := server.Listen(spec)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := server.ServeDispatcher("mountpoint/server", testServerDisp{&testServer{}}); err != nil {
+ t.Fatal(err)
+ }
+
+ // Proxy connections are started asynchronously, so we need to wait..
+ waitForMountTable := func(ch chan int, expect int) {
+ then := time.Now().Add(time.Minute)
+ for {
+ me, err := ns.Resolve(ctx, name)
+ if err == nil && len(me.Servers) == expect {
+ ch <- 1
+ return
+ }
+ if time.Now().After(then) {
+ t.Fatalf("timed out")
+ }
+ time.Sleep(100 * time.Millisecond)
+ }
+ }
+ waitForServerStatus := func(ch chan int, proxy string) {
+ then := time.Now().Add(time.Minute)
+ for {
+ status := server.Status()
+ if len(status.Proxies) == 1 && status.Proxies[0].Proxy == proxy {
+ ch <- 2
+ return
+ }
+ if time.Now().After(then) {
+ t.Fatalf("timed out")
+ }
+ time.Sleep(100 * time.Millisecond)
+ }
+ }
+ proxyEP, _ := naming.SplitAddressName(addrs[0])
+ proxiedEP, err := inaming.NewEndpoint(proxyEP)
+ if err != nil {
+ t.Fatalf("unexpected error for %q: %s", proxyEP, err)
+ }
+ proxiedEP.RID = naming.FixedRoutingID(0x555555555)
+ proxiedEP.Blessings = []string{"server"}
+ expectedNames := []string{naming.JoinAddressName(proxiedEP.String(), "suffix")}
+ if hasLocalListener {
+ expectedNames = append(expectedNames, naming.JoinAddressName(eps[0].String(), "suffix"))
+ }
+
+ // Proxy connetions are created asynchronously, so we wait for the
+ // expected number of endpoints to appear for the specified service name.
+ ch := make(chan int, 2)
+ go waitForMountTable(ch, len(expectedNames))
+ go waitForServerStatus(ch, spec.Proxy)
+ select {
+ case <-time.After(time.Minute):
+ t.Fatalf("timedout waiting for two entries in the mount table and server status")
+ case i := <-ch:
+ select {
+ case <-time.After(time.Minute):
+ t.Fatalf("timedout waiting for two entries in the mount table or server status")
+ case j := <-ch:
+ if !((i == 1 && j == 2) || (i == 2 && j == 1)) {
+ t.Fatalf("unexpected return values from waiters")
+ }
+ }
+ }
+
+ status := server.Status()
+ if got, want := status.Proxies[0].Endpoint, proxiedEP; !reflect.DeepEqual(got, want) {
+ t.Fatalf("got %q, want %q", got, want)
+ }
+
+ got := []string{}
+ for _, s := range verifyMount(t, ctx, ns, name) {
+ got = append(got, s)
+ }
+ sort.Strings(got)
+ sort.Strings(expectedNames)
+ if !reflect.DeepEqual(got, expectedNames) {
+ t.Errorf("got %v, want %v", got, expectedNames)
+ }
+
+ if hasLocalListener {
+ // Listen will publish both the local and proxied endpoint with the
+ // mount table, given that we're trying to test the proxy, we remove
+ // the local endpoint from the mount table entry! We have to remove both
+ // the tcp and the websocket address.
+ sep := eps[0].String()
+ ns.Unmount(ctx, "mountpoint/server", sep)
+ }
+
+ addrs = verifyMount(t, ctx, ns, name)
+ if len(addrs) != 1 {
+ t.Fatalf("failed to lookup proxy: addrs %v", addrs)
+ }
+
+ // Proxied endpoint should be published and RPC should succeed (through proxy).
+ // Additionally, any server authorizaton options must only apply to the end server
+ // and not the proxy.
+ const expected = `method:"Echo",suffix:"suffix",arg:"batman"`
+ if result, err := makeCall(options.ServerPublicKey{serverKey}); result != expected || err != nil {
+ t.Fatalf("Got (%v, %v) want (%v, nil)", result, err, expected)
+ }
+
+ // Proxy dies, calls should fail and the name should be unmounted.
+ if err := proxy.Stop(ctx); err != nil {
+ t.Fatal(err)
+ }
+
+ if result, err := makeCall(options.NoRetry{}); err == nil || (!strings.HasPrefix(err.Error(), "RESOLVE") && !strings.Contains(err.Error(), "EOF")) {
+ t.Fatalf(`Got (%v, %v) want ("", "RESOLVE: <err>" or "EOF") as proxy is down`, result, err)
+ }
+
+ for {
+ if _, err := ns.Resolve(ctx, name); err != nil {
+ break
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ verifyMountMissing(t, ctx, ns, name)
+
+ status = server.Status()
+ if len(status.Proxies) != 1 || status.Proxies[0].Proxy != spec.Proxy || !verror.Is(status.Proxies[0].Error, verror.ErrNoServers.ID) {
+ t.Fatalf("proxy status is incorrect: %v", status.Proxies)
+ }
+
+ // Proxy restarts, calls should eventually start succeeding.
+ if err := proxy.Start(t, ctx, args...); err != nil {
+ t.Fatal(err)
+ }
+
+ retries := 0
+ for {
+ if result, err := makeCall(); err == nil {
+ if result != expected {
+ t.Errorf("Got (%v, %v) want (%v, nil)", result, err, expected)
+ }
+ break
+ } else {
+ retries++
+ if retries > 10 {
+ t.Fatalf("Failed after 10 attempts: err: %s", err)
+ }
+ }
+ }
+}
+
+func verifyMount(t *testing.T, ctx *context.T, ns ns.Namespace, name string) []string {
+ me, err := ns.Resolve(ctx, name)
+ if err != nil {
+ t.Errorf("%s not found in mounttable", name)
+ return nil
+ }
+ return me.Names()
+}
+
+func verifyMountMissing(t *testing.T, ctx *context.T, ns ns.Namespace, name string) {
+ if me, err := ns.Resolve(ctx, name); err == nil {
+ names := me.Names()
+ t.Errorf("%s not supposed to be found in mounttable; got %d servers instead: %v", name, len(names), names)
+ }
+}
diff --git a/profiles/internal/rpc/test/signature_test.go b/profiles/internal/rpc/test/signature_test.go
new file mode 100644
index 0000000..b6fbf32
--- /dev/null
+++ b/profiles/internal/rpc/test/signature_test.go
@@ -0,0 +1,156 @@
+package test
+
+import (
+ "fmt"
+ "reflect"
+ "testing"
+
+ "v.io/v23"
+ "v.io/v23/context"
+ "v.io/v23/naming"
+ "v.io/v23/rpc"
+ "v.io/v23/rpc/reserved"
+ "v.io/v23/vdl"
+ "v.io/v23/vdlroot/signature"
+
+ _ "v.io/x/ref/profiles"
+ "v.io/x/ref/test"
+)
+
+func startSigServer(ctx *context.T, sig sigImpl) (string, func(), error) {
+ server, err := v23.NewServer(ctx)
+ if err != nil {
+ return "", nil, fmt.Errorf("failed to start sig server: %v", err)
+ }
+ eps, err := server.Listen(v23.GetListenSpec(ctx))
+ if err != nil {
+ return "", nil, fmt.Errorf("failed to listen: %v", err)
+ }
+ if err := server.Serve("", sig, nil); err != nil {
+ return "", nil, err
+ }
+ return eps[0].String(), func() { server.Stop() }, nil
+}
+
+type sigImpl struct{}
+
+func (sigImpl) NonStreaming0(rpc.ServerCall) error { panic("X") }
+func (sigImpl) NonStreaming1(_ rpc.ServerCall, _ string) (int64, error) { panic("X") }
+func (sigImpl) Streaming0(_ *streamStringBool) error { panic("X") }
+func (sigImpl) Streaming1(_ *streamStringBool, _ int64) (float64, error) { panic("X") }
+
+type streamStringBool struct{ rpc.StreamServerCall }
+
+func (*streamStringBool) Init(rpc.StreamServerCall) { panic("X") }
+func (*streamStringBool) RecvStream() interface {
+ Advance() bool
+ Value() string
+ Err() error
+} {
+ panic("X")
+}
+func (*streamStringBool) SendStream() interface {
+ Send(_ bool) error
+} {
+ panic("X")
+}
+
+func TestMethodSignature(t *testing.T) {
+ ctx, shutdown := test.InitForTest()
+ defer shutdown()
+ ep, stop, err := startSigServer(ctx, sigImpl{})
+ if err != nil {
+ t.Fatalf("startSigServer: %v", err)
+ }
+ defer stop()
+ name := naming.JoinAddressName(ep, "")
+
+ tests := []struct {
+ Method string
+ Want signature.Method
+ }{
+ {"NonStreaming0", signature.Method{
+ Name: "NonStreaming0",
+ }},
+ {"NonStreaming1", signature.Method{
+ Name: "NonStreaming1",
+ InArgs: []signature.Arg{{Type: vdl.StringType}},
+ OutArgs: []signature.Arg{{Type: vdl.Int64Type}},
+ }},
+ {"Streaming0", signature.Method{
+ Name: "Streaming0",
+ InStream: &signature.Arg{Type: vdl.StringType},
+ OutStream: &signature.Arg{Type: vdl.BoolType},
+ }},
+ {"Streaming1", signature.Method{
+ Name: "Streaming1",
+ InArgs: []signature.Arg{{Type: vdl.Int64Type}},
+ OutArgs: []signature.Arg{{Type: vdl.Float64Type}},
+ InStream: &signature.Arg{Type: vdl.StringType},
+ OutStream: &signature.Arg{Type: vdl.BoolType},
+ }},
+ }
+ for _, test := range tests {
+ sig, err := reserved.MethodSignature(ctx, name, test.Method)
+ if err != nil {
+ t.Errorf("call failed: %v", err)
+ }
+ if got, want := sig, test.Want; !reflect.DeepEqual(got, want) {
+ t.Errorf("%s got %#v, want %#v", test.Method, got, want)
+ }
+ }
+}
+
+func TestSignature(t *testing.T) {
+ ctx, shutdown := test.InitForTest()
+ defer shutdown()
+ ep, stop, err := startSigServer(ctx, sigImpl{})
+ if err != nil {
+ t.Fatalf("startSigServer: %v", err)
+ }
+ defer stop()
+ name := naming.JoinAddressName(ep, "")
+ sig, err := reserved.Signature(ctx, name)
+ if err != nil {
+ t.Errorf("call failed: %v", err)
+ }
+ if got, want := len(sig), 2; got != want {
+ t.Fatalf("got sig %#v len %d, want %d", sig, got, want)
+ }
+ // Check expected methods.
+ methods := signature.Interface{
+ Doc: "The empty interface contains methods not attached to any interface.",
+ Methods: []signature.Method{
+ {
+ Name: "NonStreaming0",
+ },
+ {
+ Name: "NonStreaming1",
+ InArgs: []signature.Arg{{Type: vdl.StringType}},
+ OutArgs: []signature.Arg{{Type: vdl.Int64Type}},
+ },
+ {
+ Name: "Streaming0",
+ InStream: &signature.Arg{Type: vdl.StringType},
+ OutStream: &signature.Arg{Type: vdl.BoolType},
+ },
+ {
+ Name: "Streaming1",
+ InArgs: []signature.Arg{{Type: vdl.Int64Type}},
+ OutArgs: []signature.Arg{{Type: vdl.Float64Type}},
+ InStream: &signature.Arg{Type: vdl.StringType},
+ OutStream: &signature.Arg{Type: vdl.BoolType},
+ },
+ },
+ }
+ if got, want := sig[0], methods; !reflect.DeepEqual(got, want) {
+ t.Errorf("got sig[0] %#v, want %#v", got, want)
+ }
+ // Check reserved methods.
+ if got, want := sig[1].Name, "__Reserved"; got != want {
+ t.Errorf("got sig[1].Name %q, want %q", got, want)
+ }
+ if got, want := signature.MethodNames(sig[1:2]), []string{"__Glob", "__MethodSignature", "__Signature"}; !reflect.DeepEqual(got, want) {
+ t.Fatalf("got sig[1] methods %v, want %v", got, want)
+ }
+}
diff --git a/profiles/internal/rpc/test/simple_test.go b/profiles/internal/rpc/test/simple_test.go
new file mode 100644
index 0000000..771e0e4
--- /dev/null
+++ b/profiles/internal/rpc/test/simple_test.go
@@ -0,0 +1,122 @@
+package test
+
+import (
+ "io"
+ "testing"
+ "time"
+
+ "v.io/v23"
+ "v.io/v23/rpc"
+)
+
+type simple struct {
+ done <-chan struct{}
+}
+
+func (s *simple) Sleep(call rpc.ServerCall) error {
+ select {
+ case <-s.done:
+ case <-time.After(time.Hour):
+ }
+ return nil
+}
+
+func (s *simple) Ping(call rpc.ServerCall) (string, error) {
+ return "pong", nil
+}
+
+func (s *simple) Source(call rpc.StreamServerCall, 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 *simple) Sink(call rpc.StreamServerCall) (int, error) {
+ i := 0
+ for {
+ if err := call.Recv(&i); err != nil {
+ if err == io.EOF {
+ return i, nil
+ }
+ return 0, err
+ }
+ }
+}
+
+func (s *simple) Inc(call rpc.StreamServerCall, inc int) (int, error) {
+ i := 0
+ for {
+ if err := call.Recv(&i); err != nil {
+ if err == io.EOF {
+ return i, nil
+ }
+ return 0, err
+ }
+ call.Send(i + inc)
+ }
+}
+
+func TestSimpleRPC(t *testing.T) {
+ ctx, shutdown := newCtx()
+ defer shutdown()
+ name, fn := initServer(t, ctx)
+ defer fn()
+
+ client := v23.GetClient(ctx)
+ call, err := client.StartCall(ctx, name, "Ping", nil)
+ if err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ response := ""
+ if err := call.Finish(&response); err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ if got, want := response, "pong"; got != want {
+ t.Fatalf("got %q, want %q", got, want)
+ }
+}
+
+func TestSimpleStreaming(t *testing.T) {
+ ctx, shutdown := newCtx()
+ defer shutdown()
+ name, fn := initServer(t, ctx)
+ defer fn()
+
+ inc := 1
+ call, err := v23.GetClient(ctx).StartCall(ctx, name, "Inc", []interface{}{inc})
+ 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)
+ }
+ got := -1
+ if err = call.Recv(&got); err != nil {
+ t.Fatalf("unexpected error: %s", err)
+ }
+ if want := i + inc; got != want {
+ t.Fatalf("got %d, want %d")
+ }
+ }
+ call.CloseSend()
+ final := -1
+ err = call.Finish(&final)
+ if err != nil {
+ t.Errorf("unexpected error: %#v", err)
+ }
+ if got := final; got != want {
+ t.Fatalf("got %d, want %d", got, want)
+ }
+}
diff --git a/profiles/internal/rpc/test/v23_internal_test.go b/profiles/internal/rpc/test/v23_internal_test.go
new file mode 100644
index 0000000..082fdf0
--- /dev/null
+++ b/profiles/internal/rpc/test/v23_internal_test.go
@@ -0,0 +1,31 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// This file was auto-generated via go generate.
+// DO NOT UPDATE MANUALLY
+package test
+
+import "fmt"
+import "testing"
+import "os"
+
+import "v.io/x/ref/test"
+import "v.io/x/ref/test/modules"
+
+func init() {
+ modules.RegisterChild("childPing", ``, childPing)
+ modules.RegisterChild("proxyServer", ``, proxyServer)
+}
+
+func TestMain(m *testing.M) {
+ test.Init()
+ if modules.IsModulesChildProcess() {
+ if err := modules.Dispatch(); err != nil {
+ fmt.Fprintf(os.Stderr, "modules.Dispatch failed: %v\n", err)
+ os.Exit(1)
+ }
+ return
+ }
+ os.Exit(m.Run())
+}
