ekuiper-new/internal/server/rpc_test.go

// Copyright 2023 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
//
//     http://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 server

import (
	"os"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/suite"

	"github.com/lf-edge/ekuiper/internal/meta"
	"github.com/lf-edge/ekuiper/internal/pkg/model"
	"github.com/lf-edge/ekuiper/internal/plugin/native"
	"github.com/lf-edge/ekuiper/internal/plugin/portable"
	"github.com/lf-edge/ekuiper/internal/schema"
	"github.com/lf-edge/ekuiper/internal/service"
)

type ServerTestSuite struct {
	suite.Suite
	s *Server
}

func (suite *ServerTestSuite) SetupTest() {
	suite.s = new(Server)
	nativeManager, _ = native.InitManager()
	portableManager, _ = portable.InitManager()
	serviceManager, _ = service.InitManager()
	_ = schema.InitRegistry()
	meta.InitYamlConfigManager()
}

func (suite *ServerTestSuite) TestStream() {
	sql := `Create Stream test () WITH (DATASOURCE="../internal/server/rpc_test_data/test.json", FORMAT="JSON", type="file");`
	var reply string
	err := suite.s.Stream(sql, &reply)
	assert.Nil(suite.T(), err)
	assert.Equal(suite.T(), "Stream test is created.\n", reply)

	reply = ""
	sql = "show streams;"
	err = suite.s.Stream(sql, &reply)
	assert.Nil(suite.T(), err)

	reply = ""
	sql = "SELECT * FROM test;"
	err = suite.s.CreateQuery(sql, &reply)
	assert.Nil(suite.T(), err)
	assert.Equal(suite.T(), "Query was submit successfully.", reply)

	var result string = ""
	for i := 0; i < 5; i++ {
		var queryresult string
		time.Sleep(time.Second)
		err = suite.s.GetQueryResult("test", &queryresult)
		assert.Nil(suite.T(), err)
		result += queryresult
	}
	assert.Equal(suite.T(), "[{\"humidity\":50,\"id\":1,\"temperature\":20}]\n[{\"humidity\":51,\"id\":2,\"temperature\":21}]\n[{\"humidity\":52,\"id\":3,\"temperature\":22}]\n[{\"humidity\":53,\"id\":4,\"temperature\":23}]", result)
	stopQuery()
}

func (suite *ServerTestSuite) TestRule() {
	sql := `Create Stream test () WITH (DATASOURCE="../internal/server/rpc_test_data/test.json", FORMAT="JSON", type="file");`
	var reply string
	err := suite.s.Stream(sql, &reply)
	assert.Nil(suite.T(), err)
	assert.Equal(suite.T(), "Stream test is created.\n", reply)

	reply = ""
	rule := `{
			  "sql": "SELECT * from test;",
			  "actions": [{
				"file": {
				  "path": "../internal/server/rpc_test_data/data/result.txt",
				  "interval": 5000,
				  "fileType": "lines",
				  "format": "json"
				}
			  }]
	}`
	ruleId := "myRule"
	args := &model.RPCArgDesc{Name: ruleId, Json: rule}
	err = suite.s.ValidateRule(args, &reply)
	assert.Nil(suite.T(), err)
	assert.Equal(suite.T(), "The rule has been successfully validated and is confirmed to be correct.", reply)

	reply = ""
	rule = `{
			  "sql": "SELECT * from test;"
			}`
	args = &model.RPCArgDesc{Name: ruleId, Json: rule}
	err = suite.s.ValidateRule(args, &reply)
	assert.Nil(suite.T(), err)
	assert.Equal(suite.T(), "invalid rule json: Missing rule actions.", reply)

	reply = ""
	rule = `{
			  "sql": "SELECT * from test;",
			  "actions": [{
				"file": {
				  "path": "../internal/server/rpc_test_data/data/result.txt",
				  "interval": 5000,
				  "fileType": "lines",
				  "format": "json"
				}
			  }]
	}`
	args = &model.RPCArgDesc{Name: ruleId, Json: rule}
	err = suite.s.CreateRule(args, &reply)
	assert.Nil(suite.T(), err)
	assert.Equal(suite.T(), "Rule myRule was created successfully, please use 'bin/kuiper getstatus rule myRule' command to get rule status.", reply)

	reply = ""
	err = suite.s.GetStatusRule(ruleId, &reply)
	assert.Nil(suite.T(), err)

	reply = ""
	err = suite.s.ShowRules(1, &reply)
	assert.Nil(suite.T(), err)

	reply = ""
	err = suite.s.DescRule(ruleId, &reply)
	assert.Nil(suite.T(), err)
	assert.Equal(suite.T(), "{\n  \"sql\": \"SELECT * from test;\",\n  \"actions\": [\n    {\n      \"file\": {\n        \"path\": \"../internal/server/rpc_test_data/data/result.txt\",\n        \"interval\": 5000,\n        \"fileType\": \"lines\",\n        \"format\": \"json\"\n      }\n    }\n  ]\n}\n", reply)

	reply = ""
	err = suite.s.GetTopoRule(ruleId, &reply)
	assert.Nil(suite.T(), err)
	assert.Equal(suite.T(), "{\n  \"sources\": [\n    \"source_test\"\n  ],\n  \"edges\": {\n    \"op_2_project\": [\n      \"sink_file_0\"\n    ],\n    \"source_test\": [\n      \"op_2_project\"\n    ]\n  }\n}", reply)

	reply = ""
	err = suite.s.StopRule(ruleId, &reply)
	assert.Nil(suite.T(), err)
	assert.Equal(suite.T(), "Rule myRule was stopped.", reply)

	reply = ""
	err = suite.s.StartRule(ruleId, &reply)
	assert.Nil(suite.T(), err)
	assert.Equal(suite.T(), "Rule myRule was started", reply)

	reply = ""
	err = suite.s.RestartRule(ruleId, &reply)
	assert.Nil(suite.T(), err)
	assert.Equal(suite.T(), "Rule myRule was restarted.", reply)

	reply = ""
	err = suite.s.DropRule(ruleId, &reply)
	assert.Nil(suite.T(), err)
	assert.Equal(suite.T(), "Rule myRule is dropped.", reply)
}

