update manager_test.go

pkg/transport/manager_test.go

@@ -45,12 +45,38 @@ func TestNewManager(t *testing.T) {
 	nEnv := snettest.NewEnv(t, keys, []string{dmsg.Type})
 	defer nEnv.Teardown()
 
-	m0, m1, tp0, tp1, err := transport.CreateTransportPair(tpDisc, keys, nEnv, "dmsg")
+	// Prepare tp manager 0.
+	pk0, sk0 := keys[0].PK, keys[0].SK
+	ls0 := transport.InMemoryTransportLogStore()
+	m0, err := transport.NewManager(nEnv.Nets[0], &transport.ManagerConfig{
+		PubKey:          pk0,
+		SecKey:          sk0,
+		DiscoveryClient: tpDisc,
+		LogStore:        ls0,
+	})
+	require.NoError(t, err)
+	go m0.Serve(context.TODO())
 	defer func() { require.NoError(t, m0.Close()) }()
-	defer func() { require.NoError(t, m1.Close()) }()
 
+	// Prepare tp manager 1.
+	pk1, sk1 := keys[1].PK, keys[1].SK
+	ls1 := transport.InMemoryTransportLogStore()
+	m2, err := transport.NewManager(nEnv.Nets[1], &transport.ManagerConfig{
+		PubKey:          pk1,
+		SecKey:          sk1,
+		DiscoveryClient: tpDisc,
+		LogStore:        ls1,
+	})
 	require.NoError(t, err)
-	require.NotNil(t, tp0)
+	go m2.Serve(context.TODO())
+	defer func() { require.NoError(t, m2.Close()) }()
+
+	// Create data transport between manager 1 & manager 2.
+	tp2, err := m2.SaveTransport(context.TODO(), pk0, "dmsg")
+	require.NoError(t, err)
+	tp1 := m0.Transport(transport.MakeTransportID(pk0, pk1, "dmsg"))
+	require.NotNil(t, tp1)
+
 	fmt.Println("transports created")
 
 	totalSent2 := 0
@@ -63,8 +89,7 @@ func TestNewManager(t *testing.T) {
 			totalSent2 += i
 			rID := routing.RouteID(i)
 			payload := cipher.RandByte(i)
-			packet := routing.MakeDataPacket(rID, payload)
-			require.NoError(t, tp1.WritePacket(context.TODO(), packet))
+			require.NoError(t, tp2.WritePacket(context.TODO(), routing.MakeDataPacket(rID, payload)))
 
 			recv, err := m0.ReadPacket()
 			require.NoError(t, err)
@@ -77,10 +102,9 @@ func TestNewManager(t *testing.T) {
 			totalSent1 += i
 			rID := routing.RouteID(i)
 			payload := cipher.RandByte(i)
-			packet := routing.MakeDataPacket(rID, payload)
-			require.NoError(t, tp0.WritePacket(context.TODO(), packet))
+			require.NoError(t, tp1.WritePacket(context.TODO(), routing.MakeDataPacket(rID, payload)))
 
-			recv, err := m1.ReadPacket()
+			recv, err := m2.ReadPacket()
 			require.NoError(t, err)
 			require.Equal(t, rID, recv.RouteID())
 			require.Equal(t, uint16(i), recv.Size())
@@ -94,12 +118,12 @@ func TestNewManager(t *testing.T) {
 		// 1.5x log write interval just to be safe.
 		time.Sleep(time.Second * 9 / 2)
 
-		entry1, err := m0.Conf.LogStore.Entry(tp0.Entry.ID)
+		entry1, err := ls0.Entry(tp1.Entry.ID)
 		require.NoError(t, err)
 		assert.Equal(t, uint64(totalSent1), entry1.SentBytes)
 		assert.Equal(t, uint64(totalSent2), entry1.RecvBytes)
 
-		entry2, err := m1.Conf.LogStore.Entry(tp1.Entry.ID)
+		entry2, err := ls1.Entry(tp2.Entry.ID)
 		require.NoError(t, err)
 		assert.Equal(t, uint64(totalSent2), entry2.SentBytes)
 		assert.Equal(t, uint64(totalSent1), entry2.RecvBytes)
@@ -109,18 +133,17 @@ func TestNewManager(t *testing.T) {
 	t.Run("check_delete_tp", func(t *testing.T) {
 
 		// Make transport ID.
-		tpID := transport.MakeTransportID(m0.Conf.PubKey, m1.Conf.PubKey, "dmsg")
+		tpID := transport.MakeTransportID(pk0, pk1, "dmsg")
 
 		// Ensure transports are registered properly in tp discovery.
 		entry, err := tpDisc.GetTransportByID(context.TODO(), tpID)
 		require.NoError(t, err)
-		assert.Equal(t, transport.SortEdges(m0.Conf.PubKey, m1.Conf.PubKey), entry.Entry.Edges)
+		assert.Equal(t, transport.SortEdges(pk0, pk1), entry.Entry.Edges)
 		assert.True(t, entry.IsUp)
 
-		m1.DeleteTransport(tp1.Entry.ID)
-		entry, err = tpDisc.GetTransportByID(context.TODO(), tpID)
-		require.NoError(t, err)
-		assert.False(t, entry.IsUp)
+		m2.DeleteTransport(tp2.Entry.ID)
+		_, err = tpDisc.GetTransportByID(context.TODO(), tpID)
+		require.Contains(t, err.Error(), "not found")
 	})
 }