// Copyright 2025 Google LLC

//

// 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 spannerdriver

import (

"context"

"database/sql"

"fmt"

"reflect"

"testing"

"time"

"cloud.google.com/go/longrunning/autogen/longrunningpb"

"cloud.google.com/go/spanner"

"cloud.google.com/go/spanner/admin/database/apiv1/databasepb"

"cloud.google.com/go/spanner/apiv1/spannerpb"

"github.com/googleapis/go-sql-spanner/connectionstate"

"github.com/googleapis/go-sql-spanner/testutil"

"google.golang.org/grpc/codes"

"google.golang.org/grpc/status"

"google.golang.org/protobuf/proto"

"google.golang.org/protobuf/types/known/anypb"

"google.golang.org/protobuf/types/known/emptypb"

)

func TestBeginTx(t *testing.T) {

t.Parallel()

db, server, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

tx, _ := db.BeginTx(ctx, &sql.TxOptions{})

_, _ = tx.ExecContext(ctx, testutil.UpdateBarSetFoo)

_ = tx.Rollback()

requests := server.TestSpanner.DrainRequestsFromServer()

beginRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.BeginTransactionRequest{}))

if g, w := len(beginRequests), 0; g != w {

t.Fatalf("begin requests count mismatch\n Got: %v\nWant: %v", g, w)

}

executeRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.ExecuteSqlRequest{}))

if g, w := len(executeRequests), 1; g != w {

t.Fatalf("execute requests count mismatch\n Got: %v\nWant: %v", g, w)

}

request := executeRequests[0].(*spannerpb.ExecuteSqlRequest)

if request.GetTransaction() == nil || request.GetTransaction().GetBegin() == nil {

t.Fatal("missing begin transaction on ExecuteSqlRequest")

}

if g, w := request.GetTransaction().GetBegin().GetIsolationLevel(), spannerpb.TransactionOptions_ISOLATION_LEVEL_UNSPECIFIED; g != w {

t.Fatalf("begin isolation level mismatch\n Got: %v\nWant: %v", g, w)

}

}

func TestTwoTransactionsOnOneConn(t *testing.T) {

t.Parallel()

db, _, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

c, _ := db.Conn(ctx)

tx1, err := c.BeginTx(ctx, &sql.TxOptions{})

defer func() { _ = tx1.Rollback() }()

if err != nil {

t.Fatal(err)

}

_, err = c.BeginTx(ctx, &sql.TxOptions{})

if g, w := spanner.ErrCode(err), codes.FailedPrecondition; g != w {

t.Fatalf("BeginTx error code mismatch\n Got: %v\nWant: %v", g, w)

}

}

func TestEmptyTransaction(t *testing.T) {

t.Parallel()

db, server, teardown := setupTestDBConnection(t)

defer teardown()

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

defer cancel()

db.SetMaxOpenConns(1)

c, err := db.Conn(ctx)

if err != nil {

t.Fatal(err)

}

defer silentClose(c)

// Run twice to ensure that there is no connection leak.

for range 2 {

tx, err := c.BeginTx(ctx, &sql.TxOptions{})

if err != nil {

t.Fatal(err)

}

if err := tx.Commit(); err != nil {

t.Fatal(err)

}

// An empty transaction should be a no-op and not lead to any requests being sent to Spanner.

requests := server.TestSpanner.DrainRequestsFromServer()

beginRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.BeginTransactionRequest{}))

if g, w := len(beginRequests), 0; g != w {

t.Fatalf("begin requests count mismatch\n Got: %v\nWant: %v", g, w)

}

commitRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.CommitRequest{}))

if g, w := len(commitRequests), 0; g != w {

t.Fatalf("commit requests count mismatch\n Got: %v\nWant: %v", g, w)

}

}

}

func TestEmptyTransactionUsingSql(t *testing.T) {

t.Parallel()

db, server, teardown := setupTestDBConnection(t)

defer teardown()

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

defer cancel()

db.SetMaxOpenConns(1)

c, err := db.Conn(ctx)

if err != nil {

t.Fatal(err)

}

defer silentClose(c)

// Run twice to ensure that there is no connection leak.

for range 2 {

if _, err := c.ExecContext(ctx, "begin"); err != nil {

t.Fatal(err)

}

if _, err := c.ExecContext(ctx, "commit"); err != nil {

t.Fatal(err)

}

// An empty transaction should be a no-op and not lead to any requests being sent to Spanner.

requests := server.TestSpanner.DrainRequestsFromServer()

beginRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.BeginTransactionRequest{}))

