dmaap-mediator-producer/internal/jobs/jobs_test.go

   1 // -
   2 //   ========================LICENSE_START=================================
   3 //   O-RAN-SC
   4 //   %%
   5 //   Copyright (C) 2021: Nordix Foundation
   6 //   %%
   7 //   Licensed under the Apache License, Version 2.0 (the "License");
   8 //   you may not use this file except in compliance with the License.
   9 //   You may obtain a copy of the License at
  10 //
  11 //        http://www.apache.org/licenses/LICENSE-2.0
  12 //
  13 //   Unless required by applicable law or agreed to in writing, software
  14 //   distributed under the License is distributed on an "AS IS" BASIS,
  15 //   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16 //   See the License for the specific language governing permissions and
  17 //   limitations under the License.
  18 //   ========================LICENSE_END===================================
  19 //
  20
  21 package jobs
  22
  23 import (
  24         "bytes"
  25         "fmt"
  26         "io/ioutil"
  27         "net/http"
  28         "strconv"
  29         "sync"
  30         "testing"
  31         "time"
  32
  33         "github.com/confluentinc/confluent-kafka-go/kafka"
  34         "github.com/stretchr/testify/mock"
  35         "github.com/stretchr/testify/require"
  36         "oransc.org/nonrtric/dmaapmediatorproducer/internal/config"
  37         "oransc.org/nonrtric/dmaapmediatorproducer/internal/kafkaclient"
  38         "oransc.org/nonrtric/dmaapmediatorproducer/mocks"
  39 )
  40
  41 func TestJobsManagerLoadTypesFromConfiguration_shouldReturnSliceOfTypesAndProvideSupportedTypes(t *testing.T) {
  42         assertions := require.New(t)
  43
  44         managerUnderTest := NewJobsManagerImpl(nil, "", kafkaclient.KafkaFactoryImpl{}, nil)
  45
  46         wantedDMaaPType := config.TypeDefinition{
  47                 Identity:      "type1",
  48                 DMaaPTopicURL: "events/unauthenticated.SEC_FAULT_OUTPUT/dmaapmediatorproducer/type1",
  49         }
  50         wantedKafkaType := config.TypeDefinition{
  51                 Identity:        "type2",
  52                 KafkaInputTopic: "topic",
  53         }
  54         wantedTypes := []config.TypeDefinition{wantedDMaaPType, wantedKafkaType}
  55
  56         types := managerUnderTest.LoadTypesFromConfiguration(wantedTypes)
  57
  58         assertions.EqualValues(wantedTypes, types)
  59
  60         supportedTypes := managerUnderTest.GetSupportedTypes()
  61         assertions.ElementsMatch([]string{"type1", "type2"}, supportedTypes)
  62         assertions.Equal(dMaaPSource, managerUnderTest.allTypes["type1"].jobsHandler.sourceType)
  63         assertions.Equal(kafkaSource, managerUnderTest.allTypes["type2"].jobsHandler.sourceType)
  64 }
  65
  66 func TestJobsManagerAddJobWhenTypeIsSupported_shouldAddJobToChannel(t *testing.T) {
  67         assertions := require.New(t)
  68         managerUnderTest := NewJobsManagerImpl(nil, "", kafkaclient.KafkaFactoryImpl{}, nil)
  69         wantedJob := JobInfo{
  70                 Owner:            "owner",
  71                 LastUpdated:      "now",
  72                 InfoJobIdentity:  "job1",
  73                 TargetUri:        "target",
  74                 InfoJobData:      Parameters{},
  75                 InfoTypeIdentity: "type1",
  76         }
  77         jobsHandler := jobsHandler{
  78                 addJobCh: make(chan JobInfo)}
  79         managerUnderTest.allTypes["type1"] = TypeData{
  80                 Identity:    "type1",
  81                 jobsHandler: &jobsHandler,
  82         }
  83
  84         var err error
  85         go func() {
  86                 err = managerUnderTest.AddJobFromRESTCall(wantedJob)
  87         }()
  88
  89         assertions.Nil(err)
  90         addedJob := <-jobsHandler.addJobCh
  91         assertions.Equal(wantedJob, addedJob)
  92 }
  93
  94 func TestJobsManagerAddJobWhenTypeIsNotSupported_shouldReturnError(t *testing.T) {
  95         assertions := require.New(t)
  96         managerUnderTest := NewJobsManagerImpl(nil, "", kafkaclient.KafkaFactoryImpl{}, nil)
  97         jobInfo := JobInfo{
  98                 InfoTypeIdentity: "type1",
  99         }
 100
 101         err := managerUnderTest.AddJobFromRESTCall(jobInfo)
 102         assertions.NotNil(err)
 103         assertions.Equal("type not supported: type1", err.Error())
 104 }
 105
 106 func TestJobsManagerAddJobWhenJobIdMissing_shouldReturnError(t *testing.T) {
 107         assertions := require.New(t)
 108         managerUnderTest := NewJobsManagerImpl(nil, "", kafkaclient.KafkaFactoryImpl{}, nil)
 109         managerUnderTest.allTypes["type1"] = TypeData{
 110                 Identity: "type1",
 111         }
 112
 113         jobInfo := JobInfo{
 114                 InfoTypeIdentity: "type1",
 115         }
 116         err := managerUnderTest.AddJobFromRESTCall(jobInfo)
 117         assertions.NotNil(err)
 118         assertions.Equal("missing required job identity: {    {{0 0}} type1 }", err.Error())
 119 }
 120
 121 func TestJobsManagerAddJobWhenTargetUriMissing_shouldReturnError(t *testing.T) {
 122         assertions := require.New(t)
 123         managerUnderTest := NewJobsManagerImpl(nil, "", kafkaclient.KafkaFactoryImpl{}, nil)
 124         managerUnderTest.allTypes["type1"] = TypeData{
 125                 Identity: "type1",
 126         }
 127
 128         jobInfo := JobInfo{
 129                 InfoTypeIdentity: "type1",
 130                 InfoJobIdentity:  "job1",
 131         }
 132         err := managerUnderTest.AddJobFromRESTCall(jobInfo)
 133         assertions.NotNil(err)
 134         assertions.Equal("missing required target URI: {  job1  {{0 0}} type1 }", err.Error())
 135 }
 136
 137 func TestJobsManagerDeleteJob_shouldSendDeleteToChannel(t *testing.T) {
 138         assertions := require.New(t)
 139         managerUnderTest := NewJobsManagerImpl(nil, "", kafkaclient.KafkaFactoryImpl{}, nil)
 140         jobsHandler := jobsHandler{
 141                 deleteJobCh: make(chan string)}
 142         managerUnderTest.allTypes["type1"] = TypeData{
 143                 Identity:    "type1",
 144                 jobsHandler: &jobsHandler,
 145         }
 146
 147         go managerUnderTest.DeleteJobFromRESTCall("job2")
 148
 149         assertions.Equal("job2", <-jobsHandler.deleteJobCh)
 150 }
 151
 152 func TestStartJobsManagerAddDMaaPJob_shouldStartPollAndDistributeMessages(t *testing.T) {
 153         assertions := require.New(t)
 154
 155         called := false
 156         dMaaPMessages := `[{"message": {"data": "dmaap"}}]`
 157         pollClientMock := NewTestClient(func(req *http.Request) *http.Response {
 158                 if req.URL.String() == "http://mrAddr/topicUrl" {
 159                         assertions.Equal(req.Method, "GET")
 160                         body := "[]"
 161                         if !called {
 162                                 called = true
 163                                 body = dMaaPMessages
 164                         }
 165                         return &http.Response{
 166                                 StatusCode: http.StatusOK,
 167                                 Body:       ioutil.NopCloser(bytes.NewReader([]byte(body))),
 168                                 Header:     make(http.Header), // Must be set to non-nil value or it panics
 169                         }
 170                 }
 171                 t.Error("Wrong call to client: ", req)
 172                 t.Fail()
 173                 return nil
 174         })
 175
 176         wg := sync.WaitGroup{}
 177         distributeClientMock := NewTestClient(func(req *http.Request) *http.Response {
 178                 if req.URL.String() == "http://consumerHost/dmaaptarget" {
 179                         assertions.Equal(req.Method, "POST")
 180                         assertions.Equal(dMaaPMessages, getBodyAsString(req, t))
 181                         assertions.Equal("application/json", req.Header.Get("Content-Type"))
 182                         wg.Done()
 183                         return &http.Response{
 184                                 StatusCode: 200,
 185                                 Body:       ioutil.NopCloser(bytes.NewBufferString(`OK`)),
 186                                 Header:     make(http.Header), // Must be set to non-nil value or it panics
 187                         }
 188                 }
 189                 t.Error("Wrong call to client: ", req)
 190                 t.Fail()
 191                 return nil
 192         })
 193         dMaaPTypeDef := config.TypeDefinition{
 194                 Identity:      "type1",
 195                 DMaaPTopicURL: "/topicUrl",
 196         }
 197         dMaaPJobsHandler := newJobsHandler(dMaaPTypeDef, "http://mrAddr", nil, pollClientMock, distributeClientMock)
 198
 199         jobsManager := NewJobsManagerImpl(pollClientMock, "http://mrAddr", kafkaclient.KafkaFactoryImpl{}, distributeClientMock)
 200         jobsManager.allTypes["type1"] = TypeData{
 201                 Identity:    "type1",
 202                 jobsHandler: dMaaPJobsHandler,
 203         }
 204         jobsManager.StartJobsForAllTypes()
 205
 206         dMaaPJobInfo := JobInfo{
 207                 InfoTypeIdentity: "type1",
 208                 InfoJobIdentity:  "job1",
 209                 TargetUri:        "http://consumerHost/dmaaptarget",
 210         }
 211
 212         wg.Add(1) // Wait till the distribution has happened
 213         err := jobsManager.AddJobFromRESTCall(dMaaPJobInfo)
 214         assertions.Nil(err)
 215
 216         if waitTimeout(&wg, 2*time.Second) {
 217                 t.Error("Not all calls to server were made")
 218                 t.Fail()
 219         }
 220 }
 221
 222 func TestStartJobsManagerAddKafkaJob_shouldStartPollAndDistributeMessages(t *testing.T) {
 223         assertions := require.New(t)
 224
 225         kafkaMessages := `1`
 226         wg := sync.WaitGroup{}
 227         distributeClientMock := NewTestClient(func(req *http.Request) *http.Response {
 228                 if req.URL.String() == "http://consumerHost/kafkatarget" {
 229                         assertions.Equal(req.Method, "POST")
 230                         assertions.Equal(kafkaMessages, getBodyAsString(req, t))
 231                         assertions.Equal("text/plain", req.Header.Get("Content-Type"))
 232                         wg.Done()
 233                         return &http.Response{
 234                                 StatusCode: 200,
 235                                 Body:       ioutil.NopCloser(bytes.NewBufferString(`OK`)),
 236                                 Header:     make(http.Header), // Must be set to non-nil value or it panics
 237                         }
 238                 }
 239                 t.Error("Wrong call to client: ", req)
 240                 t.Fail()
 241                 return nil
 242         })
 243
 244         kafkaTypeDef := config.TypeDefinition{
 245                 Identity:        "type2",
 246                 KafkaInputTopic: "topic",
 247         }
 248         kafkaFactoryMock := mocks.KafkaFactory{}
 249         kafkaConsumerMock := mocks.KafkaConsumer{}
 250         kafkaConsumerMock.On("Commit").Return([]kafka.TopicPartition{}, error(nil))
 251         kafkaConsumerMock.On("Subscribe", mock.Anything).Return(error(nil))
 252         kafkaConsumerMock.On("ReadMessage", mock.Anything).Return(&kafka.Message{
 253                 Value: []byte(kafkaMessages),
 254         }, error(nil)).Once()
 255         kafkaConsumerMock.On("ReadMessage", mock.Anything).Return(nil, fmt.Errorf("Just to stop"))
 256         kafkaFactoryMock.On("NewKafkaConsumer", mock.Anything).Return(kafkaConsumerMock, nil)
 257         kafkaJobsHandler := newJobsHandler(kafkaTypeDef, "", kafkaFactoryMock, nil, distributeClientMock)
 258
 259         jobsManager := NewJobsManagerImpl(nil, "", kafkaFactoryMock, distributeClientMock)
 260         jobsManager.allTypes["type2"] = TypeData{
 261                 Identity:    "type2",
 262                 jobsHandler: kafkaJobsHandler,
 263         }
 264
 265         jobsManager.StartJobsForAllTypes()
 266
 267         kafkaJobInfo := JobInfo{
 268                 InfoTypeIdentity: "type2",
 269                 InfoJobIdentity:  "job2",
 270                 TargetUri:        "http://consumerHost/kafkatarget",
 271         }
 272
 273         wg.Add(1) // Wait till the distribution has happened
 274         err := jobsManager.AddJobFromRESTCall(kafkaJobInfo)
 275         assertions.Nil(err)
 276
 277         if waitTimeout(&wg, 2*time.Second) {
 278                 t.Error("Not all calls to server were made")
 279                 t.Fail()
 280         }
 281 }
 282
 283 func TestJobsHandlerDeleteJob_shouldDeleteJobFromJobsMap(t *testing.T) {
 284         jobToDelete := newJob(JobInfo{}, nil)
 285         go jobToDelete.start()
 286         typeDef := config.TypeDefinition{
 287                 Identity:      "type1",
 288                 DMaaPTopicURL: "/topicUrl",
 289         }
 290         jobsHandler := newJobsHandler(typeDef, "http://mrAddr", kafkaclient.KafkaFactoryImpl{}, nil, nil)
 291         jobsHandler.jobs["job1"] = jobToDelete
 292
 293         go jobsHandler.monitorManagementChannels()
 294
 295         jobsHandler.deleteJobCh <- "job1"
 296
 297         deleted := false
 298         for i := 0; i < 100; i++ {
 299                 if len(jobsHandler.jobs) == 0 {
 300                         deleted = true
 301                         break
 302                 }
 303                 time.Sleep(time.Microsecond) // Need to drop control to let the job's goroutine do the job
 304         }
 305         require.New(t).True(deleted, "Job not deleted")
 306 }
 307
 308 func TestJobsHandlerEmptyJobMessageBufferWhenItIsFull(t *testing.T) {
 309         job := newJob(JobInfo{
 310                 InfoJobIdentity: "job",
 311         }, nil)
 312
 313         typeDef := config.TypeDefinition{
 314                 Identity:      "type1",
 315                 DMaaPTopicURL: "/topicUrl",
 316         }
 317         jobsHandler := newJobsHandler(typeDef, "http://mrAddr", kafkaclient.KafkaFactoryImpl{}, nil, nil)
 318         jobsHandler.jobs["job1"] = job
 319
 320         fillMessagesBuffer(job.messagesChannel)
 321
 322         jobsHandler.distributeMessages([]byte("sent msg"))
 323
 324         require.New(t).Len(job.messagesChannel, 0)
 325 }
 326
 327 func TestKafkaPollingAgentTimedOut_shouldResultInEMptyMessages(t *testing.T) {
 328         assertions := require.New(t)
 329
 330         kafkaFactoryMock := mocks.KafkaFactory{}
 331         kafkaConsumerMock := mocks.KafkaConsumer{}
 332         kafkaConsumerMock.On("Commit").Return([]kafka.TopicPartition{}, error(nil))
 333         kafkaConsumerMock.On("Subscribe", mock.Anything).Return(error(nil))
 334         kafkaConsumerMock.On("ReadMessage", mock.Anything).Return(nil, kafka.NewError(kafka.ErrTimedOut, "", false))
 335         kafkaFactoryMock.On("NewKafkaConsumer", mock.Anything).Return(kafkaConsumerMock, nil)
 336
 337         pollingAgentUnderTest := newKafkaPollingAgent(kafkaFactoryMock, "")
 338         messages, err := pollingAgentUnderTest.pollMessages()
 339
 340         assertions.Equal([]byte(""), messages)
 341         assertions.Nil(err)
 342 }
 343
 344 func TestJobWithoutParameters_shouldSendOneMessageAtATime(t *testing.T) {
 345         assertions := require.New(t)
 346
 347         wg := sync.WaitGroup{}
 348         messageNo := 1
 349         distributeClientMock := NewTestClient(func(req *http.Request) *http.Response {
 350                 if req.URL.String() == "http://consumerHost/target" {
 351                         assertions.Equal(req.Method, "POST")
 352                         assertions.Equal(fmt.Sprint("message", messageNo), getBodyAsString(req, t))
 353                         messageNo++
 354                         assertions.Equal("text/plain", req.Header.Get("Content-Type"))
 355                         wg.Done()
 356                         return &http.Response{
 357                                 StatusCode: 200,
 358                                 Body:       ioutil.NopCloser(bytes.NewBufferString(`OK`)),
 359                                 Header:     make(http.Header), // Must be set to non-nil value or it panics
 360                         }
 361                 }
 362                 t.Error("Wrong call to client: ", req)
 363                 t.Fail()
 364                 return nil
 365         })
 366
 367         jobUnderTest := newJob(JobInfo{
 368                 sourceType: kafkaSource,
 369                 TargetUri:  "http://consumerHost/target",
 370         }, distributeClientMock)
 371
 372         wg.Add(2)
 373         go jobUnderTest.start()
 374
 375         jobUnderTest.messagesChannel <- []byte("message1")
 376         jobUnderTest.messagesChannel <- []byte("message2")
 377
 378         if waitTimeout(&wg, 2*time.Second) {
 379                 t.Error("Not all calls to server were made")
 380                 t.Fail()
 381         }
 382 }
 383
 384 func TestJobWithBufferedParameters_shouldSendMessagesTogether(t *testing.T) {
 385         assertions := require.New(t)
 386
 387         wg := sync.WaitGroup{}
 388         distributeClientMock := NewTestClient(func(req *http.Request) *http.Response {
 389                 if req.URL.String() == "http://consumerHost/target" {
 390                         assertions.Equal(req.Method, "POST")
 391                         assertions.Equal(`["{\"data\": 1}","{\"data\": 2}","ABCDEFGH"]`, getBodyAsString(req, t))
 392                         assertions.Equal("application/json", req.Header.Get("Content-Type"))
 393                         wg.Done()
 394                         return &http.Response{
 395                                 StatusCode: 200,
 396                                 Body:       ioutil.NopCloser(bytes.NewBufferString(`OK`)),
 397                                 Header:     make(http.Header), // Must be set to non-nil value or it panics
 398                         }
 399                 }
 400                 t.Error("Wrong call to client: ", req)
 401                 t.Fail()
 402                 return nil
 403         })
 404
 405         jobUnderTest := newJob(JobInfo{
 406                 TargetUri: "http://consumerHost/target",
 407                 InfoJobData: Parameters{
 408                         BufferTimeout: BufferTimeout{
 409                                 MaxSize:            5,
 410                                 MaxTimeMiliseconds: 200,
 411                         },
 412                 },
 413         }, distributeClientMock)
 414
 415         wg.Add(1)
 416         go jobUnderTest.start()
 417
 418         go func() {
 419                 jobUnderTest.messagesChannel <- []byte(`{"data": 1}`)
 420                 jobUnderTest.messagesChannel <- []byte(`{"data": 2}`)
 421                 jobUnderTest.messagesChannel <- []byte("ABCDEFGH")
 422         }()
 423
 424         if waitTimeout(&wg, 2*time.Second) {
 425                 t.Error("Not all calls to server were made")
 426                 t.Fail()
 427         }
 428 }
 429
 430 func TestJobReadMoreThanBufferSizeMessages_shouldOnlyReturnMaxSizeNoOfMessages(t *testing.T) {
 431         assertions := require.New(t)
 432
 433         jobUnderTest := newJob(JobInfo{}, nil)
 434
 435         go func() {
 436                 for i := 0; i < 4; i++ {
 437                         jobUnderTest.messagesChannel <- []byte(strconv.Itoa(i))
 438                 }
 439         }()
 440
 441         msgs := jobUnderTest.read(BufferTimeout{
 442                 MaxSize:            2,
 443                 MaxTimeMiliseconds: 200,
 444         })
 445
 446         assertions.Equal([]byte("[\"0\",\"1\"]"), msgs)
 447 }
 448 func TestJobReadBufferedWhenTimeout_shouldOnlyReturnMessagesSentBeforeTimeout(t *testing.T) {
 449         assertions := require.New(t)
 450
 451         jobUnderTest := newJob(JobInfo{}, nil)
 452
 453         go func() {
 454                 for i := 0; i < 4; i++ {
 455                         time.Sleep(10 * time.Millisecond)
 456                         jobUnderTest.messagesChannel <- []byte(strconv.Itoa(i))
 457                 }
 458         }()
 459
 460         msgs := jobUnderTest.read(BufferTimeout{
 461                 MaxSize:            2,
 462                 MaxTimeMiliseconds: 30,
 463         })
 464
 465         assertions.Equal([]byte("[\"0\",\"1\"]"), msgs)
 466 }
 467
 468 func fillMessagesBuffer(mc chan []byte) {
 469         for i := 0; i < cap(mc); i++ {
 470                 mc <- []byte("msg")
 471         }
 472 }
 473
 474 type RoundTripFunc func(req *http.Request) *http.Response
 475
 476 func (f RoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
 477         return f(req), nil
 478 }
 479
 480 //NewTestClient returns *http.Client with Transport replaced to avoid making real calls
 481 func NewTestClient(fn RoundTripFunc) *http.Client {
 482         return &http.Client{
 483                 Transport: RoundTripFunc(fn),
 484         }
 485 }
 486
 487 // waitTimeout waits for the waitgroup for the specified max timeout.
 488 // Returns true if waiting timed out.
 489 func waitTimeout(wg *sync.WaitGroup, timeout time.Duration) bool {
 490         c := make(chan struct{})
 491         go func() {
 492                 defer close(c)
 493                 wg.Wait()
 494         }()
 495         select {
 496         case <-c:
 497                 return false // completed normally
 498         case <-time.After(timeout):
 499                 return true // timed out
 500         }
 501 }
 502
 503 func getBodyAsString(req *http.Request, t *testing.T) string {
 504         buf := new(bytes.Buffer)
 505         if _, err := buf.ReadFrom(req.Body); err != nil {
 506                 t.Fail()
 507         }
 508         return buf.String()
 509 }