Add test for Conn.ExecuteSelectStreaming

client/client_test.go

@@ -2,38 +2,16 @@ package client
 
 import (
 	"encoding/json"
-	"flag"
 	"fmt"
 	"strings"
-	"testing"
 
-	"github.com/go-mysql-org/go-mysql/test_util/test_keys"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/errors"
 
 	"github.com/go-mysql-org/go-mysql/mysql"
+	"github.com/go-mysql-org/go-mysql/test_util/test_keys"
 )
 
-var testHost = flag.String("host", "127.0.0.1", "MySQL server host")
-
-// We cover the whole range of MySQL server versions using docker-compose to bind them to different ports for testing.
-// MySQL is constantly updating auth plugin to make it secure:
-// starting from MySQL 8.0.4, a new auth plugin is introduced, causing plain password auth to fail with error:
-// ERROR 1251 (08004): Client does not support authentication protocol requested by server; consider upgrading MySQL client
-// Hint: use docker-compose to start corresponding MySQL docker containers and add the their ports here
-var testPort = flag.String("port", "3306", "MySQL server port") // choose one or more form 5561,5641,3306,5722,8003,8012,8013, e.g. '3306,5722,8003'
-var testUser = flag.String("user", "root", "MySQL user")
-var testPassword = flag.String("pass", "", "MySQL password")
-var testDB = flag.String("db", "test", "MySQL test database")
-
-func Test(t *testing.T) {
-	segs := strings.Split(*testPort, ",")
-	for _, seg := range segs {
-		Suite(&clientTestSuite{port: seg})
-	}
-	TestingT(t)
-}
-
 type clientTestSuite struct {
 	c    *Conn
 	port string

client/common_test.go

@@ -0,0 +1,30 @@
+package client
+
+import (
+	"flag"
+	"strings"
+	"testing"
+
+	. "github.com/pingcap/check"
+)
+
+var testHost = flag.String("host", "127.0.0.1", "MySQL server host")
+
+// We cover the whole range of MySQL server versions using docker-compose to bind them to different ports for testing.
+// MySQL is constantly updating auth plugin to make it secure:
+// starting from MySQL 8.0.4, a new auth plugin is introduced, causing plain password auth to fail with error:
+// ERROR 1251 (08004): Client does not support authentication protocol requested by server; consider upgrading MySQL client
+// Hint: use docker-compose to start corresponding MySQL docker containers and add the their ports here
+var testPort = flag.String("port", "3306", "MySQL server port") // choose one or more form 5561,5641,3306,5722,8003,8012,8013, e.g. '3306,5722,8003'
+var testUser = flag.String("user", "root", "MySQL user")
+var testPassword = flag.String("pass", "", "MySQL password")
+var testDB = flag.String("db", "test", "MySQL test database")
+
+func Test(t *testing.T) {
+	segs := strings.Split(*testPort, ",")
+	for _, seg := range segs {
+		Suite(&clientTestSuite{port: seg})
+		Suite(&connTestSuite{port: seg})
+	}
+	TestingT(t)
+}

client/conn_test.go

@@ -0,0 +1,126 @@
+package client
+
+import (
+	"fmt"
+
+	. "github.com/pingcap/check"
+
+	"github.com/go-mysql-org/go-mysql/mysql"
+)
+
+type connTestSuite struct {
+	c    *Conn
+	port string
+}
+
+func (s *connTestSuite) SetUpSuite(c *C) {
+	var err error
+	addr := fmt.Sprintf("%s:%s", *testHost, s.port)
+	s.c, err = Connect(addr, *testUser, *testPassword, "")
+	if err != nil {
+		c.Fatal(err)
+	}
+
+	_, err = s.c.Execute("CREATE DATABASE IF NOT EXISTS " + *testDB)
+	c.Assert(err, IsNil)
+
+	_, err = s.c.Execute("USE " + *testDB)
+	c.Assert(err, IsNil)
+
+	s.testExecute_CreateTable(c)
+}
+
+func (s *connTestSuite) TearDownSuite(c *C) {
+	if s.c == nil {
+		return
+	}
+
+	s.testExecute_DropTable(c)
+
+	if s.c != nil {
+		s.c.Close()
+	}
+}
+
+var (
+	testExecuteSelectStreamingRows      = [...]string{"foo", "helloworld", "bar", "", "spam"}
+	testExecuteSelectStreamingTablename = "execute_plain_table"
+)
+
+func (s *connTestSuite) testExecute_CreateTable(c *C) {
+	str := `CREATE TABLE IF NOT EXISTS ` + testExecuteSelectStreamingTablename + ` (
+          id INT UNSIGNED NOT NULL,
+          str VARCHAR(256),
+          PRIMARY KEY (id)
+        ) ENGINE=InnoDB DEFAULT CHARSET=utf8`
+
+	result, err := s.c.Execute(str)
+	c.Assert(err, IsNil)
+	result.Close()
+
+	result, err = s.c.Execute(`TRUNCATE TABLE ` + testExecuteSelectStreamingTablename)
+	c.Assert(err, IsNil)
+	result.Close()
+
+	stmt, err := s.c.Prepare(`INSERT INTO ` + testExecuteSelectStreamingTablename + ` (id, str) VALUES (?, ?)`)
+	c.Assert(err, IsNil)
+	defer stmt.Close()
+
+	for id, str := range testExecuteSelectStreamingRows {
+		result, err := stmt.Execute(id, str)
+		c.Assert(err, IsNil)
+		result.Close()
+	}
+}
+
+func (s *connTestSuite) testExecute_DropTable(c *C) {
+	_, err := s.c.Execute(`drop table if exists ` + testExecuteSelectStreamingTablename)
+	c.Assert(err, IsNil)
+}
+
+func (s *connTestSuite) TestExecuteSelectStreaming(c *C) {
+	var (
+		expectedRowId                int64
+		perResultCallbackCalledTimes int
+		result                       mysql.Result
+	)
+
+	const colsInResult = 2 // id, str
+
+	err := s.c.ExecuteSelectStreaming(`SELECT id, str FROM `+testExecuteSelectStreamingTablename+` ORDER BY id`,
+		&result,
+		func(row []mysql.FieldValue) error {
+			// Check number of columns
+			c.Assert(row, HasLen, colsInResult)
+			// Check type of columns
+			c.Assert(row[0].Type, Equals, mysql.FieldValueType(mysql.FieldValueTypeUnsigned))
+			c.Assert(row[1].Type, Equals, mysql.FieldValueType(mysql.FieldValueTypeString))
+
+			id := row[0].AsInt64()
+			str := row[1].AsString()
+
+			// Check order of rows
+			c.Assert(id, Equals, expectedRowId)
+			// Check string values (protection from incorrect reuse of memory)
+			c.Assert(string(str), Equals, testExecuteSelectStreamingRows[id])
+
+			expectedRowId++
+
+			return nil
+		}, func(result *mysql.Result) error {
+			// result.Resultset must be defined at this point
+			c.Assert(result.Resultset, NotNil)
+			// Check number of columns
+			c.Assert(result.Resultset.Fields, HasLen, colsInResult)
+
+			perResultCallbackCalledTimes++
+			return nil
+		})
+	c.Assert(err, IsNil)
+
+	// Check total rows count
+	c.Assert(expectedRowId, Equals, int64(len(testExecuteSelectStreamingRows)))
+
+	// Check perResultCallback call count
+	c.Assert(perResultCallbackCalledTimes, Equals, 1)
+}