if g, w := len(beginRequests), 0; g != w {

t.Fatalf("begin requests count mismatch\n Got: %v\nWant: %v", g, w)

}

commitRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.CommitRequest{}))

if g, w := len(commitRequests), 0; g != w {

t.Fatalf("commit requests count mismatch\n Got: %v\nWant: %v", g, w)

}

}

}

func TestTwoQueriesOnOneConn(t *testing.T) {

t.Parallel()

db, _, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

c, _ := db.Conn(ctx)

defer silentClose(c)

for range 2 {

r, err := c.QueryContext(context.Background(), testutil.SelectFooFromBar)

if err != nil {

t.Fatal(err)

}

_ = r.Next()

defer silentClose(r)

}

}

func TestExplicitBeginTx(t *testing.T) {

t.Parallel()

db, server, teardown := setupTestDBConnectionWithConnectorConfig(t, ConnectorConfig{

Project: "p",

Instance: "i",

Database: "d",

BeginTransactionOption: spanner.ExplicitBeginTransaction,

})

defer teardown()

ctx := context.Background()

for _, readOnly := range []bool{true, false} {

tx, err := db.BeginTx(ctx, &sql.TxOptions{ReadOnly: readOnly})

if err != nil {

t.Fatal(err)

}

res, err := tx.QueryContext(ctx, testutil.SelectFooFromBar)

if err != nil {

t.Fatal(err)

}

for res.Next() {

}

if err := res.Err(); err != nil {

t.Fatal(err)

}

if err := tx.Rollback(); err != nil {

t.Fatal(err)

}

requests := server.TestSpanner.DrainRequestsFromServer()

beginRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.BeginTransactionRequest{}))

if g, w := len(beginRequests), 1; g != w {

t.Fatalf("begin requests count mismatch\n Got: %v\nWant: %v", g, w)

}

executeRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.ExecuteSqlRequest{}))

if g, w := len(executeRequests), 1; g != w {

t.Fatalf("execute requests count mismatch\n Got: %v\nWant: %v", g, w)

}

request := executeRequests[0].(*spannerpb.ExecuteSqlRequest)

if request.GetTransaction() == nil || request.GetTransaction().GetId() == nil {

t.Fatal("missing transaction id on ExecuteSqlRequest")

}

}

}

func TestExecuteBegin(t *testing.T) {

t.Parallel()

db, server, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

for _, end := range []string{"rollback", "commit"} {

c, err := db.Conn(ctx)

if err != nil {

t.Fatal(err)

}

if _, err := c.ExecContext(ctx, "begin transaction"); err != nil {

t.Fatal(err)

}

if _, err := c.ExecContext(ctx, testutil.UpdateBarSetFoo); err != nil {

t.Fatal(err)

}

if _, err := c.ExecContext(ctx, end); err != nil {

t.Fatal(err)

}

if err := c.Close(); err != nil {

t.Fatal(err)

}

requests := server.TestSpanner.DrainRequestsFromServer()

beginRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.BeginTransactionRequest{}))

if g, w := len(beginRequests), 0; g != w {

t.Fatalf("begin requests count mismatch\n Got: %v\nWant: %v", g, w)

}

executeRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.ExecuteSqlRequest{}))

if g, w := len(executeRequests), 1; g != w {

t.Fatalf("execute requests count mismatch\n Got: %v\nWant: %v", g, w)

}

request := executeRequests[0].(*spannerpb.ExecuteSqlRequest)

if request.GetTransaction() == nil || request.GetTransaction().GetBegin() == nil {

t.Fatal("missing begin transaction on ExecuteSqlRequest")

}

commitRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.CommitRequest{}))

rollbackRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.RollbackRequest{}))

if end == "commit" {

if g, w := len(commitRequests), 1; g != w {

t.Fatalf("commit requests count mismatch\n Got: %v\nWant: %v", g, w)

}

} else if end == "rollback" {

if g, w := len(rollbackRequests), 1; g != w {

t.Fatalf("rollback requests count mismatch\n Got: %v\nWant: %v", g, w)

}

}

}

}

