8
8
"io/ioutil"
9
9
"log"
10
10
"net/http"
11
+ "os"
11
12
"strconv"
13
+ "strings"
12
14
"testing"
13
15
"time"
14
16
@@ -62,55 +64,99 @@ func SendMessage(t *testing.T, port int, topic string, method string, body []byt
62
64
}
63
65
64
66
func TestConsumer (t * testing.T ) {
65
- consumerTest (t , false , false , false )
67
+ consumerTest (t , nil )
66
68
}
67
69
68
70
func TestConsumerTLS (t * testing.T ) {
69
- consumerTest (t , false , false , true )
71
+ consumerTest (t , func (c * Config ) {
72
+ c .TlsV1 = true
73
+ c .TlsConfig = & tls.Config {
74
+ InsecureSkipVerify : true ,
75
+ }
76
+ })
70
77
}
71
78
72
79
func TestConsumerDeflate (t * testing.T ) {
73
- consumerTest (t , true , false , false )
80
+ consumerTest (t , func (c * Config ) {
81
+ c .Deflate = true
82
+ })
74
83
}
75
84
76
85
func TestConsumerSnappy (t * testing.T ) {
77
- consumerTest (t , false , true , false )
86
+ consumerTest (t , func (c * Config ) {
87
+ c .Snappy = true
88
+ })
78
89
}
79
90
80
91
func TestConsumerTLSDeflate (t * testing.T ) {
81
- consumerTest (t , true , false , true )
92
+ consumerTest (t , func (c * Config ) {
93
+ c .TlsV1 = true
94
+ c .TlsConfig = & tls.Config {
95
+ InsecureSkipVerify : true ,
96
+ }
97
+ c .Deflate = true
98
+ })
82
99
}
83
100
84
101
func TestConsumerTLSSnappy (t * testing.T ) {
85
- consumerTest (t , false , true , true )
102
+ consumerTest (t , func (c * Config ) {
103
+ c .TlsV1 = true
104
+ c .TlsConfig = & tls.Config {
105
+ InsecureSkipVerify : true ,
106
+ }
107
+ c .Snappy = true
108
+ })
86
109
}
87
110
88
- func consumerTest (t * testing.T , deflate bool , snappy bool , tlsv1 bool ) {
89
-
90
- topicName := "rdr_test"
91
- if deflate {
92
- topicName = topicName + "_deflate"
93
- } else if snappy {
94
- topicName = topicName + "_snappy"
111
+ func TestConsumerTLSClientCert (t * testing.T ) {
112
+ envDl := os .Getenv ("NSQ_DOWNLOAD" )
113
+ if strings .HasPrefix (envDl , "nsq-0.2.24" ) || strings .HasPrefix (envDl , "nsq-0.2.27" ) {
114
+ t .Log ("skipping due to older nsqd" )
115
+ return
95
116
}
96
- if tlsv1 {
97
- topicName = topicName + "_tls"
117
+ cert , _ := tls .LoadX509KeyPair ("./test/client.pem" , "./test/client.key" )
118
+ consumerTest (t , func (c * Config ) {
119
+ c .TlsV1 = true
120
+ c .TlsConfig = & tls.Config {
121
+ Certificates : []tls.Certificate {cert },
122
+ InsecureSkipVerify : true ,
123
+ }
124
+ })
125
+ }
126
+
127
+ func TestConsumerTLSClientCertViaSet (t * testing.T ) {
128
+ envDl := os .Getenv ("NSQ_DOWNLOAD" )
129
+ if strings .HasPrefix (envDl , "nsq-0.2.24" ) || strings .HasPrefix (envDl , "nsq-0.2.27" ) {
130
+ t .Log ("skipping due to older nsqd" )
131
+ return
98
132
}
99
- topicName = topicName + strconv .Itoa (int (time .Now ().Unix ()))
133
+ consumerTest (t , func (c * Config ) {
134
+ c .Set ("tls_v1" , true )
135
+ c .Set ("tls_cert" , "./test/client.pem" )
136
+ c .Set ("tls_key" , "./test/client.key" )
137
+ c .Set ("tls_insecure_skip_verify" , true )
138
+ })
139
+ }
100
140
141
+ func consumerTest (t * testing.T , cb func (c * Config )) {
101
142
config := NewConfig ()
102
143
// so that the test can simulate reaching max requeues and a call to LogFailedMessage
103
144
config .DefaultRequeueDelay = 0
104
145
// so that the test wont timeout from backing off
105
146
config .MaxBackoffDuration = time .Millisecond * 50
106
- config .Deflate = deflate
107
- config .Snappy = snappy
108
- config .TlsV1 = tlsv1
109
- if tlsv1 {
110
- config .TlsConfig = & tls.Config {
111
- InsecureSkipVerify : true ,
112
- }
147
+ if cb != nil {
148
+ cb (config )
149
+ }
150
+ topicName := "rdr_test"
151
+ if config .Deflate {
152
+ topicName = topicName + "_deflate"
153
+ } else if config .Snappy {
154
+ topicName = topicName + "_snappy"
113
155
}
156
+ if config .TlsV1 {
157
+ topicName = topicName + "_tls"
158
+ }
159
+ topicName = topicName + strconv .Itoa (int (time .Now ().Unix ()))
114
160
q , _ := NewConsumer (topicName , "ch" , config )
115
161
q .SetLogger (nullLogger , LogLevelInfo )
116
162
@@ -128,12 +174,12 @@ func consumerTest(t *testing.T, deflate bool, snappy bool, tlsv1 bool) {
128
174
addr := "127.0.0.1:4150"
129
175
err := q .ConnectToNSQD (addr )
130
176
if err != nil {
131
- t .Fatalf (err . Error () )
177
+ t .Fatal (err )
132
178
}
133
179
134
180
err = q .ConnectToNSQD (addr )
135
181
if err == nil {
136
- t .Fatalf ("should not be able to connect to the same NSQ twice" )
182
+ t .Fatal ("should not be able to connect to the same NSQ twice" )
137
183
}
138
184
139
185
<- q .StopChan
0 commit comments