source_test.go

// Copyright 2024 EMQ Technologies Co., Ltd.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package sql

import (
	"errors"
	"fmt"
	"testing"
	"time"

	"github.com/lf-edge/ekuiper/contract/v2/api"
	"github.com/pingcap/failpoint"
	"github.com/stretchr/testify/require"

	"github.com/lf-edge/ekuiper/v2/extensions/impl/sql/client"
	"github.com/lf-edge/ekuiper/v2/extensions/impl/sql/testx"
	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
	"github.com/lf-edge/ekuiper/v2/pkg/connection"
	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
	"github.com/lf-edge/ekuiper/v2/pkg/modules"
	"github.com/lf-edge/ekuiper/v2/pkg/store"
)

func init() {
	modules.RegisterConnection("sql", client.CreateConnection)
}

var (
	address = "localhost"
	port    = 33061
)

func TestProvisionErr(t *testing.T) {
	ctx := context.Background()
	testcases := []struct {
		props map[string]interface{}
		err   error
	}{
		{
			props: map[string]interface{}{},
			err:   errors.New("interval should be defined"),
		},
		{
			props: map[string]interface{}{
				"interval": "0s",
			},
			err: errors.New("interval should be defined"),
		},
		{
			props: map[string]interface{}{
				"interval": "1s",
			},
			err: errors.New("dburl should be defined"),
		},
		{
			props: map[string]interface{}{
				"interval": "1s",
				"dburl":    "123",
			},
			err: errors.New("dburl.Parse 123 fail with error: parse driver err:invalid database scheme, support drivers:[mysql]"),
		},
	}
	for _, tc := range testcases {
		err := GetSource().Provision(ctx, tc.props)
		require.Equal(t, tc.err, err)
	}
}

func TestProvisionMockErr(t *testing.T) {
	ctx := context.Background()
	props := map[string]interface{}{
		"interval": "1s",
		"dburl":    "mysql:https://root:@127.0.0.1:4000/test",
	}
	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/io/sql/MapToStructErr", "return(true)")
	err := GetSource().Provision(ctx, props)
	require.Error(t, err)
	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/io/sql/MapToStructErr")

	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/io/sql/GetQueryGeneratorErr", "return(true)")
	err = GetSource().Provision(ctx, props)
	require.Error(t, err)
	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/io/sql/GetQueryGeneratorErr")
}

func TestSQLConnectionConnect(t *testing.T) {
	connection.InitConnectionManager4Test()
	ctx := mockContext.NewMockContext("1", "2")
	s, err := testx.SetupEmbeddedMysqlServer(address, port)
	require.NoError(t, err)
	defer func() {
		s.Close()
	}()
	props := map[string]interface{}{
		"interval": "1s",
		"dburl":    fmt.Sprintf("mysql:https://root:@%v:%v/test", address, port),
		"templateSqlQueryCfg": map[string]interface{}{
			"templateSql": "select a,b from t",
		},
	}
	sqlSource := GetSource()
	require.NoError(t, sqlSource.Provision(ctx, props))
	require.NoError(t, sqlSource.Connect(ctx))
	sqlConnector, ok := sqlSource.(*SQLSourceConnector)
	require.True(t, ok)
	expectedData := map[string]any{
		"a": int64(1),
		"b": int64(1),
	}
	sqlConnector.queryData(ctx, time.Now(), func(ctx api.StreamContext, data any, meta map[string]any, ts time.Time) {
		require.Equal(t, expectedData, data)
	}, func(ctx api.StreamContext, err error) {
		require.NoError(t, err)
	})

	// query data Error
	testcases := []struct {
		path string
	}{
		{
			path: "StatementErr",
		},
		{
			path: "QueryErr",
		},
		{
			path: "ColumnTypesErr",
		},
		{
			path: "ScanErr",
		},
	}
	for _, tc := range testcases {
		fp := "github.com/lf-edge/ekuiper/v2/internal/io/sql/" + tc.path
		failpoint.Enable(fp, "return(true)")
		sqlConnector.queryData(ctx, time.Now(), func(ctx api.StreamContext, data any, meta map[string]any, ts time.Time) {}, func(ctx api.StreamContext, err error) {
			require.Error(t, err)
		})
		failpoint.Disable(fp)
	}
	sqlSource.Close(ctx)
}

