func TestTxPoolExecuteRollback(t *testing.T) {
	sql := "alter table test_table add test_column int"
	db := fakesqldb.Register()
	db.AddQuery(sql, &sqltypes.Result{})
	db.AddQuery("begin", &sqltypes.Result{})
	db.AddQuery("rollback", &sqltypes.Result{})

	txPool := newTxPool(false)
	appParams := sqldb.ConnParams{Engine: db.Name}
	dbaParams := sqldb.ConnParams{Engine: db.Name}
	txPool.Open(&appParams, &dbaParams)
	defer txPool.Close()
	ctx := context.Background()
	transactionID, err := txPool.Begin(ctx)
	if err != nil {
		t.Fatal(err)
	}
	txConn, err := txPool.Get(transactionID, "for query")
	if err != nil {
		t.Fatal(err)
	}
	defer txPool.Rollback(ctx, transactionID)
	txConn.RecordQuery(sql)
	_, err = txConn.Exec(ctx, sql, 1, true)
	txConn.Recycle()
	if err != nil {
		t.Fatalf("got error: %v", err)
	}
}

func TestTxPoolExecFailDueToConnFail(t *testing.T) {
	db := fakesqldb.Register()
	db.AddQuery("begin", &sqltypes.Result{})

	txPool := newTxPool(false)
	appParams := sqldb.ConnParams{Engine: db.Name}
	dbaParams := sqldb.ConnParams{Engine: db.Name}
	txPool.Open(&appParams, &dbaParams)
	defer txPool.Close()
	ctx := context.Background()
	txPool.Begin(ctx)
	sql := "alter table test_table add test_column int"

	transactionID, err := txPool.Begin(ctx)
	if err != nil {
		t.Fatal(err)
	}
	txConn, err := txPool.Get(transactionID, "for query")
	if err != nil {
		t.Fatal(err)
	}
	db.EnableConnFail()
	_, err = txConn.Exec(ctx, sql, 1, true)
	txConn.Recycle()
	if err == nil {
		t.Fatalf("exec should fail because of a conn error")
	}
}

func TestTxPoolCommitFail(t *testing.T) {
	db := fakesqldb.Register()
	sql := fmt.Sprintf("alter table test_table add test_column int")
	db.AddQuery("begin", &sqltypes.Result{})
	db.AddQuery(sql, &sqltypes.Result{})
	db.AddRejectedQuery("commit", errRejected)
	txPool := newTxPool(false)
	appParams := sqldb.ConnParams{Engine: db.Name}
	dbaParams := sqldb.ConnParams{Engine: db.Name}
	txPool.Open(&appParams, &dbaParams)
	defer txPool.Close()
	ctx := context.Background()
	transactionID, err := txPool.Begin(ctx)
	if err != nil {
		t.Fatal(err)
	}
	txConn, err := txPool.Get(transactionID, "for query")
	if err != nil {
		t.Fatal(err)
	}
	_, err = txConn.Exec(ctx, sql, 1, true)
	txConn.Recycle()
	if err != nil {
		t.Fatalf("got exec error: %v", err)
	}
	err = txPool.Commit(ctx, transactionID)
	want := "error: error: rejected"
	if err == nil || err.Error() != want {
		t.Errorf("Commit: %v, want %s", err, want)
	}
}

func TestCachePoolStatsURL(t *testing.T) {
	cache := fakecacheservice.Register()
	fakesqldb.Register()
	rowCacheConfig := RowCacheConfig{
		Binary:      "ls",
		Connections: 100,
	}
	cachePool := newTestCachePool(rowCacheConfig, false)
	idleTimeout := 1 * time.Second
	cachePool.idleTimeout = idleTimeout
	cachePool.Open()
	request, _ := http.NewRequest("GET", fmt.Sprintf("%sstats", cachePool.statsURL), nil)
	response := httptest.NewRecorder()
	cachePool.ServeHTTP(response, request)
	// any memcache calls should fail
	cache.EnableCacheServiceError()
	response = httptest.NewRecorder()
	cachePool.ServeHTTP(response, request)
	cache.DisableCacheServiceError()
	cachePool.Close()
	response = httptest.NewRecorder()
	cachePool.ServeHTTP(response, request)
	body, _ := ioutil.ReadAll(response.Body)
	matcher := regexp.MustCompile("closed")
	if !matcher.Match(body) {
		t.Fatalf("stats page should contain 'closed', but got %s", string(body))
	}
}

