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 TestGetTypes_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 handlerUnderTest := NewJobHandlerImpl(fname, nil, nil)
48 os.RemoveAll(typesDir)
49 handlerUnderTest.clearAll()
51 if err = os.WriteFile(fname, []byte(typeDefinition), 0666); err != nil {
52 t.Errorf("Unable to create temporary config file for types due to: %v", err)
54 types, err := handlerUnderTest.GetTypes()
55 wantedType := config.TypeDefinition{
57 DmaapTopicURL: "events/unauthenticated.SEC_FAULT_OUTPUT/dmaapmediatorproducer/type1",
59 wantedTypes := []config.TypeDefinition{wantedType}
60 assertions.EqualValues(wantedTypes, types)
63 supportedTypes := handlerUnderTest.GetSupportedTypes()
64 assertions.EqualValues([]string{"type1"}, supportedTypes)
67 func TestAddJobWhenTypeIsSupported_shouldAddJobToAllJobsMap(t *testing.T) {
68 assertions := require.New(t)
69 handlerUnderTest := NewJobHandlerImpl("", nil, nil)
73 InfoJobIdentity: "job1",
76 InfoTypeIdentity: "type1",
78 handlerUnderTest.allTypes["type1"] = TypeData{
80 Jobs: map[string]JobInfo{"job1": wantedJob},
83 handlerUnderTest.clearAll()
86 err := handlerUnderTest.AddJob(wantedJob)
88 assertions.Equal(1, len(handlerUnderTest.allTypes["type1"].Jobs))
89 assertions.Equal(wantedJob, handlerUnderTest.allTypes["type1"].Jobs["job1"])
92 func TestAddJobWhenTypeIsNotSupported_shouldReturnError(t *testing.T) {
93 assertions := require.New(t)
94 handlerUnderTest := NewJobHandlerImpl("", nil, nil)
96 InfoTypeIdentity: "type1",
99 err := handlerUnderTest.AddJob(jobInfo)
100 assertions.NotNil(err)
101 assertions.Equal("type not supported: type1", err.Error())
104 func TestAddJobWhenJobIdMissing_shouldReturnError(t *testing.T) {
105 assertions := require.New(t)
106 handlerUnderTest := NewJobHandlerImpl("", nil, nil)
107 handlerUnderTest.allTypes["type1"] = TypeData{
111 handlerUnderTest.clearAll()
115 InfoTypeIdentity: "type1",
117 err := handlerUnderTest.AddJob(jobInfo)
118 assertions.NotNil(err)
119 assertions.Equal("missing required job identity: { <nil> type1}", err.Error())
122 func TestAddJobWhenTargetUriMissing_shouldReturnError(t *testing.T) {
123 assertions := require.New(t)
124 handlerUnderTest := NewJobHandlerImpl("", nil, nil)
125 handlerUnderTest.allTypes["type1"] = TypeData{
129 handlerUnderTest.clearAll()
133 InfoTypeIdentity: "type1",
134 InfoJobIdentity: "job1",
136 err := handlerUnderTest.AddJob(jobInfo)
137 assertions.NotNil(err)
138 assertions.Equal("missing required target URI: { job1 <nil> type1}", err.Error())
141 func TestDeleteJob(t *testing.T) {
142 assertions := require.New(t)
143 handlerUnderTest := NewJobHandlerImpl("", nil, nil)
144 jobToKeep := JobInfo{
145 InfoJobIdentity: "job1",
146 InfoTypeIdentity: "type1",
148 jobToDelete := JobInfo{
149 InfoJobIdentity: "job2",
150 InfoTypeIdentity: "type1",
152 handlerUnderTest.allTypes["type1"] = TypeData{
154 Jobs: map[string]JobInfo{"job1": jobToKeep, "job2": jobToDelete},
157 handlerUnderTest.clearAll()
160 handlerUnderTest.DeleteJob("job2")
161 assertions.Equal(1, len(handlerUnderTest.allTypes["type1"].Jobs))
162 assertions.Equal(jobToKeep, handlerUnderTest.allTypes["type1"].Jobs["job1"])
165 func TestPollAndDistributeMessages(t *testing.T) {
166 assertions := require.New(t)
168 wg := sync.WaitGroup{}
169 messages := `[{"message": {"data": "data"}}]`
170 pollClientMock := NewTestClient(func(req *http.Request) *http.Response {
171 if req.URL.String() == "http://mrAddr/topicUrl" {
172 assertions.Equal(req.Method, "GET")
173 wg.Done() // Signal that the poll call has been made
174 return &http.Response{
176 Body: ioutil.NopCloser(bytes.NewReader([]byte(messages))),
177 Header: make(http.Header), // Must be set to non-nil value or it panics
180 t.Error("Wrong call to client: ", req)
184 distributeClientMock := NewTestClient(func(req *http.Request) *http.Response {
185 if req.URL.String() == "http://consumerHost/target" {
186 assertions.Equal(req.Method, "POST")
187 assertions.Equal(messages, getBodyAsString(req))
188 assertions.Equal("application/json; charset=utf-8", req.Header.Get("Content-Type"))
189 wg.Done() // Signal that the distribution call has been made
190 return &http.Response{
192 Body: ioutil.NopCloser(bytes.NewBufferString(`OK`)),
193 Header: make(http.Header), // Must be set to non-nil value or it panics
196 t.Error("Wrong call to client: ", req)
200 handlerUnderTest := NewJobHandlerImpl("", pollClientMock, distributeClientMock)
202 InfoTypeIdentity: "type1",
203 InfoJobIdentity: "job1",
204 TargetUri: "http://consumerHost/target",
206 handlerUnderTest.allTypes["type1"] = TypeData{
208 DMaaPTopicURL: "topicUrl",
209 Jobs: map[string]JobInfo{"job1": jobInfo},
212 handlerUnderTest.clearAll()
215 wg.Add(2) // Two calls should be made to the server, one to poll and one to distribute
216 handlerUnderTest.pollAndDistributeMessages("http://mrAddr")
218 if waitTimeout(&wg, 100*time.Millisecond) {
219 t.Error("Not all calls to server were made")
224 type RoundTripFunc func(req *http.Request) *http.Response
226 func (f RoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
230 //NewTestClient returns *http.Client with Transport replaced to avoid making real calls
231 func NewTestClient(fn RoundTripFunc) *http.Client {
233 Transport: RoundTripFunc(fn),
237 // waitTimeout waits for the waitgroup for the specified max timeout.
238 // Returns true if waiting timed out.
239 func waitTimeout(wg *sync.WaitGroup, timeout time.Duration) bool {
240 c := make(chan struct{})
247 return false // completed normally
248 case <-time.After(timeout):
249 return true // timed out
253 func getBodyAsString(req *http.Request) string {
254 buf := new(bytes.Buffer)
255 buf.ReadFrom(req.Body)