2 // ========================LICENSE_START=================================
5 // Copyright (C) 2021: Nordix Foundation
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
11 // http://www.apache.org/licenses/LICENSE-2.0
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===================================
33 "github.com/stretchr/testify/require"
34 "oransc.org/nonrtric/dmaapmediatorproducer/internal/config"
37 const typeDefinition = `{"types": [{"id": "type1", "dmaapTopicUrl": "events/unauthenticated.SEC_FAULT_OUTPUT/dmaapmediatorproducer/type1"}]}`
39 func TestJobsManagerGetTypes_filesOkShouldReturnSliceOfTypesAndProvideSupportedTypes(t *testing.T) {
40 assertions := require.New(t)
41 typesDir, err := os.MkdirTemp("", "configs")
43 t.Errorf("Unable to create temporary directory for types due to: %v", err)
45 fname := filepath.Join(typesDir, "type_config.json")
46 managerUnderTest := NewJobsManagerImpl(fname, nil, "", nil)
48 os.RemoveAll(typesDir)
50 if err = os.WriteFile(fname, []byte(typeDefinition), 0666); err != nil {
51 t.Errorf("Unable to create temporary config file for types due to: %v", err)
53 types, err := managerUnderTest.LoadTypesFromConfiguration()
54 wantedType := config.TypeDefinition{
56 DmaapTopicURL: "events/unauthenticated.SEC_FAULT_OUTPUT/dmaapmediatorproducer/type1",
58 wantedTypes := []config.TypeDefinition{wantedType}
59 assertions.EqualValues(wantedTypes, types)
62 supportedTypes := managerUnderTest.GetSupportedTypes()
63 assertions.EqualValues([]string{"type1"}, supportedTypes)
66 func TestJobsManagerAddJobWhenTypeIsSupported_shouldAddJobToChannel(t *testing.T) {
67 assertions := require.New(t)
68 managerUnderTest := NewJobsManagerImpl("", nil, "", nil)
72 InfoJobIdentity: "job1",
75 InfoTypeIdentity: "type1",
77 jobsHandler := jobsHandler{
78 addJobCh: make(chan JobInfo)}
79 managerUnderTest.allTypes["type1"] = TypeData{
81 jobsHandler: &jobsHandler,
86 err = managerUnderTest.AddJobFromRESTCall(wantedJob)
90 addedJob := <-jobsHandler.addJobCh
91 assertions.Equal(wantedJob, addedJob)
94 func TestJobsManagerAddJobWhenTypeIsNotSupported_shouldReturnError(t *testing.T) {
95 assertions := require.New(t)
96 managerUnderTest := NewJobsManagerImpl("", nil, "", nil)
98 InfoTypeIdentity: "type1",
101 err := managerUnderTest.AddJobFromRESTCall(jobInfo)
102 assertions.NotNil(err)
103 assertions.Equal("type not supported: type1", err.Error())
106 func TestJobsManagerAddJobWhenJobIdMissing_shouldReturnError(t *testing.T) {
107 assertions := require.New(t)
108 managerUnderTest := NewJobsManagerImpl("", nil, "", nil)
109 managerUnderTest.allTypes["type1"] = TypeData{
114 InfoTypeIdentity: "type1",
116 err := managerUnderTest.AddJobFromRESTCall(jobInfo)
117 assertions.NotNil(err)
118 assertions.Equal("missing required job identity: { <nil> type1}", err.Error())
121 func TestJobsManagerAddJobWhenTargetUriMissing_shouldReturnError(t *testing.T) {
122 assertions := require.New(t)
123 managerUnderTest := NewJobsManagerImpl("", nil, "", nil)
124 managerUnderTest.allTypes["type1"] = TypeData{
129 InfoTypeIdentity: "type1",
130 InfoJobIdentity: "job1",
132 err := managerUnderTest.AddJobFromRESTCall(jobInfo)
133 assertions.NotNil(err)
134 assertions.Equal("missing required target URI: { job1 <nil> type1}", err.Error())
137 func TestJobsManagerDeleteJob_shouldSendDeleteToChannel(t *testing.T) {
138 assertions := require.New(t)
139 managerUnderTest := NewJobsManagerImpl("", nil, "", nil)
140 jobsHandler := jobsHandler{
141 deleteJobCh: make(chan string)}
142 managerUnderTest.allTypes["type1"] = TypeData{
144 jobsHandler: &jobsHandler,
147 go managerUnderTest.DeleteJobFromRESTCall("job2")
149 assertions.Equal("job2", <-jobsHandler.deleteJobCh)
152 func TestAddJobToJobsManager_shouldStartPollAndDistributeMessages(t *testing.T) {
153 assertions := require.New(t)
156 messages := `[{"message": {"data": "data"}}]`
157 pollClientMock := NewTestClient(func(req *http.Request) *http.Response {
158 if req.URL.String() == "http://mrAddr/topicUrl" {
159 assertions.Equal(req.Method, "GET")
165 return &http.Response{
167 Body: ioutil.NopCloser(bytes.NewReader([]byte(body))),
168 Header: make(http.Header), // Must be set to non-nil value or it panics
171 t.Error("Wrong call to client: ", req)
176 wg := sync.WaitGroup{}
177 distributeClientMock := NewTestClient(func(req *http.Request) *http.Response {
178 if req.URL.String() == "http://consumerHost/target" {
179 assertions.Equal(req.Method, "POST")
180 assertions.Equal(messages, getBodyAsString(req, t))
181 assertions.Equal("application/json", req.Header.Get("Content-Type"))
183 return &http.Response{
185 Body: ioutil.NopCloser(bytes.NewBufferString(`OK`)),
186 Header: make(http.Header), // Must be set to non-nil value or it panics
189 t.Error("Wrong call to client: ", req)
193 jobsHandler := newJobsHandler("type1", "/topicUrl", pollClientMock, distributeClientMock)
195 jobsManager := NewJobsManagerImpl("", pollClientMock, "http://mrAddr", distributeClientMock)
196 jobsManager.allTypes["type1"] = TypeData{
197 DMaaPTopicURL: "/topicUrl",
199 jobsHandler: jobsHandler,
202 jobsManager.StartJobsForAllTypes()
205 InfoTypeIdentity: "type1",
206 InfoJobIdentity: "job1",
207 TargetUri: "http://consumerHost/target",
210 wg.Add(1) // Wait till the distribution has happened
211 err := jobsManager.AddJobFromRESTCall(jobInfo)
214 if waitTimeout(&wg, 2*time.Second) {
215 t.Error("Not all calls to server were made")
220 func TestJobsHandlerDeleteJob_shouldDeleteJobFromJobsMap(t *testing.T) {
221 jobToDelete := newJob(JobInfo{}, nil)
222 go jobToDelete.start()
223 jobsHandler := newJobsHandler("type1", "/topicUrl", nil, nil)
224 jobsHandler.jobs["job1"] = jobToDelete
226 go jobsHandler.monitorManagementChannels()
228 jobsHandler.deleteJobCh <- "job1"
231 for i := 0; i < 100; i++ {
232 if len(jobsHandler.jobs) == 0 {
236 time.Sleep(time.Microsecond) // Need to drop control to let the job's goroutine do the job
238 require.New(t).True(deleted, "Job not deleted")
241 func TestJobsHandlerEmptyJobMessageBufferWhenItIsFull(t *testing.T) {
242 job := newJob(JobInfo{
243 InfoJobIdentity: "job",
246 jobsHandler := newJobsHandler("type1", "/topicUrl", nil, nil)
247 jobsHandler.jobs["job1"] = job
249 fillMessagesBuffer(job.messagesChannel)
251 jobsHandler.distributeMessages([]byte("sent msg"))
253 require.New(t).Len(job.messagesChannel, 0)
256 func fillMessagesBuffer(mc chan []byte) {
257 for i := 0; i < cap(mc); i++ {
262 type RoundTripFunc func(req *http.Request) *http.Response
264 func (f RoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
268 //NewTestClient returns *http.Client with Transport replaced to avoid making real calls
269 func NewTestClient(fn RoundTripFunc) *http.Client {
271 Transport: RoundTripFunc(fn),
275 // waitTimeout waits for the waitgroup for the specified max timeout.
276 // Returns true if waiting timed out.
277 func waitTimeout(wg *sync.WaitGroup, timeout time.Duration) bool {
278 c := make(chan struct{})
285 return false // completed normally
286 case <-time.After(timeout):
287 return true // timed out
291 func getBodyAsString(req *http.Request, t *testing.T) string {
292 buf := new(bytes.Buffer)
293 if _, err := buf.ReadFrom(req.Body); err != nil {