func TestTxPoolRollbackNonBusy(t *testing.T) {
	db := fakesqldb.Register()
	db.AddQuery("begin", &sqltypes.Result{})
	db.AddQuery("rollback", &sqltypes.Result{})

	txPool := newTxPool(false)
	appParams := sqldb.ConnParams{Engine: db.Name}
	dbaParams := sqldb.ConnParams{Engine: db.Name}
	txPool.Open(&appParams, &dbaParams)
	defer txPool.Close()
	ctx := context.Background()
	txid1, err := txPool.Begin(ctx)
	if err != nil {
		t.Fatal(err)
	}
	_, err = txPool.Begin(ctx)
	if err != nil {
		t.Fatal(err)
	}
	conn1, err := txPool.Get(txid1, "for query")
	if err != nil {
		t.Fatal(err)
	}
	// This should rollback only txid2.
	txPool.RollbackNonBusy(ctx)
	if sz := txPool.activePool.Size(); sz != 1 {
		t.Errorf("txPool.activePool.Size(): %d, want 1", sz)
	}
	conn1.Recycle()
	// This should rollback txid1.
	txPool.RollbackNonBusy(ctx)
	if sz := txPool.activePool.Size(); sz != 0 {
		t.Errorf("txPool.activePool.Size(): %d, want 0", sz)
	}
}

func TestTableInfoWithoutRowCacheViaNoPKColumn(t *testing.T) {
	fakecacheservice.Register()
	db := fakesqldb.Register()
	db.AddQuery("show index from `test_table`", &sqltypes.Result{})
	db.AddQuery("select * from `test_table` where 1 != 1", &sqltypes.Result{
		Fields: []*querypb.Field{{
			Name: "pk",
			Type: sqltypes.Int32,
		}},
	})
	db.AddQuery("describe `test_table`", &sqltypes.Result{
		RowsAffected: 1,
		Rows: [][]sqltypes.Value{
			{
				sqltypes.MakeString([]byte("pk")),
				sqltypes.MakeString([]byte("int")),
				sqltypes.MakeString([]byte{}),
				sqltypes.MakeString([]byte{}),
				sqltypes.MakeString([]byte("1")),
				sqltypes.MakeString([]byte{}),
			},
		},
	})

	cachePool := newTestTableInfoCachePool()
	cachePool.Open()
	defer cachePool.Close()
	tableInfo, err := newTestTableInfo(cachePool, "USER_TABLE", "test table", db)
	if err != nil {
		t.Fatalf("failed to create a test table info")
	}
	if tableInfo.Cache != nil {
		t.Fatalf("table info's rowcache should be disabled")
	}
}

func TestCachePoolState(t *testing.T) {
	fakecacheservice.Register()
	fakesqldb.Register()
	rowCacheConfig := RowCacheConfig{
		Binary:      "ls",
		Connections: 100,
	}
	cachePool := newTestCachePool(rowCacheConfig, true)
	idleTimeout := 1 * time.Second
	cachePool.idleTimeout = idleTimeout
	cachePool.Open()
	cachePool.memcacheStats.update()
	defer cachePool.Close()
	if cachePool.Available() <= 0 {
		t.Fatalf("cache pool should have connections available")
	}
	if cachePool.Capacity() <= 0 {
		t.Fatalf("cache pool should have positive capacity")
	}
	if cachePool.MaxCap() <= 0 {
		t.Fatalf("cache pool should have positive max cap")
	}
	if cachePool.WaitCount() > 0 {
		t.Fatalf("cache pool has never waited for a connection, WaitCount should return 0")
	}
	if cachePool.WaitTime() > 0 {
		t.Fatalf("cache pool has never waited for a connection, WaitTime should return 0")
	}
	if cachePool.IdleTimeout() != idleTimeout {
		t.Fatalf("cache pool's idle timeout does not match the specified one")
	}
	if len(cachePool.StatsJSON()) <= 0 {
		t.Fatalf("cache pool stats json should return non empty result")
	}
}

