1 files changed, 130 insertions, 0 deletions
diff --git a/src/go/collectors/go.d.plugin/agent/discovery/sd/pipeline/sim_test.go b/src/go/collectors/go.d.plugin/agent/discovery/sd/pipeline/sim_test.go
new file mode 100644
index 000000000..23a120751
--- /dev/null
+++ b/src/go/collectors/go.d.plugin/agent/discovery/sd/pipeline/sim_test.go
@@ -0,0 +1,130 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package pipeline
+
+import (
+	"context"
+	"sort"
+	"testing"
+	"time"
+
+	"github.com/netdata/netdata/go/go.d.plugin/agent/confgroup"
+	"github.com/netdata/netdata/go/go.d.plugin/agent/discovery/sd/model"
+	"github.com/netdata/netdata/go/go.d.plugin/logger"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gopkg.in/yaml.v2"
+)
+
+type discoverySim struct {
+	config            string
+	discoverers       []model.Discoverer
+	wantClassifyCalls int
+	wantComposeCalls  int
+	wantConfGroups    []*confgroup.Group
+}
+
+func (sim discoverySim) run(t *testing.T) {
+	t.Helper()
+
+	var cfg Config
+	err := yaml.Unmarshal([]byte(sim.config), &cfg)
+	require.Nilf(t, err, "cfg unmarshal")
+
+	clr, err := newTargetClassificator(cfg.Classify)
+	require.Nil(t, err, "newTargetClassificator")
+
+	cmr, err := newConfigComposer(cfg.Compose)
+	require.Nil(t, err, "newConfigComposer")
+
+	mockClr := &mockClassificator{clr: clr}
+	mockCmr := &mockComposer{cmr: cmr}
+
+	accum := newAccumulator()
+	accum.sendEvery = time.Second * 2
+
+	pl := &Pipeline{
+		Logger:      logger.New(),
+		discoverers: sim.discoverers,
+		accum:       accum,
+		clr:         mockClr,
+		cmr:         mockCmr,
+		configs:     make(map[string]map[uint64][]confgroup.Config),
+	}
+
+	pl.accum.Logger = pl.Logger
+	clr.Logger = pl.Logger
+	cmr.Logger = pl.Logger
+
+	groups := sim.collectGroups(t, pl)
+
+	sortConfigGroups(groups)
+	sortConfigGroups(sim.wantConfGroups)
+
+	assert.Equal(t, sim.wantConfGroups, groups)
+	assert.Equalf(t, sim.wantClassifyCalls, mockClr.calls, "classify calls")
+	assert.Equalf(t, sim.wantComposeCalls, mockCmr.calls, "compose calls")
+}
+
+func (sim discoverySim) collectGroups(t *testing.T, pl *Pipeline) []*confgroup.Group {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	in := make(chan []*confgroup.Group)
+	done := make(chan struct{})
+
+	go func() { defer close(done); pl.Run(ctx, in) }()
+
+	timeout := time.Second * 10
+	var groups []*confgroup.Group
+
+	func() {
+		for {
+			select {
+			case inGroups := <-in:
+				groups = append(groups, inGroups...)
+			case <-done:
+				return
+			case <-time.After(timeout):
+				t.Logf("discovery timed out after %s, got %d groups, expected %d, some events are skipped",
+					timeout, len(groups), len(sim.wantConfGroups))
+				return
+			}
+		}
+	}()
+
+	return groups
+}
+
+type mockClassificator struct {
+	calls int
+	clr   *targetClassificator
+}
+
+func (m *mockClassificator) classify(tgt model.Target) model.Tags {
+	m.calls++
+	return m.clr.classify(tgt)
+}
+
+type mockComposer struct {
+	calls int
+	cmr   *configComposer
+}
+
+func (m *mockComposer) compose(tgt model.Target) []confgroup.Config {
+	m.calls++
+	return m.cmr.compose(tgt)
+}
+
+func sortConfigGroups(groups []*confgroup.Group) {
+	sort.Slice(groups, func(i, j int) bool {
+		return groups[i].Source < groups[j].Source
+	})
+
+	for _, g := range groups {
+		sort.Slice(g.Configs, func(i, j int) bool {
+			return g.Configs[i].Name() < g.Configs[j].Name()
+		})
+	}
+}