private/testplanet: allow modifying created database

Instead of providing the database from outside to testplanet create it
inside and then allow wrapping and modifying it. This is more convenient
to use.

Change-Id: I9b8f69e6e0a19ff984b4e2bfe927c9100c77bc6c

private/testplanet/planet.go

@@ -55,6 +55,12 @@ type Config struct {
 	NonParallel bool
 }
 
+// DatabaseConfig defines connection strings for database.
+type DatabaseConfig struct {
+	SatelliteDB        string
+	SatellitePointerDB string
+}
+
 // Planet is a full storj system setup.
 type Planet struct {
 	id        string
@@ -113,7 +119,7 @@ func (peer *closablePeer) Close() error {
 }
 
 // NewCustom creates a new full system with the specified configuration.
-func NewCustom(log *zap.Logger, config Config) (*Planet, error) {
+func NewCustom(log *zap.Logger, config Config, satelliteDatabases satellitedbtest.SatelliteDatabases) (*Planet, error) {
 	// Clear error in the beginning to avoid issues down the line.
 	if err := satellitedbtest.PostgresDefined(); err != nil {
 		return nil, err
@@ -158,7 +164,7 @@ func NewCustom(log *zap.Logger, config Config) (*Planet, error) {
 		return nil, errs.Combine(err, planet.Shutdown())
 	}
 
-	planet.Satellites, err = planet.newSatellites(config.SatelliteCount)
+	planet.Satellites, err = planet.newSatellites(config.SatelliteCount, satelliteDatabases)
 	if err != nil {
 		return nil, errs.Combine(err, planet.Shutdown())
 	}

private/testplanet/reconfigure.go

@@ -18,14 +18,15 @@ import (
 
 // Reconfigure allows to change node configurations
 type Reconfigure struct {
-	NewSatelliteDB        func(log *zap.Logger, index int) (satellite.DB, error)
+	SatelliteDB        func(log *zap.Logger, index int, db satellite.DB) (satellite.DB, error)
-	NewSatellitePointerDB func(log *zap.Logger, index int) (metainfo.PointerDB, error)
+	SatellitePointerDB func(log *zap.Logger, index int, db metainfo.PointerDB) (metainfo.PointerDB, error)
-	Satellite             func(log *zap.Logger, index int, config *satellite.Config)
+	Satellite          func(log *zap.Logger, index int, config *satellite.Config)
+
 	ReferralManagerServer func(log *zap.Logger) pbgrpc.ReferralManagerServer
 
-	NewStorageNodeDB func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error)
+	StorageNodeDB func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error)
-	StorageNode      func(index int, config *storagenode.Config)
+	StorageNode   func(index int, config *storagenode.Config)
-	UniqueIPCount    int
+	UniqueIPCount int
 
 	Identities func(log *zap.Logger, version storj.IDVersion) *testidentity.Identities
 }

private/testplanet/run.go

@@ -6,12 +6,9 @@ package testplanet
 import (
 	"testing"
 
-	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
 
 	"storj.io/common/testcontext"
-	"storj.io/storj/satellite"
-	"storj.io/storj/satellite/metainfo"
 	"storj.io/storj/satellite/satellitedb/satellitedbtest"
 )
 
@@ -31,19 +28,12 @@ func Run(t *testing.T, config Config, test func(t *testing.T, ctx *testcontext.C
 			if satelliteDB.MasterDB.URL == "" {
 				t.Skipf("Database %s connection string not provided. %s", satelliteDB.MasterDB.Name, satelliteDB.MasterDB.Message)
 			}
-
 			planetConfig := config
-			planetConfig.Reconfigure.NewSatelliteDB = func(log *zap.Logger, index int) (satellite.DB, error) {
+			if planetConfig.Name == "" {
-				return satellitedbtest.CreateMasterDB(ctx, t, "S", index, satelliteDB.MasterDB)
+				planetConfig.Name = t.Name()
 			}
 
-			if satelliteDB.PointerDB.URL != "" {
+			planet, err := NewCustom(zaptest.NewLogger(t), config, satelliteDB)
-				planetConfig.Reconfigure.NewSatellitePointerDB = func(log *zap.Logger, index int) (metainfo.PointerDB, error) {
-					return satellitedbtest.CreatePointerDB(ctx, t, "P", index, satelliteDB.PointerDB)
-				}
-			}
-
-			planet, err := NewCustom(zaptest.NewLogger(t), planetConfig)
 			if err != nil {
 				t.Fatalf("%+v", err)
 			}

private/testplanet/satellite.go

@@ -54,6 +54,7 @@ import (
 	"storj.io/storj/satellite/repair/checker"
 	"storj.io/storj/satellite/repair/irreparable"
 	"storj.io/storj/satellite/repair/repairer"
+	"storj.io/storj/satellite/satellitedb/satellitedbtest"
 	"storj.io/storj/satellite/vouchers"
 	"storj.io/storj/storage/redis/redisserver"
 )
@@ -229,7 +230,7 @@ func (system *Satellite) Run(ctx context.Context) (err error) {
 func (system *Satellite) PrivateAddr() string { return system.API.Server.PrivateAddr().String() }
 
 // newSatellites initializes satellites
-func (planet *Planet) newSatellites(count int) ([]*Satellite, error) {
+func (planet *Planet) newSatellites(count int, satelliteDatabases satellitedbtest.SatelliteDatabases) ([]*Satellite, error) {
 	var xs []*Satellite
 	defer func() {
 		for _, x := range xs {
@@ -251,33 +252,40 @@ func (planet *Planet) newSatellites(count int) ([]*Satellite, error) {
 			return nil, err
 		}
 
-		var db satellite.DB
+		db, err := satellitedbtest.CreateMasterDB(context.TODO(), log.Named("db"), planet.config.Name, "S", i, satelliteDatabases.MasterDB)
-		if planet.config.Reconfigure.NewSatelliteDB != nil {
-			db, err = planet.config.Reconfigure.NewSatelliteDB(log.Named("db"), i)
-		} else {
-			return nil, errs.New("NewSatelliteDB not defined")
-		}
 		if err != nil {
 			return nil, err
 		}
+
+		if planet.config.Reconfigure.SatelliteDB != nil {
+			var newdb satellite.DB
+			newdb, err = planet.config.Reconfigure.SatelliteDB(log.Named("db"), i, db)
+			if err != nil {
+				return nil, errs.Combine(err, db.Close())
+			}
+			db = newdb
+		}
 		planet.databases = append(planet.databases, db)
 
-		var pointerDB metainfo.PointerDB
+		pointerDB, err := satellitedbtest.CreatePointerDB(context.TODO(), log.Named("pointerdb"), planet.config.Name, "P", i, satelliteDatabases.PointerDB)
-		if planet.config.Reconfigure.NewSatellitePointerDB != nil {
-			pointerDB, err = planet.config.Reconfigure.NewSatellitePointerDB(log.Named("pointerdb"), i)
-		} else {
-			return nil, errs.New("NewSatellitePointerDB not defined")
-		}
 		if err != nil {
 			return nil, err
 		}
+
+		if planet.config.Reconfigure.SatellitePointerDB != nil {
+			var newPointerDB metainfo.PointerDB
+			newPointerDB, err = planet.config.Reconfigure.SatellitePointerDB(log.Named("pointerdb"), i, pointerDB)
+			if err != nil {
+				return nil, errs.Combine(err, pointerDB.Close())
+			}
+			pointerDB = newPointerDB
+		}
 		planet.databases = append(planet.databases, pointerDB)
 
 		redis, err := redisserver.Mini()
 		if err != nil {
 			return nil, err
 		}
-
 		planet.databases = append(planet.databases, redis)
 
 		config := satellite.Config{

private/testplanet/storagenode.go

@@ -197,8 +197,8 @@ func (planet *Planet) newStorageNodes(count int, whitelistedSatellites storj.Nod
 			return nil, err
 		}
 
-		if planet.config.Reconfigure.NewStorageNodeDB != nil {
+		if planet.config.Reconfigure.StorageNodeDB != nil {
-			db, err = planet.config.Reconfigure.NewStorageNodeDB(i, db, planet.log)
+			db, err = planet.config.Reconfigure.StorageNodeDB(i, db, planet.log)
 			if err != nil {
 				return nil, err
 			}

satellite/audit/reverify_test.go

@@ -1080,7 +1080,7 @@ func TestReverifySlowDownload(t *testing.T) {
 	testplanet.Run(t, testplanet.Config{
 		SatelliteCount: 1, StorageNodeCount: 4, UplinkCount: 1,
 		Reconfigure: testplanet.Reconfigure{
-			NewStorageNodeDB: func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error) {
+			StorageNodeDB: func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error) {
 				return testblobs.NewSlowDB(log.Named("slowdb"), db), nil
 			},
 			Satellite: func(log *zap.Logger, index int, config *satellite.Config) {
@@ -1179,7 +1179,7 @@ func TestReverifyUnknownError(t *testing.T) {
 	testplanet.Run(t, testplanet.Config{
 		SatelliteCount: 1, StorageNodeCount: 4, UplinkCount: 1,
 		Reconfigure: testplanet.Reconfigure{
-			NewStorageNodeDB: func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error) {
+			StorageNodeDB: func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error) {
 				return testblobs.NewBadDB(log.Named("baddb"), db), nil
 			},
 			Satellite: testplanet.ReconfigureRS(2, 2, 4, 4),

satellite/audit/verifier_test.go

@@ -285,7 +285,7 @@ func TestDownloadSharesDownloadTimeout(t *testing.T) {
 	testplanet.Run(t, testplanet.Config{
 		SatelliteCount: 1, StorageNodeCount: 1, UplinkCount: 1,
 		Reconfigure: testplanet.Reconfigure{
-			NewStorageNodeDB: func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error) {
+			StorageNodeDB: func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error) {
 				return testblobs.NewSlowDB(log.Named("slowdb"), db), nil
 			},
 		},
@@ -769,7 +769,7 @@ func TestVerifierSlowDownload(t *testing.T) {
 	testplanet.Run(t, testplanet.Config{
 		SatelliteCount: 1, StorageNodeCount: 4, UplinkCount: 1,
 		Reconfigure: testplanet.Reconfigure{
-			NewStorageNodeDB: func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error) {
+			StorageNodeDB: func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error) {
 				return testblobs.NewSlowDB(log.Named("slowdb"), db), nil
 			},
 			Satellite: func(log *zap.Logger, index int, config *satellite.Config) {
@@ -832,7 +832,7 @@ func TestVerifierUnknownError(t *testing.T) {
 	testplanet.Run(t, testplanet.Config{
 		SatelliteCount: 1, StorageNodeCount: 4, UplinkCount: 1,
 		Reconfigure: testplanet.Reconfigure{
-			NewStorageNodeDB: func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error) {
+			StorageNodeDB: func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error) {
 				return testblobs.NewBadDB(log.Named("baddb"), db), nil
 			},
 			Satellite: testplanet.ReconfigureRS(2, 2, 4, 4),

satellite/gracefulexit/endpoint_test.go

@@ -241,7 +241,7 @@ func TestRecvTimeout(t *testing.T) {
 		StorageNodeCount: successThreshold + 1,
 		UplinkCount:      1,
 		Reconfigure: testplanet.Reconfigure{
-			NewStorageNodeDB: func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error) {
+			StorageNodeDB: func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error) {
 				return testblobs.NewSlowDB(log.Named("slowdb"), db), nil
 			},
 			Satellite: func(logger *zap.Logger, index int, config *satellite.Config) {

satellite/metainfo/piecedeletion/service_test.go

@@ -314,7 +314,7 @@ func TestService_DeletePieces_Timeout(t *testing.T) {
 	testplanet.Run(t, testplanet.Config{
 		SatelliteCount: 1, StorageNodeCount: 4, UplinkCount: 1,
 		Reconfigure: testplanet.Reconfigure{
-			NewStorageNodeDB: func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error) {
+			StorageNodeDB: func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error) {
 				return testblobs.NewSlowDB(log.Named("slowdb"), db), nil
 			},
 			Satellite: func(log *zap.Logger, index int, config *satellite.Config) {

satellite/satellitedb/satellitedbtest/run.go

@@ -7,6 +7,7 @@ package satellitedbtest
 
 import (
 	"context"
+	"fmt"
 	"strconv"
 	"strings"
 	"testing"
@@ -95,23 +96,21 @@ func (db *tempMasterDB) TestDBAccess() *dbx.DB {
 }
 
 // CreateMasterDB creates a new satellite database for testing
-func CreateMasterDB(ctx context.Context, t testing.TB, category string, index int, dbInfo Database) (db satellite.DB, err error) {
+func CreateMasterDB(ctx context.Context, log *zap.Logger, name string, category string, index int, dbInfo Database) (db satellite.DB, err error) {
 	if dbInfo.URL == "" {
-		t.Fatalf("Database %s connection string not provided. %s", dbInfo.Name, dbInfo.Message)
+		return nil, fmt.Errorf("Database %s connection string not provided. %s", dbInfo.Name, dbInfo.Message)
 	}
 
 	schemaSuffix := SchemaSuffix()
-	t.Log("schema-suffix ", schemaSuffix)
+	log.Debug("creating", zap.String("suffix", schemaSuffix))
-
+	schema := SchemaName(name, category, index, schemaSuffix)
-	log := zaptest.NewLogger(t)
-	schema := SchemaName(t.Name(), category, index, schemaSuffix)
 
 	tempDB, err := tempdb.OpenUnique(ctx, dbInfo.URL, schema)
 	if err != nil {
 		return nil, err
 	}
 
-	return CreateMasterDBOnTopOf(log.Named("db"), tempDB)
+	return CreateMasterDBOnTopOf(log, tempDB)
 }
 
 // CreateMasterDBOnTopOf creates a new satellite database on top of an already existing
@@ -133,23 +132,22 @@ func (db *tempPointerDB) Close() error {
 }
 
 // CreatePointerDB creates a new satellite pointer database for testing
-func CreatePointerDB(ctx context.Context, t testing.TB, category string, index int, dbInfo Database) (db metainfo.PointerDB, err error) {
+func CreatePointerDB(ctx context.Context, log *zap.Logger, name string, category string, index int, dbInfo Database) (db metainfo.PointerDB, err error) {
 	if dbInfo.URL == "" {
-		t.Fatalf("Database %s connection string not provided. %s", dbInfo.Name, dbInfo.Message)
+		return nil, fmt.Errorf("Database %s connection string not provided. %s", dbInfo.Name, dbInfo.Message)
 	}
 
 	schemaSuffix := SchemaSuffix()
-	t.Log("schema-suffix ", schemaSuffix)
+	log.Debug("creating", zap.String("suffix", schemaSuffix))
 
-	log := zaptest.NewLogger(t)
+	schema := SchemaName(name, category, index, schemaSuffix)
-	schema := SchemaName(t.Name(), category, index, schemaSuffix)
 
 	tempDB, err := tempdb.OpenUnique(ctx, dbInfo.URL, schema)
 	if err != nil {
 		return nil, err
 	}
 
-	return CreatePointerDBOnTopOf(ctx, log.Named("pointerdb"), tempDB)
+	return CreatePointerDBOnTopOf(ctx, log, tempDB)
 }
 
 // CreatePointerDBOnTopOf creates a new satellite database on top of an already existing
@@ -174,7 +172,7 @@ func Run(t *testing.T, test func(ctx *testcontext.Context, t *testing.T, db sate
 				t.Skipf("Database %s connection string not provided. %s", dbInfo.MasterDB.Name, dbInfo.MasterDB.Message)
 			}
 
-			db, err := CreateMasterDB(ctx, t, "T", 0, dbInfo.MasterDB)
+			db, err := CreateMasterDB(ctx, zaptest.NewLogger(t), t.Name(), "T", 0, dbInfo.MasterDB)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -208,7 +206,7 @@ func Bench(b *testing.B, bench func(b *testing.B, db satellite.DB)) {
 			ctx := testcontext.New(b)
 			defer ctx.Cleanup()
 
-			db, err := CreateMasterDB(ctx, b, "X", 0, dbInfo.MasterDB)
+			db, err := CreateMasterDB(ctx, zap.NewNop(), b.Name(), "X", 0, dbInfo.MasterDB)
 			if err != nil {
 				b.Fatal(err)
 			}

storagenode/gracefulexit/worker_test.go

@@ -101,7 +101,7 @@ func TestWorkerTimeout(t *testing.T) {
 		StorageNodeCount: successThreshold + 1,
 		UplinkCount:      1,
 		Reconfigure: testplanet.Reconfigure{
-			NewStorageNodeDB: func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error) {
+			StorageNodeDB: func(index int, db storagenode.DB, log *zap.Logger) (storagenode.DB, error) {
 				return testblobs.NewSlowDB(log.Named("slowdb"), db), nil
 			},
 			Satellite: testplanet.ReconfigureRS(2, 3, successThreshold, successThreshold),