blob: 7753d66dbe6c5fe233c12ddcc034bc01e84bfe5b [file] [log] [blame]
khenaidoo59ce9dd2019-11-11 13:05:32 -05001/*
2 * Copyright 2019-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 */
16
khenaidoob6238b32020-04-07 12:07:36 -040017package kafka
khenaidoo59ce9dd2019-11-11 13:05:32 -050018
19import (
Neha Sharma94f16a92020-06-26 04:17:55 +000020 "context"
serkant.uluderyab38671c2019-11-01 09:35:38 -070021 "github.com/opencord/voltha-lib-go/v3/pkg/kafka"
22 ic "github.com/opencord/voltha-protos/v3/go/inter_container"
23 "github.com/stretchr/testify/assert"
Neha Sharma94f16a92020-06-26 04:17:55 +000024 "testing"
25 "time"
khenaidoo59ce9dd2019-11-11 13:05:32 -050026)
27
khenaidoo59ce9dd2019-11-11 13:05:32 -050028func TestKafkaClientCreateTopic(t *testing.T) {
Neha Sharma94f16a92020-06-26 04:17:55 +000029 ctx := context.Background()
khenaidoo59ce9dd2019-11-11 13:05:32 -050030 cTkc := NewKafkaClient()
31 topic := kafka.Topic{Name: "myTopic"}
Neha Sharma94f16a92020-06-26 04:17:55 +000032 err := cTkc.CreateTopic(ctx, &topic, 1, 1)
khenaidoo59ce9dd2019-11-11 13:05:32 -050033 assert.Nil(t, err)
Neha Sharma94f16a92020-06-26 04:17:55 +000034 err = cTkc.CreateTopic(ctx, &topic, 1, 1)
khenaidoo59ce9dd2019-11-11 13:05:32 -050035 assert.NotNil(t, err)
36}
37
38func TestKafkaClientDeleteTopic(t *testing.T) {
39 cTkc := NewKafkaClient()
40 topic := kafka.Topic{Name: "myTopic"}
Neha Sharma94f16a92020-06-26 04:17:55 +000041 err := cTkc.DeleteTopic(context.Background(), &topic)
khenaidoo59ce9dd2019-11-11 13:05:32 -050042 assert.Nil(t, err)
43}
44
45func TestKafkaClientSubscribeSend(t *testing.T) {
46 cTkc := NewKafkaClient()
47 topic := kafka.Topic{Name: "myTopic"}
Neha Sharma94f16a92020-06-26 04:17:55 +000048 ch, err := cTkc.Subscribe(context.Background(), &topic)
khenaidoo59ce9dd2019-11-11 13:05:32 -050049 assert.Nil(t, err)
50 assert.NotNil(t, ch)
51 testCh := make(chan bool)
52 maxWait := 5 * time.Millisecond
53 msg := &ic.InterContainerMessage{
54 Header: &ic.Header{Id: "1234", ToTopic: topic.Name},
55 Body: nil,
56 }
57 timer := time.NewTimer(maxWait)
58 defer timer.Stop()
59 go func() {
60 select {
61 case val, ok := <-ch:
62 assert.True(t, ok)
63 assert.Equal(t, val, msg)
64 testCh <- true
65 case <-timer.C:
66 testCh <- false
67 }
68 }()
Neha Sharma94f16a92020-06-26 04:17:55 +000069 err = cTkc.Send(context.Background(), msg, &topic)
khenaidoo59ce9dd2019-11-11 13:05:32 -050070 assert.Nil(t, err)
71 res := <-testCh
72 assert.True(t, res)
73}
74
75func TestKafkaClientUnSubscribe(t *testing.T) {
76 cTkc := NewKafkaClient()
77 topic := kafka.Topic{Name: "myTopic"}
Neha Sharma94f16a92020-06-26 04:17:55 +000078 ch, err := cTkc.Subscribe(context.Background(), &topic)
khenaidoo59ce9dd2019-11-11 13:05:32 -050079 assert.Nil(t, err)
80 assert.NotNil(t, ch)
Neha Sharma94f16a92020-06-26 04:17:55 +000081 err = cTkc.UnSubscribe(context.Background(), &topic, ch)
khenaidoo59ce9dd2019-11-11 13:05:32 -050082 assert.Nil(t, err)
83}
84
85func TestKafkaClientStop(t *testing.T) {
86 cTkc := NewKafkaClient()
87 topic := kafka.Topic{Name: "myTopic"}
Neha Sharma94f16a92020-06-26 04:17:55 +000088 ch, err := cTkc.Subscribe(context.Background(), &topic)
khenaidoo59ce9dd2019-11-11 13:05:32 -050089 assert.Nil(t, err)
90 assert.NotNil(t, ch)
Neha Sharma94f16a92020-06-26 04:17:55 +000091 err = cTkc.UnSubscribe(context.Background(), &topic, ch)
khenaidoo59ce9dd2019-11-11 13:05:32 -050092 assert.Nil(t, err)
Neha Sharma94f16a92020-06-26 04:17:55 +000093 cTkc.Stop(context.Background())
khenaidoo59ce9dd2019-11-11 13:05:32 -050094}