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.

218 lines
5.8 KiB

  1. package p2p_test
  2. import (
  3. "io"
  4. "net"
  5. "testing"
  6. "time"
  7. "github.com/fortytw2/leaktest"
  8. "github.com/stretchr/testify/require"
  9. "github.com/tendermint/tendermint/internal/p2p"
  10. "github.com/tendermint/tendermint/internal/p2p/conn"
  11. "github.com/tendermint/tendermint/libs/log"
  12. )
  13. // Transports are mainly tested by common tests in transport_test.go, we
  14. // register a transport factory here to get included in those tests.
  15. func init() {
  16. testTransports["mconn"] = func(t *testing.T) p2p.Transport {
  17. transport := p2p.NewMConnTransport(
  18. log.TestingLogger(),
  19. conn.DefaultMConnConfig(),
  20. []*p2p.ChannelDescriptor{{ID: byte(chID), Priority: 1}},
  21. p2p.MConnTransportOptions{},
  22. )
  23. err := transport.Listen(p2p.Endpoint{
  24. Protocol: p2p.MConnProtocol,
  25. IP: net.IPv4(127, 0, 0, 1),
  26. Port: 0, // assign a random port
  27. })
  28. require.NoError(t, err)
  29. t.Cleanup(func() {
  30. require.NoError(t, transport.Close())
  31. })
  32. return transport
  33. }
  34. }
  35. func TestMConnTransport_AcceptBeforeListen(t *testing.T) {
  36. transport := p2p.NewMConnTransport(
  37. log.TestingLogger(),
  38. conn.DefaultMConnConfig(),
  39. []*p2p.ChannelDescriptor{{ID: byte(chID), Priority: 1}},
  40. p2p.MConnTransportOptions{
  41. MaxAcceptedConnections: 2,
  42. },
  43. )
  44. t.Cleanup(func() {
  45. _ = transport.Close()
  46. })
  47. _, err := transport.Accept()
  48. require.Error(t, err)
  49. require.NotEqual(t, io.EOF, err) // io.EOF should be returned after Close()
  50. }
  51. func TestMConnTransport_AcceptMaxAcceptedConnections(t *testing.T) {
  52. transport := p2p.NewMConnTransport(
  53. log.TestingLogger(),
  54. conn.DefaultMConnConfig(),
  55. []*p2p.ChannelDescriptor{{ID: byte(chID), Priority: 1}},
  56. p2p.MConnTransportOptions{
  57. MaxAcceptedConnections: 2,
  58. },
  59. )
  60. t.Cleanup(func() {
  61. _ = transport.Close()
  62. })
  63. err := transport.Listen(p2p.Endpoint{
  64. Protocol: p2p.MConnProtocol,
  65. IP: net.IPv4(127, 0, 0, 1),
  66. })
  67. require.NoError(t, err)
  68. require.NotEmpty(t, transport.Endpoints())
  69. endpoint := transport.Endpoints()[0]
  70. // Start a goroutine to just accept any connections.
  71. acceptCh := make(chan p2p.Connection, 10)
  72. go func() {
  73. for {
  74. conn, err := transport.Accept()
  75. if err != nil {
  76. return
  77. }
  78. acceptCh <- conn
  79. }
  80. }()
  81. // The first two connections should be accepted just fine.
  82. dial1, err := transport.Dial(ctx, endpoint)
  83. require.NoError(t, err)
  84. defer dial1.Close()
  85. accept1 := <-acceptCh
  86. defer accept1.Close()
  87. require.Equal(t, dial1.LocalEndpoint(), accept1.RemoteEndpoint())
  88. dial2, err := transport.Dial(ctx, endpoint)
  89. require.NoError(t, err)
  90. defer dial2.Close()
  91. accept2 := <-acceptCh
  92. defer accept2.Close()
  93. require.Equal(t, dial2.LocalEndpoint(), accept2.RemoteEndpoint())
  94. // The third connection will be dialed successfully, but the accept should
  95. // not go through.
  96. dial3, err := transport.Dial(ctx, endpoint)
  97. require.NoError(t, err)
  98. defer dial3.Close()
  99. select {
  100. case <-acceptCh:
  101. require.Fail(t, "unexpected accept")
  102. case <-time.After(time.Second):
  103. }
  104. // However, once either of the other connections are closed, the accept
  105. // goes through.
  106. require.NoError(t, accept1.Close())
  107. accept3 := <-acceptCh
  108. defer accept3.Close()
  109. require.Equal(t, dial3.LocalEndpoint(), accept3.RemoteEndpoint())
  110. }
  111. func TestMConnTransport_Listen(t *testing.T) {
  112. testcases := []struct {
  113. endpoint p2p.Endpoint
  114. ok bool
  115. }{
  116. // Valid v4 and v6 addresses, with mconn and tcp protocols.
  117. {p2p.Endpoint{Protocol: p2p.MConnProtocol, IP: net.IPv4zero}, true},
  118. {p2p.Endpoint{Protocol: p2p.MConnProtocol, IP: net.IPv4(127, 0, 0, 1)}, true},
  119. {p2p.Endpoint{Protocol: p2p.MConnProtocol, IP: net.IPv6zero}, true},
  120. {p2p.Endpoint{Protocol: p2p.MConnProtocol, IP: net.IPv6loopback}, true},
  121. {p2p.Endpoint{Protocol: p2p.TCPProtocol, IP: net.IPv4zero}, true},
  122. // Invalid endpoints.
  123. {p2p.Endpoint{}, false},
  124. {p2p.Endpoint{Protocol: p2p.MConnProtocol, Path: "foo"}, false},
  125. {p2p.Endpoint{Protocol: p2p.MConnProtocol, IP: net.IPv4zero, Path: "foo"}, false},
  126. }
  127. for _, tc := range testcases {
  128. tc := tc
  129. t.Run(tc.endpoint.String(), func(t *testing.T) {
  130. t.Cleanup(leaktest.Check(t))
  131. transport := p2p.NewMConnTransport(
  132. log.TestingLogger(),
  133. conn.DefaultMConnConfig(),
  134. []*p2p.ChannelDescriptor{{ID: byte(chID), Priority: 1}},
  135. p2p.MConnTransportOptions{},
  136. )
  137. // Transport should not listen on any endpoints yet.
  138. require.Empty(t, transport.Endpoints())
  139. // Start listening, and check any expected errors.
  140. err := transport.Listen(tc.endpoint)
  141. if !tc.ok {
  142. require.Error(t, err)
  143. return
  144. }
  145. require.NoError(t, err)
  146. // Check the endpoint.
  147. endpoints := transport.Endpoints()
  148. require.Len(t, endpoints, 1)
  149. endpoint := endpoints[0]
  150. require.Equal(t, p2p.MConnProtocol, endpoint.Protocol)
  151. if tc.endpoint.IP.IsUnspecified() {
  152. require.True(t, endpoint.IP.IsUnspecified(),
  153. "expected unspecified IP, got %v", endpoint.IP)
  154. } else {
  155. require.True(t, tc.endpoint.IP.Equal(endpoint.IP),
  156. "expected %v, got %v", tc.endpoint.IP, endpoint.IP)
  157. }
  158. require.NotZero(t, endpoint.Port)
  159. require.Empty(t, endpoint.Path)
  160. dialedChan := make(chan struct{})
  161. var peerConn p2p.Connection
  162. go func() {
  163. // Dialing the endpoint should work.
  164. var err error
  165. peerConn, err = transport.Dial(ctx, endpoint)
  166. require.NoError(t, err)
  167. close(dialedChan)
  168. }()
  169. conn, err := transport.Accept()
  170. require.NoError(t, err)
  171. _ = conn.Close()
  172. <-dialedChan
  173. time.Sleep(time.Minute)
  174. // closing the connection should not error
  175. require.NoError(t, peerConn.Close())
  176. // try to read from the connection should error
  177. _, _, err = peerConn.ReceiveMessage()
  178. require.Error(t, err)
  179. // Trying to listen again should error.
  180. err = transport.Listen(tc.endpoint)
  181. require.Error(t, err)
  182. // close the transport
  183. _ = transport.Close()
  184. // Dialing the closed endpoint should error
  185. _, err = transport.Dial(ctx, endpoint)
  186. require.Error(t, err)
  187. })
  188. }
  189. }