You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

144 lines
3.7 KiB

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