test: add tests for manager

internal/manager/manager.go

@@ -3,6 +3,7 @@ package manager
 import (
 	"context"
 	"crypto/tls"
+	"errors"
 	k2kconfig "github.com/raczu/kube2kafka/internal/config"
 	"github.com/raczu/kube2kafka/pkg/exporter"
 	"github.com/raczu/kube2kafka/pkg/kube/watcher"
@@ -13,6 +14,10 @@ import (
 	"sync"
 )
 
+var (
+	ErrManagerNotSetup = errors.New("manager was not set up")
+)
+
 type Manager struct {
 	watcher    *watcher.Watcher
 	processor  *processor.Processor
@@ -98,17 +103,20 @@ func (m *Manager) Setup() error {
 }
 
 func (m *Manager) Start(ctx context.Context) error {
+	if m.watcher == nil || m.processor == nil || m.exporter == nil {
+		return ErrManagerNotSetup
+	}
+
 	m.logger.Info("starting manager...")
-	ch := make(chan error, 1)
+	errs := make(chan error, 1)
 	subctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
 	m.wg.Add(3)
 	go func() {
 		defer m.wg.Done()
 		if err := m.watcher.Watch(subctx); err != nil {
-			ch <- err
-			return
+			errs <- err
 		}
 	}()
 
@@ -120,19 +128,25 @@ func (m *Manager) Start(ctx context.Context) error {
 	go func() {
 		defer m.wg.Done()
 		if err := m.exporter.Export(subctx); err != nil {
-			ch <- err
+			errs <- err
 		}
 	}()
 
 	var err error
 	select {
-	case <-subctx.Done():
-		m.logger.Info("context canceled, shutting down manager...")
-	case err = <-ch:
-		m.logger.Error("one of the components failed", zap.Error(err))
+	case <-ctx.Done():
+		m.logger.Info("context canceled, stopping manager...")
+	case err = <-errs:
+		m.logger.Error("one of the components failed, stopping manager...")
 		cancel()
 	}
 
 	m.wg.Wait()
+	// Ensure that error is not lost even if the context was canceled.
+	close(errs)
+	if e, ok := <-errs; ok {
+		err = e
+	}
+
 	return err
 }

internal/manager/manager_suite_test.go

@@ -0,0 +1,13 @@
+package manager_test
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestManager(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Manager Suite")
+}