func TestSchemaInfoOpenFailedDueToTableInfoErr(t *testing.T) {
	fakecacheservice.Register()
	db := fakesqldb.Register()
	for query, result := range getSchemaInfoBaseTestQueries() {
		db.AddQuery(query, result)
	}
	db.AddQuery(baseShowTables, &mproto.QueryResult{
		RowsAffected: 1,
		Rows: [][]sqltypes.Value{
			createTestTableBaseShowTable("test_table"),
		},
	})
	db.AddQuery("describe `test_table`", &mproto.QueryResult{
		// this will cause NewTableInfo error
		RowsAffected: math.MaxUint64,
	})
	schemaInfo := newTestSchemaInfo(10, 1*time.Second, 1*time.Second, false)
	appParams := sqldb.ConnParams{}
	dbaParams := sqldb.ConnParams{}
	cachePool := newTestSchemaInfoCachePool(false, schemaInfo.queryServiceStats)
	cachePool.Open()
	defer cachePool.Close()
	defer handleAndVerifyTabletError(
		t,
		"schema info Open should fail because NewTableInfo failed",
		ErrFail,
	)
	schemaInfo.Open(&appParams, &dbaParams, []SchemaOverride{}, cachePool, false)
}

func TestSchemaInfoStatsURL(t *testing.T) {
	db := fakesqldb.Register()
	for query, result := range getSchemaInfoTestSupportedQueries() {
		db.AddQuery(query, result)
	}
	query := "select * from test_table_01"
	db.AddQuery("select * from test_table_01 where 1 != 1", &sqltypes.Result{})
	schemaInfo := newTestSchemaInfo(10, 1*time.Second, 1*time.Second, false)
	dbaParams := sqldb.ConnParams{Engine: db.Name}
	schemaInfo.Open(&dbaParams, true)
	defer schemaInfo.Close()
	// warm up cache
	ctx := context.Background()
	logStats := newLogStats("GetPlanStats", ctx)
	schemaInfo.GetPlan(ctx, logStats, query)

	request, _ := http.NewRequest("GET", schemaInfo.endpoints[debugQueryPlansKey], nil)
	response := httptest.NewRecorder()
	schemaInfo.ServeHTTP(response, request)

	request, _ = http.NewRequest("GET", schemaInfo.endpoints[debugQueryStatsKey], nil)
	response = httptest.NewRecorder()
	schemaInfo.ServeHTTP(response, request)

	request, _ = http.NewRequest("GET", schemaInfo.endpoints[debugSchemaKey], nil)
	response = httptest.NewRecorder()
	schemaInfo.ServeHTTP(response, request)

	request, _ = http.NewRequest("GET", "/debug/unknown", nil)
	response = httptest.NewRecorder()
	schemaInfo.ServeHTTP(response, request)
}

// TestInitMasterShardChecks makes sure the safety checks work
func TestInitMasterShardChecks(t *testing.T) {
	ctx := context.Background()
	db := fakesqldb.Register()
	ts := zktestserver.New(t, []string{"cell1", "cell2"})
	wr := wrangler.New(logutil.NewConsoleLogger(), ts, tmclient.NewTabletManagerClient())

	master := NewFakeTablet(t, wr, "cell1", 0, topodatapb.TabletType_MASTER, db)

	// InitShardMaster with an unknown tablet
	if err := wr.InitShardMaster(ctx, master.Tablet.Keyspace, master.Tablet.Shard, &topodatapb.TabletAlias{
		Cell: master.Tablet.Alias.Cell,
		Uid:  master.Tablet.Alias.Uid + 1,
	}, false /*force*/, 10*time.Second); err == nil || !strings.Contains(err.Error(), "is not in the shard") {
		t.Errorf("InitShardMaster with unknown alias returned wrong error: %v", err)
	}

	// InitShardMaster with two masters in the shard, no force flag
	// (master2 needs to run InitTablet with -force, as it is the second
	// master in the same shard)
	master2 := NewFakeTablet(t, wr, "cell1", 1, topodatapb.TabletType_MASTER, db, ForceInitTablet())
	if err := wr.InitShardMaster(ctx, master2.Tablet.Keyspace, master2.Tablet.Shard, master2.Tablet.Alias, false /*force*/, 10*time.Second); err == nil || !strings.Contains(err.Error(), "is not the only master in the shard") {
		t.Errorf("InitShardMaster with two masters returned wrong error: %v", err)
	}

	// InitShardMaster where the new master fails (use force flag
	// as we have 2 masters). We force the failure by making the
	// SQL commands executed on the master unexpected by the test fixture
	master.StartActionLoop(t, wr)
	defer master.StopActionLoop(t)
	master2.StartActionLoop(t, wr)
	defer master2.StopActionLoop(t)
	if err := wr.InitShardMaster(ctx, master.Tablet.Keyspace, master.Tablet.Shard, master.Tablet.Alias, true /*force*/, 10*time.Second); err == nil || !strings.Contains(err.Error(), "unexpected extra query") {
		t.Errorf("InitShardMaster with new master failing in new master InitMaster returned wrong error: %v", err)
	}
}

