Poll MR and send messages to consumers

[nonrtric.git] / dmaap-mediator-producer / internal / jobs / jobs_test.go

diff --git a/dmaap-mediator-producer/internal/jobs/jobs_test.go b/dmaap-mediator-producer/internal/jobs/jobs_test.go
index 0941033..b53d85e 100644 (file)
--- a/dmaap-mediator-producer/internal/jobs/jobs_test.go
+++ b/dmaap-mediator-producer/internal/jobs/jobs_test.go
@@ -21,14 +21,21 @@
 package jobs
 
 import (
+       "bytes"
+       "io/ioutil"
+       "net/http"
        "os"
        "path/filepath"
        "testing"
+       "time"
 
+       "github.com/stretchr/testify/mock"
        "github.com/stretchr/testify/require"
+       "oransc.org/nonrtric/dmaapmediatorproducer/internal/restclient"
+       "oransc.org/nonrtric/dmaapmediatorproducer/mocks"
 )
 
-const type1Schema = `{"title": "Type 1"}`
+const typeDefinition = `{"id": "type1", "dmaapTopic": "unauthenticated.SEC_FAULT_OUTPUT", "schema": {"title": "Type 1"}}`
 
 func TestGetTypes_filesOkShouldReturnSliceOfTypesAndProvideSupportedTypes(t *testing.T) {
        assertions := require.New(t)
@@ -42,13 +49,15 @@ func TestGetTypes_filesOkShouldReturnSliceOfTypesAndProvideSupportedTypes(t *tes
        })
        typeDir = typesDir
        fname := filepath.Join(typesDir, "type1.json")
-       if err = os.WriteFile(fname, []byte(type1Schema), 0666); err != nil {
+       if err = os.WriteFile(fname, []byte(typeDefinition), 0666); err != nil {
                t.Errorf("Unable to create temporary files for types due to: %v", err)
        }
        types, err := GetTypes()
        wantedType := Type{
-               TypeId: "type1",
-               Schema: type1Schema,
+               TypeId:     "type1",
+               DMaaPTopic: "unauthenticated.SEC_FAULT_OUTPUT",
+               Schema:     `{"title":"Type 1"}`,
+               Jobs:       make(map[string]JobInfo),
        }
        wantedTypes := []*Type{&wantedType}
        assertions.EqualValues(wantedTypes, types)
@@ -60,11 +69,7 @@ func TestGetTypes_filesOkShouldReturnSliceOfTypesAndProvideSupportedTypes(t *tes
 
 func TestAddJobWhenTypeIsSupported_shouldAddJobToAllJobsMap(t *testing.T) {
        assertions := require.New(t)
-       allJobs["type1"] = make(map[string]JobInfo)
-       t.Cleanup(func() {
-               clearAll()
-       })
-       jobInfo := JobInfo{
+       wantedJob := JobInfo{
                Owner:            "owner",
                LastUpdated:      "now",
                InfoJobIdentity:  "job1",
@@ -72,11 +77,18 @@ func TestAddJobWhenTypeIsSupported_shouldAddJobToAllJobsMap(t *testing.T) {
                InfoJobData:      "{}",
                InfoTypeIdentity: "type1",
        }
+       allJobs["type1"] = Type{
+               TypeId: "type1",
+               Jobs:   map[string]JobInfo{"job1": wantedJob},
+       }
+       t.Cleanup(func() {
+               clearAll()
+       })
 
-       err := AddJob(jobInfo)
+       err := AddJob(wantedJob)
        assertions.Nil(err)
-       assertions.Equal(1, len(allJobs["type1"]))
-       assertions.Equal(jobInfo, allJobs["type1"]["job1"])
+       assertions.Equal(1, len(allJobs["type1"].Jobs))
+       assertions.Equal(wantedJob, allJobs["type1"].Jobs["job1"])
 }
 
 func TestAddJobWhenTypeIsNotSupported_shouldReturnError(t *testing.T) {
@@ -92,7 +104,9 @@ func TestAddJobWhenTypeIsNotSupported_shouldReturnError(t *testing.T) {
 
 func TestAddJobWhenJobIdMissing_shouldReturnError(t *testing.T) {
        assertions := require.New(t)
-       allJobs["type1"] = make(map[string]JobInfo)
+       allJobs["type1"] = Type{
+               TypeId: "type1",
+       }
        t.Cleanup(func() {
                clearAll()
        })
@@ -107,7 +121,9 @@ func TestAddJobWhenJobIdMissing_shouldReturnError(t *testing.T) {
 
 func TestAddJobWhenTargetUriMissing_shouldReturnError(t *testing.T) {
        assertions := require.New(t)
-       allJobs["type1"] = make(map[string]JobInfo)
+       allJobs["type1"] = Type{
+               TypeId: "type1",
+       }
        jobInfo := JobInfo{
                InfoTypeIdentity: "type1",
                InfoJobIdentity:  "job1",
@@ -118,3 +134,53 @@ func TestAddJobWhenTargetUriMissing_shouldReturnError(t *testing.T) {
        assertions.Equal("missing required target URI: {  job1   type1}", err.Error())
        clearAll()
 }
+
+func TestPollAndDistributeMessages(t *testing.T) {
+       assertions := require.New(t)
+       jobInfo := JobInfo{
+               InfoTypeIdentity: "type1",
+               InfoJobIdentity:  "job1",
+               TargetUri:        "http://consumerHost/target",
+       }
+       allJobs["type1"] = Type{
+               TypeId:     "type1",
+               DMaaPTopic: "topic",
+               Jobs:       map[string]JobInfo{"job1": jobInfo},
+       }
+       t.Cleanup(func() {
+               clearAll()
+       })
+
+       body := ioutil.NopCloser(bytes.NewReader([]byte(`[{"message": {"data": "data"}}]`)))
+       clientMock := mocks.HTTPClient{}
+       clientMock.On("Get", mock.Anything).Return(&http.Response{
+               StatusCode: http.StatusOK,
+               Body:       body,
+       }, nil)
+
+       clientMock.On("Do", mock.Anything).Return(&http.Response{
+               StatusCode: http.StatusOK,
+       }, nil)
+
+       restclient.Client = &clientMock
+
+       pollAndDistributeMessages("http://mrAddr")
+
+       time.Sleep(100 * time.Millisecond)
+
+       var actualRequest *http.Request
+       clientMock.AssertCalled(t, "Get", "http://mrAddr/events/topic/users/dmaapmediatorproducer")
+       clientMock.AssertNumberOfCalls(t, "Get", 1)
+
+       clientMock.AssertCalled(t, "Do", mock.MatchedBy(func(req *http.Request) bool {
+               actualRequest = req
+               return true
+       }))
+       assertions.Equal(http.MethodPost, actualRequest.Method)
+       assertions.Equal("consumerHost", actualRequest.URL.Host)
+       assertions.Equal("/target", actualRequest.URL.Path)
+       assertions.Equal("application/json; charset=utf-8", actualRequest.Header.Get("Content-Type"))
+       actualBody, _ := ioutil.ReadAll(actualRequest.Body)
+       assertions.Equal([]byte(`[{"message": {"data": "data"}}]`), actualBody)
+       clientMock.AssertNumberOfCalls(t, "Do", 1)
+}