<commit_before>// 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.

package test

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

	"v.io/v23"
	"v.io/v23/context"
	"v.io/v23/flow"
	fmessage "v.io/v23/flow/message"
	"v.io/v23/naming"
	"v.io/v23/options"
	"v.io/v23/rpc"
	"v.io/v23/security"
	"v.io/v23/vdlroot/signature"
	"v.io/v23/verror"

	"v.io/x/ref"
	"v.io/x/ref/internal/logger"
	lsecurity "v.io/x/ref/lib/security"
	_ "v.io/x/ref/runtime/factories/generic"
	"v.io/x/ref/runtime/internal/flow/protocols/debug"
	inaming "v.io/x/ref/runtime/internal/naming"
	irpc "v.io/x/ref/runtime/internal/rpc"
	"v.io/x/ref/runtime/internal/rpc/stream/message"
	"v.io/x/ref/runtime/internal/testing/mocks/mocknet"
	"v.io/x/ref/services/mounttable/mounttablelib"
	"v.io/x/ref/test"
	"v.io/x/ref/test/expect"
	"v.io/x/ref/test/modules"
	"v.io/x/ref/test/testutil"
)

//go:generate jiri test generate .

var rootMT = modules.Register(func(env *modules.Env, args ...string) error {
	seclevel := options.SecurityConfidential
	if len(args) == 1 && args[0] == "nosec" {
		seclevel = options.SecurityNone
	}
	return runRootMT(seclevel, env, args...)
}, "rootMT")

func runRootMT(seclevel options.SecurityLevel, env *modules.Env, args ...string) error {
	ctx, shutdown := v23.Init()
	defer shutdown()
	if seclevel == options.SecurityNone && ref.RPCTransitionState() >= ref.XServers {
		ls := v23.GetListenSpec(ctx)
		for i := range ls.Addrs {
			ls.Addrs[i].Protocol, ls.Addrs[i].Address = debug.WrapAddress(
				ls.Addrs[i].Protocol, ls.Addrs[i].Address)
		}
		ctx = v23.WithListenSpec(ctx, ls)
	}
	mt, err := mounttablelib.NewMountTableDispatcher(ctx, "", "", "mounttable")
	if err != nil {
		return fmt.Errorf("mounttablelib.NewMountTableDispatcher failed: %s", err)
	}
	ctx, server, err := v23.WithNewDispatchingServer(ctx, "", mt, options.ServesMountTable(true), seclevel)
	if err != nil {
		return fmt.Errorf("root failed: %v", err)
	}
	fmt.Fprintf(env.Stdout, "PID=%d\n", os.Getpid())
	for _, ep := range server.Status().Endpoints {
		fmt.Fprintf(env.Stdout, "MT_NAME=%s\n", ep.Name())
	}
	modules.WaitForEOF(env.Stdin)
	return nil
}

type treeDispatcher struct{ id string }

func (d treeDispatcher) Lookup(_ *context.T, suffix string) (interface{}, security.Authorizer, error) {
	return &echoServerObject{d.id, suffix}, nil, nil
}

type echoServerObject struct {
	id, suffix string
}

func (es *echoServerObject) Echo(_ *context.T, _ rpc.ServerCall, m string) (string, error) {
	if len(es.suffix) > 0 {
		return fmt.Sprintf("%s.%s: %s\n", es.id, es.suffix, m), nil
	}
	return fmt.Sprintf("%s: %s\n", es.id, m), nil
}

func (es *echoServerObject) Sleep(_ *context.T, _ rpc.ServerCall, d string) error {
	duration, err := time.ParseDuration(d)
	if err != nil {
		return err
	}
	time.Sleep(duration)
	return nil
}

var echoServer = modules.Register(func(env *modules.Env, args ...string) error {
	ctx, shutdown := v23.Init()
	defer shutdown()

	id, mp := args[0], args[1]
	disp := &treeDispatcher{id: id}
	ctx, server, err := v23.WithNewDispatchingServer(ctx, mp, disp)
	if err != nil {
		return err
	}
	fmt.Fprintf(env.Stdout, "PID=%d\n", os.Getpid())
	for _, ep := range server.Status().Endpoints {
		fmt.Fprintf(env.Stdout, "NAME=%s\n", ep.Name())
	}
	modules.WaitForEOF(env.Stdin)
	return nil
}, "echoServer")

var echoClient = modules.Register(func(env *modules.Env, args ...string) error {
	ctx, shutdown := v23.Init()
	defer shutdown()

	name := args[0]
	args = args[1:]
	client := v23.GetClient(ctx)
	for _, a := range args {
		var r string
		if err := client.Call(ctx, name, "Echo", []interface{}{a}, []interface{}{&r}); err != nil {
			return err
		}
		fmt.Fprintf(env.Stdout, r)
	}
	return nil
}, "echoClient")