func (suite *ServerTestSuite) TestImportAndExport() {
	file := "rpc_test_data/import.json"
	var reply string
	err := suite.s.Import(file, &reply)
	assert.Nil(suite.T(), err)
	assert.Equal(suite.T(), "imported 1 streams, 0 tables and 1 rules", reply)

	reply = ""
	file = "rpc_test_data/export.json"
	err = suite.s.Export(file, &reply)
	assert.Nil(suite.T(), err)
	os.Remove(file)
}

func (suite *ServerTestSuite) TestConfigurarion() {
	importArg := model.ImportDataDesc{
		FileName: "rpc_test_data/import_configuration.json",
		Stop:     false,
		Partial:  false,
	}
	var reply string
	err := suite.s.ImportConfiguration(&importArg, &reply)
	assert.Nil(suite.T(), err)
	assert.Equal(suite.T(), "{\n  \"ErrorMsg\": \"\",\n  \"ConfigResponse\": {\n    \"streams\": {},\n    \"tables\": {},\n    \"rules\": {},\n    \"nativePlugins\": {},\n    \"portablePlugins\": {},\n    \"sourceConfig\": {},\n    \"sinkConfig\": {},\n    \"connectionConfig\": {},\n    \"Service\": {},\n    \"Schema\": {},\n    \"uploads\": {}\n  }\n}", reply)

	reply = ""
	err = suite.s.GetStatusImport(1, &reply)
	assert.Nil(suite.T(), err)
	assert.Equal(suite.T(), "{\n  \"streams\": {},\n  \"tables\": {},\n  \"rules\": {},\n  \"nativePlugins\": {},\n  \"portablePlugins\": {},\n  \"sourceConfig\": {},\n  \"sinkConfig\": {},\n  \"connectionConfig\": {},\n  \"Service\": {},\n  \"Schema\": {},\n  \"uploads\": {}\n}", reply)

	reply = ""
	exportArg := model.ExportDataDesc{
		FileName: "rpc_test_data/export_configuration.json",
		Rules:    []string{},
	}
	err = suite.s.ExportConfiguration(&exportArg, &reply)
	assert.Nil(suite.T(), err)
	assert.Equal(suite.T(), "export configuration success", reply)
	os.Remove("rpc_test_data/export_configuration.json")
}

func (suite *ServerTestSuite) TearDownTest() {
	// Clean up
	sql := "DROP STREAM test;"
	var reply string
	_ = suite.s.Stream(sql, &reply)
	_ = suite.s.DropRule("myRule", &reply)
}

func TestServerTestSuite(t *testing.T) {
	suite.Run(t, new(ServerTestSuite))
}