func TestEndTransactionWithoutBegin(t *testing.T) {

t.Parallel()

db, _, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

for _, end := range []string{"rollback", "commit"} {

c, err := db.Conn(ctx)

if err != nil {

t.Fatal(err)

}

_, err = c.ExecContext(ctx, end)

if g, w := spanner.ErrCode(err), codes.FailedPrecondition; g != w {

t.Fatalf("error code mismatch\n Got: %v\nWant: %v", g, w)

}

if err := c.Close(); err != nil {

t.Fatal(err)

}

}

}

func TestBeginTxWithIsolationLevel(t *testing.T) {

t.Parallel()

db, server, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

for _, level := range []sql.IsolationLevel{

sql.LevelDefault,

sql.LevelSnapshot,

sql.LevelRepeatableRead,

sql.LevelSerializable,

} {

originalLevel := level

for _, disableRetryAborts := range []bool{true, false} {

if disableRetryAborts {

level = WithDisableRetryAborts(originalLevel)

} else {

level = originalLevel

}

tx, _ := db.BeginTx(ctx, &sql.TxOptions{

Isolation: level,

})

_, _ = tx.ExecContext(ctx, testutil.UpdateBarSetFoo)

_ = tx.Rollback()

requests := server.TestSpanner.DrainRequestsFromServer()

beginRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.BeginTransactionRequest{}))

if g, w := len(beginRequests), 0; g != w {

t.Fatalf("begin requests count mismatch\n Got: %v\nWant: %v", g, w)

}

executeRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.ExecuteSqlRequest{}))

if g, w := len(executeRequests), 1; g != w {

t.Fatalf("execute requests count mismatch\n Got: %v\nWant: %v", g, w)

}

request := executeRequests[0].(*spannerpb.ExecuteSqlRequest)

if request.GetTransaction() == nil || request.GetTransaction().GetBegin() == nil {

t.Fatalf("execute request does not have a begin transaction")

}

wantIsolationLevel, _ := toProtoIsolationLevel(originalLevel)

if g, w := request.GetTransaction().GetBegin().GetIsolationLevel(), wantIsolationLevel; g != w {

t.Fatalf("begin isolation level mismatch\n Got: %v\nWant: %v", g, w)

}

}

}

}

func TestBeginTxWithInvalidIsolationLevel(t *testing.T) {

t.Parallel()

db, _, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

for _, level := range []sql.IsolationLevel{

sql.LevelReadUncommitted,

sql.LevelReadCommitted,

sql.LevelWriteCommitted,

sql.LevelLinearizable,

} {

originalLevel := level

for _, disableRetryAborts := range []bool{true, false} {

if disableRetryAborts {

level = WithDisableRetryAborts(originalLevel)

} else {

level = originalLevel

}

_, err := db.BeginTx(ctx, &sql.TxOptions{

Isolation: level,

})

if err == nil {

t.Fatalf("BeginTx should have failed with invalid isolation level: %v", level)

}

}

}

}

func TestDefaultIsolationLevel(t *testing.T) {

t.Parallel()

for _, level := range []sql.IsolationLevel{

sql.LevelDefault,

sql.LevelSnapshot,

sql.LevelRepeatableRead,

sql.LevelSerializable,

} {

db, server, teardown := setupTestDBConnectionWithParams(t, fmt.Sprintf("isolationLevel=%v", level))

defer teardown()

ctx := context.Background()

conn, err := db.Conn(ctx)

if err != nil {

t.Fatal(err)

}

if err := conn.Raw(func(driverConn interface{}) error {

spannerConn, ok := driverConn.(SpannerConn)

if !ok {

return fmt.Errorf("expected spanner conn, got %T", driverConn)

}

if spannerConn.IsolationLevel() != level {

return fmt.Errorf("expected isolation level %v, got %v", level, spannerConn.IsolationLevel())

}

return nil

}); err != nil {

t.Fatal(err)

}

originalLevel := level

for _, disableRetryAborts := range []bool{true, false} {

if disableRetryAborts {

level = WithDisableRetryAborts(originalLevel)

} else {

level = originalLevel

}

// Note: No isolation level is passed in here, so it will use the default.

tx, _ := db.BeginTx(ctx, &sql.TxOptions{})

_, _ = tx.ExecContext(ctx, testutil.UpdateBarSetFoo)

_ = tx.Rollback()

requests := server.TestSpanner.DrainRequestsFromServer()

beginRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.BeginTransactionRequest{}))

if g, w := len(beginRequests), 0; g != w {

t.Fatalf("begin requests count mismatch\n Got: %v\nWant: %v", g, w)

}

executeRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.ExecuteSqlRequest{}))

