blob: 09286adfcd56dc663d4d7c0e1270d539f4e30b9f [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) {
Kent Hagerman3a402302020-01-31 15:03:53 -050024 actualResult := newInterContainerProxy()
Neha Sharmadd9af392020-04-28 09:03:57 +000025 assert.Equal(t, actualResult.kafkaAddress, DefaultKafkaAddress)
Scott Baker2c1c4822019-10-16 11:02:41 -070026 assert.Equal(t, actualResult.defaultRequestHandlerInterface, interface{}(nil))
27}
28
Neha Sharmadd9af392020-04-28 09:03:57 +000029func TestKafkaProxyOptionAddress(t *testing.T) {
30 actualResult := newInterContainerProxy(InterContainerAddress("10.20.30.40:1020"))
31 assert.Equal(t, actualResult.kafkaAddress, "10.20.30.40:1020")
Scott Baker2c1c4822019-10-16 11:02:41 -070032 assert.Equal(t, actualResult.defaultRequestHandlerInterface, interface{}(nil))
33}
34
35func TestKafkaProxyOptionTopic(t *testing.T) {
Kent Hagerman3a402302020-01-31 15:03:53 -050036 actualResult := newInterContainerProxy(DefaultTopic(&Topic{Name: "Adapter"}))
Neha Sharmadd9af392020-04-28 09:03:57 +000037 assert.Equal(t, actualResult.kafkaAddress, DefaultKafkaAddress)
Scott Baker2c1c4822019-10-16 11:02:41 -070038 assert.Equal(t, actualResult.defaultRequestHandlerInterface, interface{}(nil))
Matteo Scandolof346a2d2020-01-24 13:14:54 -080039 assert.Equal(t, actualResult.defaultTopic.Name, "Adapter")
Scott Baker2c1c4822019-10-16 11:02:41 -070040}
41
42type myInterface struct {
43}
44
Scott Baker2c1c4822019-10-16 11:02:41 -070045func TestKafkaProxyOptionTargetInterface(t *testing.T) {
46 var m *myInterface
Kent Hagerman3a402302020-01-31 15:03:53 -050047 actualResult := newInterContainerProxy(RequestHandlerInterface(m))
Neha Sharmadd9af392020-04-28 09:03:57 +000048 assert.Equal(t, actualResult.kafkaAddress, DefaultKafkaAddress)
Scott Baker2c1c4822019-10-16 11:02:41 -070049 assert.Equal(t, actualResult.defaultRequestHandlerInterface, m)
50}
51
52func TestKafkaProxyChangeAllOptions(t *testing.T) {
53 var m *myInterface
Kent Hagerman3a402302020-01-31 15:03:53 -050054 actualResult := newInterContainerProxy(
Neha Sharmadd9af392020-04-28 09:03:57 +000055 InterContainerAddress("10.20.30.40:1020"),
Scott Baker2c1c4822019-10-16 11:02:41 -070056 DefaultTopic(&Topic{Name: "Adapter"}),
57 RequestHandlerInterface(m))
Neha Sharmadd9af392020-04-28 09:03:57 +000058 assert.Equal(t, actualResult.kafkaAddress, "10.20.30.40:1020")
Scott Baker2c1c4822019-10-16 11:02:41 -070059 assert.Equal(t, actualResult.defaultRequestHandlerInterface, m)
Matteo Scandolof346a2d2020-01-24 13:14:54 -080060 assert.Equal(t, actualResult.defaultTopic.Name, "Adapter")
Scott Baker2c1c4822019-10-16 11:02:41 -070061}
Scott Baker104b67d2019-10-29 15:56:27 -070062
63func TestKafkaProxyEnableLivenessChannel(t *testing.T) {
64 var m *myInterface
65
66 // Note: This doesn't actually start the client
67 client := NewSaramaClient()
68
Kent Hagerman3a402302020-01-31 15:03:53 -050069 probe := newInterContainerProxy(
Neha Sharmadd9af392020-04-28 09:03:57 +000070 InterContainerAddress("10.20.30.40:1020"),
Scott Baker104b67d2019-10-29 15:56:27 -070071 DefaultTopic(&Topic{Name: "Adapter"}),
72 RequestHandlerInterface(m),
73 MsgClient(client),
74 )
75
Scott Bakere6685952020-06-23 04:05:39 +000076 ch := probe.EnableLivenessChannel(true)
Scott Baker104b67d2019-10-29 15:56:27 -070077
78 // The channel should have one "true" message on it
79 assert.NotEmpty(t, ch)
80
81 select {
82 case stuff := <-ch:
83 assert.True(t, stuff)
84 default:
85 t.Error("Failed to read from the channel")
86 }
87}