+++ /dev/null
-// -
-// ========================LICENSE_START=================================
-// O-RAN-SC
-// %%
-// Copyright (C) 2022: Nordix Foundation
-// %%
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-// ========================LICENSE_END===================================
-//
-
-package main
-
-import (
- "bytes"
- "fmt"
- "io/ioutil"
- "net/http"
- "os/exec"
- "sync"
- "testing"
- "time"
-
- "github.com/stretchr/testify/require"
- "oransc.org/nonrtric/dmaapmediatorproducer/internal/config"
- "oransc.org/nonrtric/dmaapmediatorproducer/internal/jobs"
- "oransc.org/nonrtric/dmaapmediatorproducer/internal/kafkaclient"
-)
-
-// This is not a real test, just a way to get the Swagger documentation generated automatically.
-// Hence there are no assertions in this test.
-func TestGenerateSwaggerDocs(t *testing.T) {
- cmd := exec.Command("./generate_swagger_docs.sh")
-
- err := cmd.Run()
- if err != nil {
- fmt.Println("Error generating Swagger:", err)
- }
-}
-
-func TestValidateConfiguration(t *testing.T) {
- assertions := require.New(t)
-
- validConfig := config.Config{
- InfoProducerHost: "host",
- DMaaPMRAddress: "address",
- KafkaBootstrapServers: "servers",
- ProducerCertPath: "path",
- ProducerKeyPath: "path",
- }
- assertions.Nil(validateConfiguration(&validConfig))
-
- missingProducerHost := config.Config{
- DMaaPMRAddress: "address",
- KafkaBootstrapServers: "servers",
- ProducerCertPath: "path",
- ProducerKeyPath: "path",
- }
- assertions.Contains(validateConfiguration(&missingProducerHost).Error(), "INFO_PRODUCER_HOST")
-
- missingCert := config.Config{
- InfoProducerHost: "host",
- DMaaPMRAddress: "address",
- KafkaBootstrapServers: "servers",
- ProducerKeyPath: "path",
- }
- assertions.Contains(validateConfiguration(&missingCert).Error(), "PRODUCER_CERT")
-
- missingCertKey := config.Config{
- InfoProducerHost: "host",
- DMaaPMRAddress: "address",
- KafkaBootstrapServers: "servers",
- ProducerCertPath: "path",
- }
- assertions.Contains(validateConfiguration(&missingCertKey).Error(), "PRODUCER_KEY")
-
- missingMRAddress := config.Config{
- InfoProducerHost: "host",
- KafkaBootstrapServers: "servers",
- ProducerCertPath: "path",
- ProducerKeyPath: "path",
- }
- assertions.Nil(validateConfiguration(&missingMRAddress))
-
- missingKafkaServers := config.Config{
- InfoProducerHost: "host",
- DMaaPMRAddress: "address",
- ProducerCertPath: "path",
- ProducerKeyPath: "path",
- }
- assertions.Nil(validateConfiguration(&missingKafkaServers))
-
- missingMRAddressdAndKafkaServers := config.Config{
- InfoProducerHost: "host",
- ProducerCertPath: "path",
- ProducerKeyPath: "path",
- }
- assertions.Contains(validateConfiguration(&missingMRAddressdAndKafkaServers).Error(), "DMAAP_MR_ADDR")
- assertions.Contains(validateConfiguration(&missingMRAddressdAndKafkaServers).Error(), "KAFKA_BOOTSRAP_SERVERS")
-}
-
-func TestRegisterTypesAndProducer(t *testing.T) {
- assertions := require.New(t)
-
- wg := sync.WaitGroup{}
- clientMock := NewTestClient(func(req *http.Request) *http.Response {
- if req.URL.String() == configuration.InfoCoordinatorAddress+"/data-producer/v1/info-types/STD_Fault_Messages" {
- assertions.Equal(req.Method, "PUT")
- body := getBodyAsString(req, t)
- assertions.Contains(body, "info_job_data_schema")
- assertions.Equal("application/json", req.Header.Get("Content-Type"))
- wg.Done()
- return &http.Response{
- StatusCode: 200,
- Body: ioutil.NopCloser(bytes.NewBufferString(`OK`)),
- Header: make(http.Header), // Must be set to non-nil value or it panics
- }
- } else if req.URL.String() == configuration.InfoCoordinatorAddress+"/data-producer/v1/info-types/Kafka_TestTopic" {
- assertions.Equal(req.Method, "PUT")
- body := getBodyAsString(req, t)
- assertions.Contains(body, "info_job_data_schema")
- assertions.Equal("application/json", req.Header.Get("Content-Type"))
- wg.Done()
- return &http.Response{
- StatusCode: 200,
- Body: ioutil.NopCloser(bytes.NewBufferString(`OK`)),
- Header: make(http.Header), // Must be set to non-nil value or it panics
- }
- } else if req.URL.String() == configuration.InfoCoordinatorAddress+"/data-producer/v1/info-producers/DMaaP_Mediator_Producer" {
- assertions.Equal(req.Method, "PUT")
- body := getBodyAsString(req, t)
- assertions.Contains(body, "callbackAddress/health_check")
- assertions.Contains(body, "callbackAddress/info_job")
- assertions.Contains(body, "Kafka_TestTopic")
- assertions.Contains(body, "STD_Fault_Messages")
- assertions.Equal("application/json", req.Header.Get("Content-Type"))
- wg.Done()
- return &http.Response{
- StatusCode: 200,
- Body: ioutil.NopCloser(bytes.NewBufferString(`OK`)),
- Header: make(http.Header), // Must be set to non-nil value or it panics
- }
- }
- t.Error("Wrong call to client: ", req)
- t.Fail()
- return nil
- })
- jobsManager := jobs.NewJobsManagerImpl(clientMock, configuration.DMaaPMRAddress, kafkaclient.KafkaFactoryImpl{}, nil)
-
- wg.Add(3)
- err := registerTypesAndProducer(jobsManager, configuration.InfoCoordinatorAddress, "callbackAddress", clientMock)
-
- assertions.Nil(err)
-
- if waitTimeout(&wg, 2*time.Second) {
- t.Error("Not all calls to server were made")
- t.Fail()
- }
-}
-
-type RoundTripFunc func(req *http.Request) *http.Response
-
-func (f RoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
- return f(req), nil
-}
-
-//NewTestClient returns *http.Client with Transport replaced to avoid making real calls
-func NewTestClient(fn RoundTripFunc) *http.Client {
- return &http.Client{
- Transport: RoundTripFunc(fn),
- }
-}
-
-func getBodyAsString(req *http.Request, t *testing.T) string {
- buf := new(bytes.Buffer)
- if _, err := buf.ReadFrom(req.Body); err != nil {
- t.Fail()
- }
- return buf.String()
-}
-
-// waitTimeout waits for the waitgroup for the specified max timeout.
-// Returns true if waiting timed out.
-func waitTimeout(wg *sync.WaitGroup, timeout time.Duration) bool {
- c := make(chan struct{})
- go func() {
- defer close(c)
- wg.Wait()
- }()
- select {
- case <-c:
- return false // completed normally
- case <-time.After(timeout):
- return true // timed out
- }
-}
Code Review / nonrtric.git / blobdiff