func TestDBConnStream(t *testing.T) {
	db := fakesqldb.Register()
	testUtils := newTestUtils()
	sql := "select * from test_table limit 1000"
	expectedResult := &mproto.QueryResult{
		RowsAffected: 0,
		Rows: [][]sqltypes.Value{
			[]sqltypes.Value{sqltypes.MakeString([]byte("123"))},
		},
	}
	db.AddQuery(sql, expectedResult)
	connPool := testUtils.newConnPool()
	appParams := &sqldb.ConnParams{Engine: db.Name}
	dbaParams := &sqldb.ConnParams{Engine: db.Name}
	connPool.Open(appParams, dbaParams)
	defer connPool.Close()
	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(10*time.Second))
	defer cancel()
	queryServiceStats := NewQueryServiceStats("", false)
	dbConn, err := NewDBConn(connPool, appParams, dbaParams, queryServiceStats)
	defer dbConn.Close()
	var result mproto.QueryResult
	err = dbConn.Stream(
		ctx, sql, func(r *mproto.QueryResult) error {
			result = *r
			return nil
		}, 10)
	if err != nil {
		t.Fatalf("should not get an error, err: %v", err)
	}
	testUtils.checkEqual(t, expectedResult, &result)
}

func TestSchemaInfoDropTable(t *testing.T) {
	fakecacheservice.Register()
	db := fakesqldb.Register()
	for query, result := range getSchemaInfoTestSupportedQueries() {
		db.AddQuery(query, result)
	}
	existingTable := "test_table_01"
	createOrDropTableQuery := fmt.Sprintf("%s and table_name = '%s'", baseShowTables, existingTable)
	db.AddQuery(createOrDropTableQuery, &mproto.QueryResult{
		RowsAffected: 1,
		Rows: [][]sqltypes.Value{
			createTestTableBaseShowTable(existingTable),
		},
	})
	schemaInfo := newTestSchemaInfo(10, 1*time.Second, 1*time.Second, false)
	appParams := sqldb.ConnParams{}
	dbaParams := sqldb.ConnParams{}
	cachePool := newTestSchemaInfoCachePool(false, schemaInfo.queryServiceStats)
	cachePool.Open()
	defer cachePool.Close()
	schemaInfo.Open(&appParams, &dbaParams, getSchemaInfoTestSchemaOverride(), cachePool, false)
	tableInfo := schemaInfo.GetTable(existingTable)
	if tableInfo == nil {
		t.Fatalf("table: %s should exist", existingTable)
	}
	schemaInfo.DropTable(existingTable)
	tableInfo = schemaInfo.GetTable(existingTable)
	if tableInfo != nil {
		t.Fatalf("table: %s should not exist", existingTable)
	}
	schemaInfo.Close()
}