if g, w := len(executeRequests), 1; g != w {

t.Fatalf("execute requests count mismatch\n Got: %v\nWant: %v", g, w)

}

request := executeRequests[0].(*spannerpb.ExecuteSqlRequest)

if request.GetTransaction() == nil || request.GetTransaction().GetBegin() == nil {

t.Fatalf("ExecuteSqlRequest should have a Begin transaction")

}

wantIsolationLevel, _ := toProtoIsolationLevel(originalLevel)

if g, w := request.GetTransaction().GetBegin().GetIsolationLevel(), wantIsolationLevel; g != w {

t.Fatalf("begin isolation level mismatch\n Got: %v\nWant: %v", g, w)

}

}

}

}

func TestSetIsolationLevel(t *testing.T) {

t.Parallel()

db, _, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

// Repeat twice to ensure that the state is reset after closing the connection.

for i := 0; i < 2; i++ {

conn, err := db.Conn(ctx)

if err != nil {

t.Fatal(err)

}

var level sql.IsolationLevel

_ = conn.Raw(func(driverConn interface{}) error {

level = driverConn.(SpannerConn).IsolationLevel()

return nil

})

if g, w := level, sql.LevelDefault; g != w {

t.Fatalf("isolation level mismatch\n Got: %v\nWant: %v", g, w)

}

_ = conn.Raw(func(driverConn interface{}) error {

return driverConn.(SpannerConn).SetIsolationLevel(sql.LevelSnapshot)

})

_ = conn.Raw(func(driverConn interface{}) error {

level = driverConn.(SpannerConn).IsolationLevel()

return nil

})

if g, w := level, sql.LevelSnapshot; g != w {

t.Fatalf("isolation level mismatch\n Got: %v\nWant: %v", g, w)

}

_ = conn.Close()

}

}

func TestIsolationLevelAutoCommit(t *testing.T) {

t.Parallel()

db, server, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

for _, level := range []sql.IsolationLevel{

sql.LevelDefault,

sql.LevelSnapshot,

sql.LevelRepeatableRead,

sql.LevelSerializable,

} {

spannerLevel, _ := toProtoIsolationLevel(level)

_, _ = db.ExecContext(ctx, testutil.UpdateBarSetFoo, ExecOptions{TransactionOptions: spanner.TransactionOptions{

IsolationLevel: spannerLevel,

}})

requests := server.TestSpanner.DrainRequestsFromServer()

executeRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.ExecuteSqlRequest{}))

if g, w := len(executeRequests), 1; g != w {

t.Fatalf("execute requests count mismatch\n Got: %v\nWant: %v", g, w)

}

request := executeRequests[0].(*spannerpb.ExecuteSqlRequest)

wantIsolationLevel, _ := toProtoIsolationLevel(level)

if g, w := request.Transaction.GetBegin().GetIsolationLevel(), wantIsolationLevel; g != w {

t.Fatalf("begin isolation level mismatch\n Got: %v\nWant: %v", g, w)

}

}

}

func TestDefaultReadLockMode(t *testing.T) {

t.Parallel()

for mode, name := range spannerpb.TransactionOptions_ReadWrite_ReadLockMode_name {

readLockMode := spannerpb.TransactionOptions_ReadWrite_ReadLockMode(mode)

db, server, teardown := setupTestDBConnectionWithParams(t, fmt.Sprintf("read_lock_mode=%v", name))

defer teardown()

ctx := context.Background()

c, err := db.Conn(ctx)

if err != nil {

t.Fatal(err)

}

if err := c.Raw(func(driverConn interface{}) error {

spannerConn, ok := driverConn.(*conn)

if !ok {

return fmt.Errorf("expected spanner conn, got %T", driverConn)

}

if spannerConn.ReadLockMode() != readLockMode {

return fmt.Errorf("expected read lock mode %v, got %v", readLockMode, spannerConn.ReadLockMode())

}

return nil

}); err != nil {

t.Fatal(err)

}

tx, _ := db.BeginTx(ctx, &sql.TxOptions{})

_, _ = tx.ExecContext(ctx, testutil.UpdateBarSetFoo)

_ = tx.Rollback()

requests := server.TestSpanner.DrainRequestsFromServer()

beginRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.BeginTransactionRequest{}))

