blob: c3eace736bc8df5a9188d9f9c0a9ab06e750cb49 [file] [log] [blame]
Scott Baker2c1c4822019-10-16 11:02:41 -07001/*
2 * Copyright 2018-present Open Networking Foundation
3
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7
8 * http://www.apache.org/licenses/LICENSE-2.0
9
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16package kafka
17
18import (
19 "github.com/stretchr/testify/assert"
20 "testing"
21)
22
23func TestDefaultKafkaProxy(t *testing.T) {
24 actualResult, error := NewInterContainerProxy()
25 assert.Equal(t, error, nil)
26 assert.Equal(t, actualResult.kafkaHost, DefaultKafkaHost)
27 assert.Equal(t, actualResult.kafkaPort, DefaultKafkaPort)
28 assert.Equal(t, actualResult.defaultRequestHandlerInterface, interface{}(nil))
29}
30
31func TestKafkaProxyOptionHost(t *testing.T) {
32 actualResult, error := NewInterContainerProxy(InterContainerHost("10.20.30.40"))
33 assert.Equal(t, error, nil)
34 assert.Equal(t, actualResult.kafkaHost, "10.20.30.40")
35 assert.Equal(t, actualResult.kafkaPort, DefaultKafkaPort)
36 assert.Equal(t, actualResult.defaultRequestHandlerInterface, interface{}(nil))
37}
38
39func TestKafkaProxyOptionPort(t *testing.T) {
40 actualResult, error := NewInterContainerProxy(InterContainerPort(1020))
41 assert.Equal(t, error, nil)
42 assert.Equal(t, actualResult.kafkaHost, DefaultKafkaHost)
43 assert.Equal(t, actualResult.kafkaPort, 1020)
44 assert.Equal(t, actualResult.defaultRequestHandlerInterface, interface{}(nil))
45}
46
47func TestKafkaProxyOptionTopic(t *testing.T) {
48 actualResult, error := NewInterContainerProxy(DefaultTopic(&Topic{Name: "Adapter"}))
49 assert.Equal(t, error, nil)
50 assert.Equal(t, actualResult.kafkaHost, DefaultKafkaHost)
51 assert.Equal(t, actualResult.kafkaPort, DefaultKafkaPort)
52 assert.Equal(t, actualResult.defaultRequestHandlerInterface, interface{}(nil))
53 assert.Equal(t, actualResult.DefaultTopic.Name, "Adapter")
54}
55
56type myInterface struct {
57}
58
59func (m *myInterface) doSomething() {
60}
61
62func TestKafkaProxyOptionTargetInterface(t *testing.T) {
63 var m *myInterface
64 actualResult, error := NewInterContainerProxy(RequestHandlerInterface(m))
65 assert.Equal(t, error, nil)
66 assert.Equal(t, actualResult.kafkaHost, DefaultKafkaHost)
67 assert.Equal(t, actualResult.kafkaPort, DefaultKafkaPort)
68 assert.Equal(t, actualResult.defaultRequestHandlerInterface, m)
69}
70
71func TestKafkaProxyChangeAllOptions(t *testing.T) {
72 var m *myInterface
73 actualResult, error := NewInterContainerProxy(
74 InterContainerHost("10.20.30.40"),
75 InterContainerPort(1020),
76 DefaultTopic(&Topic{Name: "Adapter"}),
77 RequestHandlerInterface(m))
78 assert.Equal(t, error, nil)
79 assert.Equal(t, actualResult.kafkaHost, "10.20.30.40")
80 assert.Equal(t, actualResult.kafkaPort, 1020)
81 assert.Equal(t, actualResult.defaultRequestHandlerInterface, m)
82 assert.Equal(t, actualResult.DefaultTopic.Name, "Adapter")
83}
Scott Baker104b67d2019-10-29 15:56:27 -070084
85func TestKafkaProxyEnableLivenessChannel(t *testing.T) {
86 var m *myInterface
87
88 // Note: This doesn't actually start the client
89 client := NewSaramaClient()
90
91 probe, err := NewInterContainerProxy(
92 InterContainerHost("10.20.30.40"),
93 InterContainerPort(1020),
94 DefaultTopic(&Topic{Name: "Adapter"}),
95 RequestHandlerInterface(m),
96 MsgClient(client),
97 )
98
99 assert.Nil(t, err)
100
101 ch := probe.EnableLivenessChannel(true)
102
103 // The channel should have one "true" message on it
104 assert.NotEmpty(t, ch)
105
106 select {
107 case stuff := <-ch:
108 assert.True(t, stuff)
109 default:
110 t.Error("Failed to read from the channel")
111 }
112}