func TestSchemaInfoGetPlanPanicDuetoEmptyQuery(t *testing.T) {
	fakecacheservice.Register()
	db := fakesqldb.Register()
	for query, result := range getSchemaInfoTestSupportedQueries() {
		db.AddQuery(query, result)
	}
	schemaInfo := newTestSchemaInfo(10, 10*time.Second, 10*time.Second, false)
	appParams := sqldb.ConnParams{}
	dbaParams := sqldb.ConnParams{}
	cachePool := newTestSchemaInfoCachePool(false, schemaInfo.queryServiceStats)
	cachePool.Open()
	defer cachePool.Close()
	schemaOverrides := getSchemaInfoTestSchemaOverride()
	// test cache type RW
	schemaInfo.Open(&appParams, &dbaParams, schemaOverrides, cachePool, true)
	defer schemaInfo.Close()

	ctx := context.Background()
	logStats := newLogStats("GetPlanStats", ctx)
	defer handleAndVerifyTabletError(
		t,
		"schema info GetPlan should fail because of empty query",
		ErrFail,
	)
	schemaInfo.GetPlan(ctx, logStats, "")
}

func TestSchemaInfoDropTable(t *testing.T) {
	db := fakesqldb.Register()
	for query, result := range getSchemaInfoTestSupportedQueries() {
		db.AddQuery(query, result)
	}
	existingTable := "test_table_01"
	createOrDropTableQuery := fmt.Sprintf("%s and table_name = '%s'", baseShowTables, existingTable)
	db.AddQuery(createOrDropTableQuery, &sqltypes.Result{
		RowsAffected: 1,
		Rows: [][]sqltypes.Value{
			createTestTableBaseShowTable(existingTable),
		},
	})
	schemaInfo := newTestSchemaInfo(10, 1*time.Second, 1*time.Second, false)
	dbaParams := sqldb.ConnParams{Engine: db.Name}
	schemaInfo.Open(&dbaParams, false)
	tableInfo := schemaInfo.GetTable(existingTable)
	if tableInfo == nil {
		t.Fatalf("table: %s should exist", existingTable)
	}
	schemaInfo.DropTable(existingTable)
	tableInfo = schemaInfo.GetTable(existingTable)
	if tableInfo != nil {
		t.Fatalf("table: %s should not exist", existingTable)
	}
	schemaInfo.Close()
}

func TestSchemaInfoCreateOrUpdateTableFailedDuetoExecErr(t *testing.T) {
	fakecacheservice.Register()
	db := fakesqldb.Register()
	for query, result := range getSchemaInfoTestSupportedQueries() {
		db.AddQuery(query, result)
	}
	createOrDropTableQuery := fmt.Sprintf("%s and table_name = '%s'", baseShowTables, "test_table")
	db.AddQuery(createOrDropTableQuery, &mproto.QueryResult{
		// make this query fail
		RowsAffected: math.MaxUint64,
		Rows: [][]sqltypes.Value{
			createTestTableBaseShowTable("test_table"),
		},
	})
	schemaInfo := newTestSchemaInfo(10, 1*time.Second, 1*time.Second, false)
	appParams := sqldb.ConnParams{}
	dbaParams := sqldb.ConnParams{}
	cachePool := newTestSchemaInfoCachePool(false, schemaInfo.queryServiceStats)
	cachePool.Open()
	defer cachePool.Close()
	defer handleAndVerifyTabletError(
		t,
		"CreateOrUpdateTable should fail because it could not tables from MySQL",
		ErrFail,
	)
	schemaInfo.Open(&appParams, &dbaParams, getSchemaInfoTestSchemaOverride(), cachePool, false)
	defer schemaInfo.Close()
	schemaInfo.CreateOrUpdateTable(context.Background(), "test_table")
}

func TestTableInfoInvalidCardinalityInIndex(t *testing.T) {
	fakecacheservice.Register()
	db := fakesqldb.Register()
	for query, result := range getTestTableInfoQueries() {
		db.AddQuery(query, result)
	}
	db.AddQuery("show index from `test_table`", &sqltypes.Result{
		RowsAffected: 1,
		Rows: [][]sqltypes.Value{
			{
				sqltypes.MakeString([]byte{}),
				sqltypes.MakeString([]byte{}),
				sqltypes.MakeString([]byte("PRIMARY")),
				sqltypes.MakeString([]byte{}),
				sqltypes.MakeString([]byte("pk")),
				sqltypes.MakeString([]byte{}),
				sqltypes.MakeString([]byte("invalid")),
			},
		},
	})
	cachePool := newTestTableInfoCachePool()
	cachePool.Open()
	defer cachePool.Close()
	tableInfo, err := newTestTableInfo(cachePool, "USER_TABLE", "test table", db)
	if err != nil {
		t.Fatalf("failed to create a table info: %v", err)
	}
	if len(tableInfo.PKColumns) != 1 {
		t.Fatalf("table should have one PK column although the cardinality is invalid")
	}
}