if g, w := len(beginRequests), 0; g != w {

t.Fatalf("begin requests count mismatch\n Got: %v\nWant: %v", g, w)

}

executeRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.ExecuteSqlRequest{}))

if g, w := len(executeRequests), 1; g != w {

t.Fatalf("execute requests count mismatch\n Got: %v\nWant: %v", g, w)

}

request := executeRequests[0].(*spannerpb.ExecuteSqlRequest)

if request.GetTransaction() == nil || request.GetTransaction().GetBegin() == nil {

t.Fatalf("ExecuteSqlRequest should have a Begin transaction")

}

if g, w := request.GetTransaction().GetBegin().GetReadWrite().GetReadLockMode(), readLockMode; g != w {

t.Fatalf("begin read lock mode mismatch\n Got: %v\nWant: %v", g, w)

}

}

}

func TestSetReadLockMode(t *testing.T) {

t.Parallel()

db, _, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

// Repeat twice to ensure that the state is reset after closing the connection.

for i := 0; i < 2; i++ {

c, err := db.Conn(ctx)

if err != nil {

t.Fatal(err)

}

var readLockMode spannerpb.TransactionOptions_ReadWrite_ReadLockMode

_ = c.Raw(func(driverConn interface{}) error {

readLockMode = driverConn.(*conn).ReadLockMode()

return nil

})

if g, w := readLockMode, spannerpb.TransactionOptions_ReadWrite_READ_LOCK_MODE_UNSPECIFIED; g != w {

t.Fatalf("read lock mode mismatch\n Got: %v\nWant: %v", g, w)

}

_ = c.Raw(func(driverConn interface{}) error {

return driverConn.(SpannerConn).SetReadLockMode(spannerpb.TransactionOptions_ReadWrite_OPTIMISTIC)

})

_ = c.Raw(func(driverConn interface{}) error {

readLockMode = driverConn.(SpannerConn).ReadLockMode()

return nil

})

if g, w := readLockMode, spannerpb.TransactionOptions_ReadWrite_OPTIMISTIC; g != w {

t.Fatalf("read lock mode mismatch\n Got: %v\nWant: %v", g, w)

}

_ = c.Close()

}

}

func TestReadLockModeAutoCommit(t *testing.T) {

t.Parallel()

db, server, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

for mode := range spannerpb.TransactionOptions_ReadWrite_ReadLockMode_name {

readLockMode := spannerpb.TransactionOptions_ReadWrite_ReadLockMode(mode)

_, _ = db.ExecContext(ctx, testutil.UpdateBarSetFoo, ExecOptions{TransactionOptions: spanner.TransactionOptions{

ReadLockMode: readLockMode,

}})

requests := server.TestSpanner.DrainRequestsFromServer()

executeRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.ExecuteSqlRequest{}))

if g, w := len(executeRequests), 1; g != w {

t.Fatalf("execute requests count mismatch\n Got: %v\nWant: %v", g, w)

}

request := executeRequests[0].(*spannerpb.ExecuteSqlRequest)

if g, w := request.Transaction.GetBegin().GetReadWrite().GetReadLockMode(), readLockMode; g != w {

t.Fatalf("begin read lock mode mismatch\n Got: %v\nWant: %v", g, w)

}

}

}

func TestSetLocalReadLockMode(t *testing.T) {

t.Parallel()

db, server, teardown := setupTestDBConnection(t)

// Make sure we only have one connection in the pool.

db.SetMaxOpenConns(1)

defer teardown()

ctx := context.Background()

tx, err := db.BeginTx(ctx, nil)

if err != nil {

t.Fatal(err)

}

if _, err := tx.ExecContext(ctx, "set local read_lock_mode='optimistic'"); err != nil {

t.Fatal(err)

}

if _, err := tx.ExecContext(ctx, testutil.UpdateBarSetFoo); err != nil {

t.Fatal(err)

}

if err := tx.Commit(); err != nil {

t.Fatal(err)

}

requests := server.TestSpanner.DrainRequestsFromServer()

beginRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.BeginTransactionRequest{}))

