chore: improve code coverage (#350)

* chore: improve code coverage for registry package

* chore: test for empty discover plugin

* chore: add test for invalid labels in compass sink

* chore: add build plugins label to tests for tableau extractor

* chore: add build plugins label to tests for bigquery extractor

* chore: add test for new stats client in metrics

* chore: add build plugins label to tests for bigquery extractor

* fix(lint): revert back one commit

* chore: improve test coverage for reistry package to 100%

* chore: improve code-coverage for agent package

* chore: remove unused error file from recipe

* chore: add test for checking invalid template path

agent/agent_test.go

@@ -776,6 +776,91 @@ func TestAgentRunMultiple(t *testing.T) {
 	})
 }
 
+func TestValidate(t *testing.T) {
+	t.Run("should return error if plugins in recipe not found in Factory", func(t *testing.T) {
+		r := agent.NewAgent(agent.Config{
+			ExtractorFactory: registry.NewExtractorFactory(),
+			ProcessorFactory: registry.NewProcessorFactory(),
+			SinkFactory:      registry.NewSinkFactory(),
+			Logger:           utils.Logger,
+		})
+		var expectedErrs []error
+		errs := r.Validate(validRecipe)
+		expectedErrs = append(expectedErrs, plugins.NotFoundError{Type: plugins.PluginTypeExtractor, Name: "test-extractor"})
+		expectedErrs = append(expectedErrs, plugins.NotFoundError{Type: plugins.PluginTypeSink, Name: "test-sink"})
+		expectedErrs = append(expectedErrs, plugins.NotFoundError{Type: plugins.PluginTypeProcessor, Name: "test-processor"})
+		assert.Equal(t, 3, len(errs))
+		assert.Equal(t, expectedErrs, errs)
+	})
+	t.Run("", func(t *testing.T) {
+		var invalidRecipe = recipe.Recipe{
+			Name: "sample",
+			Source: recipe.PluginRecipe{
+				Name: "test-extractor",
+				Config: map[string]interface{}{
+					"proc-foo": "proc-bar",
+				},
+			},
+			Processors: []recipe.PluginRecipe{
+				{
+					Name: "test-processor",
+					Config: map[string]interface{}{
+						"proc-foo": "proc-bar",
+					},
+				},
+			},
+			Sinks: []recipe.PluginRecipe{
+				{
+					Name: "test-sink",
+					Config: map[string]interface{}{
+						"url": "http://localhost:3000/data",
+					},
+				},
+			},
+		}
+
+		extr := mocks.NewExtractor()
+		err := plugins.InvalidConfigError{}
+		extr.On("Validate", invalidRecipe.Source.Config).Return(err).Once()
+		ef := registry.NewExtractorFactory()
+		if err := ef.Register("test-extractor", newExtractor(extr)); err != nil {
+			t.Fatal(err)
+		}
+
+		proc := mocks.NewProcessor()
+		proc.On("Validate", invalidRecipe.Processors[0].Config).Return(err).Once()
+		defer proc.AssertExpectations(t)
+		pf := registry.NewProcessorFactory()
+		if err := pf.Register("test-processor", newProcessor(proc)); err != nil {
+			t.Fatal(err)
+		}
+
+		sink := mocks.NewSink()
+		sink.On("Validate", invalidRecipe.Sinks[0].Config).Return(err).Once()
+		defer sink.AssertExpectations(t)
+		sf := registry.NewSinkFactory()
+		if err := sf.Register("test-sink", newSink(sink)); err != nil {
+			t.Fatal(err)
+		}
+
+		r := agent.NewAgent(agent.Config{
+			ExtractorFactory: ef,
+			ProcessorFactory: pf,
+			SinkFactory:      sf,
+			Logger:           utils.Logger,
+			Monitor:          newMockMonitor(),
+		})
+
+		var expectedErrs []error
+		errs := r.Validate(invalidRecipe)
+		assert.Equal(t, 3, len(errs))
+		expectedErrs = append(expectedErrs, enrichInvalidConfigError(err, invalidRecipe.Source.Name, plugins.PluginTypeExtractor))
+		expectedErrs = append(expectedErrs, enrichInvalidConfigError(err, invalidRecipe.Sinks[0].Name, plugins.PluginTypeSink))
+		expectedErrs = append(expectedErrs, enrichInvalidConfigError(err, invalidRecipe.Processors[0].Name, plugins.PluginTypeProcessor))
+		assert.Equal(t, expectedErrs, errs)
+	})
+}
+
 func newExtractor(extr plugins.Extractor) func() plugins.Extractor {
 	return func() plugins.Extractor {
 		return extr
@@ -821,3 +906,16 @@ type panicProcessor struct {
 func (p *panicProcessor) Process(_ context.Context, _ models.Record) (dst models.Record, err error) {
 	panic("panicking")
 }
+
+// enrichInvalidConfigError enrich the error with plugin information
+func enrichInvalidConfigError(err error, pluginName string, pluginType plugins.PluginType) error {
+	if errors.As(err, &plugins.InvalidConfigError{}) {
+		icErr := err.(plugins.InvalidConfigError)
+		icErr.PluginName = pluginName
+		icErr.Type = pluginType
+
+		return icErr
+	}
+
+	return err
+}

metrics/statsd_test.go

@@ -2,11 +2,18 @@ package metrics_test
 
 import (
 	"fmt"
+	"log"
+	"os"
 	"testing"
 
+	"github.com/odpf/meteor/test/utils"
+
 	"github.com/odpf/meteor/agent"
 	"github.com/odpf/meteor/metrics"
 	"github.com/odpf/meteor/recipe"
+	"github.com/ory/dockertest/v3"
+	"github.com/ory/dockertest/v3/docker"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 )
 
@@ -26,6 +33,47 @@ func (c *mockStatsdClient) Increment(name string) {
 	c.Called(name)
 }
 
+var port = "8125"
+
+func TestMain(m *testing.M) {
+	// setup test
+	opts := dockertest.RunOptions{
+		Repository: "statsd/statsd",
+		Tag:        "latest",
+		Env: []string{
+			"MYSQL_ALLOW_EMPTY_PASSWORD=true",
+		},
+		ExposedPorts: []string{"8125", port},
+		PortBindings: map[docker.Port][]docker.PortBinding{
+			"8125": {
+				{HostIP: "0.0.0.0", HostPort: port},
+			},
+		},
+	}
+	// exponential backoff-retry, because the application in the container might not be ready to accept connections yet
+	retryFn := func(resource *dockertest.Resource) (err error) {
+		c, err := metrics.NewStatsdClient("127.0.0.1:" + port)
+		if err != nil {
+			return
+		}
+		c.Open()
+		return
+	}
+	purgeFn, err := utils.CreateContainer(opts, retryFn)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// run tests
+	code := m.Run()
+
+	// clean tests
+	if err := purgeFn(); err != nil {
+		log.Fatal(err)
+	}
+	os.Exit(code)
+}
+
 func TestStatsdMonitorRecordRun(t *testing.T) {
 	statsdPrefix := "testprefix"
 
@@ -117,3 +165,14 @@ func TestStatsdMonitorRecordRun(t *testing.T) {
 		monitor.RecordRun(agent.Run{Recipe: recipe, DurationInMs: duration, RecordCount: 2, Success: true})
 	})
 }
+
+func TestNewStatsClient(t *testing.T) {
+	t.Run("should return error for invalid address", func(t *testing.T) {
+		_, err := metrics.NewStatsdClient("127.0.0.1")
+		assert.Error(t, err)
+	})
+	t.Run("should return error for invalid port", func(t *testing.T) {
+		_, err := metrics.NewStatsdClient("127.0.0.1:81A5")
+		assert.Error(t, err)
+	})
+}

plugins/external/discover_test.go

@@ -6,11 +6,15 @@ package plugins
 import (
 	"testing"
 
+	"github.com/odpf/meteor/registry"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestDiscoverPlugins(t *testing.T) {
 	// TODO: add test
+	factory := registry.NewProcessorFactory()
+	_, err := DiscoverPlugins(factory)
+	assert.Nil(t, err)
 }
 
 // once we already setup a test for DiscoverPlugins this test will not be needed

plugins/extractors/bigquery/profile_test.go

@@ -1,3 +1,6 @@
+//go:build plugins
+// +build plugins
+
 package bigquery
 
 import (

plugins/extractors/tableau/builder_test.go

@@ -1,3 +1,6 @@
+//go:build plugins
+// +build plugins
+
 package tableau
 
 import (

plugins/extractors/tableau/client_test.go

@@ -1,3 +1,6 @@
+//go:build plugins
+// +build plugins
+
 package tableau
 
 import (

plugins/sinks/compass/sink.go

@@ -6,15 +6,16 @@ import (
 	_ "embed"
 	"encoding/json"
 	"fmt"
+	"io/ioutil"
+	"net/http"
+	"strings"
+
 	"github.com/odpf/meteor/models"
 	"github.com/odpf/meteor/plugins"
 	"github.com/odpf/meteor/registry"
 	"github.com/odpf/meteor/utils"
 	"github.com/odpf/salt/log"
 	"github.com/pkg/errors"
-	"io/ioutil"
-	"net/http"
-	"strings"
 )
 
 //go:embed README.md

plugins/sinks/compass/sink_test.go

@@ -97,6 +97,83 @@ func TestSink(t *testing.T) {
 		}
 	})
 
+	t.Run("should return error for various invalid labels", func(t *testing.T) {
+		testData := &assetsv1beta1.User{
+			Resource: &commonv1beta1.Resource{
+				Urn:         "my-topic-urn",
+				Name:        "my-topic",
+				Service:     "kafka",
+				Type:        "topic",
+				Description: "topic information",
+			},
+			Properties: &facetsv1beta1.Properties{
+				Attributes: utils.TryParseMapToProto(map[string]interface{}{
+					"attrA": "valueAttrA",
+					"attrB": "valueAttrB",
+				}),
+				Labels: map[string]string{
+					"labelA": "valueLabelA",
+					"labelB": "valueLabelB",
+				},
+			},
+		}
+		testPayload := compass.RequestPayload{
+			Asset: compass.Asset{
+				URN:         "my-topic-urn",
+				Name:        "my-topic",
+				Service:     "kafka",
+				Type:        "topic",
+				Description: "topic information",
+			},
+		}
+		invalidConfigs := []map[string]interface{}{
+			{
+				"host": host,
+				"labels": map[string]string{
+					"foo": "$properties.attributes",
+				},
+			},
+			{
+				"host": host,
+				"labels": map[string]string{
+					"foo": "$properties.attributes.12",
+				},
+			},
+			{
+				"host": host,
+				"labels": map[string]string{
+					"foo": "$properties.attributes.attrC",
+				},
+			},
+			{
+				"host": host,
+				"labels": map[string]string{
+					"foo": "$invalid.attributes.attrC",
+				},
+			},
+			{
+				"host": host,
+				"labels": map[string]string{
+					"bar": "$properties.labels.labelC",
+				},
+			},
+		}
+		for _, c := range invalidConfigs {
+			client := newMockHTTPClient(c, http.MethodPatch, url, testPayload)
+			client.SetupResponse(200, "")
+			ctx := context.TODO()
+			compassSink := compass.New(client, testUtils.Logger)
+			err := compassSink.Init(ctx, c)
+			if err != nil {
+				t.Fatal(err)
+			}
+			err = compassSink.Sink(ctx, []models.Record{models.NewRecord(testData)})
+			assert.Error(t, err)
+			fmt.Println(err)
+		}
+
+	})
+
 	successTestCases := []struct {
 		description string
 		data        models.Metadata

recipe/generator_test.go

@@ -14,6 +14,29 @@ import (
 )
 
 func TestFromTemplate(t *testing.T) {
+	t.Run("should throw error for invalid template path", func(t *testing.T) {
+		templatePath := "./testdata/template.yaml"
+		outputDir := "./test/temp"
+		bytes, err := ioutil.ReadFile("./testdata/generator/data-3.yaml")
+		if err != nil {
+			fmt.Println(fmt.Errorf("error reading data: %w", err))
+			return
+		}
+
+		var data []recipe.TemplateData
+		if err := yaml.Unmarshal(bytes, &data); err != nil {
+			fmt.Println(fmt.Errorf("error parsing data: %w", err))
+			return
+		}
+
+		err = recipe.FromTemplate(recipe.TemplateConfig{
+			TemplateFilePath: templatePath,
+			OutputDirPath:    outputDir,
+			Data:             data,
+		})
+		assert.Error(t, err)
+	})
+
 	t.Run("should output recipe files using template to output directory", func(t *testing.T) {
 		templatePath := "./testdata/generator/template.yaml"
 		outputDir := "./testdata/generator/temp"