Merge pull request #48 from lukaszbudnik/verify-migrations-action

Verify migrations action

core/core.go

@@ -2,14 +2,15 @@ package core
 
 import (
 	"fmt"
+	"log"
+
 	"github.com/lukaszbudnik/migrator/config"
 	"github.com/lukaszbudnik/migrator/db"
 	"github.com/lukaszbudnik/migrator/loader"
 	"github.com/lukaszbudnik/migrator/migrations"
 	"github.com/lukaszbudnik/migrator/notifications"
 	"github.com/lukaszbudnik/migrator/types"
 	"github.com/lukaszbudnik/migrator/utils"
-	"log"
 )
 
 const (
@@ -45,7 +46,6 @@ type ExecuteFlags struct {
 func GetDiskMigrations(config *config.Config, createLoader func(*config.Config) loader.Loader) []types.Migration {
 	loader := createLoader(config)
 	diskMigrations := loader.GetDiskMigrations()
-	log.Printf("Read disk migrations: %d", len(diskMigrations))
 	return diskMigrations
 }
 
@@ -56,7 +56,6 @@ func GetDBTenants(config *config.Config, createConnector func(*config.Config) db
 	connector.Init()
 	defer connector.Dispose()
 	dbTenants := connector.GetTenants()
-	log.Printf("Read DB tenants: %d", len(dbTenants))
 	return dbTenants
 }
 
@@ -67,18 +66,21 @@ func GetDBMigrations(config *config.Config, createConnector func(*config.Config)
 	connector.Init()
 	defer connector.Dispose()
 	dbMigrations := connector.GetDBMigrations()
-	log.Printf("Read DB migrations: %d", len(dbMigrations))
 	return dbMigrations
 }
 
 // ApplyMigrations is a function which applies disk migrations to DB as defined in config passed as first argument
 // and using connector created by a function passed as second argument and disk loader created by a function passed as third argument
 func ApplyMigrations(config *config.Config, createConnector func(*config.Config) db.Connector, createLoader func(*config.Config) loader.Loader) []types.Migration {
 	diskMigrations := GetDiskMigrations(config, createLoader)
+	log.Printf("Read disk migrations: %d", len(diskMigrations))
+
 	dbMigrations := GetDBMigrations(config, createConnector)
-	migrationsToApply := migrations.ComputeMigrationsToApply(diskMigrations, dbMigrations)
+	log.Printf("Read DB migrations: %d", len(dbMigrations))
 
+	migrationsToApply := migrations.ComputeMigrationsToApply(diskMigrations, dbMigrations)
 	log.Printf("Found migrations to apply: %d", len(migrationsToApply))
+
 	doApplyMigrations(migrationsToApply, config, createConnector)
 
 	notifier := notifications.CreateNotifier(config)
@@ -98,12 +100,13 @@ func ApplyMigrations(config *config.Config, createConnector func(*config.Config)
 func AddTenant(tenant string, config *config.Config, createConnector func(*config.Config) db.Connector, createLoader func(*config.Config) loader.Loader) []types.Migration {
 
 	diskMigrations := GetDiskMigrations(config, createLoader)
+	log.Printf("Read disk migrations: %d", len(diskMigrations))
 
 	// filter only tenant schemas
 	// var migrationsToApply []types.Migration
 	migrationsToApply := migrations.FilterTenantMigrations(diskMigrations)
-
 	log.Printf("Found migrations to apply: %d", len(migrationsToApply))
+
 	doAddTenantAndApplyMigrations(tenant, migrationsToApply, config, createConnector)
 
 	notifier := notifications.CreateNotifier(config)
@@ -119,6 +122,14 @@ func AddTenant(tenant string, config *config.Config, createConnector func(*confi
 	return diskMigrations
 }
 
+// VerifyMigrations loads disk and db migrations and verifies their checksums
+// see migrations.VerifyCheckSums for more information
+func VerifyMigrations(config *config.Config, createConnector func(*config.Config) db.Connector, createLoader func(*config.Config) loader.Loader) (bool, []types.Migration) {
+	diskMigrations := GetDiskMigrations(config, createLoader)
+	dbMigrations := GetDBMigrations(config, createConnector)
+	return migrations.VerifyCheckSums(diskMigrations, dbMigrations)
+}
+
 // ExecuteMigrator is a function which executes actions on resources defined in config passed as first argument action defined as second argument
 // and using connector created by a function passed as third argument and disk loader created by a function passed as fourth argument
 func ExecuteMigrator(config *config.Config, executeFlags ExecuteFlags) {
@@ -136,20 +147,34 @@ func doExecuteMigrator(config *config.Config, executeFlags ExecuteFlags, createC
 		}
 	case GetDBMigrationsAction:
 		dbMigrations := GetDBMigrations(config, createConnector)
+		log.Printf("Read DB migrations: %d", len(dbMigrations))
 		if len(dbMigrations) > 0 {
 			log.Printf("List of db migrations\n%v", utils.MigrationDBArrayToString(dbMigrations))
 		}
 	case AddTenantAction:
-		AddTenant(executeFlags.Tenant, config, createConnector, createLoader)
+		verified, offendingMigrations := VerifyMigrations(config, createConnector, createLoader)
+		if !verified {
+			log.Printf("Checksum verification failed.")
+			log.Printf("List of offending disk migrations\n%v", utils.MigrationArrayToString(offendingMigrations))
+		} else {
+			AddTenant(executeFlags.Tenant, config, createConnector, createLoader)
+		}
 	case GetDBTenantsAction:
 		dbTenants := GetDBTenants(config, createConnector)
+		log.Printf("Read DB tenants: %d", len(dbTenants))
 		if len(dbTenants) > 0 {
 			log.Printf("List of db tenants\n%v", utils.TenantArrayToString(dbTenants))
 		}
 	case ApplyAction:
-		migrationsApplied := ApplyMigrations(config, createConnector, createLoader)
-		if len(migrationsApplied) > 0 {
-			log.Printf("List of migrations applied\n%v", utils.MigrationArrayToString(migrationsApplied))
+		verified, offendingMigrations := VerifyMigrations(config, createConnector, createLoader)
+		if !verified {
+			log.Printf("Checksum verification failed.")
+			log.Printf("List of offending disk migrations\n%v", utils.MigrationArrayToString(offendingMigrations))
+		} else {
+			migrationsApplied := ApplyMigrations(config, createConnector, createLoader)
+			if len(migrationsApplied) > 0 {
+				log.Printf("List of migrations applied\n%v", utils.MigrationArrayToString(migrationsApplied))
+			}
 		}
 	}
 }

core/core_mocks.go

@@ -1,26 +1,39 @@
 package core
 
 import (
+	"time"
+
 	"github.com/lukaszbudnik/migrator/config"
 	"github.com/lukaszbudnik/migrator/db"
 	"github.com/lukaszbudnik/migrator/loader"
 	"github.com/lukaszbudnik/migrator/types"
-	"time"
 )
 
 type mockedDiskLoader struct {
 }
 
 func (m *mockedDiskLoader) GetDiskMigrations() []types.Migration {
-	m1 := types.MigrationDefinition{Name: "201602220000.sql", SourceDir: "source", File: "source/201602220000.sql", MigrationType: types.MigrationTypeSingleSchema}
-	m2 := types.MigrationDefinition{Name: "201602220001.sql", SourceDir: "source", File: "source/201602220001.sql", MigrationType: types.MigrationTypeSingleSchema}
-	return []types.Migration{{MigrationDefinition: m1, Contents: "select abc"}, {MigrationDefinition: m2, Contents: "select def"}}
+	m1 := types.Migration{Name: "201602220000.sql", SourceDir: "source", File: "source/201602220000.sql", MigrationType: types.MigrationTypeSingleSchema, Contents: "select abc", CheckSum: "abc"}
+	m2 := types.Migration{Name: "201602220001.sql", SourceDir: "source", File: "source/201602220001.sql", MigrationType: types.MigrationTypeSingleSchema, Contents: "select def", CheckSum: "def"}
+	return []types.Migration{m1, m2}
 }
 
 func createMockedDiskLoader(config *config.Config) loader.Loader {
 	return new(mockedDiskLoader)
 }
 
+type mockedBrokenCheckSumDiskLoader struct {
+}
+
+func (m *mockedBrokenCheckSumDiskLoader) GetDiskMigrations() []types.Migration {
+	m1 := types.Migration{Name: "201602220000.sql", SourceDir: "source", File: "source/201602220000.sql", MigrationType: types.MigrationTypeSingleSchema, Contents: "select abc", CheckSum: "xxx"}
+	return []types.Migration{m1}
+}
+
+func createBrokenCheckSumMockedDiskLoader(config *config.Config) loader.Loader {
+	return new(mockedBrokenCheckSumDiskLoader)
+}
+
 type mockedConnector struct {
 }
 
@@ -54,9 +67,9 @@ func (m *mockedConnector) AddTenantAndApplyMigrations(string, []types.Migration)
 }
 
 func (m *mockedConnector) GetDBMigrations() []types.MigrationDB {
-	m1 := types.MigrationDefinition{Name: "201602220000.sql", SourceDir: "source", File: "source/201602220000.sql", MigrationType: types.MigrationTypeSingleSchema}
+	m1 := types.Migration{Name: "201602220000.sql", SourceDir: "source", File: "source/201602220000.sql", MigrationType: types.MigrationTypeSingleSchema, CheckSum: "abc"}
 	d1 := time.Date(2016, 02, 22, 16, 41, 1, 123, time.UTC)
-	ms := []types.MigrationDB{{MigrationDefinition: m1, Schema: "source", Created: d1}}
+	ms := []types.MigrationDB{{Migration: m1, Schema: "source", Created: d1}}
 
 	return ms
 }

core/core_test.go

@@ -1,15 +1,10 @@
-// These are integration tests.
-// The following tests must be working in order to get this one working:
-// * config_test.go
-// * migrations_test.go
-// DB & Disk operations are mocked using xcli_mocks.go
-
 package core
 
 import (
+	"testing"
+
 	"github.com/lukaszbudnik/migrator/config"
 	"github.com/stretchr/testify/assert"
-	"testing"
 )
 
 const (
@@ -57,10 +52,26 @@ func TestApplyMigrations(t *testing.T) {
 	doExecuteMigrator(config, executeFlags, createMockedConnector, createMockedDiskLoader)
 }
 
+func TestApplyMigrationsVerificationFailed(t *testing.T) {
+	config, err := config.FromFile(configFile)
+	assert.Nil(t, err)
+	executeFlags := ExecuteFlags{}
+	executeFlags.Action = ApplyAction
+	doExecuteMigrator(config, executeFlags, createMockedConnector, createBrokenCheckSumMockedDiskLoader)
+}
+
 func TestAddTenant(t *testing.T) {
 	config, err := config.FromFile(configFile)
 	assert.Nil(t, err)
 	executeFlags := ExecuteFlags{}
 	executeFlags.Action = AddTenantAction
 	doExecuteMigrator(config, executeFlags, createMockedConnector, createMockedDiskLoader)
 }
+
+func TestAddTenantVerificationFailed(t *testing.T) {
+	config, err := config.FromFile(configFile)
+	assert.Nil(t, err)
+	executeFlags := ExecuteFlags{}
+	executeFlags.Action = AddTenantAction
+	doExecuteMigrator(config, executeFlags, createMockedConnector, createBrokenCheckSumMockedDiskLoader)
+}

coverage.sh

@@ -17,5 +17,5 @@ do
     continue
   fi
   go test -race -covermode=atomic -coverprofile=coverage-$package.txt ./$package
-  cat coverage-$package.txt | sed '/^mode/d' >> coverage.txt
+  cat coverage-$package.txt | sed '/^mode/d' | sed '/_mocks.go/d' >> coverage.txt
 done

db/db.go

@@ -138,12 +138,14 @@ func (bc *BaseConnector) GetDBMigrations() []types.MigrationDB {
 			migrationType types.MigrationType
 			schema        string
 			created       time.Time
+			contents      string
+			checksum      string
 		)
-		if err := rows.Scan(&name, &sourceDir, &filename, &migrationType, &schema, &created); err != nil {
+		if err := rows.Scan(&name, &sourceDir, &filename, &migrationType, &schema, &created, &contents, &checksum); err != nil {
 			log.Panicf("Could not read DB migration: %v", err)
 		}
-		mdef := types.MigrationDefinition{Name: name, SourceDir: sourceDir, File: filename, MigrationType: migrationType}
-		dbMigrations = append(dbMigrations, types.MigrationDB{MigrationDefinition: mdef, Schema: schema, Created: created})
+		mdef := types.Migration{Name: name, SourceDir: sourceDir, File: filename, MigrationType: migrationType, Contents: contents, CheckSum: checksum}
+		dbMigrations = append(dbMigrations, types.MigrationDB{Migration: mdef, Schema: schema, Created: created})
 	}
 
 	return dbMigrations
@@ -252,7 +254,7 @@ func (bc *BaseConnector) applyMigrationsInTx(tx *sql.Tx, tenants []string, migra
 				log.Panicf("SQL failed, transaction rollback was called: %v %v", err, contents)
 			}
 
-			_, err = tx.Stmt(insert).Exec(m.Name, m.SourceDir, m.File, m.MigrationType, s)
+			_, err = tx.Stmt(insert).Exec(m.Name, m.SourceDir, m.File, m.MigrationType, s, m.Contents, m.CheckSum)
 			if err != nil {
 				tx.Rollback()
 				log.Panicf("Failed to add migration entry, transaction rollback was called: %v", err)

db/db_dialect.go

@@ -2,8 +2,9 @@ package db
 
 import (
 	"fmt"
-	"github.com/lukaszbudnik/migrator/config"
 	"log"
+
+	"github.com/lukaszbudnik/migrator/config"
 )
 
 // Dialect returns SQL statements for given DB
@@ -22,7 +23,7 @@ type BaseDialect struct {
 }
 
 const (
-	selectMigrationsSQL      = "select name, source_dir as sd, filename, type, db_schema, created from %v.%v order by name, source_dir"
+	selectMigrationsSQL      = "select name, source_dir as sd, filename, type, db_schema, created, contents, checksum from %v.%v order by name, source_dir"
 	selectTenantsSQL         = "select name from %v.%v"
 	createMigrationsTableSQL = `
 create table if not exists %v.%v (
@@ -32,7 +33,9 @@ create table if not exists %v.%v (
   filename varchar(200) not null,
   type int not null,
   db_schema varchar(200) not null,
-  created timestamp default now()
+  created timestamp default now(),
+	contents text,
+	checksum varchar(64)
 )
 `
 	createTenantsTableSQL = `

db/db_mssql.go

@@ -11,7 +11,7 @@ type msSQLDialect struct {
 }
 
 const (
-	insertMigrationMSSQLDialectSQL    = "insert into %v.%v (name, source_dir, filename, type, db_schema) values (@p1, @p2, @p3, @p4, @p5)"
+	insertMigrationMSSQLDialectSQL    = "insert into %v.%v (name, source_dir, filename, type, db_schema, contents, checksum) values (@p1, @p2, @p3, @p4, @p5, @p6, @p7)"
 	insertTenantMSSQLDialectSQL       = "insert into %v.%v (name) values (@p1)"
 	createTenantsTableMSSQLDialectSQL = `
 IF NOT EXISTS (select * from information_schema.tables where table_schema = '%v' and table_name = '%v')
@@ -33,7 +33,9 @@ BEGIN
     filename varchar(200) not null,
     type int not null,
     db_schema varchar(200) not null,
-    created datetime default CURRENT_TIMESTAMP
+    created datetime default CURRENT_TIMESTAMP,
+		contents text,
+		checksum varchar(64)
   );
 END
 `

db/db_mysql.go

@@ -11,7 +11,7 @@ type mySQLDialect struct {
 }
 
 const (
-	insertMigrationMySQLDialectSQL = "insert into %v.%v (name, source_dir, filename, type, db_schema) values (?, ?, ?, ?, ?)"
+	insertMigrationMySQLDialectSQL = "insert into %v.%v (name, source_dir, filename, type, db_schema, contents, checksum) values (?, ?, ?, ?, ?, ?, ?)"
 	insertTenantMySQLDialectSQL    = "insert into %v.%v (name) values (?)"
 )
 

db/db_postgresql.go

@@ -11,7 +11,7 @@ type postgreSQLDialect struct {
 }
 
 const (
-	insertMigrationPostgreSQLDialectSQL = "insert into %v.%v (name, source_dir, filename, type, db_schema) values ($1, $2, $3, $4, $5)"
+	insertMigrationPostgreSQLDialectSQL = "insert into %v.%v (name, source_dir, filename, type, db_schema, contents, checksum) values ($1, $2, $3, $4, $5, $6, $7)"
 	insertTenantPostgreSQLDialectSQL    = "insert into %v.%v (name) values ($1)"
 )