if g, w := len(beginRequests), 0; g != w {

t.Fatalf("begin requests count mismatch\n Got: %v\nWant: %v", g, w)

}

executeRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.ExecuteSqlRequest{}))

if g, w := len(executeRequests), 1; g != w {

t.Fatalf("execute requests count mismatch\n Got: %v\nWant: %v", g, w)

}

request := executeRequests[0].(*spannerpb.ExecuteSqlRequest)

if request.GetTransaction() == nil || request.GetTransaction().GetBegin() == nil {

t.Fatalf("ExecuteSqlRequest should have a Begin transaction")

}

if g, w := request.GetTransaction().GetBegin().GetReadWrite().GetReadLockMode(), spannerpb.TransactionOptions_ReadWrite_OPTIMISTIC; g != w {

t.Fatalf("begin read lock mode mismatch\n Got: %v\nWant: %v", g, w)

}

// Execute another transaction without a specific read lock mode. This should then use the default.

tx, err = db.BeginTx(ctx, nil)

if err != nil {

t.Fatal(err)

}

if _, err := tx.ExecContext(ctx, testutil.UpdateBarSetFoo); err != nil {

t.Fatal(err)

}

if err := tx.Commit(); err != nil {

t.Fatal(err)

}

requests = server.TestSpanner.DrainRequestsFromServer()

beginRequests = testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.BeginTransactionRequest{}))

if g, w := len(beginRequests), 0; g != w {

t.Fatalf("begin requests count mismatch\n Got: %v\nWant: %v", g, w)

}

executeRequests = testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.ExecuteSqlRequest{}))

if g, w := len(executeRequests), 1; g != w {

t.Fatalf("execute requests count mismatch\n Got: %v\nWant: %v", g, w)

}

request = executeRequests[0].(*spannerpb.ExecuteSqlRequest)

if request.GetTransaction() == nil || request.GetTransaction().GetBegin() == nil {

t.Fatalf("ExecuteSqlRequest should have a Begin transaction")

}

if g, w := request.GetTransaction().GetBegin().GetReadWrite().GetReadLockMode(), spannerpb.TransactionOptions_ReadWrite_READ_LOCK_MODE_UNSPECIFIED; g != w {

t.Fatalf("begin read lock mode mismatch\n Got: %v\nWant: %v", g, w)

}

}

func TestTimestampBound(t *testing.T) {

t.Parallel()

db, server, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

staleness := spanner.MaxStaleness(10 * time.Second)

row := db.QueryRowContext(ctx, testutil.SelectFooFromBar, ExecOptions{TimestampBound: &staleness})

if row.Err() != nil {

t.Fatal(row.Err())

}

var val int64

if err := row.Scan(&val); err != nil {

t.Fatal(err)

}

requests := server.TestSpanner.DrainRequestsFromServer()

executeRequests := testutil.RequestsOfType(requests, reflect.TypeOf(&spannerpb.ExecuteSqlRequest{}))

if g, w := len(executeRequests), 1; g != w {

t.Fatalf("execute requests count mismatch\n Got: %v\nWant: %v", g, w)

}

request := executeRequests[0].(*spannerpb.ExecuteSqlRequest)

if g, w := request.Transaction.GetSingleUse().GetReadOnly().GetMaxStaleness().GetSeconds(), int64(10); g != w {

t.Fatalf("read staleness mismatch\n Got: %v\nWant: %v", g, w)

}

}