func setUpSQLQueryTest() *fakesqldb.DB {
	db := fakesqldb.Register()
	for query, result := range getSupportedQueries() {
		db.AddQuery(query, result)
	}
	return db
}

func TestTxPoolTransactionKiller(t *testing.T) {
	sql := "alter table test_table add test_column int"
	db := fakesqldb.Register()
	db.AddQuery(sql, &sqltypes.Result{})
	db.AddQuery("begin", &sqltypes.Result{})

	txPool := newTxPool(false)
	// make sure transaction killer will run frequent enough
	txPool.SetTimeout(time.Duration(10))
	appParams := sqldb.ConnParams{Engine: db.Name}
	dbaParams := sqldb.ConnParams{Engine: db.Name}
	txPool.Open(&appParams, &dbaParams)
	defer txPool.Close()
	ctx := context.Background()
	killCount := txPool.queryServiceStats.KillStats.Counts()["Transactions"]
	transactionID := txPool.Begin(ctx)
	txConn := txPool.Get(transactionID)
	txConn.RecordQuery(sql)
	txConn.Recycle()
	// transaction killer should kill the query
	txPool.WaitForEmpty()
	killCountDiff := txPool.queryServiceStats.KillStats.Counts()["Transactions"] - killCount
	if killCountDiff != 1 {
		t.Fatalf("query: %s should be killed by transaction killer", sql)
	}
}

func TestCachePoolStateWithoutOpen(t *testing.T) {
	fakecacheservice.Register()
	fakesqldb.Register()
	rowCacheConfig := RowCacheConfig{
		Binary:      "ls",
		Connections: 100,
	}
	cachePool := newTestCachePool(rowCacheConfig, false)
	idleTimeout := 1 * time.Second
	cachePool.idleTimeout = idleTimeout
	if cachePool.StatsJSON() != "{}" {
		t.Fatalf("cache pool StatsJSON() should return {}")
	}
	if cachePool.Capacity() != 0 {
		t.Fatalf("cache pool Capacity() should return 0")
	}
	if cachePool.Available() != 0 {
		t.Fatalf("cache pool Available() should return 0")
	}
	if cachePool.MaxCap() != 0 {
		t.Fatalf("cache pool MaxCap() should return 0")
	}
	if cachePool.WaitCount() != 0 {
		t.Fatalf("cache pool WaitCount() should return 0")
	}
	if cachePool.WaitTime() != 0 {
		t.Fatalf("cache pool WaitTime() should return 0")
	}
	if cachePool.IdleTimeout() != 0 {
		t.Fatalf("cache pool IdleTimeout() should return 0")
	}
	cachePool.Put(nil)
}

func TestTxPoolExecuteCommit(t *testing.T) {
	tableName := "test_table"
	sql := fmt.Sprintf("alter table %s add test_column int", tableName)
	db := fakesqldb.Register()
	db.AddQuery("begin", &sqltypes.Result{})
	db.AddQuery("commit", &sqltypes.Result{})
	db.AddQuery(sql, &sqltypes.Result{})

	txPool := newTxPool(true)
	txPool.SetTimeout(1 * time.Second)
	appParams := sqldb.ConnParams{Engine: db.Name}
	dbaParams := sqldb.ConnParams{Engine: db.Name}
	txPool.Open(&appParams, &dbaParams)
	defer txPool.Close()
	ctx := context.Background()
	transactionID := txPool.Begin(ctx)
	txConn := txPool.Get(transactionID)
	defer txPool.Commit(ctx, transactionID)
	txConn.RecordQuery(sql)
	_, err := txConn.Exec(ctx, sql, 1, true)
	txConn.Recycle()
	if err != nil {
		t.Fatalf("got error: %v", err)
	}
	txPool.LogActive()
	txPool.LogActive()
	// start another transaction which should be killed
	// in txPool.Close()
	_ = txPool.Begin(ctx)
}