func runMountTable(t *testing.T, ctx *context.T, args ...string) (*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(nil, rootMT, args...)
	if err != nil {
		t.Fatalf("unexpected error for root mt: %s", err)
	}
	deferFn := func() {
		sh.Cleanup(os.Stderr, os.Stderr)
	}

	root.ExpectVar("PID")
	rootName := root.ExpectVar("MT_NAME")
	if len(rootName) == 0 {
		root.Shutdown(nil, os.Stderr)
	}

	sh.SetVar(ref.EnvNamespacePrefix, 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(nil, echoClient, "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, expected int) int {
	for {
		me, err := v23.GetNamespace(ctx).Resolve(ctx, name)
		if err == nil && len(me.Servers) == expected {
			return expected
		}
		time.Sleep(10 * time.Millisecond)
	}
}

// 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 := test.V23Init()
	defer shutdown()

	sh, fn := runMountTable(t, ctx)
	defer fn()
	srv, err := sh.Start(nil, echoServer, "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), 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), 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), 100; got != want {
		t.Fatalf("got: %d, want: %d", got, want)
	}
}

func TestTimeout(t *testing.T) {
	t.Skip("https://github.com/vanadium/issues/issues/650")

	ctx, shutdown := test.V23Init()
	defer shutdown()
	client := v23.GetClient(ctx)
	ctx, _ = context.WithTimeout(ctx, 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"})
	t.Log(err)
	if verror.ErrorID(err) != verror.ErrTimeout.ID {
		t.Fatalf("wrong error: %s", err)
	}
}

func logErrors(t *testing.T, msg string, logerr, logstack, debugString bool, err error) {
	_, file, line, _ := runtime.Caller(2)
	loc := fmt.Sprintf("%s:%d", filepath.Base(file), line)
	if logerr {
		t.Logf("%s: %s: %v", loc, msg, err)
	}
	if logstack {
		t.Logf("%s: %s: %v", loc, msg, verror.Stack(err).String())
	}
	if debugString {
		t.Logf("%s: %s: %v", loc, msg, verror.DebugString(err))
	}
}

func TestStartCallErrors(t *testing.T) {
	// TODO(suharshs,mattr): This test will be enables once the new client is written and solves the flakiness.
	t.Skip("TODO(suharshs,mattr): This test is very flaky and is temporarily disabled.")
	ctx, shutdown := test.V23Init()
	defer shutdown()
	client := v23.GetClient(ctx)

	ns := v23.GetNamespace(ctx)

	logErr := func(msg string, err error) {
		logErrors(t, msg, true, false, false, err)
	}

	emptyCtx := &context.T{}
	emptyCtx = context.WithLogger(emptyCtx, logger.Global())
	_, err := client.StartCall(emptyCtx, "noname", "nomethod", nil)
	if verror.ErrorID(err) != verror.ErrBadArg.ID {
		t.Fatalf("wrong error: %s", err)
	}
	logErr("no context", err)

	// TODO(mattr): This test doesn't pass because xclient doesn't look for this
	// error.
	// p1 := security.PublicKeyAuthorizer(testutil.NewPrincipal().PublicKey())
	// p2 := security.PublicKeyAuthorizer(testutil.NewPrincipal().PublicKey())
	// _, err = client.StartCall(ctx, "noname", "nomethod", nil, options.ServerAuthorizer{p1}, options.ServerAuthorizer{p2})
	// if verror.ErrorID(err) != verror.ErrBadArg.ID {
	// 	t.Fatalf("wrong error: %s", err)
	// }
	// logErr("too many public keys", err)

	// This will fail with NoServers, but because there is no mount table
	// to communicate with. The error message should include a
	// 'connection refused' string.
	ns.SetRoots("/127.0.0.1:8101")
	_, err = client.StartCall(ctx, "noname", "nomethod", nil, options.NoRetry{})
	if verror.ErrorID(err) != verror.ErrNoServers.ID {
		t.Fatalf("wrong error: %s", err)
	}
	if want := "connection refused"; !strings.Contains(verror.DebugString(err), want) {
		t.Fatalf("wrong error: %s - doesn't contain %q", err, want)
	}
	logErr("no mount table", err)

	// This will fail with NoServers, but because there really is no
	// name registered with the mount table.
	_, shutdown = runMountTable(t, ctx)
	defer shutdown()
	_, err = client.StartCall(ctx, "noname", "nomethod", nil, options.NoRetry{})
	if verror.ErrorID(err) != verror.ErrNoServers.ID {
		t.Fatalf("wrong error: %s", err)
	}
	if unwanted := "connection refused"; strings.Contains(err.Error(), unwanted) {
		t.Fatalf("wrong error: %s - does contain %q", err, unwanted)
	}
	logErr("no name registered", err)

	// The following tests will fail with NoServers, but because there are
	// no protocols that the client and servers (mount table, and "name") share.
	nctx, nclient, err := v23.WithNewClient(ctx, irpc.PreferredProtocols([]string{"wsh"}))

	addr := naming.FormatEndpoint("nope", "127.0.0.1:1081")
	if err := ns.Mount(ctx, "name", addr, time.Minute); err != nil {
		t.Fatal(err)
	}

	// This will fail in its attempt to call ResolveStep to the mount table
	// because we are using both the new context and the new client.
	_, err = nclient.StartCall(nctx, "name", "nomethod", nil, options.NoRetry{})
	if verror.ErrorID(err) != verror.ErrNoServers.ID {
		t.Fatalf("wrong error: %s", err)
	}
	if want := "ResolveStep"; !strings.Contains(err.Error(), want) {
		t.Fatalf("wrong error: %s - doesn't contain %q", err, want)
	}
	logErr("mismatched protocols", err)

	// This will fail in its attempt to invoke the actual RPC because
	// we are using the old context (which supplies the context for the calls
	// to ResolveStep) and the new client.
	_, err = nclient.StartCall(ctx, "name", "nomethod", nil, options.NoRetry{})
	if verror.ErrorID(err) != verror.ErrNoServers.ID {
		t.Fatalf("wrong error: %s", err)
	}
	if want := "nope"; !strings.Contains(err.Error(), want) {
		t.Fatalf("wrong error: %s - doesn't contain %q", err, want)
	}
	if unwanted := "ResolveStep"; strings.Contains(err.Error(), unwanted) {
		t.Fatalf("wrong error: %s - does contain %q", err, unwanted)

	}
	logErr("mismatched protocols", err)

	// The following two tests will fail due to a timeout.
	ns.SetRoots("/203.0.113.10:8101")
	nctx, _ = context.WithTimeout(ctx, 100*time.Millisecond)
	// This call will timeout talking to the mount table.
	call, err := client.StartCall(nctx, "name", "noname", nil, options.NoRetry{})
	if verror.ErrorID(err) != verror.ErrTimeout.ID {
		t.Fatalf("wrong error: %s", err)
	}
	if call != nil {
		t.Fatalf("expected call to be nil")
	}
	logErr("timeout to mount table", err)

	// TODO(mattr): This test is temporarily disabled.  It's very flaky
	// and we're about to move to a new version of the client.  We will
	// fix it in the new client.
	// This, second test, will fail due a timeout contacting the server itself.
	// addr = naming.FormatEndpoint("tcp", "203.0.113.10:8101")

	// nctx, _ = context.WithTimeout(ctx, 100*time.Millisecond)
	// new_name := naming.JoinAddressName(addr, "")
	// call, err = client.StartCall(nctx, new_name, "noname", nil, options.NoRetry{})
	// if verror.ErrorID(err) != verror.ErrTimeout.ID {
	// 	t.Fatalf("wrong error: %s", err)
	// }
	// if call != nil {
	// 	t.Fatalf("expected call to be nil")
	// }
	// logErr("timeout to server", err)
}

func dropDataDialer(ctx *context.T, network, address string, timeout time.Duration) (net.Conn, error) {
	matcher := func(read bool, msg message.T) bool {
		// Drop and close the connection when reading the first data message.
		if _, ok := msg.(*message.Data); ok && read {
			return true
		}
		return false
	}
	opts := mocknet.Opts{
		Mode:              mocknet.V23CloseAtMessage,
		V23MessageMatcher: matcher,
	}
	return mocknet.DialerWithOpts(opts, network, address, timeout)
}

func simpleResolver(ctx *context.T, network, address string) (string, string, error) {
	return network, address, nil
}

func TestStartCallBadProtocol(t *testing.T) {
	if ref.RPCTransitionState() >= ref.XServers {
		t.Skip("This version of the test only runs under the old rpc system.")
	}
	ctx, shutdown := test.V23Init()
	defer shutdown()

	client := v23.GetClient(ctx)

	logErr := func(msg string, err error) {
		logErrors(t, msg, true, false, false, err)
	}

	ns := v23.GetNamespace(ctx)
	simpleListen := func(ctx *context.T, protocol, address string) (net.Listener, error) {
		return net.Listen(protocol, address)
	}
	rpc.RegisterProtocol("dropData", dropDataDialer, simpleResolver, simpleListen)
	// The following test will fail due to a broken connection.
	// We need to run mount table and servers with no security to use
	// the V23CloseAtMessage net.Conn mock.
	_, shutdown = runMountTable(t, ctx, "nosec")
	defer shutdown()
	roots := ns.Roots()
	brkRoot, err := mocknet.RewriteEndpointProtocol(roots[0], "dropData")
	if err != nil {
		t.Fatal(err)
	}
	ns.SetRoots(brkRoot.Name())
	nctx, _ := context.WithTimeout(ctx, 5*time.Second)
	call, err := client.StartCall(nctx, "name", "noname", nil, options.NoRetry{}, options.SecurityNone)
	if verror.ErrorID(err) != verror.ErrNoServers.ID {
		t.Errorf("wrong error: %s", verror.DebugString(err))
	}
	if call != nil {
		t.Errorf("expected call to be nil")
	}
	logErr("broken connection", err)

	// The following test will fail with because the client will set up
	// a secure connection to a server that isn't expecting one.
	nctx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()
	name, fn := initServer(t, ctx, options.SecurityNone)
	defer fn()
	call, err = client.StartCall(nctx, name, "noname", nil, options.NoRetry{})
	if verror.ErrorID(err) != verror.ErrBadProtocol.ID {
		t.Errorf("wrong error: %s", err)
	}
	if call != nil {
		t.Errorf("expected call to be nil")
	}
	logErr("insecure server", err)

	// This is the inverse, secure server, insecure client
	name, fn = initServer(t, ctx)
	defer fn()
	call, err = client.StartCall(nctx, name, "noname", nil, options.NoRetry{}, options.SecurityNone)
	if verror.ErrorID(err) != verror.ErrBadProtocol.ID {
		t.Errorf("wrong error: %s", err)
	}
	if call != nil {
		t.Errorf("expected call to be nil")
	}
	logErr("insecure client", err)
}

type closeConn struct {
	ctx *context.T
	flow.Conn
	closed chan struct{}
}

func (c *closeConn) ReadMsg() ([]byte, error) {
	buf, err := c.Conn.ReadMsg()
	if err == nil {
		if m, err := fmessage.Read(c.ctx, buf); err == nil {
			if _, ok := m.(*fmessage.Data); ok {
				close(c.closed)
				c.Conn.Close()
				return nil, io.EOF
			}
		}
	}
	return buf, err
}

func TestStartCallBadProtocol_NewRPC(t *testing.T) {
	if ref.RPCTransitionState() < ref.XServers {
		t.Skip("This version of the test only runs under the new RPC system")
	}
	ctx, shutdown := test.V23Init()
	defer shutdown()

	client := v23.GetClient(ctx)

	logErr := func(msg string, err error) {
		logErrors(t, msg, true, false, false, err)
	}

	ns := v23.GetNamespace(ctx)
	// The following test will fail due to a broken connection.
	// We need to run mount table and servers with no security to use
	// the V23CloseAtMessage net.Conn mock.
	_, shutdown = runMountTable(t, ctx, "nosec")
	defer shutdown()
	ns.SetRoots(debug.WrapName(ns.Roots()[0]))
	ch := make(chan struct{})
	nctx := debug.WithFilter(ctx, func(c flow.Conn) flow.Conn {
		return &closeConn{ctx, c, ch}
	})
	call, err := client.StartCall(nctx, "name", "noname", nil, options.NoRetry{})
	if verror.ErrorID(err) != verror.ErrNoServers.ID {
		t.Errorf("wrong error: %s", verror.DebugString(err))
	}
	if call != nil {
		t.Errorf("expected call to be nil")
	}
	logErr("broken connection", err)

	// Make sure we failed because we really did close the connection
	// with our filter
	<-ch
}

func TestStartCallSecurity(t *testing.T) {
	ctx, shutdown := test.V23Init()
	defer shutdown()
	client := v23.GetClient(ctx)

	logErr := func(msg string, err error) {
		logErrors(t, msg, true, false, false, err)
	}

	name, fn := initServer(t, ctx)
	defer fn()

	// Create a context with a new principal that doesn't match the server,
	// so that the client will not trust the server.
	ctx1, err := v23.WithPrincipal(ctx, testutil.NewPrincipal("test-blessing"))
	if err != nil {
		t.Fatal(err)
	}
	call, err := client.StartCall(ctx1, name, "noname", nil, options.NoRetry{})
	if verror.ErrorID(err) != verror.ErrNotTrusted.ID {<commit_msg>rpc/test/client_test: Make chan not block forever.<commit_after>// 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.

package test

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

	"v.io/v23"
	"v.io/v23/context"
	"v.io/v23/flow"
	fmessage "v.io/v23/flow/message"
	"v.io/v23/naming"
	"v.io/v23/options"
	"v.io/v23/rpc"
	"v.io/v23/security"
	"v.io/v23/vdlroot/signature"
	"v.io/v23/verror"

	"v.io/x/ref"
	"v.io/x/ref/internal/logger"
	lsecurity "v.io/x/ref/lib/security"
	_ "v.io/x/ref/runtime/factories/generic"
	"v.io/x/ref/runtime/internal/flow/protocols/debug"
	inaming "v.io/x/ref/runtime/internal/naming"
	irpc "v.io/x/ref/runtime/internal/rpc"
	"v.io/x/ref/runtime/internal/rpc/stream/message"
	"v.io/x/ref/runtime/internal/testing/mocks/mocknet"
	"v.io/x/ref/services/mounttable/mounttablelib"
	"v.io/x/ref/test"
	"v.io/x/ref/test/expect"
	"v.io/x/ref/test/modules"
	"v.io/x/ref/test/testutil"
)

//go:generate jiri test generate .

var rootMT = modules.Register(func(env *modules.Env, args ...string) error {
	seclevel := options.SecurityConfidential
	if len(args) == 1 && args[0] == "nosec" {
		seclevel = options.SecurityNone
	}
	return runRootMT(seclevel, env, args...)
}, "rootMT")

func runRootMT(seclevel options.SecurityLevel, env *modules.Env, args ...string) error {
	ctx, shutdown := v23.Init()
	defer shutdown()
	if seclevel == options.SecurityNone && ref.RPCTransitionState() >= ref.XServers {
		ls := v23.GetListenSpec(ctx)
		for i := range ls.Addrs {
			ls.Addrs[i].Protocol, ls.Addrs[i].Address = debug.WrapAddress(
				ls.Addrs[i].Protocol, ls.Addrs[i].Address)
		}
		ctx = v23.WithListenSpec(ctx, ls)
	}
	mt, err := mounttablelib.NewMountTableDispatcher(ctx, "", "", "mounttable")
	if err != nil {
		return fmt.Errorf("mounttablelib.NewMountTableDispatcher failed: %s", err)
	}
	ctx, server, err := v23.WithNewDispatchingServer(ctx, "", mt, options.ServesMountTable(true), seclevel)
	if err != nil {
		return fmt.Errorf("root failed: %v", err)
	}
	fmt.Fprintf(env.Stdout, "PID=%d\n", os.Getpid())
	for _, ep := range server.Status().Endpoints {
		fmt.Fprintf(env.Stdout, "MT_NAME=%s\n", ep.Name())
	}
	modules.WaitForEOF(env.Stdin)
	return nil
}

type treeDispatcher struct{ id string }

func (d treeDispatcher) Lookup(_ *context.T, suffix string) (interface{}, security.Authorizer, error) {
	return &echoServerObject{d.id, suffix}, nil, nil
}

type echoServerObject struct {
	id, suffix string
}

func (es *echoServerObject) Echo(_ *context.T, _ rpc.ServerCall, m string) (string, error) {
	if len(es.suffix) > 0 {
		return fmt.Sprintf("%s.%s: %s\n", es.id, es.suffix, m), nil
	}
	return fmt.Sprintf("%s: %s\n", es.id, m), nil
}

func (es *echoServerObject) Sleep(_ *context.T, _ rpc.ServerCall, d string) error {
	duration, err := time.ParseDuration(d)
	if err != nil {
		return err
	}
	time.Sleep(duration)
	return nil
}

var echoServer = modules.Register(func(env *modules.Env, args ...string) error {
	ctx, shutdown := v23.Init()
	defer shutdown()

	id, mp := args[0], args[1]
	disp := &treeDispatcher{id: id}
	ctx, server, err := v23.WithNewDispatchingServer(ctx, mp, disp)
	if err != nil {
		return err
	}
	fmt.Fprintf(env.Stdout, "PID=%d\n", os.Getpid())
	for _, ep := range server.Status().Endpoints {
		fmt.Fprintf(env.Stdout, "NAME=%s\n", ep.Name())
	}
	modules.WaitForEOF(env.Stdin)
	return nil
}, "echoServer")

var echoClient = modules.Register(func(env *modules.Env, args ...string) error {
	ctx, shutdown := v23.Init()
	defer shutdown()

	name := args[0]
	args = args[1:]
	client := v23.GetClient(ctx)
	for _, a := range args {
		var r string
		if err := client.Call(ctx, name, "Echo", []interface{}{a}, []interface{}{&r}); err != nil {
			return err
		}
		fmt.Fprintf(env.Stdout, r)
	}
	return nil
}, "echoClient")

func runMountTable(t *testing.T, ctx *context.T, args ...string) (*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(nil, rootMT, args...)
	if err != nil {
		t.Fatalf("unexpected error for root mt: %s", err)
	}
	deferFn := func() {
		sh.Cleanup(os.Stderr, os.Stderr)
	}

	root.ExpectVar("PID")
	rootName := root.ExpectVar("MT_NAME")
	if len(rootName) == 0 {
		root.Shutdown(nil, os.Stderr)
	}

	sh.SetVar(ref.EnvNamespacePrefix, 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(nil, echoClient, "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, expected int) int {
	for {
		me, err := v23.GetNamespace(ctx).Resolve(ctx, name)
		if err == nil && len(me.Servers) == expected {
			return expected
		}
		time.Sleep(10 * time.Millisecond)
	}
}

// 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 := test.V23Init()
	defer shutdown()

	sh, fn := runMountTable(t, ctx)
	defer fn()
	srv, err := sh.Start(nil, echoServer, "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), 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), 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), 100; got != want {
		t.Fatalf("got: %d, want: %d", got, want)
	}
}

func TestTimeout(t *testing.T) {
	t.Skip("https://github.com/vanadium/issues/issues/650")

	ctx, shutdown := test.V23Init()
	defer shutdown()
	client := v23.GetClient(ctx)
	ctx, _ = context.WithTimeout(ctx, 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"})
	t.Log(err)
	if verror.ErrorID(err) != verror.ErrTimeout.ID {
		t.Fatalf("wrong error: %s", err)
	}
}

func logErrors(t *testing.T, msg string, logerr, logstack, debugString bool, err error) {
	_, file, line, _ := runtime.Caller(2)
	loc := fmt.Sprintf("%s:%d", filepath.Base(file), line)
	if logerr {
		t.Logf("%s: %s: %v", loc, msg, err)
	}
	if logstack {
		t.Logf("%s: %s: %v", loc, msg, verror.Stack(err).String())
	}
	if debugString {
		t.Logf("%s: %s: %v", loc, msg, verror.DebugString(err))
	}
}

func TestStartCallErrors(t *testing.T) {
	// TODO(suharshs,mattr): This test will be enables once the new client is written and solves the flakiness.
	t.Skip("TODO(suharshs,mattr): This test is very flaky and is temporarily disabled.")
	ctx, shutdown := test.V23Init()
	defer shutdown()
	client := v23.GetClient(ctx)

	ns := v23.GetNamespace(ctx)

	logErr := func(msg string, err error) {
		logErrors(t, msg, true, false, false, err)
	}

	emptyCtx := &context.T{}
	emptyCtx = context.WithLogger(emptyCtx, logger.Global())
	_, err := client.StartCall(emptyCtx, "noname", "nomethod", nil)
	if verror.ErrorID(err) != verror.ErrBadArg.ID {
		t.Fatalf("wrong error: %s", err)
	}
	logErr("no context", err)

	// TODO(mattr): This test doesn't pass because xclient doesn't look for this
	// error.
	// p1 := security.PublicKeyAuthorizer(testutil.NewPrincipal().PublicKey())
	// p2 := security.PublicKeyAuthorizer(testutil.NewPrincipal().PublicKey())
	// _, err = client.StartCall(ctx, "noname", "nomethod", nil, options.ServerAuthorizer{p1}, options.ServerAuthorizer{p2})
	// if verror.ErrorID(err) != verror.ErrBadArg.ID {
	// 	t.Fatalf("wrong error: %s", err)
	// }
	// logErr("too many public keys", err)

	// This will fail with NoServers, but because there is no mount table
	// to communicate with. The error message should include a
	// 'connection refused' string.
	ns.SetRoots("/127.0.0.1:8101")
	_, err = client.StartCall(ctx, "noname", "nomethod", nil, options.NoRetry{})
	if verror.ErrorID(err) != verror.ErrNoServers.ID {
		t.Fatalf("wrong error: %s", err)
	}
	if want := "connection refused"; !strings.Contains(verror.DebugString(err), want) {
		t.Fatalf("wrong error: %s - doesn't contain %q", err, want)
	}
	logErr("no mount table", err)

	// This will fail with NoServers, but because there really is no
	// name registered with the mount table.
	_, shutdown = runMountTable(t, ctx)
	defer shutdown()
	_, err = client.StartCall(ctx, "noname", "nomethod", nil, options.NoRetry{})
	if verror.ErrorID(err) != verror.ErrNoServers.ID {
		t.Fatalf("wrong error: %s", err)
	}
	if unwanted := "connection refused"; strings.Contains(err.Error(), unwanted) {
		t.Fatalf("wrong error: %s - does contain %q", err, unwanted)
	}
	logErr("no name registered", err)

	// The following tests will fail with NoServers, but because there are
	// no protocols that the client and servers (mount table, and "name") share.
	nctx, nclient, err := v23.WithNewClient(ctx, irpc.PreferredProtocols([]string{"wsh"}))

	addr := naming.FormatEndpoint("nope", "127.0.0.1:1081")
	if err := ns.Mount(ctx, "name", addr, time.Minute); err != nil {
		t.Fatal(err)
	}

	// This will fail in its attempt to call ResolveStep to the mount table
	// because we are using both the new context and the new client.
	_, err = nclient.StartCall(nctx, "name", "nomethod", nil, options.NoRetry{})
	if verror.ErrorID(err) != verror.ErrNoServers.ID {
		t.Fatalf("wrong error: %s", err)
	}
	if want := "ResolveStep"; !strings.Contains(err.Error(), want) {
		t.Fatalf("wrong error: %s - doesn't contain %q", err, want)
	}
	logErr("mismatched protocols", err)

	// This will fail in its attempt to invoke the actual RPC because
	// we are using the old context (which supplies the context for the calls
	// to ResolveStep) and the new client.
	_, err = nclient.StartCall(ctx, "name", "nomethod", nil, options.NoRetry{})
	if verror.ErrorID(err) != verror.ErrNoServers.ID {
		t.Fatalf("wrong error: %s", err)
	}
	if want := "nope"; !strings.Contains(err.Error(), want) {
		t.Fatalf("wrong error: %s - doesn't contain %q", err, want)
	}
	if unwanted := "ResolveStep"; strings.Contains(err.Error(), unwanted) {
		t.Fatalf("wrong error: %s - does contain %q", err, unwanted)

	}
	logErr("mismatched protocols", err)

	// The following two tests will fail due to a timeout.
	ns.SetRoots("/203.0.113.10:8101")
	nctx, _ = context.WithTimeout(ctx, 100*time.Millisecond)
	// This call will timeout talking to the mount table.
	call, err := client.StartCall(nctx, "name", "noname", nil, options.NoRetry{})
	if verror.ErrorID(err) != verror.ErrTimeout.ID {
		t.Fatalf("wrong error: %s", err)
	}
	if call != nil {
		t.Fatalf("expected call to be nil")
	}
	logErr("timeout to mount table", err)

	// TODO(mattr): This test is temporarily disabled.  It's very flaky
	// and we're about to move to a new version of the client.  We will
	// fix it in the new client.
	// This, second test, will fail due a timeout contacting the server itself.
	// addr = naming.FormatEndpoint("tcp", "203.0.113.10:8101")

	// nctx, _ = context.WithTimeout(ctx, 100*time.Millisecond)
	// new_name := naming.JoinAddressName(addr, "")
	// call, err = client.StartCall(nctx, new_name, "noname", nil, options.NoRetry{})
	// if verror.ErrorID(err) != verror.ErrTimeout.ID {
	// 	t.Fatalf("wrong error: %s", err)
	// }
	// if call != nil {
	// 	t.Fatalf("expected call to be nil")
	// }
	// logErr("timeout to server", err)
}

func dropDataDialer(ctx *context.T, network, address string, timeout time.Duration) (net.Conn, error) {
	matcher := func(read bool, msg message.T) bool {
		// Drop and close the connection when reading the first data message.
		if _, ok := msg.(*message.Data); ok && read {
			return true
		}
		return false
	}
	opts := mocknet.Opts{
		Mode:              mocknet.V23CloseAtMessage,
		V23MessageMatcher: matcher,
	}
	return mocknet.DialerWithOpts(opts, network, address, timeout)
}

func simpleResolver(ctx *context.T, network, address string) (string, string, error) {
	return network, address, nil
}

func TestStartCallBadProtocol(t *testing.T) {
	if ref.RPCTransitionState() >= ref.XServers {
		t.Skip("This version of the test only runs under the old rpc system.")
	}
	ctx, shutdown := test.V23Init()
	defer shutdown()

	client := v23.GetClient(ctx)

	logErr := func(msg string, err error) {
		logErrors(t, msg, true, false, false, err)
	}

	ns := v23.GetNamespace(ctx)
	simpleListen := func(ctx *context.T, protocol, address string) (net.Listener, error) {
		return net.Listen(protocol, address)
	}
	rpc.RegisterProtocol("dropData", dropDataDialer, simpleResolver, simpleListen)
	// The following test will fail due to a broken connection.
	// We need to run mount table and servers with no security to use
	// the V23CloseAtMessage net.Conn mock.
	_, shutdown = runMountTable(t, ctx, "nosec")
	defer shutdown()
	roots := ns.Roots()
	brkRoot, err := mocknet.RewriteEndpointProtocol(roots[0], "dropData")
	if err != nil {
		t.Fatal(err)
	}
	ns.SetRoots(brkRoot.Name())
	nctx, _ := context.WithTimeout(ctx, 5*time.Second)
	call, err := client.StartCall(nctx, "name", "noname", nil, options.NoRetry{}, options.SecurityNone)
	if verror.ErrorID(err) != verror.ErrNoServers.ID {
		t.Errorf("wrong error: %s", verror.DebugString(err))
	}
	if call != nil {
		t.Errorf("expected call to be nil")
	}
	logErr("broken connection", err)

	// The following test will fail with because the client will set up
	// a secure connection to a server that isn't expecting one.
	nctx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()
	name, fn := initServer(t, ctx, options.SecurityNone)
	defer fn()
	call, err = client.StartCall(nctx, name, "noname", nil, options.NoRetry{})
	if verror.ErrorID(err) != verror.ErrBadProtocol.ID {
		t.Errorf("wrong error: %s", err)
	}
	if call != nil {
		t.Errorf("expected call to be nil")
	}
	logErr("insecure server", err)

	// This is the inverse, secure server, insecure client
	name, fn = initServer(t, ctx)
	defer fn()
	call, err = client.StartCall(nctx, name, "noname", nil, options.NoRetry{}, options.SecurityNone)
	if verror.ErrorID(err) != verror.ErrBadProtocol.ID {
		t.Errorf("wrong error: %s", err)
	}
	if call != nil {
		t.Errorf("expected call to be nil")
	}
	logErr("insecure client", err)
}

type closeConn struct {
	ctx *context.T
	flow.Conn
	closed chan struct{}
}

func (c *closeConn) ReadMsg() ([]byte, error) {
	buf, err := c.Conn.ReadMsg()
	if err == nil {
		if m, err := fmessage.Read(c.ctx, buf); err == nil {
			if _, ok := m.(*fmessage.Data); ok {
				close(c.closed)
				c.Conn.Close()
				return nil, io.EOF
			}
		}
	}
	return buf, err
}

func TestStartCallBadProtocol_NewRPC(t *testing.T) {
	if ref.RPCTransitionState() < ref.XServers {
		t.Skip("This version of the test only runs under the new RPC system")
	}
	ctx, shutdown := test.V23Init()
	defer shutdown()

	client := v23.GetClient(ctx)

	logErr := func(msg string, err error) {
		logErrors(t, msg, true, false, false, err)
	}

	ns := v23.GetNamespace(ctx)
	// The following test will fail due to a broken connection.
	// We need to run mount table and servers with no security to use
	// the V23CloseAtMessage net.Conn mock.
	_, shutdown = runMountTable(t, ctx, "nosec")
	defer shutdown()
	ns.SetRoots(debug.WrapName(ns.Roots()[0]))
	ch := make(chan struct{})
	nctx := debug.WithFilter(ctx, func(c flow.Conn) flow.Conn {
		return &closeConn{ctx, c, ch}
	})
	call, err := client.StartCall(nctx, "name", "noname", nil, options.NoRetry{})
	if verror.ErrorID(err) != verror.ErrNoServers.ID {
		t.Errorf("wrong error: %s", verror.DebugString(err))
	}
	if call != nil {
		t.Errorf("expected call to be nil")
	}
	logErr("broken connection", err)

	// Make sure we failed because we really did close the connection
	// with our filter
	select {
	case <-ch:
	case <-time.After(10 * time.Second):
		t.Errorf("timeout waiting for chan")
	}
}

func TestStartCallSecurity(t *testing.T) {
	ctx, shutdown := test.V23Init()
	defer shutdown()
	client := v23.GetClient(ctx)

	logErr := func(msg string, err error) {
		logErrors(t, msg, true, false, false, err)
	}

	name, fn := initServer(t, ctx)
	defer fn()

	// Create a context with a new principal that doesn't match the server,
	// so that the client will not trust the server.
	ctx1, err := v23.WithPrincipal(ctx, testutil.NewPrincipal("test-blessing"))
	if err != nil {
		t.Fatal(err)
	}
	call, err := client.StartCall(ctx1, name, "noname", nil, options.NoRetry{})
	if verror.ErrorID(err) != verror.ErrNotTrusted.ID {
{'fraction_non_alphanumeric': 0.16054467512818357, 'fraction_numerical': 0.012944439774733125, 'mean_word_length': 3.991888111888112, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 9, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 12, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1676619', 'n_tokens_mistral': 13668, 'n_tokens_neox': 12952, 'n_words': 3536}