func TestCreateDatabase(t *testing.T) {

t.Parallel()

ctx := context.Background()

db, server, teardown := setupTestDBConnection(t)

defer teardown()

var expectedResponse = &databasepb.Database{}

anyMsg, _ := anypb.New(expectedResponse)

server.TestDatabaseAdmin.SetResps([]proto.Message{

&longrunningpb.Operation{

Done: true,

Result: &longrunningpb.Operation_Response{Response: anyMsg},

Name: "test-operation",

},

})

conn, err := db.Conn(ctx)

if err != nil {

t.Fatal(err)

}

defer silentClose(conn)

if _, err = conn.ExecContext(ctx, "create database `foo`"); err != nil {

t.Fatalf("failed to execute CREATE DATABASE: %v", err)

}

requests := server.TestDatabaseAdmin.Reqs()

if g, w := len(requests), 1; g != w {

t.Fatalf("requests count mismatch\nGot: %v\nWant: %v", g, w)

}

if req, ok := requests[0].(*databasepb.CreateDatabaseRequest); ok {

if g, w := req.Parent, "projects/p/instances/i"; g != w {

t.Fatalf("parent mismatch\n Got: %v\nWant: %v", g, w)

}

} else {

t.Fatalf("request type mismatch, got %v", requests[0])

}

}

func TestCreateDatabaseWithExtraStatements(t *testing.T) {

t.Parallel()

ctx := context.Background()

db, server, teardown := setupTestDBConnection(t)

defer teardown()

var expectedResponse = &databasepb.Database{}

anyMsg, _ := anypb.New(expectedResponse)

server.TestDatabaseAdmin.SetResps([]proto.Message{

&longrunningpb.Operation{

Done: true,

Result: &longrunningpb.Operation_Response{Response: anyMsg},

Name: "test-operation",

},

})

conn, err := db.Conn(ctx)

if err != nil {

t.Fatal(err)

}

defer silentClose(conn)

statements := []string{

"start batch ddl",

"create database `foo`",

"create table my_table (id int64 primary key, value string(max))",

"create index my_index on my_table (value)",

"run batch",

}

for _, stmt := range statements {

if _, err := conn.ExecContext(ctx, stmt); err != nil {

t.Fatalf("failed to execute %q: %v", stmt, err)

}

}

requests := server.TestDatabaseAdmin.Reqs()

if g, w := len(requests), 1; g != w {

t.Fatalf("requests count mismatch\nGot: %v\nWant: %v", g, w)

}

if req, ok := requests[0].(*databasepb.CreateDatabaseRequest); ok {

if g, w := req.Parent, "projects/p/instances/i"; g != w {

t.Fatalf("parent mismatch\n Got: %v\nWant: %v", g, w)

}

if g, w := len(req.ExtraStatements), 2; g != w {

t.Fatalf("extra statements count mismatch\n Got: %v\nWant: %v", g, w)

} else {

wantExtraStatements := []string{

"create table my_table (id int64 primary key, value string(max))",

"create index my_index on my_table (value)",

}

if !reflect.DeepEqual(req.ExtraStatements, wantExtraStatements) {

t.Fatalf("extra statements mismatch\n Got: %v\nWant: %v", req.ExtraStatements, wantExtraStatements)

}

}

} else {

t.Fatalf("request type mismatch, got %v", requests[0])

}

}

func TestDropDatabase(t *testing.T) {

t.Parallel()

ctx := context.Background()

db, server, teardown := setupTestDBConnection(t)

defer teardown()

server.TestDatabaseAdmin.SetResps([]proto.Message{&emptypb.Empty{}})

conn, err := db.Conn(ctx)

if err != nil {

t.Fatal(err)

}

defer silentClose(conn)

if _, err = conn.ExecContext(ctx, "drop database foo"); err != nil {

t.Fatalf("failed to execute DROP DATABASE: %v", err)

}

requests := server.TestDatabaseAdmin.Reqs()

if g, w := len(requests), 1; g != w {

t.Fatalf("requests count mismatch\nGot: %v\nWant: %v", g, w)

}

if req, ok := requests[0].(*databasepb.DropDatabaseRequest); ok {

if g, w := req.Database, "projects/p/instances/i/databases/foo"; g != w {

t.Fatalf("database name mismatch\n Got: %v\nWant: %v", g, w)

}

} else {

t.Fatalf("request type mismatch, got %v", requests[0])

}

}

func TestDDLUsingQueryContext(t *testing.T) {

t.Parallel()

db, server, teardown := setupTestDBConnection(t)

defer teardown()

var expectedResponse = &emptypb.Empty{}

anyMsg, _ := anypb.New(expectedResponse)

server.TestDatabaseAdmin.SetResps([]proto.Message{

&longrunningpb.Operation{

Done: true,

Result: &longrunningpb.Operation_Response{Response: anyMsg},

Name: "test-operation",

},

})

ctx := context.Background()

// DDL statements should be able to use QueryContext.

if it, err := db.QueryContext(ctx, "CREATE TABLE Foo (Bar STRING(100))"); err != nil {

t.Fatal(err)

} else {

if it.Next() {

t.Fatalf("DDL should not return any rows")

}

}

}

