Skip to content

Commit f0a7f88

Browse files
scortierscortier
committed
refactor: e2e test
1 parent 36d2721 commit f0a7f88

3 files changed

Lines changed: 80 additions & 88 deletions

File tree

‎agent/agent_test.go‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,19 +3,18 @@ package agent_test
33
import (
44
"context"
55
"errors"
6-
"github.com/odpf/meteor/test/utils"
7-
"testing"
8-
"time"
9-
106
"github.com/odpf/meteor/agent"
117
"github.com/odpf/meteor/models"
128
"github.com/odpf/meteor/models/odpf/assets"
139
"github.com/odpf/meteor/plugins"
1410
"github.com/odpf/meteor/recipe"
1511
"github.com/odpf/meteor/registry"
1612
"github.com/odpf/meteor/test/mocks"
13+
"github.com/odpf/meteor/test/utils"
1714
"github.com/stretchr/testify/assert"
1815
"github.com/stretchr/testify/mock"
16+
"testing"
17+
"time"
1918
)
2019

2120
var mockCtx = mock.AnythingOfType("*context.emptyCtx")
@@ -208,7 +207,6 @@ func TestRunnerRun(t *testing.T) {
208207

209208
sink := mocks.NewSink()
210209
sink.On("Init", mockCtx, validRecipe.Sinks[0].Config).Return(errors.New("some error")).Once()
211-
sink.On("Close").Return(nil)
212210
defer sink.AssertExpectations(t)
213211
sf := registry.NewSinkFactory()
214212
if err := sf.Register("test-sink", newSink(sink)); err != nil {
@@ -536,6 +534,7 @@ func TestRunnerRun(t *testing.T) {
536534
sink := mocks.NewSink()
537535
sink.On("Init", mockCtx, validRecipe.Sinks[0].Config).Return(nil).Once()
538536
sink.On("Sink", mockCtx, data).Return(nil)
537+
sink.On("Close").Return(nil)
539538
defer sink.AssertExpectations(t)
540539
sf := registry.NewSinkFactory()
541540
if err := sf.Register("test-sink", newSink(sink)); err != nil {
@@ -587,6 +586,7 @@ func TestRunnerRun(t *testing.T) {
587586
sink.On("Init", mockCtx, validRecipe.Sinks[0].Config).Return(nil).Once()
588587
sink.On("Sink", mockCtx, data).Return(plugins.NewRetryError(err)).Once()
589588
sink.On("Sink", mockCtx, data).Return(nil)
589+
sink.On("Close").Return(nil)
590590
defer sink.AssertExpectations(t)
591591
sf := registry.NewSinkFactory()
592592
if err := sf.Register("test-sink", newSink(sink)); err != nil {
@@ -636,6 +636,7 @@ func TestRunnerRunMultiple(t *testing.T) {
636636
sink := mocks.NewSink()
637637
sink.On("Init", mockCtx, validRecipe.Sinks[0].Config).Return(nil)
638638
sink.On("Sink", mockCtx, data).Return(nil)
639+
sink.On("Close").Return(nil)
639640
defer sink.AssertExpectations(t)
640641
sf := registry.NewSinkFactory()
641642
if err := sf.Register("test-sink", newSink(sink)); err != nil {

‎plugins/sinks/kafka/sink.go‎

Lines changed: 0 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -86,9 +86,6 @@ func (s *Sink) Close() (err error) {
8686
}
8787

8888
func (s *Sink) push(ctx context.Context, payload interface{}) error {
89-
// struct needs to be cast to pointer to implement proto methods
90-
payload = castModelToPointer(payload)
91-
9289
kafkaValue, err := s.buildValue(payload)
9390
if err != nil {
9491
return err
@@ -190,13 +187,6 @@ func (s *Sink) getTopLevelKeyFromPath(keyPath string) (string, error) {
190187
return keyPaths[1], nil
191188
}
192189

193-
func castModelToPointer(value interface{}) interface{} {
194-
vp := reflect.New(reflect.TypeOf(value))
195-
vp.Elem().Set(reflect.ValueOf(value))
196-
197-
return vp.Interface()
198-
}
199-
200190
func createWriter(config Config) *kafka.Writer {
201191
brokers := strings.Split(config.Brokers, ",")
202192
return &kafka.Writer{

‎test/e2e/e2e_test.go‎

Lines changed: 74 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// +build integration
1+
//+build integration
22

33
package e2e_test
44

@@ -7,6 +7,8 @@ import (
77
"database/sql"
88
"fmt"
99
"github.com/odpf/meteor/test/utils"
10+
"github.com/ory/dockertest/v3"
11+
"github.com/ory/dockertest/v3/docker"
1012
"log"
1113
"net"
1214
"os"
@@ -23,8 +25,6 @@ import (
2325
_ "github.com/odpf/meteor/plugins/extractors"
2426
_ "github.com/odpf/meteor/plugins/processors"
2527
_ "github.com/odpf/meteor/plugins/sinks"
26-
"github.com/ory/dockertest/v3"
27-
"github.com/ory/dockertest/v3/docker"
2828
"github.com/pkg/errors"
2929
"github.com/segmentio/kafka-go"
3030
"github.com/stretchr/testify/assert"
@@ -88,75 +88,8 @@ func TestMain(m *testing.M) {
8888
os.Exit(code)
8989
}
9090

91-
// kafkaDockerSetup sets up a kafka docker container
92-
func kafkaDockerSetup() (purge func() error, err error) {
93-
// kafka setup test
94-
kafkaOpts := dockertest.RunOptions{
95-
Repository: "moeenz/docker-kafka-kraft",
96-
Tag: "latest",
97-
Env: []string{
98-
"KRAFT_CONTAINER_HOST_NAME=kafka",
99-
},
100-
ExposedPorts: []string{"9093"},
101-
PortBindings: map[docker.Port][]docker.PortBinding{
102-
"9093": {
103-
{HostIP: "localhost", HostPort: "9093"},
104-
},
105-
},
106-
}
107-
// exponential backoff-retry, because the application in the container might not be ready to accept connections yet
108-
kafkaRetryFn := func(resource *dockertest.Resource) (err error) {
109-
// create client
110-
if conn, err = kafka.Dial("tcp", brokerHost); err != nil {
111-
return errors.Wrap(err, "failed to kafka create client")
112-
}
113-
if broker, err = conn.Controller(); err != nil {
114-
return errors.Wrap(err, "failed to generate broker request")
115-
}
116-
return
117-
}
118-
purgeContainer, err := utils.CreateContainer(kafkaOpts, kafkaRetryFn)
119-
if err != nil {
120-
log.Fatal(err)
121-
}
122-
123-
return purgeContainer, nil
124-
}
125-
126-
// mysqlDockerSetup sets up a mysql docker container
127-
func mysqlDockerSetup() (purge func() error, err error) {
128-
// mysql setup test
129-
mysqlOpts := dockertest.RunOptions{
130-
Repository: "mysql",
131-
Tag: "latest",
132-
Env: []string{
133-
"MYSQL_ROOT_PASSWORD=" + pass,
134-
},
135-
ExposedPorts: []string{"3306"},
136-
PortBindings: map[docker.Port][]docker.PortBinding{
137-
"3306": {
138-
{HostIP: "0.0.0.0", HostPort: "3306"},
139-
},
140-
},
141-
}
142-
// exponential backoff-retry, because the application in the container might not be ready to accept connections yet
143-
mysqlRetryFn := func(resource *dockertest.Resource) (err error) {
144-
db, err = sql.Open("mysql", fmt.Sprintf("root:%s@tcp(%s)/", pass, mysqlHost))
145-
if err != nil {
146-
return errors.Wrap(err, "failed to create mysql client")
147-
}
148-
return db.Ping()
149-
}
150-
purgeContainer, err := utils.CreateContainer(mysqlOpts, mysqlRetryFn)
151-
if err != nil {
152-
log.Fatal(err)
153-
}
154-
155-
return purgeContainer, nil
156-
}
157-
158-
// TestRecipe tests the recipe from source to sink completely
159-
func TestRecipe(t *testing.T) {
91+
// TestMySqlToKafka tests the recipe from source to sink completely
92+
func TestMySqlToKafka(t *testing.T) {
16093
err := setupKafka()
16194
if err != nil {
16295
t.Fatal(err)
@@ -218,6 +151,7 @@ func listenToTopic(ctx context.Context, topic string, data *[]*assets.Table) err
218151
msg, err := reader.ReadMessage(ctx)
219152
if err != nil {
220153
break
154+
221155
}
222156
var convertMsg assets.Table
223157
if err := proto.Unmarshal(msg.Value, &convertMsg); err != nil {
@@ -229,7 +163,7 @@ func listenToTopic(ctx context.Context, topic string, data *[]*assets.Table) err
229163
return nil
230164
}
231165

232-
// setupKafka intializes kafka broker with topic and partition
166+
// setupKafka initializes kafka broker with topic and partition
233167
func setupKafka() error {
234168
conn, err := kafka.DialLeader(context.TODO(), "tcp", net.JoinHostPort(broker.Host, strconv.Itoa(broker.Port)), testTopic, partition)
235169
if err != nil {
@@ -293,6 +227,73 @@ func execute(db *sql.DB, queries []string) (err error) {
293227
return
294228
}
295229

230+
// kafkaDockerSetup sets up a kafka docker container
231+
func kafkaDockerSetup() (purge func() error, err error) {
232+
// kafka setup test
233+
kafkaOpts := dockertest.RunOptions{
234+
Repository: "moeenz/docker-kafka-kraft",
235+
Tag: "latest",
236+
Env: []string{
237+
"KRAFT_CONTAINER_HOST_NAME=kafka",
238+
},
239+
ExposedPorts: []string{"9093"},
240+
PortBindings: map[docker.Port][]docker.PortBinding{
241+
"9093": {
242+
{HostIP: "localhost", HostPort: "9093"},
243+
},
244+
},
245+
}
246+
// exponential backoff-retry, because the application in the container might not be ready to accept connections yet
247+
kafkaRetryFn := func(resource *dockertest.Resource) (err error) {
248+
// create client
249+
if conn, err = kafka.Dial("tcp", brokerHost); err != nil {
250+
return errors.Wrap(err, "failed to kafka create client")
251+
}
252+
if broker, err = conn.Controller(); err != nil {
253+
return errors.Wrap(err, "failed to generate broker request")
254+
}
255+
return
256+
}
257+
purgeContainer, err := utils.CreateContainer(kafkaOpts, kafkaRetryFn)
258+
if err != nil {
259+
log.Fatal(err)
260+
}
261+
262+
return purgeContainer, nil
263+
}
264+
265+
// mysqlDockerSetup sets up a mysql docker container
266+
func mysqlDockerSetup() (purge func() error, err error) {
267+
// mysql setup test
268+
mysqlOpts := dockertest.RunOptions{
269+
Repository: "mysql",
270+
Tag: "latest",
271+
Env: []string{
272+
"MYSQL_ROOT_PASSWORD=" + pass,
273+
},
274+
ExposedPorts: []string{"3306"},
275+
PortBindings: map[docker.Port][]docker.PortBinding{
276+
"3306": {
277+
{HostIP: "0.0.0.0", HostPort: "3306"},
278+
},
279+
},
280+
}
281+
// exponential backoff-retry, because the application in the container might not be ready to accept connections yet
282+
mysqlRetryFn := func(resource *dockertest.Resource) (err error) {
283+
db, err = sql.Open("mysql", fmt.Sprintf("root:%s@tcp(%s)/", pass, mysqlHost))
284+
if err != nil {
285+
return errors.Wrap(err, "failed to create mysql client")
286+
}
287+
return db.Ping()
288+
}
289+
purgeContainer, err := utils.CreateContainer(mysqlOpts, mysqlRetryFn)
290+
if err != nil {
291+
log.Fatal(err)
292+
}
293+
294+
return purgeContainer, nil
295+
}
296+
296297
// getExpectedTables returns the expected tables
297298
func getExpectedTables() []*assets.Table {
298299
return []*assets.Table{

0 commit comments

Comments
 (0)