func readSchema() ([]*schema.Migration, error) {
	const filename = "schema.sql"
	contents, err := ioutil.ReadFile(filename)
	if err != nil {
		return nil, fmt.Errorf("failed to read sql file from '%s': %w", filename, err)
	}
	return []*schema.Migration{{
		ID:     time.Now().Local().String() + " db schema",
		Script: string(contents),
	}}, nil
}
readSchema loads the indexing database schema file
package psql
import (
	"context"
	"database/sql"
	"flag"
	"fmt"
	"io/ioutil"
	"log"
	"os"
	"os/signal"
	"testing"
	"time"
	"github.com/adlio/schema"
	"github.com/gogo/protobuf/proto"
	"github.com/ory/dockertest"
	"github.com/ory/dockertest/docker"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	abci "github.com/tendermint/tendermint/abci/types"
	"github.com/tendermint/tendermint/internal/state/indexer"
	"github.com/tendermint/tendermint/types"
	_ "github.com/lib/pq"
)
var _ indexer.EventSink = (*EventSink)(nil)
var (
	doPauseAtExit = flag.Bool("pause-at-exit", false,
		"If true, pause the test until interrupted at shutdown, to allow debugging")
	testDB func() *sql.DB
)
const (
	user     = "postgres"
	password = "secret"
	port     = "5432"
	dsn      = "postgres://%s:%s@localhost:%s/%s?sslmode=disable"
	dbName   = "postgres"
	chainID  = "test-chainID"
	viewBlockEvents = "block_events"
	viewTxEvents    = "tx_events"
)
func TestMain(m *testing.M) {
	flag.Parse()
	pool, err := dockertest.NewPool(os.Getenv("DOCKER_URL"))
	if err != nil {
		log.Fatalf("Creating docker pool: %v", err)
	}
	resource, err := pool.RunWithOptions(&dockertest.RunOptions{
		Repository: "postgres",
		Tag:        "13",
		Env: []string{
			"POSTGRES_USER=" + user,
			"POSTGRES_PASSWORD=" + password,
			"POSTGRES_DB=" + dbName,
			"listen_addresses = '*'",
		},
		ExposedPorts: []string{port},
	}, func(config *docker.HostConfig) {
		config.AutoRemove = true
		config.RestartPolicy = docker.RestartPolicy{
			Name: "no",
		}
	})
	if err != nil {
		log.Fatalf("Starting docker pool: %v", err)
	}
	if *doPauseAtExit {
		log.Print("Pause at exit is enabled, containers will not expire")
	} else {
		const expireSeconds = 60
		_ = resource.Expire(expireSeconds)
		log.Printf("Container expiration set to %d seconds", expireSeconds)
	}
	conn := fmt.Sprintf(dsn, user, password, resource.GetPort(port+"/tcp"), dbName)
	var db *sql.DB
	if err := pool.Retry(func() error {
		sink, err := NewEventSink(conn, chainID)
		if err != nil {
			return err
		}
		db = sink.DB()  
		return db.Ping()
	}); err != nil {
		log.Fatalf("Connecting to database: %v", err)
	}
	if err := resetDatabase(db); err != nil {
		log.Fatalf("Flushing database: %v", err)
	}
	sm, err := readSchema()
	if err != nil {
		log.Fatalf("Reading schema: %v", err)
	} else if err := schema.NewMigrator().Apply(db, sm); err != nil {
		log.Fatalf("Applying schema: %v", err)
	}
	testDB = func() *sql.DB { return db }
	code := m.Run()
	if *doPauseAtExit {
		log.Print("Testing complete, pausing for inspection. Send SIGINT to resume teardown")
		waitForInterrupt()
		log.Print("(resuming)")
	}
	log.Print("Shutting down database")
	if err := pool.Purge(resource); err != nil {
		log.Printf("WARNING: Purging pool failed: %v", err)
	}
	if err := db.Close(); err != nil {
		log.Printf("WARNING: Closing database failed: %v", err)
	}
	os.Exit(code)
}
func TestType(t *testing.T) {
	psqlSink := &EventSink{store: testDB(), chainID: chainID}
	assert.Equal(t, indexer.PSQL, psqlSink.Type())
}
func TestIndexing(t *testing.T) {
	t.Run("IndexBlockEvents", func(t *testing.T) {
		indexer := &EventSink{store: testDB(), chainID: chainID}
		require.NoError(t, indexer.IndexBlockEvents(newTestBlockHeader()))
		verifyBlock(t, 1)
		verifyBlock(t, 2)
		verifyNotImplemented(t, "hasBlock", func() (bool, error) { return indexer.HasBlock(1) })
		verifyNotImplemented(t, "hasBlock", func() (bool, error) { return indexer.HasBlock(2) })
		verifyNotImplemented(t, "block search", func() (bool, error) {
			v, err := indexer.SearchBlockEvents(context.Background(), nil)
			return v != nil, err
		})
		require.NoError(t, verifyTimeStamp(tableBlocks))
		require.NoError(t, indexer.IndexBlockEvents(newTestBlockHeader()))
	})
	t.Run("IndexTxEvents", func(t *testing.T) {
		indexer := &EventSink{store: testDB(), chainID: chainID}
		txResult := txResultWithEvents([]abci.Event{
			makeIndexedEvent("account.number", "1"),
			makeIndexedEvent("account.owner", "Ivan"),
			makeIndexedEvent("account.owner", "Yulieta"),
			{Type: "", Attributes: []abci.EventAttribute{{Key: "not_allowed", Value: "Vlad", Index: true}}},
		})
		require.NoError(t, indexer.IndexTxEvents([]*abci.TxResult{txResult}))
		txr, err := loadTxResult(types.Tx(txResult.Tx).Hash())
		require.NoError(t, err)
		assert.Equal(t, txResult, txr)
		require.NoError(t, verifyTimeStamp(tableTxResults))
		require.NoError(t, verifyTimeStamp(viewTxEvents))
		verifyNotImplemented(t, "getTxByHash", func() (bool, error) {
			txr, err := indexer.GetTxByHash(types.Tx(txResult.Tx).Hash())
			return txr != nil, err
		})
		verifyNotImplemented(t, "tx search", func() (bool, error) {
			txr, err := indexer.SearchTxEvents(context.Background(), nil)
			return txr != nil, err
		})
		err = indexer.IndexTxEvents([]*abci.TxResult{txResult})
		require.NoError(t, err)
	})
}
func TestStop(t *testing.T) {
	indexer := &EventSink{store: testDB()}
	require.NoError(t, indexer.Stop())
}
func newTestBlockHeader() types.EventDataNewBlockHeader {
	return types.EventDataNewBlockHeader{
		Header: types.Header{Height: 1},
		ResultBeginBlock: abci.ResponseBeginBlock{
			Events: []abci.Event{
				makeIndexedEvent("begin_event.proposer", "FCAA001"),
				makeIndexedEvent("thingy.whatzit", "O.O"),
			},
		},
		ResultEndBlock: abci.ResponseEndBlock{
			Events: []abci.Event{
				makeIndexedEvent("end_event.foo", "100"),
				makeIndexedEvent("thingy.whatzit", "-.O"),
			},
		},
	}
}