func TestDDLUsingQueryContextInReadOnlyTx(t *testing.T) {

t.Parallel()

db, _, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

tx, err := db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})

if err != nil {

t.Fatal(err)

}

defer func() { _ = tx.Rollback() }()

// DDL statements should not use the query context in a read-only transaction.

_, err = tx.QueryContext(ctx, "CREATE TABLE Foo (Bar STRING(100))")

if err == nil {

t.Fatal("expected error for DDL statement using QueryContext in read-only transaction, got nil")

}

if g, w := err.Error(), `spanner: code = "FailedPrecondition", desc = "cannot execute DDL as part of a transaction"`; g != w {

t.Fatalf("error mismatch\n Got: %v\nWant: %v", g, w)

}

}

func TestDDLUsingQueryContextInReadWriteTransaction(t *testing.T) {

t.Parallel()

db, _, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

tx, err := db.BeginTx(ctx, &sql.TxOptions{})

if err != nil {

t.Fatal(err)

}

defer func() { _ = tx.Rollback() }()

// DDL statements should not use the query context in a read-write transaction.

_, err = tx.QueryContext(ctx, "CREATE TABLE Foo (Bar STRING(100))")

if err == nil {

t.Fatal("expected error for DDL statement using QueryContext in read-write transaction, got nil")

}

if g, w := err.Error(), `spanner: code = "FailedPrecondition", desc = "cannot execute DDL as part of a transaction"`; g != w {

t.Fatalf("error mismatch\n Got: %v\nWant: %v", g, w)

}

}

func TestRunDmlBatch(t *testing.T) {

t.Parallel()

db, _, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

conn, err := db.Conn(ctx)

if err != nil {

t.Fatal(err)

}

defer silentClose(conn)

if err := conn.Raw(func(driverConn interface{}) error {

spannerConn, _ := driverConn.(SpannerConn)

return spannerConn.StartBatchDML()

}); err != nil {

t.Fatal(err)

}

// Buffer two DML statements.

for range 2 {

if _, err := conn.ExecContext(ctx, testutil.UpdateBarSetFoo); err != nil {

t.Fatal(err)

}

}

var res SpannerResult

if err := conn.Raw(func(driverConn interface{}) (err error) {

spannerConn, _ := driverConn.(SpannerConn)

res, err = spannerConn.RunDmlBatch(ctx)

return err

}); err != nil {

t.Fatal(err)

}

affected, err := res.BatchRowsAffected()

if err != nil {

t.Fatal(err)

}

if g, w := affected, []int64{testutil.UpdateBarSetFooRowCount, testutil.UpdateBarSetFooRowCount}; !reflect.DeepEqual(g, w) {

t.Fatalf("affected mismatch\n Got: %v\nWant: %v", g, w)

}

}

func TestSetRetryAbortsInternallyInInactiveTransaction(t *testing.T) {

t.Parallel()

db, _, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

tx, err := db.BeginTx(ctx, &sql.TxOptions{})

if err != nil {

t.Fatal(err)

}

if _, err := tx.ExecContext(ctx, "set retry_aborts_internally = false"); err != nil {

t.Fatal(err)

}

_ = tx.Rollback()

}

func TestSetRetryAbortsInternallyInActiveTransaction(t *testing.T) {

t.Parallel()

db, _, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()

tx, err := db.BeginTx(ctx, &sql.TxOptions{})

if err != nil {

t.Fatal(err)

}

if _, err := tx.ExecContext(ctx, testutil.UpdateBarSetFoo); err != nil {

t.Fatal(err)

}

// It is (now) allowed to change this value during a transaction, but it won't have an effect on the

// running transaction.

if _, err := tx.ExecContext(ctx, "set retry_aborts_internally = false"); err != nil {

t.Fatal(err)

}

_ = tx.Rollback()

}

func TestSetLocalRetryAbortsInternally(t *testing.T) {

t.Parallel()

db, server, teardown := setupTestDBConnection(t)

defer teardown()

ctx := context.Background()