func TestSQLConnectionErr(t *testing.T) {
	connection.InitConnectionManager4Test()
	ctx := mockContext.NewMockContext("1", "2")
	props := map[string]interface{}{
		"interval": "1s",
		"dburl":    "mysql:https://mock:@mock:123/mock",
		"templateSqlQueryCfg": map[string]interface{}{
			"templateSql": "select a,b from t",
		},
	}
	sqlSource := GetSource()
	require.NoError(t, sqlSource.Provision(ctx, props))
	require.NoError(t, sqlSource.Connect(ctx))
}

func TestSQLSourceRewind(t *testing.T) {
	connection.InitConnectionManager4Test()
	ctx := mockContext.NewMockContext("1", "2")
	s, err := testx.SetupEmbeddedMysqlServer(address, port)
	require.NoError(t, err)
	defer func() {
		s.Close()
	}()
	props := map[string]interface{}{
		"interval": "1s",
		"dburl":    fmt.Sprintf("mysql:https://root:@%v:%v/test", address, port),
		"internalSqlQueryCfg": map[string]interface{}{
			"table": "t",
			"limit": 1,
			"indexFields": []map[string]interface{}{
				{
					"indexField":     "a",
					"indexValue":     0,
					"indexFieldType": "bigint",
				},
				{
					"indexField":     "b",
					"indexValue":     0,
					"indexFieldType": "bigint",
				},
			},
		},
	}
	sqlSource := GetSource()
	require.NoError(t, sqlSource.Provision(ctx, props))
	require.NoError(t, sqlSource.Connect(ctx))
	sqlConnector, ok := sqlSource.(*SQLSourceConnector)
	require.True(t, ok)
	expectedData := map[string]any{
		"a": int64(1),
		"b": int64(1),
	}
	dataChan := make(chan any, 1)
	sqlConnector.Pull(ctx, time.Now(), func(ctx api.StreamContext, data any, meta map[string]any, ts time.Time) {
		dataChan <- data
	}, func(ctx api.StreamContext, err error) {})
	require.Equal(t, expectedData, <-dataChan)
	state, err := sqlConnector.GetOffset()
	require.NoError(t, err)
	expectState := store.NewIndexFieldWrap([]*store.IndexField{
		{
			IndexFieldName:     "a",
			IndexFieldValue:    int64(1),
			IndexFieldDataType: "bigint",
		},
		{
			IndexFieldName:     "b",
			IndexFieldValue:    int64(1),
			IndexFieldDataType: "bigint",
		},
	}...).GetStore()
	require.Equal(t, expectState, state)
}

func TestSQLReconnect(t *testing.T) {
	connection.InitConnectionManager4Test()
	ctx := mockContext.NewMockContext("1", "2")
	props := map[string]interface{}{
		"interval": "1s",
		"dburl":    fmt.Sprintf("mysql:https://root:@%v:%v/test", address, port),
		"templateSqlQueryCfg": map[string]interface{}{
			"templateSql": "select a,b from t",
		},
	}
	sqlSource := GetSource()
	require.NoError(t, sqlSource.Provision(ctx, props))
	require.NoError(t, sqlSource.Connect(ctx))
	sqlConnector, ok := sqlSource.(*SQLSourceConnector)
	require.True(t, ok)
	sqlConnector.queryData(ctx, time.Now(), func(ctx api.StreamContext, data any, meta map[string]any, ts time.Time) {}, func(ctx api.StreamContext, err error) {})
	require.True(t, sqlConnector.needReconnect)

	// start server then reconnect
	s, err := testx.SetupEmbeddedMysqlServer(address, port)
	require.NoError(t, err)
	defer s.Close()
	sqlConnector.queryData(ctx, time.Now(), func(ctx api.StreamContext, data any, meta map[string]any, ts time.Time) {}, func(ctx api.StreamContext, err error) {})
	require.False(t, sqlConnector.needReconnect)
}

func TestSQLConfURL(t *testing.T) {
	testcases := []struct {
		got *SQLConf
		exp *SQLConf
	}{
		{
			got: &SQLConf{
				DBUrl: "123",
				URL:   "321",
			},
			exp: &SQLConf{
				DBUrl: "123",
			},
		},
		{
			got: &SQLConf{
				URL: "321",
			},
			exp: &SQLConf{
				DBUrl: "321",
			},
		},
	}
	for _, tc := range testcases {
		require.NoError(t, tc.got.resolveDBURL())
		require.Equal(t, tc.exp, tc.got)
	}
}