| // 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 manager |
| |
| import ( |
| "bufio" |
| "strings" |
| "testing" |
| "time" |
| |
| "v.io/v23/context" |
| "v.io/v23/flow" |
| "v.io/v23/naming" |
| _ "v.io/x/ref/runtime/factories/fake" |
| "v.io/x/ref/runtime/internal/flow/conn" |
| "v.io/x/ref/runtime/internal/flow/flowtest" |
| "v.io/x/ref/test" |
| "v.io/x/ref/test/goroutines" |
| ) |
| |
| const leakWaitTime = 250 * time.Millisecond |
| |
| func TestDirectConnection(t *testing.T) { |
| defer goroutines.NoLeaks(t, leakWaitTime)() |
| ctx, shutdown := test.V23Init() |
| ctx, cancel := context.WithCancel(ctx) |
| |
| am := New(ctx, naming.FixedRoutingID(0x5555), nil, 0, 0, nil) |
| if _, err := am.Listen(ctx, "tcp", "127.0.0.1:0"); err != nil { |
| t.Fatal(err) |
| } |
| dm := New(ctx, naming.FixedRoutingID(0x1111), nil, 0, 0, nil) |
| |
| testFlows(t, ctx, dm, am, flowtest.AllowAllPeersAuthorizer{}) |
| |
| cancel() |
| <-am.Closed() |
| <-dm.Closed() |
| shutdown() |
| } |
| |
| func TestDialCachedConn(t *testing.T) { |
| defer goroutines.NoLeaks(t, leakWaitTime)() |
| ctx, shutdown := test.V23Init() |
| ctx, cancel := context.WithCancel(ctx) |
| |
| am := New(ctx, naming.FixedRoutingID(0x5555), nil, 0, 0, nil) |
| if _, err := am.Listen(ctx, "tcp", "127.0.0.1:0"); err != nil { |
| t.Fatal(err) |
| } |
| |
| dm := New(ctx, naming.FixedRoutingID(0x1111), nil, 0, 0, nil) |
| // At first the cache should be empty. |
| if got, want := len(dm.(*manager).cache.addrCache), 0; got != want { |
| t.Fatalf("got cache size %v, want %v", got, want) |
| } |
| // After dialing a connection the cache should hold one connection. |
| testFlows(t, ctx, dm, am, flowtest.AllowAllPeersAuthorizer{}) |
| if got, want := len(dm.(*manager).cache.addrCache), 1; got != want { |
| t.Fatalf("got cache size %v, want %v", got, want) |
| } |
| old := dm.(*manager).cache.ridCache[am.RoutingID()][0] |
| // After dialing another connection the cache should still hold one connection |
| // because the connections should be reused. |
| testFlows(t, ctx, dm, am, flowtest.AllowAllPeersAuthorizer{}) |
| if got, want := len(dm.(*manager).cache.ridCache[am.RoutingID()]), 1; got != want { |
| t.Errorf("got cache size %v, want %v", got, want) |
| } |
| if c := dm.(*manager).cache.ridCache[am.RoutingID()][0]; c != old { |
| t.Errorf("got kv want %v", c, old) |
| } |
| |
| cancel() |
| <-am.Closed() |
| <-dm.Closed() |
| shutdown() |
| } |
| |
| func TestBidirectionalListeningEndpoint(t *testing.T) { |
| defer goroutines.NoLeaks(t, leakWaitTime)() |
| ctx, shutdown := test.V23Init() |
| ctx, cancel := context.WithCancel(ctx) |
| |
| am := New(ctx, naming.FixedRoutingID(0x5555), nil, 0, 0, nil) |
| if _, err := am.Listen(ctx, "tcp", "127.0.0.1:0"); err != nil { |
| t.Fatal(err) |
| } |
| |
| dm := New(ctx, naming.FixedRoutingID(0x1111), nil, 0, 0, nil) |
| testFlows(t, ctx, dm, am, flowtest.AllowAllPeersAuthorizer{}) |
| // Now am should be able to make a flow to dm even though dm is not listening. |
| testFlows(t, ctx, am, dm, flowtest.AllowAllPeersAuthorizer{}) |
| |
| cancel() |
| <-am.Closed() |
| <-dm.Closed() |
| shutdown() |
| } |
| |
| func TestPublicKeyOnlyClientBlessings(t *testing.T) { |
| defer goroutines.NoLeaks(t, leakWaitTime)() |
| ctx, shutdown := test.V23Init() |
| ctx, cancel := context.WithCancel(ctx) |
| |
| am := New(ctx, naming.FixedRoutingID(0x5555), nil, 0, 0, nil) |
| if _, err := am.Listen(ctx, "tcp", "127.0.0.1:0"); err != nil { |
| t.Fatal(err) |
| } |
| nulldm := New(ctx, naming.NullRoutingID, nil, 0, 0, nil) |
| _, af := testFlows(t, ctx, nulldm, am, flowtest.AllowAllPeersAuthorizer{}) |
| // Ensure that the remote blessings of the underlying conn of the accepted blessings |
| // only has the public key of the client and no certificates. |
| if rBlessings := af.Conn().(*conn.Conn).RemoteBlessings(); len(rBlessings.String()) > 0 || rBlessings.PublicKey() == nil { |
| t.Errorf("got %v, want no-cert blessings", rBlessings) |
| } |
| dm := New(ctx, naming.FixedRoutingID(0x1111), nil, 0, 0, nil) |
| _, af = testFlows(t, ctx, dm, am, flowtest.AllowAllPeersAuthorizer{}) |
| // Ensure that the remote blessings of the underlying conn of the accepted flow are |
| // non-zero if we did specify a RoutingID. |
| if rBlessings := af.Conn().(*conn.Conn).RemoteBlessings(); len(rBlessings.String()) == 0 { |
| t.Errorf("got %v, want full blessings", rBlessings) |
| } |
| |
| cancel() |
| <-am.Closed() |
| <-dm.Closed() |
| <-nulldm.Closed() |
| shutdown() |
| } |
| |
| func TestStopListening(t *testing.T) { |
| defer goroutines.NoLeaks(t, leakWaitTime)() |
| ctx, shutdown := test.V23Init() |
| ctx, cancel := context.WithCancel(ctx) |
| |
| am := New(ctx, naming.FixedRoutingID(0x5555), nil, 0, 0, nil) |
| if _, err := am.Listen(ctx, "tcp", "127.0.0.1:0"); err != nil { |
| t.Fatal(err) |
| } |
| dm := New(ctx, naming.FixedRoutingID(0x1111), nil, 0, 0, nil) |
| testFlows(t, ctx, dm, am, flowtest.AllowAllPeersAuthorizer{}) |
| |
| lameEP := am.Status().Endpoints[0] |
| am.StopListening(ctx) |
| |
| if f, err := dm.Dial(ctx, lameEP, flowtest.AllowAllPeersAuthorizer{}, 0); err == nil { |
| t.Errorf("dialing a lame duck should fail, but didn't %#v.", f) |
| } |
| |
| cancel() |
| <-am.Closed() |
| <-dm.Closed() |
| shutdown() |
| } |
| |
| func testFlows(t testing.TB, ctx *context.T, dm, am flow.Manager, auth flow.PeerAuthorizer) (df, af flow.Flow) { |
| ep := am.Status().Endpoints[0] |
| var err error |
| df, err = dm.Dial(ctx, ep, auth, 0) |
| if err != nil { |
| t.Fatal(err) |
| } |
| want := "do you read me?" |
| writeLine(df, want) |
| af, err = am.Accept(ctx) |
| if err != nil { |
| t.Fatal(err) |
| } |
| |
| got, err := readLine(af) |
| if err != nil { |
| t.Error(err) |
| } |
| if got != want { |
| t.Errorf("got %v, want %v", got, want) |
| } |
| |
| want = "i read you" |
| if err := writeLine(af, want); err != nil { |
| t.Error(err) |
| } |
| got, err = readLine(df) |
| if err != nil { |
| t.Error(err) |
| } |
| if got != want { |
| t.Errorf("got %v, want %v", got, want) |
| } |
| return |
| } |
| |
| func readLine(f flow.Flow) (string, error) { |
| s, err := bufio.NewReader(f).ReadString('\n') |
| return strings.TrimRight(s, "\n"), err |
| } |
| |
| func writeLine(f flow.Flow, data string) error { |
| data += "\n" |
| _, err := f.Write([]byte(data)) |
| return err |
| } |
| |
| func BenchmarkDialCachedConn(b *testing.B) { |
| defer goroutines.NoLeaks(b, leakWaitTime)() |
| ctx, shutdown := test.V23Init() |
| |
| am := New(ctx, naming.FixedRoutingID(0x5555), nil, 0, 0, nil) |
| if _, err := am.Listen(ctx, "tcp", "127.0.0.1:0"); err != nil { |
| b.Fatal(err) |
| } |
| dm := New(ctx, naming.FixedRoutingID(0x1111), nil, 0, 0, nil) |
| // At first the cache should be empty. |
| if got, want := len(dm.(*manager).cache.addrCache), 0; got != want { |
| b.Fatalf("got cache size %v, want %v", got, want) |
| } |
| // After dialing a connection the cache should hold one connection. |
| auth := flowtest.AllowAllPeersAuthorizer{} |
| testFlows(b, ctx, dm, am, auth) |
| if got, want := len(dm.(*manager).cache.addrCache), 1; got != want { |
| b.Fatalf("got cache size %v, want %v", got, want) |
| } |
| ep := am.Status().Endpoints[0] |
| b.ResetTimer() |
| for i := 0; i < b.N; i++ { |
| if _, err := dm.Dial(ctx, ep, auth, 0); err != nil { |
| b.Fatal(err) |
| } |
| } |
| b.StopTimer() |
| shutdown() |
| <-am.Closed() |
| <-dm.Closed() |
| } |
