p2p/connection/connection_test.go

   1 package connection
   2
   3 import (
   4         "net"
   5         "testing"
   6         "time"
   7
   8         "github.com/stretchr/testify/assert"
   9         "github.com/stretchr/testify/require"
  10         "github.com/tendermint/tmlibs/log"
  11 )
  12
  13 func createMConnection(conn net.Conn) *MConnection {
  14         onReceive := func(chID byte, msgBytes []byte) {
  15         }
  16         onError := func(r interface{}) {
  17         }
  18         c := createMConnectionWithCallbacks(conn, onReceive, onError)
  19         c.SetLogger(log.TestingLogger())
  20         return c
  21 }
  22
  23 func createMConnectionWithCallbacks(conn net.Conn, onReceive func(chID byte, msgBytes []byte), onError func(r interface{})) *MConnection {
  24         chDescs := []*ChannelDescriptor{&ChannelDescriptor{ID: 0x01, Priority: 1, SendQueueCapacity: 1}}
  25         c := NewMConnectionWithConfig(conn, chDescs, onReceive, onError, DefaultMConnConfig())
  26         c.SetLogger(log.TestingLogger())
  27         return c
  28 }
  29
  30 func TestMConnectionSend(t *testing.T) {
  31         assert, require := assert.New(t), require.New(t)
  32
  33         server, client := net.Pipe()
  34         defer server.Close()
  35         defer client.Close()
  36
  37         mconn := createMConnection(client)
  38         _, err := mconn.Start()
  39         require.Nil(err)
  40         defer mconn.Stop()
  41
  42         msg := "Ant-Man"
  43         assert.True(mconn.Send(0x01, msg))
  44         // Note: subsequent Send/TrySend calls could pass because we are reading from
  45         // the send queue in a separate goroutine.
  46         server.Read(make([]byte, len(msg)))
  47         assert.True(mconn.CanSend(0x01))
  48
  49         msg = "Spider-Man"
  50         assert.True(mconn.TrySend(0x01, msg))
  51         server.Read(make([]byte, len(msg)))
  52
  53         assert.False(mconn.CanSend(0x05), "CanSend should return false because channel is unknown")
  54         assert.False(mconn.Send(0x05, "Absorbing Man"), "Send should return false because channel is unknown")
  55 }
  56
  57 func TestMConnectionReceive(t *testing.T) {
  58         assert, require := assert.New(t), require.New(t)
  59
  60         server, client := net.Pipe()
  61         defer server.Close()
  62         defer client.Close()
  63
  64         receivedCh := make(chan []byte)
  65         errorsCh := make(chan interface{})
  66         onReceive := func(chID byte, msgBytes []byte) {
  67                 receivedCh <- msgBytes
  68         }
  69         onError := func(r interface{}) {
  70                 errorsCh <- r
  71         }
  72         mconn1 := createMConnectionWithCallbacks(client, onReceive, onError)
  73         _, err := mconn1.Start()
  74         require.Nil(err)
  75         defer mconn1.Stop()
  76
  77         mconn2 := createMConnection(server)
  78         _, err = mconn2.Start()
  79         require.Nil(err)
  80         defer mconn2.Stop()
  81
  82         msg := "Cyclops"
  83         assert.True(mconn2.Send(0x01, msg))
  84
  85         select {
  86         case receivedBytes := <-receivedCh:
  87                 assert.Equal([]byte(msg), receivedBytes[2:]) // first 3 bytes are internal
  88         case err := <-errorsCh:
  89                 t.Fatalf("Expected %s, got %+v", msg, err)
  90         case <-time.After(500 * time.Millisecond):
  91                 t.Fatalf("Did not receive %s message in 500ms", msg)
  92         }
  93 }
  94
  95 func TestMConnectionStopsAndReturnsError(t *testing.T) {
  96         assert, require := assert.New(t), require.New(t)
  97
  98         server, client := net.Pipe()
  99         defer server.Close()
 100         defer client.Close()
 101
 102         receivedCh := make(chan []byte)
 103         errorsCh := make(chan interface{})
 104         onReceive := func(chID byte, msgBytes []byte) {
 105                 receivedCh <- msgBytes
 106         }
 107         onError := func(r interface{}) {
 108                 errorsCh <- r
 109         }
 110         mconn := createMConnectionWithCallbacks(client, onReceive, onError)
 111         _, err := mconn.Start()
 112         require.Nil(err)
 113         defer mconn.Stop()
 114
 115         client.Close()
 116
 117         select {
 118         case receivedBytes := <-receivedCh:
 119                 t.Fatalf("Expected error, got %v", receivedBytes)
 120         case err := <-errorsCh:
 121                 assert.NotNil(err)
 122                 assert.False(mconn.IsRunning())
 123         case <-time.After(500 * time.Millisecond):
 124                 t.Fatal("Did not receive error in 500ms")
 125         }
 126 }