From 97e3e44a5003dbb370b79ae6218f6ba1f0039aab Mon Sep 17 00:00:00 2001 From: Aditya Singh Sisodiya Date: Fri, 8 Oct 2021 13:31:56 +0530 Subject: [PATCH] feat: add e2e testing (#243) * feat: add e2e recipe testing * refactor: shift e2e to test folder * refactor: restructure test folder * fix: close method to close any connection * refactor: e2e test * feat: add make test-e2e command * chore: update test yaml * refactor: change build tag * chore: add count --- .github/workflows/test.yml | 2 +- Makefile | 3 + agent/agent.go | 6 + agent/agent_test.go | 51 +-- agent/stream.go | 14 +- plugins/extractors/bigquery/bigquery_test.go | 6 +- plugins/extractors/bigtable/bigtable_test.go | 9 +- .../extractors/cassandra/cassandra_test.go | 10 +- .../extractors/clickhouse/clickhouse_test.go | 6 +- plugins/extractors/couchdb/couchdb_test.go | 8 +- plugins/extractors/csv/csv_test.go | 8 +- plugins/extractors/elastic/elastic_test.go | 6 +- plugins/extractors/gcs/gcs_test.go | 4 +- plugins/extractors/grafana/grafana_test.go | 8 +- plugins/extractors/kafka/kafka_test.go | 6 +- plugins/extractors/metabase/metabase_test.go | 8 +- plugins/extractors/mongodb/mongodb_test.go | 8 +- plugins/extractors/mssql/mssql_test.go | 8 +- plugins/extractors/mysql/mysql_test.go | 8 +- plugins/extractors/postgres/postgres_test.go | 8 +- plugins/extractors/superset/superset_test.go | 8 +- plugins/plugin.go | 3 + plugins/sinks/columbus/sink.go | 2 + plugins/sinks/columbus/sink_test.go | 14 +- plugins/sinks/console/sink.go | 2 + plugins/sinks/kafka/sink.go | 16 +- test/e2e/e2e_test.go | 357 ++++++++++++++++++ test/e2e/mysql_kafka.yml | 15 + test/mocks/plugin.go | 5 + test/{ => utils}/dockertest.go | 2 +- test/{ => utils}/logger.go | 2 +- 31 files changed, 509 insertions(+), 104 deletions(-) create mode 100644 test/e2e/e2e_test.go create mode 100644 test/e2e/mysql_kafka.yml rename test/{ => utils}/dockertest.go (98%) rename test/{ => utils}/logger.go (91%) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 7ca098045..6b2200699 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -14,4 +14,4 @@ jobs: - name: Install packages run: go mod tidy - name: Run Test - run: make test \ No newline at end of file + run: make test diff --git a/Makefile b/Makefile index 1e42402a7..87931d1c3 100644 --- a/Makefile +++ b/Makefile @@ -18,6 +18,9 @@ test: test-coverage: test go tool cover -html=coverage.out +test-e2e: + go test ./test/e2e -tags=integration -count=1 + generate-proto: ## regenerate protos @echo " > cloning protobuf from odpf/proton" @echo " > generating protobuf" diff --git a/agent/agent.go b/agent/agent.go index 873c7c3a8..03e64045c 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -254,6 +254,12 @@ func (r *Agent) setupSink(ctx context.Context, sr recipe.SinkRecipe, stream *str return err }, defaultBatchSize) + stream.onClose(func() { + if err = sink.Close(); err != nil { + r.logger.Warn("error closing sink", "sink", sr.Name, "error", err) + } + }) + return } diff --git a/agent/agent_test.go b/agent/agent_test.go index dc1703379..c6bf99fb2 100644 --- a/agent/agent_test.go +++ b/agent/agent_test.go @@ -3,10 +3,6 @@ package agent_test import ( "context" "errors" - "github.com/odpf/meteor/test" - "testing" - "time" - "github.com/odpf/meteor/agent" "github.com/odpf/meteor/models" "github.com/odpf/meteor/models/odpf/assets" @@ -14,8 +10,11 @@ import ( "github.com/odpf/meteor/recipe" "github.com/odpf/meteor/registry" "github.com/odpf/meteor/test/mocks" + "github.com/odpf/meteor/test/utils" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" + "testing" + "time" ) var mockCtx = mock.AnythingOfType("*context.emptyCtx") @@ -43,7 +42,7 @@ func TestRunnerRun(t *testing.T) { ExtractorFactory: registry.NewExtractorFactory(), ProcessorFactory: registry.NewProcessorFactory(), SinkFactory: registry.NewSinkFactory(), - Logger: test.Logger, + Logger: utils.Logger, }) run := r.Run(validRecipe) assert.IsType(t, agent.Run{}, run) @@ -67,7 +66,7 @@ func TestRunnerRun(t *testing.T) { ExtractorFactory: registry.NewExtractorFactory(), ProcessorFactory: pf, SinkFactory: sf, - Logger: test.Logger, + Logger: utils.Logger, }) run := r.Run(validRecipe) assert.Error(t, run.Error) @@ -92,7 +91,7 @@ func TestRunnerRun(t *testing.T) { ExtractorFactory: ef, ProcessorFactory: registry.NewProcessorFactory(), SinkFactory: sf, - Logger: test.Logger, + Logger: utils.Logger, }) run := r.Run(validRecipe) assert.Error(t, run.Error) @@ -119,7 +118,7 @@ func TestRunnerRun(t *testing.T) { ExtractorFactory: ef, ProcessorFactory: pf, SinkFactory: registry.NewSinkFactory(), - Logger: test.Logger, + Logger: utils.Logger, }) run := r.Run(validRecipe) assert.Error(t, run.Error) @@ -150,7 +149,7 @@ func TestRunnerRun(t *testing.T) { ExtractorFactory: ef, ProcessorFactory: pf, SinkFactory: sf, - Logger: test.Logger, + Logger: utils.Logger, }) run := r.Run(validRecipe) assert.Error(t, run.Error) @@ -183,7 +182,7 @@ func TestRunnerRun(t *testing.T) { ExtractorFactory: ef, ProcessorFactory: pf, SinkFactory: sf, - Logger: test.Logger, + Logger: utils.Logger, }) run := r.Run(validRecipe) assert.Error(t, run.Error) @@ -218,7 +217,7 @@ func TestRunnerRun(t *testing.T) { ExtractorFactory: ef, ProcessorFactory: pf, SinkFactory: sf, - Logger: test.Logger, + Logger: utils.Logger, }) run := r.Run(validRecipe) assert.Error(t, run.Error) @@ -243,6 +242,7 @@ func TestRunnerRun(t *testing.T) { sink := mocks.NewSink() sink.On("Init", mockCtx, validRecipe.Sinks[0].Config).Return(nil).Once() + sink.On("Close").Return(nil) defer sink.AssertExpectations(t) sf := registry.NewSinkFactory() if err := sf.Register("test-sink", newSink(sink)); err != nil { @@ -253,7 +253,7 @@ func TestRunnerRun(t *testing.T) { ExtractorFactory: ef, ProcessorFactory: pf, SinkFactory: sf, - Logger: test.Logger, + Logger: utils.Logger, }) run := r.Run(validRecipe) assert.Error(t, run.Error) @@ -277,6 +277,7 @@ func TestRunnerRun(t *testing.T) { sink := mocks.NewSink() sink.On("Init", mockCtx, validRecipe.Sinks[0].Config).Return(nil).Once() + sink.On("Close").Return(nil) defer sink.AssertExpectations(t) sf := registry.NewSinkFactory() if err := sf.Register("test-sink", newSink(sink)); err != nil { @@ -287,7 +288,7 @@ func TestRunnerRun(t *testing.T) { ExtractorFactory: ef, ProcessorFactory: pf, SinkFactory: sf, - Logger: test.Logger, + Logger: utils.Logger, }) run := r.Run(validRecipe) assert.Error(t, run.Error) @@ -318,6 +319,7 @@ func TestRunnerRun(t *testing.T) { sink := mocks.NewSink() sink.On("Init", mockCtx, validRecipe.Sinks[0].Config).Return(nil).Once() + sink.On("Close").Return(nil) defer sink.AssertExpectations(t) sf := registry.NewSinkFactory() if err := sf.Register("test-sink", newSink(sink)); err != nil { @@ -328,7 +330,7 @@ func TestRunnerRun(t *testing.T) { ExtractorFactory: ef, ProcessorFactory: pf, SinkFactory: sf, - Logger: test.Logger, + Logger: utils.Logger, }) run := r.Run(validRecipe) assert.Error(t, run.Error) @@ -358,6 +360,7 @@ func TestRunnerRun(t *testing.T) { sink := mocks.NewSink() sink.On("Init", mockCtx, validRecipe.Sinks[0].Config).Return(nil).Once() + sink.On("Close").Return(nil) defer sink.AssertExpectations(t) sf := registry.NewSinkFactory() if err := sf.Register("test-sink", newSink(sink)); err != nil { @@ -368,7 +371,7 @@ func TestRunnerRun(t *testing.T) { ExtractorFactory: ef, ProcessorFactory: pf, SinkFactory: sf, - Logger: test.Logger, + Logger: utils.Logger, }) run := r.Run(validRecipe) assert.Error(t, run.Error) @@ -400,6 +403,7 @@ func TestRunnerRun(t *testing.T) { sink := mocks.NewSink() sink.On("Init", mockCtx, validRecipe.Sinks[0].Config).Return(nil).Once() sink.On("Sink", mockCtx, data).Return(errors.New("some error")) + sink.On("Close").Return(nil) defer sink.AssertExpectations(t) sf := registry.NewSinkFactory() if err := sf.Register("test-sink", newSink(sink)); err != nil { @@ -410,7 +414,7 @@ func TestRunnerRun(t *testing.T) { ExtractorFactory: ef, ProcessorFactory: pf, SinkFactory: sf, - Logger: test.Logger, + Logger: utils.Logger, }) run := r.Run(validRecipe) assert.NoError(t, run.Error) @@ -442,6 +446,7 @@ func TestRunnerRun(t *testing.T) { sink := mocks.NewSink() sink.On("Init", mockCtx, validRecipe.Sinks[0].Config).Return(nil).Once() sink.On("Sink", mockCtx, data).Return(errors.New("some error")) + sink.On("Close").Return(nil) defer sink.AssertExpectations(t) sf := registry.NewSinkFactory() if err := sf.Register("test-sink", newSink(sink)); err != nil { @@ -452,7 +457,7 @@ func TestRunnerRun(t *testing.T) { ExtractorFactory: ef, ProcessorFactory: pf, SinkFactory: sf, - Logger: test.Logger, + Logger: utils.Logger, StopOnSinkError: true, }) run := r.Run(validRecipe) @@ -485,6 +490,7 @@ func TestRunnerRun(t *testing.T) { sink := mocks.NewSink() sink.On("Init", mockCtx, validRecipe.Sinks[0].Config).Return(nil).Once() sink.On("Sink", mockCtx, data).Return(nil) + sink.On("Close").Return(nil) defer sink.AssertExpectations(t) sf := registry.NewSinkFactory() if err := sf.Register("test-sink", newSink(sink)); err != nil { @@ -495,7 +501,7 @@ func TestRunnerRun(t *testing.T) { ExtractorFactory: ef, ProcessorFactory: pf, SinkFactory: sf, - Logger: test.Logger, + Logger: utils.Logger, }) run := r.Run(validRecipe) assert.NoError(t, run.Error) @@ -528,6 +534,7 @@ func TestRunnerRun(t *testing.T) { sink := mocks.NewSink() sink.On("Init", mockCtx, validRecipe.Sinks[0].Config).Return(nil).Once() sink.On("Sink", mockCtx, data).Return(nil) + sink.On("Close").Return(nil) defer sink.AssertExpectations(t) sf := registry.NewSinkFactory() if err := sf.Register("test-sink", newSink(sink)); err != nil { @@ -544,7 +551,7 @@ func TestRunnerRun(t *testing.T) { ProcessorFactory: pf, SinkFactory: sf, Monitor: monitor, - Logger: test.Logger, + Logger: utils.Logger, }) run := r.Run(validRecipe) assert.NoError(t, run.Error) @@ -579,6 +586,7 @@ func TestRunnerRun(t *testing.T) { sink.On("Init", mockCtx, validRecipe.Sinks[0].Config).Return(nil).Once() sink.On("Sink", mockCtx, data).Return(plugins.NewRetryError(err)).Once() sink.On("Sink", mockCtx, data).Return(nil) + sink.On("Close").Return(nil) defer sink.AssertExpectations(t) sf := registry.NewSinkFactory() if err := sf.Register("test-sink", newSink(sink)); err != nil { @@ -589,7 +597,7 @@ func TestRunnerRun(t *testing.T) { ExtractorFactory: ef, ProcessorFactory: pf, SinkFactory: sf, - Logger: test.Logger, + Logger: utils.Logger, MaxRetries: 2, // need to retry "at least" 2 times since Sink returns RetryError twice RetryInitialInterval: 1 * time.Millisecond, // this is to override default retry interval to reduce test time }) @@ -628,6 +636,7 @@ func TestRunnerRunMultiple(t *testing.T) { sink := mocks.NewSink() sink.On("Init", mockCtx, validRecipe.Sinks[0].Config).Return(nil) sink.On("Sink", mockCtx, data).Return(nil) + sink.On("Close").Return(nil) defer sink.AssertExpectations(t) sf := registry.NewSinkFactory() if err := sf.Register("test-sink", newSink(sink)); err != nil { @@ -638,7 +647,7 @@ func TestRunnerRunMultiple(t *testing.T) { ExtractorFactory: ef, ProcessorFactory: pf, SinkFactory: sf, - Logger: test.Logger, + Logger: utils.Logger, }) runs := r.RunMultiple(recipeList) diff --git a/agent/stream.go b/agent/stream.go index dc3b04dd7..a6fc1f4f8 100644 --- a/agent/stream.go +++ b/agent/stream.go @@ -18,6 +18,7 @@ type subscriber struct { type stream struct { middlewares []streamMiddleware subscribers []*subscriber + onCloses []func() closed bool err error } @@ -38,6 +39,13 @@ func (s *stream) subscribe(callback func(batchedData []models.Record) error, bat return s } +// onClose() is used to register callback for after stream is closed. +func (s *stream) onClose(callback func()) *stream { + s.onCloses = append(s.onCloses, callback) + + return s +} + // broadcast() will start listening to emitter for any pushed data. // This process is blocking, so most times you would want to call this inside a goroutine. func (s *stream) broadcast() error { @@ -55,7 +63,7 @@ func (s *stream) broadcast() error { batch := newBatch(l.batchSize) // listen to channel and emit data to subscriber callback if batch is full for d := range l.channel { - if err := batch.add(d); err != nil { + if err := batch.add(d); err != nil { s.closeWithError(err) } if batch.isFull() { @@ -116,6 +124,10 @@ func (s *stream) Close() { close(l.channel) } s.closed = true + + for _, onClose := range s.onCloses { + onClose() + } } func (s *stream) runMiddlewares(d models.Record) (res models.Record, err error) { diff --git a/plugins/extractors/bigquery/bigquery_test.go b/plugins/extractors/bigquery/bigquery_test.go index b240d2304..520cc69e9 100644 --- a/plugins/extractors/bigquery/bigquery_test.go +++ b/plugins/extractors/bigquery/bigquery_test.go @@ -4,17 +4,17 @@ package bigquery_test import ( "context" + "github.com/odpf/meteor/test/utils" "testing" "github.com/odpf/meteor/plugins" "github.com/odpf/meteor/plugins/extractors/bigquery" - "github.com/odpf/meteor/test" "github.com/stretchr/testify/assert" ) func TestInit(t *testing.T) { t.Run("should return error if config is invalid", func(t *testing.T) { - extr := bigquery.New(test.Logger) + extr := bigquery.New(utils.Logger) ctx, cancel := context.WithCancel(context.Background()) defer cancel() err := extr.Init(ctx, map[string]interface{}{ @@ -24,7 +24,7 @@ func TestInit(t *testing.T) { assert.Equal(t, plugins.InvalidConfigError{}, err) }) t.Run("should not return invalid config error if config is valid", func(t *testing.T) { - extr := bigquery.New(test.Logger) + extr := bigquery.New(utils.Logger) ctx, cancel := context.WithCancel(context.Background()) defer cancel() err := extr.Init(ctx, map[string]interface{}{ diff --git a/plugins/extractors/bigtable/bigtable_test.go b/plugins/extractors/bigtable/bigtable_test.go index 67bb7d9ef..27cc35e2b 100644 --- a/plugins/extractors/bigtable/bigtable_test.go +++ b/plugins/extractors/bigtable/bigtable_test.go @@ -4,6 +4,7 @@ package bigtable_test import ( "context" + "github.com/odpf/meteor/test/utils" "log" "os" "testing" @@ -11,8 +12,6 @@ import ( "cloud.google.com/go/bigtable" "github.com/odpf/meteor/plugins" bt "github.com/odpf/meteor/plugins/extractors/bigtable" - "github.com/odpf/meteor/test" - "github.com/ory/dockertest/v3" "github.com/ory/dockertest/v3/docker" "github.com/stretchr/testify/assert" @@ -38,7 +37,7 @@ func TestMain(m *testing.M) { _, err = bigtable.NewAdminClient(context.Background(), "dev", "dev") return } - purgeFn, err := test.CreateContainer(opts, retryFn) + purgeFn, err := utils.CreateContainer(opts, retryFn) if err != nil { log.Fatal("", err) } @@ -54,7 +53,7 @@ func TestMain(m *testing.M) { func TestInit(t *testing.T) { t.Run("should return error if no project_id in config", func(t *testing.T) { - err := bt.New(test.Logger).Init(context.TODO(), map[string]interface{}{ + err := bt.New(utils.Logger).Init(context.TODO(), map[string]interface{}{ "wrong-config": "sample-project", }) @@ -62,7 +61,7 @@ func TestInit(t *testing.T) { }) t.Run("should return error if project_id is empty", func(t *testing.T) { - err := bt.New(test.Logger).Init(context.TODO(), map[string]interface{}{ + err := bt.New(utils.Logger).Init(context.TODO(), map[string]interface{}{ "project_id": "", }) diff --git a/plugins/extractors/cassandra/cassandra_test.go b/plugins/extractors/cassandra/cassandra_test.go index ead175dab..ea4a0b0b3 100644 --- a/plugins/extractors/cassandra/cassandra_test.go +++ b/plugins/extractors/cassandra/cassandra_test.go @@ -5,6 +5,7 @@ package cassandra_test import ( "context" "fmt" + "github.com/odpf/meteor/test/utils" "log" "os" "testing" @@ -16,7 +17,6 @@ import ( "github.com/odpf/meteor/models/odpf/assets/facets" "github.com/odpf/meteor/plugins" "github.com/odpf/meteor/plugins/extractors/cassandra" - "github.com/odpf/meteor/test" "github.com/odpf/meteor/test/mocks" "github.com/ory/dockertest/v3" "github.com/ory/dockertest/v3/docker" @@ -70,7 +70,7 @@ func TestMain(m *testing.M) { } return nil } - purgeFn, err := test.CreateContainer(opts, retryFn) + purgeFn, err := utils.CreateContainer(opts, retryFn) if err != nil { log.Fatal(err) } @@ -103,7 +103,7 @@ func TestEmptyHosts(t *testing.T) { // TestInit tests the configs func TestInit(t *testing.T) { t.Run("should return error for invalid configs", func(t *testing.T) { - err := cassandra.New(test.Logger).Init(context.TODO(), map[string]interface{}{ + err := cassandra.New(utils.Logger).Init(context.TODO(), map[string]interface{}{ "password": pass, "host": host, }) @@ -116,7 +116,7 @@ func TestInit(t *testing.T) { func TestExtract(t *testing.T) { t.Run("should extract and output tables metadata along with its columns", func(t *testing.T) { ctx := context.TODO() - extr := cassandra.New(test.Logger) + extr := cassandra.New(utils.Logger) err := extr.Init(ctx, map[string]interface{}{ "user_id": user, @@ -175,7 +175,7 @@ func execute(queries []string) (err error) { // newExtractor returns a new extractor func newExtractor() *cassandra.Extractor { - return cassandra.New(test.Logger) + return cassandra.New(utils.Logger) } // getExpected returns the expected result diff --git a/plugins/extractors/clickhouse/clickhouse_test.go b/plugins/extractors/clickhouse/clickhouse_test.go index 901f1076d..42b697d49 100644 --- a/plugins/extractors/clickhouse/clickhouse_test.go +++ b/plugins/extractors/clickhouse/clickhouse_test.go @@ -5,6 +5,7 @@ package clickhouse_test import ( "context" "fmt" + "github.com/odpf/meteor/test/utils" "log" "os" "testing" @@ -18,7 +19,6 @@ import ( "github.com/odpf/meteor/models/odpf/assets/facets" "github.com/odpf/meteor/plugins" "github.com/odpf/meteor/plugins/extractors/clickhouse" - "github.com/odpf/meteor/test" "github.com/odpf/meteor/test/mocks" "github.com/ory/dockertest/v3" "github.com/ory/dockertest/v3/docker" @@ -65,7 +65,7 @@ func TestMain(m *testing.M) { } return db.Ping() } - purgeFn, err := test.CreateContainer(opts, retryFn) + purgeFn, err := utils.CreateContainer(opts, retryFn) if err != nil { log.Fatal(err) } @@ -206,5 +206,5 @@ func execute(db *sql.DB, queries []string) (err error) { } func newExtractor() *clickhouse.Extractor { - return clickhouse.New(test.Logger) + return clickhouse.New(utils.Logger) } diff --git a/plugins/extractors/couchdb/couchdb_test.go b/plugins/extractors/couchdb/couchdb_test.go index 887d589f5..dbc442f9e 100644 --- a/plugins/extractors/couchdb/couchdb_test.go +++ b/plugins/extractors/couchdb/couchdb_test.go @@ -5,6 +5,7 @@ package couchdb_test import ( "context" "fmt" + "github.com/odpf/meteor/test/utils" "log" "os" "strconv" @@ -14,7 +15,6 @@ import ( "github.com/go-kivik/kivik" "github.com/odpf/meteor/plugins" "github.com/odpf/meteor/plugins/extractors/couchdb" - "github.com/odpf/meteor/test" "github.com/odpf/meteor/test/mocks" "github.com/ory/dockertest/v3" "github.com/ory/dockertest/v3/docker" @@ -60,7 +60,7 @@ func TestMain(m *testing.M) { _, err = client.Ping(context.TODO()) return } - purgeFn, err := test.CreateContainer(opts, retryFn) + purgeFn, err := utils.CreateContainer(opts, retryFn) if err != nil { log.Fatal(err) } @@ -81,7 +81,7 @@ func TestMain(m *testing.M) { func TestInit(t *testing.T) { t.Run("should return error for invalid configs", func(t *testing.T) { - err := couchdb.New(test.Logger).Init(context.TODO(), map[string]interface{}{ + err := couchdb.New(utils.Logger).Init(context.TODO(), map[string]interface{}{ "password": "pass", "host": host, }) @@ -93,7 +93,7 @@ func TestInit(t *testing.T) { func TestExtract(t *testing.T) { t.Run("should extract and output tables metadata along with its columns", func(t *testing.T) { ctx := context.TODO() - extr := couchdb.New(test.Logger) + extr := couchdb.New(utils.Logger) err := extr.Init(ctx, map[string]interface{}{ "user_id": user, diff --git a/plugins/extractors/csv/csv_test.go b/plugins/extractors/csv/csv_test.go index 72fdd4f8c..007b6e844 100644 --- a/plugins/extractors/csv/csv_test.go +++ b/plugins/extractors/csv/csv_test.go @@ -4,6 +4,7 @@ package csv_test import ( "context" + "github.com/odpf/meteor/test/utils" "testing" "github.com/odpf/meteor/models" @@ -12,7 +13,6 @@ import ( "github.com/odpf/meteor/models/odpf/assets/facets" "github.com/odpf/meteor/plugins" "github.com/odpf/meteor/plugins/extractors/csv" - "github.com/odpf/meteor/test" "github.com/odpf/meteor/test/mocks" "github.com/stretchr/testify/assert" ) @@ -20,7 +20,7 @@ import ( func TestInit(t *testing.T) { t.Run("should return error if fileName and directory both are empty", func(t *testing.T) { config := map[string]interface{}{} - err := csv.New(test.Logger).Init( + err := csv.New(utils.Logger).Init( context.TODO(), config) assert.Equal(t, plugins.InvalidConfigError{}, err) @@ -30,7 +30,7 @@ func TestInit(t *testing.T) { func TestExtract(t *testing.T) { t.Run("should extract data if path is a file", func(t *testing.T) { ctx := context.TODO() - extr := csv.New(test.Logger) + extr := csv.New(utils.Logger) err := extr.Init(ctx, map[string]interface{}{ "path": "./testdata/test.csv", }) @@ -64,7 +64,7 @@ func TestExtract(t *testing.T) { t.Run("should extract data from all files if path is a dir", func(t *testing.T) { ctx := context.TODO() - extr := csv.New(test.Logger) + extr := csv.New(utils.Logger) err := extr.Init(ctx, map[string]interface{}{ "path": "./testdata", }) diff --git a/plugins/extractors/elastic/elastic_test.go b/plugins/extractors/elastic/elastic_test.go index a936ed6fe..b893b2117 100644 --- a/plugins/extractors/elastic/elastic_test.go +++ b/plugins/extractors/elastic/elastic_test.go @@ -6,6 +6,7 @@ import ( "context" "encoding/json" "fmt" + "github.com/odpf/meteor/test/utils" "log" "net/http" "os" @@ -20,7 +21,6 @@ import ( "github.com/odpf/meteor/models/odpf/assets/facets" "github.com/odpf/meteor/plugins" "github.com/odpf/meteor/plugins/extractors/elastic" - "github.com/odpf/meteor/test" "github.com/odpf/meteor/test/mocks" "github.com/ory/dockertest/v3" "github.com/ory/dockertest/v3/docker" @@ -83,7 +83,7 @@ func TestMain(m *testing.M) { } return } - purgeFn, err := test.CreateContainer(opts, retryFn) + purgeFn, err := utils.CreateContainer(opts, retryFn) if err != nil { log.Fatal(err) } @@ -184,7 +184,7 @@ func jsonStruct(doc MeteorMockElasticDocs) string { } func newExtractor() *elastic.Extractor { - return elastic.New(test.Logger) + return elastic.New(utils.Logger) } func getExpectedVal() []models.Record { diff --git a/plugins/extractors/gcs/gcs_test.go b/plugins/extractors/gcs/gcs_test.go index 3735ac36c..fcfbeb7ba 100644 --- a/plugins/extractors/gcs/gcs_test.go +++ b/plugins/extractors/gcs/gcs_test.go @@ -4,17 +4,17 @@ package gcs_test import ( "context" + "github.com/odpf/meteor/test/utils" "testing" "github.com/odpf/meteor/plugins" "github.com/odpf/meteor/plugins/extractors/gcs" - "github.com/odpf/meteor/test" "github.com/stretchr/testify/assert" ) func TestInit(t *testing.T) { t.Run("should return error if no project_id in config", func(t *testing.T) { - err := gcs.New(test.Logger).Init(context.TODO(), map[string]interface{}{ + err := gcs.New(utils.Logger).Init(context.TODO(), map[string]interface{}{ "wrong-config": "sample-project", }) diff --git a/plugins/extractors/grafana/grafana_test.go b/plugins/extractors/grafana/grafana_test.go index 4c0d52aa1..db4e0e194 100644 --- a/plugins/extractors/grafana/grafana_test.go +++ b/plugins/extractors/grafana/grafana_test.go @@ -5,6 +5,7 @@ package grafana_test import ( "context" "fmt" + "github.com/odpf/meteor/test/utils" "net/http" "net/http/httptest" "os" @@ -15,7 +16,6 @@ import ( "github.com/odpf/meteor/models/odpf/assets/common" "github.com/odpf/meteor/plugins" "github.com/odpf/meteor/plugins/extractors/grafana" - "github.com/odpf/meteor/test" "github.com/odpf/meteor/test/mocks" "github.com/stretchr/testify/assert" ) @@ -34,7 +34,7 @@ func TestMain(m *testing.M) { func TestInit(t *testing.T) { t.Run("should return error if for empty base_url in config", func(t *testing.T) { - err := grafana.New(test.Logger).Init(context.TODO(), map[string]interface{}{ + err := grafana.New(utils.Logger).Init(context.TODO(), map[string]interface{}{ "base_url": "", "api_key": "qwerty123", }) @@ -43,7 +43,7 @@ func TestInit(t *testing.T) { }) t.Run("should return error if for empty api_key in config", func(t *testing.T) { - err := grafana.New(test.Logger).Init(context.TODO(), map[string]interface{}{ + err := grafana.New(utils.Logger).Init(context.TODO(), map[string]interface{}{ "base_url": testServer.URL, "api_key": "", }) @@ -117,7 +117,7 @@ func TestExtract(t *testing.T) { } ctx := context.TODO() - extractor := grafana.New(test.Logger) + extractor := grafana.New(utils.Logger) err := extractor.Init(ctx, map[string]interface{}{ "base_url": testServer.URL, "api_key": "qwerty123", diff --git a/plugins/extractors/kafka/kafka_test.go b/plugins/extractors/kafka/kafka_test.go index d28d0f882..839c91719 100644 --- a/plugins/extractors/kafka/kafka_test.go +++ b/plugins/extractors/kafka/kafka_test.go @@ -4,6 +4,7 @@ package kafka_test import ( "context" + "github.com/odpf/meteor/test/utils" "log" "net" @@ -16,7 +17,6 @@ import ( "github.com/odpf/meteor/models/odpf/assets/common" "github.com/odpf/meteor/plugins" "github.com/odpf/meteor/plugins/extractors/kafka" - "github.com/odpf/meteor/test" "github.com/odpf/meteor/test/mocks" "github.com/ory/dockertest/v3" "github.com/ory/dockertest/v3/docker" @@ -60,7 +60,7 @@ func TestMain(m *testing.M) { return } - purgeContainer, err := test.CreateContainer(opts, retryFn) + purgeContainer, err := utils.CreateContainer(opts, retryFn) if err != nil { log.Fatal(err) } @@ -161,7 +161,7 @@ func setup(broker kafkaLib.Broker) (err error) { } func newExtractor() *kafka.Extractor { - return kafka.New(test.Logger) + return kafka.New(utils.Logger) } // This function compares two slices without concerning about the order diff --git a/plugins/extractors/metabase/metabase_test.go b/plugins/extractors/metabase/metabase_test.go index e01f3dc5b..2b1addf3f 100644 --- a/plugins/extractors/metabase/metabase_test.go +++ b/plugins/extractors/metabase/metabase_test.go @@ -7,6 +7,7 @@ import ( "context" "encoding/json" "fmt" + "github.com/odpf/meteor/test/utils" "io/ioutil" "log" "net/http" @@ -18,7 +19,6 @@ import ( "github.com/odpf/meteor/models/odpf/assets" "github.com/odpf/meteor/plugins" "github.com/odpf/meteor/plugins/extractors/metabase" - "github.com/odpf/meteor/test" "github.com/odpf/meteor/test/mocks" "github.com/ory/dockertest/v3" "github.com/ory/dockertest/v3/docker" @@ -82,7 +82,7 @@ func TestMain(m *testing.M) { } // Exponential backoff-retry for container to be resy to accept connections - purgeFn, err := test.CreateContainer(opts, retryFn) + purgeFn, err := utils.CreateContainer(opts, retryFn) if err != nil { log.Fatal(err) } @@ -102,7 +102,7 @@ func TestMain(m *testing.M) { func TestInit(t *testing.T) { t.Run("should return error for invalid config", func(t *testing.T) { - err := metabase.New(test.Logger).Init(context.TODO(), map[string]interface{}{ + err := metabase.New(utils.Logger).Init(context.TODO(), map[string]interface{}{ "user_id": "user", "host": host, }) @@ -114,7 +114,7 @@ func TestInit(t *testing.T) { func TestExtract(t *testing.T) { t.Run("should return dashboard model", func(t *testing.T) { ctx := context.TODO() - extr := metabase.New(test.Logger) + extr := metabase.New(utils.Logger) err := extr.Init(ctx, map[string]interface{}{ "user_id": email, "password": pass, diff --git a/plugins/extractors/mongodb/mongodb_test.go b/plugins/extractors/mongodb/mongodb_test.go index 4a89b8ece..329ddd296 100644 --- a/plugins/extractors/mongodb/mongodb_test.go +++ b/plugins/extractors/mongodb/mongodb_test.go @@ -5,6 +5,7 @@ package mongodb_test import ( "context" "fmt" + "github.com/odpf/meteor/test/utils" "log" "os" "testing" @@ -14,7 +15,6 @@ import ( "github.com/odpf/meteor/models/odpf/assets/common" "github.com/odpf/meteor/plugins" "github.com/odpf/meteor/plugins/extractors/mongodb" - "github.com/odpf/meteor/test" "github.com/odpf/meteor/test/mocks" "github.com/ory/dockertest/v3" "github.com/ory/dockertest/v3/docker" @@ -68,7 +68,7 @@ func TestMain(m *testing.M) { return client.Ping(ctx, nil) } - purgeFn, err := test.CreateContainer(opts, retryFn) + purgeFn, err := utils.CreateContainer(opts, retryFn) if err != nil { log.Fatal(err) } @@ -92,7 +92,7 @@ func TestMain(m *testing.M) { func TestInit(t *testing.T) { t.Run("should return error for invalid", func(t *testing.T) { - err := mongodb.New(test.Logger).Init(context.TODO(), map[string]interface{}{ + err := mongodb.New(utils.Logger).Init(context.TODO(), map[string]interface{}{ "password": pass, "host": host, }) @@ -104,7 +104,7 @@ func TestInit(t *testing.T) { func TestExtract(t *testing.T) { t.Run("should extract and output tables metadata along with its columns", func(t *testing.T) { ctx := context.TODO() - extr := mongodb.New(test.Logger) + extr := mongodb.New(utils.Logger) err := extr.Init(ctx, map[string]interface{}{ "user_id": user, diff --git a/plugins/extractors/mssql/mssql_test.go b/plugins/extractors/mssql/mssql_test.go index 26cf8535f..38b1fd2c5 100644 --- a/plugins/extractors/mssql/mssql_test.go +++ b/plugins/extractors/mssql/mssql_test.go @@ -6,6 +6,7 @@ import ( "context" "database/sql" "fmt" + "github.com/odpf/meteor/test/utils" "log" "os" "testing" @@ -17,7 +18,6 @@ import ( "github.com/odpf/meteor/models/odpf/assets/facets" "github.com/odpf/meteor/plugins" "github.com/odpf/meteor/plugins/extractors/mssql" - "github.com/odpf/meteor/test" "github.com/odpf/meteor/test/mocks" "github.com/ory/dockertest/v3" "github.com/ory/dockertest/v3/docker" @@ -58,7 +58,7 @@ func TestMain(m *testing.M) { } return db.Ping() } - purgeFn, err := test.CreateContainer(opts, retryFn) + purgeFn, err := utils.CreateContainer(opts, retryFn) if err != nil { log.Fatal(err) } @@ -79,7 +79,7 @@ func TestMain(m *testing.M) { func TestInit(t *testing.T) { t.Run("should error for invalid configurations", func(t *testing.T) { - err := mssql.New(test.Logger).Init(context.TODO(), map[string]interface{}{ + err := mssql.New(utils.Logger).Init(context.TODO(), map[string]interface{}{ "password": "pass", "host": host, }) @@ -91,7 +91,7 @@ func TestInit(t *testing.T) { func TestExtract(t *testing.T) { t.Run("should extract and output tables metadata along with its columns", func(t *testing.T) { ctx := context.TODO() - extr := mssql.New(test.Logger) + extr := mssql.New(utils.Logger) err := extr.Init(ctx, map[string]interface{}{ "user_id": user, diff --git a/plugins/extractors/mysql/mysql_test.go b/plugins/extractors/mysql/mysql_test.go index 849e92087..b4637f208 100644 --- a/plugins/extractors/mysql/mysql_test.go +++ b/plugins/extractors/mysql/mysql_test.go @@ -5,6 +5,7 @@ package mysql_test import ( "context" "fmt" + "github.com/odpf/meteor/test/utils" "log" "os" "testing" @@ -18,7 +19,6 @@ import ( "github.com/odpf/meteor/models/odpf/assets/facets" "github.com/odpf/meteor/plugins" "github.com/odpf/meteor/plugins/extractors/mysql" - "github.com/odpf/meteor/test" "github.com/odpf/meteor/test/mocks" "github.com/ory/dockertest/v3" "github.com/ory/dockertest/v3/docker" @@ -58,7 +58,7 @@ func TestMain(m *testing.M) { } return db.Ping() } - purgeFn, err := test.CreateContainer(opts, retryFn) + purgeFn, err := utils.CreateContainer(opts, retryFn) if err != nil { log.Fatal(err) } @@ -79,7 +79,7 @@ func TestMain(m *testing.M) { func TestInit(t *testing.T) { t.Run("should return error for invalid configs", func(t *testing.T) { - err := mysql.New(test.Logger).Init(context.TODO(), map[string]interface{}{ + err := mysql.New(utils.Logger).Init(context.TODO(), map[string]interface{}{ "password": "pass", "host": host, }) @@ -91,7 +91,7 @@ func TestInit(t *testing.T) { func TestExtract(t *testing.T) { t.Run("should extract and output tables metadata along with its columns", func(t *testing.T) { ctx := context.TODO() - extr := mysql.New(test.Logger) + extr := mysql.New(utils.Logger) err := extr.Init(ctx, map[string]interface{}{ "user_id": user, diff --git a/plugins/extractors/postgres/postgres_test.go b/plugins/extractors/postgres/postgres_test.go index 758c4e745..f59f70196 100644 --- a/plugins/extractors/postgres/postgres_test.go +++ b/plugins/extractors/postgres/postgres_test.go @@ -5,6 +5,7 @@ package postgres_test import ( "context" "fmt" + "github.com/odpf/meteor/test/utils" "log" "os" "testing" @@ -15,7 +16,6 @@ import ( "github.com/odpf/meteor/models/odpf/assets" "github.com/odpf/meteor/plugins" "github.com/odpf/meteor/plugins/extractors/postgres" - "github.com/odpf/meteor/test" "github.com/odpf/meteor/test/mocks" "github.com/ory/dockertest/v3" "github.com/ory/dockertest/v3/docker" @@ -52,7 +52,7 @@ func TestMain(m *testing.M) { } return db.Ping() } - purgeFn, err := test.CreateContainer(opts, retryFn) + purgeFn, err := utils.CreateContainer(opts, retryFn) if err != nil { log.Fatal(err) } @@ -73,7 +73,7 @@ func TestMain(m *testing.M) { func TestInit(t *testing.T) { t.Run("should return error for invalid config", func(t *testing.T) { - err := postgres.New(test.Logger).Init(context.TODO(), map[string]interface{}{ + err := postgres.New(utils.Logger).Init(context.TODO(), map[string]interface{}{ "password": "pass", "host": host, }) @@ -85,7 +85,7 @@ func TestInit(t *testing.T) { func TestExtract(t *testing.T) { t.Run("should return mockdata we generated with postgres", func(t *testing.T) { ctx := context.TODO() - extr := postgres.New(test.Logger) + extr := postgres.New(utils.Logger) err := extr.Init(ctx, map[string]interface{}{ "user_id": user, diff --git a/plugins/extractors/superset/superset_test.go b/plugins/extractors/superset/superset_test.go index 724b4e108..3bb6945e8 100644 --- a/plugins/extractors/superset/superset_test.go +++ b/plugins/extractors/superset/superset_test.go @@ -7,6 +7,7 @@ import ( "context" "encoding/json" "fmt" + "github.com/odpf/meteor/test/utils" "io" "io/ioutil" "log" @@ -18,7 +19,6 @@ import ( "github.com/odpf/meteor/models/odpf/assets" "github.com/odpf/meteor/plugins" "github.com/odpf/meteor/plugins/extractors/superset" - "github.com/odpf/meteor/test" "github.com/odpf/meteor/test/mocks" "github.com/ory/dockertest/v3" "github.com/ory/dockertest/v3/docker" @@ -83,7 +83,7 @@ func TestMain(m *testing.M) { } // exponential backoff-retry, because the application in the container might not be ready to accept connections yet - purgeFn, err := test.CreateContainer(opts, retryFn) + purgeFn, err := utils.CreateContainer(opts, retryFn) if err != nil { log.Fatal(err) } @@ -104,7 +104,7 @@ func TestMain(m *testing.M) { // TestInit tests the configs func TestInit(t *testing.T) { t.Run("should return error for invalid config", func(t *testing.T) { - err := superset.New(test.Logger).Init(context.TODO(), map[string]interface{}{ + err := superset.New(utils.Logger).Init(context.TODO(), map[string]interface{}{ "user_id": "user", "host": host, }) @@ -116,7 +116,7 @@ func TestInit(t *testing.T) { func TestExtract(t *testing.T) { t.Run("should return dashboard model", func(t *testing.T) { ctx := context.TODO() - extr := superset.New(test.Logger) + extr := superset.New(utils.Logger) err := extr.Init(ctx, map[string]interface{}{ "username": user, "password": pass, diff --git a/plugins/plugin.go b/plugins/plugin.go index 481e27e57..4f60bca04 100644 --- a/plugins/plugin.go +++ b/plugins/plugin.go @@ -55,6 +55,9 @@ type Processor interface { type Syncer interface { Plugin Sink(ctx context.Context, batch []models.Record) (err error) + + // Close will be called once after everything is done + Close() error } // ParseInfo parses the plugin's meta.yaml file and returns an plugin Info struct. diff --git a/plugins/sinks/columbus/sink.go b/plugins/sinks/columbus/sink.go index 08a170799..46bb1d5c9 100644 --- a/plugins/sinks/columbus/sink.go +++ b/plugins/sinks/columbus/sink.go @@ -92,6 +92,8 @@ func (s *Sink) Sink(ctx context.Context, batch []models.Record) (err error) { return } +func (s *Sink) Close() (err error) { return } + func (s *Sink) buildColumbusPayload(metadata models.Metadata) (interface{}, error) { // skip if mapping is not defined if s.config.Mapping == nil { diff --git a/plugins/sinks/columbus/sink_test.go b/plugins/sinks/columbus/sink_test.go index 13fde8616..8ee96b091 100644 --- a/plugins/sinks/columbus/sink_test.go +++ b/plugins/sinks/columbus/sink_test.go @@ -5,6 +5,7 @@ import ( "context" "errors" "fmt" + "github.com/odpf/meteor/test/utils" "io/ioutil" "net/http" "testing" @@ -15,7 +16,6 @@ import ( "github.com/odpf/meteor/models/odpf/assets/facets" "github.com/odpf/meteor/plugins" "github.com/odpf/meteor/plugins/sinks/columbus" - "github.com/odpf/meteor/test" "github.com/stretchr/testify/assert" ) @@ -55,7 +55,7 @@ func TestInit(t *testing.T) { } for i, config := range invalidConfigs { t.Run(fmt.Sprintf("test invalid config #%d", i+1), func(t *testing.T) { - columbusSink := columbus.New(newmockHTTPClient(http.MethodGet, url, requestPayload), test.Logger) + columbusSink := columbus.New(newmockHTTPClient(http.MethodGet, url, requestPayload), utils.Logger) err := columbusSink.Init(context.TODO(), config) assert.Equal(t, plugins.InvalidConfigError{Type: plugins.PluginTypeSink}, err) @@ -70,7 +70,7 @@ func TestSink(t *testing.T) { client.SetupResponse(200, "") ctx := context.TODO() - columbusSink := columbus.New(client, test.Logger) + columbusSink := columbus.New(client, utils.Logger) err := columbusSink.Init(ctx, map[string]interface{}{ "host": host, "type": columbusType, @@ -95,7 +95,7 @@ func TestSink(t *testing.T) { client.SetupResponse(404, columbusError) ctx := context.TODO() - columbusSink := columbus.New(client, test.Logger) + columbusSink := columbus.New(client, utils.Logger) err := columbusSink.Init(ctx, map[string]interface{}{ "host": host, "type": "my-type", @@ -118,7 +118,7 @@ func TestSink(t *testing.T) { client.SetupResponse(code, `{"reason":"internal server error"}`) ctx := context.TODO() - columbusSink := columbus.New(client, test.Logger) + columbusSink := columbus.New(client, utils.Logger) err := columbusSink.Init(ctx, map[string]interface{}{ "host": host, "type": "my-type", @@ -141,7 +141,7 @@ func TestSink(t *testing.T) { client.SetupResponse(200, `{"success": true}`) ctx := context.TODO() - columbusSink := columbus.New(client, test.Logger) + columbusSink := columbus.New(client, utils.Logger) err := columbusSink.Init(ctx, map[string]interface{}{ "host": host, "type": "my-type", @@ -174,7 +174,7 @@ func TestSink(t *testing.T) { client.SetupResponse(200, "") ctx := context.TODO() - columbusSink := columbus.New(client, test.Logger) + columbusSink := columbus.New(client, utils.Logger) err := columbusSink.Init(ctx, map[string]interface{}{ "host": host, "type": columbusType, diff --git a/plugins/sinks/console/sink.go b/plugins/sinks/console/sink.go index 5f5ed64fc..a757d6527 100644 --- a/plugins/sinks/console/sink.go +++ b/plugins/sinks/console/sink.go @@ -49,6 +49,8 @@ func (s *Sink) Sink(ctx context.Context, batch []models.Record) (err error) { return nil } +func (s *Sink) Close() (err error) { return } + func (s *Sink) process(value interface{}) error { jsonBytes, err := json.Marshal(value) if err != nil { diff --git a/plugins/sinks/kafka/sink.go b/plugins/sinks/kafka/sink.go index 9a227e118..b2bba9edd 100644 --- a/plugins/sinks/kafka/sink.go +++ b/plugins/sinks/kafka/sink.go @@ -72,8 +72,6 @@ func (s *Sink) Init(ctx context.Context, configMap map[string]interface{}) (err } func (s *Sink) Sink(ctx context.Context, batch []models.Record) (err error) { - defer s.writer.Close() - for _, record := range batch { if err := s.push(ctx, record.Data()); err != nil { return err @@ -83,10 +81,11 @@ func (s *Sink) Sink(ctx context.Context, batch []models.Record) (err error) { return } -func (s *Sink) push(ctx context.Context, payload interface{}) error { - // struct needs to be cast to pointer to implement proto methods - payload = castModelToPointer(payload) +func (s *Sink) Close() (err error) { + return s.writer.Close() +} +func (s *Sink) push(ctx context.Context, payload interface{}) error { kafkaValue, err := s.buildValue(payload) if err != nil { return err @@ -188,13 +187,6 @@ func (s *Sink) getTopLevelKeyFromPath(keyPath string) (string, error) { return keyPaths[1], nil } -func castModelToPointer(value interface{}) interface{} { - vp := reflect.New(reflect.TypeOf(value)) - vp.Elem().Set(reflect.ValueOf(value)) - - return vp.Interface() -} - func createWriter(config Config) *kafka.Writer { brokers := strings.Split(config.Brokers, ",") return &kafka.Writer{ diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go new file mode 100644 index 000000000..166ece476 --- /dev/null +++ b/test/e2e/e2e_test.go @@ -0,0 +1,357 @@ +//+build integration + +package e2e_test + +import ( + "context" + "database/sql" + "fmt" + "github.com/odpf/meteor/test/utils" + "github.com/ory/dockertest/v3" + "github.com/ory/dockertest/v3/docker" + "log" + "net" + "os" + "strconv" + "strings" + "testing" + "time" + + "github.com/odpf/meteor/cmd" + "github.com/odpf/meteor/config" + "github.com/odpf/meteor/models/odpf/assets" + "github.com/odpf/meteor/models/odpf/assets/common" + "github.com/odpf/meteor/models/odpf/assets/facets" + _ "github.com/odpf/meteor/plugins/extractors" + _ "github.com/odpf/meteor/plugins/processors" + _ "github.com/odpf/meteor/plugins/sinks" + "github.com/pkg/errors" + "github.com/segmentio/kafka-go" + "github.com/stretchr/testify/assert" + "google.golang.org/protobuf/proto" +) + +var ( + db *sql.DB + conn *kafka.Conn + broker kafka.Broker +) + +const ( + testDB = "test_db" + user = "test_user" + pass = "admin" + mysqlHost = "localhost:3306" + brokerHost = "localhost:9093" + testTopic = "topic-a" + partition = 0 +) + +func TestMain(m *testing.M) { + // generate purge function + mysqlPurgeContainer, err := mysqlDockerSetup() + if err != nil { + return + } + kafkaPurgeContainer, err := kafkaDockerSetup() + if err != nil { + return + } + + // setup and populate data for testing + if err := setupMySQL(); err != nil { + log.Fatal(err) + } + if err := setupKafka(); err != nil { + log.Fatal(err) + } + + // run tests + code := m.Run() + + // clean tests + if err = conn.Close(); err != nil { + return + } + if err = db.Close(); err != nil { + return + } + + // purge containers + if err := mysqlPurgeContainer(); err != nil { + log.Fatal(err) + } + if err := kafkaPurgeContainer(); err != nil { + log.Fatal(err) + } + + os.Exit(code) +} + +// TestMySqlToKafka tests the recipe from source to sink completely +func TestMySqlToKafka(t *testing.T) { + err := setupKafka() + if err != nil { + t.Fatal(err) + } + + var sinkData []*assets.Table + ctx, cancel := context.WithCancel(context.TODO()) + go func() { + err = listenToTopic(ctx, testTopic, &sinkData) + if err != nil { + t.Error(err) + } + }() + + // run mysql_kafka.yml file + cfg, err := config.Load() + if err != nil { + t.Error(err) + } + command := cmd.New(utils.Logger, nil, cfg) + command.SetArgs([]string{"run", "mysql_kafka.yml"}) + if err := command.Execute(); err != nil { + if strings.HasPrefix(err.Error(), "unknown command ") { + if !strings.HasSuffix(err.Error(), "\n") { + t.Fatal(err) + } + t.Fatal(err) + } else { + t.Fatal(err) + } + } + + time.Sleep(2 * time.Second) // this is to wait consumer to finish adding data to sinkData + cancel() // cancel will cancel context, hinting the consumer to end + time.Sleep(100 * time.Millisecond) // this is to give time for the consumer to closing all its connections + + expected := getExpectedTables() + assert.Equal(t, len(getExpectedTables()), len(sinkData)) + for tableNum := 0; tableNum < len(getExpectedTables()); tableNum++ { + assert.Equal(t, expected[tableNum].Resource.Urn, sinkData[tableNum].Resource.Urn) + assert.Equal(t, expected[tableNum].Resource.Name, sinkData[tableNum].Resource.Name) + assert.Equal(t, len(expected[tableNum].Schema.Columns), len(sinkData[tableNum].Schema.Columns)) + } +} + +// listenToTopic listens to a topic and stores the data in sinkData +func listenToTopic(ctx context.Context, topic string, data *[]*assets.Table) error { + reader := kafka.NewReader(kafka.ReaderConfig{ + Brokers: []string{brokerHost}, + Topic: topic, + }) + defer func(reader *kafka.Reader) { + if err := reader.Close(); err != nil { + return + } + }(reader) + + for { + msg, err := reader.ReadMessage(ctx) + if err != nil { + break + + } + var convertMsg assets.Table + if err := proto.Unmarshal(msg.Value, &convertMsg); err != nil { + return errors.Wrap(err, "failed to parse kafka message") + } + *data = append(*data, &convertMsg) + } + + return nil +} + +// setupKafka initializes kafka broker with topic and partition +func setupKafka() error { + conn, err := kafka.DialLeader(context.TODO(), "tcp", net.JoinHostPort(broker.Host, strconv.Itoa(broker.Port)), testTopic, partition) + if err != nil { + return errors.Wrap(err, "failed to setup kafka connection") + } + defer func(conn *kafka.Conn) { + if err := conn.Close(); err != nil { + return + } + }(conn) + + if err := conn.DeleteTopics(testTopic); err != nil { + return errors.Wrap(err, "failed to delete topic") + } + if err := conn.CreateTopics(kafka.TopicConfig{ + Topic: testTopic, + NumPartitions: 1, + ReplicationFactor: 1, + }); err != nil { + return errors.Wrap(err, "failed to create topic") + } + + return nil +} + +// setupMySQL initializes mysql database +func setupMySQL() (err error) { + // create database, user and grant access + if err = execute(db, []string{ + fmt.Sprintf("DROP DATABASE IF EXISTS %s", testDB), + fmt.Sprintf("CREATE DATABASE %s", testDB), + fmt.Sprintf("USE %s;", testDB), + fmt.Sprintf(`CREATE USER IF NOT EXISTS '%s'@'%%' IDENTIFIED BY '%s';`, user, pass), + fmt.Sprintf(`GRANT ALL PRIVILEGES ON *.* TO '%s'@'%%';`, user), + }); err != nil { + return errors.Wrap(err, "failed to create database") + } + + // create and populate tables + if err = execute(db, []string{ + "CREATE TABLE applicant (applicant_id int, last_name varchar(255), first_name varchar(255));", + "INSERT INTO applicant VALUES (1, 'test1', 'test11');", + "CREATE TABLE jobs (job_id int, job varchar(255), department varchar(255));", + "INSERT INTO jobs VALUES (2, 'test2', 'test22');", + }); err != nil { + return errors.Wrap(err, "failed to populate database") + } + + return +} + +// execute executes a list of sql statements +func execute(db *sql.DB, queries []string) (err error) { + for _, query := range queries { + _, err = db.Exec(query) + if err != nil { + return + } + } + + return +} + +// kafkaDockerSetup sets up a kafka docker container +func kafkaDockerSetup() (purge func() error, err error) { + // kafka setup test + kafkaOpts := dockertest.RunOptions{ + Repository: "moeenz/docker-kafka-kraft", + Tag: "latest", + Env: []string{ + "KRAFT_CONTAINER_HOST_NAME=kafka", + }, + ExposedPorts: []string{"9093"}, + PortBindings: map[docker.Port][]docker.PortBinding{ + "9093": { + {HostIP: "localhost", HostPort: "9093"}, + }, + }, + } + // exponential backoff-retry, because the application in the container might not be ready to accept connections yet + kafkaRetryFn := func(resource *dockertest.Resource) (err error) { + // create client + if conn, err = kafka.Dial("tcp", brokerHost); err != nil { + return errors.Wrap(err, "failed to kafka create client") + } + if broker, err = conn.Controller(); err != nil { + return errors.Wrap(err, "failed to generate broker request") + } + return + } + purgeContainer, err := utils.CreateContainer(kafkaOpts, kafkaRetryFn) + if err != nil { + log.Fatal(err) + } + + return purgeContainer, nil +} + +// mysqlDockerSetup sets up a mysql docker container +func mysqlDockerSetup() (purge func() error, err error) { + // mysql setup test + mysqlOpts := dockertest.RunOptions{ + Repository: "mysql", + Tag: "latest", + Env: []string{ + "MYSQL_ROOT_PASSWORD=" + pass, + }, + ExposedPorts: []string{"3306"}, + PortBindings: map[docker.Port][]docker.PortBinding{ + "3306": { + {HostIP: "0.0.0.0", HostPort: "3306"}, + }, + }, + } + // exponential backoff-retry, because the application in the container might not be ready to accept connections yet + mysqlRetryFn := func(resource *dockertest.Resource) (err error) { + db, err = sql.Open("mysql", fmt.Sprintf("root:%s@tcp(%s)/", pass, mysqlHost)) + if err != nil { + return errors.Wrap(err, "failed to create mysql client") + } + return db.Ping() + } + purgeContainer, err := utils.CreateContainer(mysqlOpts, mysqlRetryFn) + if err != nil { + log.Fatal(err) + } + + return purgeContainer, nil +} + +// getExpectedTables returns the expected tables +func getExpectedTables() []*assets.Table { + return []*assets.Table{ + { + Resource: &common.Resource{ + Urn: testDB + ".applicant", + Name: "applicant", + }, + Schema: &facets.Columns{ + Columns: []*facets.Column{ + { + Name: "applicant_id", + DataType: "int", + IsNullable: true, + Length: 0, + }, + { + Name: "first_name", + DataType: "varchar", + IsNullable: true, + Length: 255, + }, + { + Name: "last_name", + DataType: "varchar", + IsNullable: true, + Length: 255, + }, + }, + }, + }, + { + Resource: &common.Resource{ + Urn: testDB + ".jobs", + Name: "jobs", + }, + Schema: &facets.Columns{ + Columns: []*facets.Column{ + { + Name: "department", + DataType: "varchar", + IsNullable: true, + Length: 255, + }, + { + Name: "job", + DataType: "varchar", + IsNullable: true, + Length: 255, + }, + { + Name: "job_id", + DataType: "int", + IsNullable: true, + Length: 0, + }, + }, + }, + }, + } +} diff --git a/test/e2e/mysql_kafka.yml b/test/e2e/mysql_kafka.yml new file mode 100644 index 000000000..26d7dff6a --- /dev/null +++ b/test/e2e/mysql_kafka.yml @@ -0,0 +1,15 @@ +name: sample +source: + type: mysql + config: + user_id: root + password: admin + host: 127.0.0.1:3306 +sinks: + - name: kafka + config: + brokers: "localhost:9093" + topic: "topic-a" +processors: + - name: enrich + config: diff --git a/test/mocks/plugin.go b/test/mocks/plugin.go index b93d3cba4..6e12f7f9e 100644 --- a/test/mocks/plugin.go +++ b/test/mocks/plugin.go @@ -76,6 +76,11 @@ func (m *Sink) Sink(ctx context.Context, batch []models.Record) error { return args.Error(0) } +func (m *Sink) Close() error { + args := m.Called() + return args.Error(0) +} + type Emitter struct { data []models.Record } diff --git a/test/dockertest.go b/test/utils/dockertest.go similarity index 98% rename from test/dockertest.go rename to test/utils/dockertest.go index e7311a566..4f2ad1051 100644 --- a/test/dockertest.go +++ b/test/utils/dockertest.go @@ -1,4 +1,4 @@ -package test +package utils import ( "fmt" diff --git a/test/logger.go b/test/utils/logger.go similarity index 91% rename from test/logger.go rename to test/utils/logger.go index f3d7c7189..602b3896e 100644 --- a/test/logger.go +++ b/test/utils/logger.go @@ -1,4 +1,4 @@ -package test +package utils import ( "io/ioutil"