Tests: integration test for database (#1)

* Tests: integration test for database

* Fix: remove unused test

* Fix: close fixtures database connection

* Fix: close database connection

* Upgrade libraries and change ORM

cmd/metadata/channel.go

@@ -125,7 +125,7 @@ func (channel Channel) parseToken(ctx context.Context, token *pb.Token) error {
 }
 
 func (channel Channel) saveToken(ctx context.Context, metadata storage.TokenMetadata) error {
-	tx, err := channel.storage.Transactable.BeginTransaction(ctx)
+	tx, err := postgres.BeginTransaction(ctx, channel.storage.Transactable)
 	if err != nil {
 		return err
 	}
@@ -136,14 +136,11 @@ func (channel Channel) saveToken(ctx context.Context, metadata storage.TokenMeta
 	}
 
 	if metadata.ContractID > 0 {
-		if _, err := tx.Exec(
-			ctx,
-			`INSERT INTO address (id, hash) VALUES (?,?) ON CONFLICT (id) DO NOTHING`,
-			metadata.Contract.ID,
-			metadata.Contract.Hash,
-		); err != nil {
+		if err := tx.SaveAddress(ctx, &storage.Address{
+			ID:   metadata.Contract.ID,
+			Hash: metadata.Contract.Hash,
+		}); err != nil {
 			return tx.HandleError(ctx, err)
-
 		}
 	}
 

cmd/metadata/filler.go

@@ -337,6 +337,10 @@ func handlerFillerError(err error) error {
 }
 
 func (f Filler) getJsonSchema(ctx context.Context, hash []byte) (abi.JsonSchema, error) {
+	log.Info().
+		Hex("contract", hash).
+		Msg("try to get json schema")
+
 	cacheKey := fmt.Sprintf("%x:schema", hash)
 	item, err := f.cache.Fetch(cacheKey, time.Hour, func() (interface{}, error) {
 		var intSchema abi.JsonSchema

cmd/metadata/main.go

@@ -5,6 +5,7 @@ import (
 	"os"
 	"os/signal"
 	"runtime"
+	"strconv"
 	"syscall"
 
 	ipfs "github.com/dipdup-io/ipfs-tools"
@@ -50,6 +51,29 @@ func main() {
 	}
 	runtime.GOMAXPROCS(cfg.Metadata.MaxCPU)
 
+	if cfg.LogLevel == "" {
+		cfg.LogLevel = zerolog.LevelInfoValue
+	}
+
+	logLevel, err := zerolog.ParseLevel(cfg.LogLevel)
+	if err != nil {
+		log.Panic().Err(err).Msg("parsing log level")
+		return
+	}
+	zerolog.SetGlobalLevel(logLevel)
+	zerolog.CallerMarshalFunc = func(pc uintptr, file string, line int) string {
+		short := file
+		for i := len(file) - 1; i > 0; i-- {
+			if file[i] == '/' {
+				short = file[i+1:]
+				break
+			}
+		}
+		file = short
+		return file + ":" + strconv.Itoa(line)
+	}
+	log.Logger = log.Logger.With().Caller().Logger()
+
 	ctx, cancel := context.WithCancel(context.Background())
 
 	pg, err := postgres.Create(ctx, cfg.Database)

cmd/metadata/receiver.go

@@ -98,7 +98,7 @@ func (r *Receiver) work(ctx context.Context) {
 			}
 			tasks, err := r.storage.GetByStatus(ctx, storage.StatusFilled, 100, 0, r.maxAttempts, r.delay)
 			if err != nil {
-				log.Err(err).Msg("receiving receiving tasks")
+				log.Err(err).Msg("receiving tasks")
 				continue
 			}
 

internal/storage/address.go

@@ -2,6 +2,7 @@ package storage
 
 import (
 	"github.com/dipdup-net/indexer-sdk/pkg/storage"
+	"github.com/uptrace/bun"
 )
 
 // IAddress -
@@ -11,11 +12,10 @@ type IAddress interface {
 
 // Address -
 type Address struct {
-	// nolint
-	tableName struct{} `pg:"address" comment:"Table with starknet addresses."`
+	bun.BaseModel `bun:"table:address" comment:"Table with starknet addresses."`
 
-	ID   uint64 `pg:"id,type:bigint,pk,notnull" comment:"Unique internal identity"`
-	Hash []byte `pg:",unique:address_hash" comment:"Address hash."`
+	ID   uint64 `bun:"id,type:bigint,pk,notnull" comment:"Unique internal identity"`
+	Hash []byte `bun:",unique:address_hash" comment:"Address hash."`
 }
 
 // TableName -

internal/storage/postgres/address.go

@@ -12,7 +12,7 @@ type Address struct {
 }
 
 // NewAddress -
-func NewAddress(db *database.PgGo) *Address {
+func NewAddress(db *database.Bun) *Address {
 	return &Address{
 		Table: postgres.NewTable[*storage.Address](db),
 	}

internal/storage/postgres/core.go

@@ -2,15 +2,15 @@ package postgres
 
 import (
 	"context"
+	"database/sql"
 
 	models "github.com/dipdup-io/starknet-metadata/internal/storage"
 	"github.com/dipdup-net/go-lib/config"
 	"github.com/dipdup-net/go-lib/database"
 	"github.com/dipdup-net/indexer-sdk/pkg/storage/postgres"
-	"github.com/go-pg/pg/v10"
-	"github.com/go-pg/pg/v10/orm"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog/log"
+	"github.com/uptrace/bun"
 )
 
 // Storage -
@@ -20,6 +20,7 @@ type Storage struct {
 	Address       models.IAddress
 	TokenMetadata models.ITokenMetadata
 	State         models.IState
+	Tx            Transaction
 }
 
 // Create -
@@ -39,15 +40,13 @@ func Create(ctx context.Context, cfg config.Database) (Storage, error) {
 	return s, nil
 }
 
-func initDatabase(ctx context.Context, conn *database.PgGo) error {
+func initDatabase(ctx context.Context, conn *database.Bun) error {
 	if err := createTypes(ctx, conn); err != nil {
 		return err
 	}
 
 	for _, data := range models.Models {
-		if err := conn.DB().WithContext(ctx).Model(data).CreateTable(&orm.CreateTableOptions{
-			IfNotExists: true,
-		}); err != nil {
+		if _, err := conn.DB().NewCreateTable().IfNotExists().Model(data).Exec(ctx); err != nil {
 			if err := conn.Close(); err != nil {
 				return err
 			}
@@ -66,16 +65,16 @@ func initDatabase(ctx context.Context, conn *database.PgGo) error {
 	return createIndices(ctx, conn)
 }
 
-func createIndices(ctx context.Context, conn *database.PgGo) error {
+func createIndices(ctx context.Context, conn *database.Bun) error {
 	log.Info().Msg("creating indexes...")
-	return conn.DB().RunInTransaction(ctx, func(tx *pg.Tx) error {
+	return conn.DB().RunInTx(ctx, &sql.TxOptions{}, func(ctx context.Context, tx bun.Tx) error {
 		return nil
 	})
 }
 
-func createTypes(ctx context.Context, conn *database.PgGo) error {
+func createTypes(ctx context.Context, conn *database.Bun) error {
 	log.Info().Msg("creating custom types...")
-	return conn.DB().RunInTransaction(ctx, func(tx *pg.Tx) error {
+	return conn.DB().RunInTx(ctx, &sql.TxOptions{}, func(ctx context.Context, tx bun.Tx) error {
 		if _, err := tx.ExecContext(
 			ctx,
 			`DO $$

internal/storage/postgres/fixtures/address.yml

@@ -0,0 +1,8 @@
+- id: 1
+  hash: 0x049d36570d4e46f48e99674bd3fcc84644ddd6b96f7c741b1562b82f9e004dc7
+- id: 2
+  hash: 0x053c91253bc9682c04929ca02ed00b3e423f6710d2ee7e0d5ebb06f3ecf368a8
+- id: 3
+  hash: 0x04eafcdde3136a2db2aec1de26e43ac9e04b11d6a9e879c92cb2f9ef1f0be815
+- id: 4
+  hash: 0x046bfa580e4fa55a38eaa7f51a3469f86b336eed59a6136a07b7adcd095b0eb2

internal/storage/postgres/fixtures/state.yml

@@ -0,0 +1,4 @@
+- id: 1
+  name: test
+  last_height: 100
+  last_time: '2023-08-15T11:29:06+00:00'

internal/storage/postgres/fixtures/token_metadata.yml

@@ -0,0 +1,44 @@
+- id: 1
+  created_at: 1691750161
+  updated_at: 1691750161
+  update_id: 10
+  contract_id: 1
+  token_id: 0
+  type: erc20
+  status: new
+  attempts: 0
+- id: 2
+  created_at: 1691750161
+  updated_at: 1691750161
+  update_id: 11
+  contract_id: 2
+  token_id: 0
+  type: erc20
+  status: filled
+  uri: token_2_uri
+  attempts: 1
+- id: 3
+  created_at: 1691750161
+  updated_at: 1691750161
+  update_id: 12
+  contract_id: 3
+  token_id: 1
+  type: erc721
+  status: success
+  uri: token_2_uri
+  metadata:
+    name: Token 1
+    symbol: TKN
+    decimals: 18
+  attempts: 1
+- id: 4
+  created_at: 1691750161
+  updated_at: 1691750161
+  update_id: 13
+  contract_id: 4
+  token_id: 14
+  type: erc1155
+  status: failed
+  uri: token_3_uri
+  attempts: 5
+  error: invalid URI

internal/storage/postgres/state.go

@@ -14,14 +14,14 @@ type State struct {
 }
 
 // NewState -
-func NewState(db *database.PgGo) *State {
+func NewState(db *database.Bun) *State {
 	return &State{
 		Table: postgres.NewTable[*storage.State](db),
 	}
 }
 
 // ByName -
 func (s *State) ByName(ctx context.Context, name string) (state storage.State, err error) {
-	err = s.DB().ModelContext(ctx, &state).Where("name = ?", name).Select(&state)
+	err = s.DB().NewSelect().Model(&state).Where("name = ?", name).Scan(ctx)
 	return
 }

internal/storage/postgres/storage_test.go

@@ -0,0 +1,126 @@
+package postgres
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+	"time"
+
+	"github.com/dipdup-io/starknet-metadata/internal/storage"
+	"github.com/dipdup-net/go-lib/config"
+	"github.com/dipdup-net/go-lib/database"
+	"github.com/go-testfixtures/testfixtures/v3"
+	_ "github.com/lib/pq"
+	"github.com/stretchr/testify/suite"
+)
+
+type TestSuite struct {
+	suite.Suite
+	psqlContainer *database.PostgreSQLContainer
+	storage       Storage
+}
+
+// SetupSuite -
+func (s *TestSuite) SetupSuite() {
+	ctx, ctxCancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer ctxCancel()
+
+	psqlContainer, err := database.NewPostgreSQLContainer(ctx, database.PostgreSQLContainerConfig{
+		User:     "user",
+		Password: "password",
+		Database: "db_test",
+		Port:     5432,
+		Image:    "postgres:15",
+	})
+	s.Require().NoError(err)
+	s.psqlContainer = psqlContainer
+
+	s.storage, err = Create(ctx, config.Database{
+		Kind:     config.DBKindPostgres,
+		User:     s.psqlContainer.Config.User,
+		Database: s.psqlContainer.Config.Database,
+		Password: s.psqlContainer.Config.Password,
+		Host:     s.psqlContainer.Config.Host,
+		Port:     s.psqlContainer.MappedPort().Int(),
+	})
+	s.Require().NoError(err)
+
+	db, err := sql.Open("postgres", s.psqlContainer.GetDSN())
+	s.Require().NoError(err)
+
+	fixtures, err := testfixtures.New(
+		testfixtures.Database(db),
+		testfixtures.Dialect("postgres"),
+		testfixtures.Directory("fixtures"),
+	)
+	s.Require().NoError(err)
+	s.Require().NoError(fixtures.Load())
+	s.Require().NoError(db.Close())
+}
+
+func (s *TestSuite) TearDownSuite() {
+	ctx, ctxCancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer ctxCancel()
+
+	s.Require().NoError(s.storage.Close())
+	s.Require().NoError(s.psqlContainer.Terminate(ctx))
+}
+
+func (s *TestSuite) TestTokenMetadataGetByStatus() {
+	for _, status := range []storage.Status{
+		storage.StatusFailed,
+		storage.StatusFilled,
+		storage.StatusNew,
+		storage.StatusSuccess,
+	} {
+		ctx, ctxCancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer ctxCancel()
+		response, err := s.storage.TokenMetadata.GetByStatus(ctx, status, 1, 0, 0, 0)
+		s.Require().NoError(err)
+
+		s.Require().Len(response, 1)
+		s.Require().Equal(response[0].Status, status)
+		s.Require().Len(response[0].Contract.Hash, 32)
+	}
+}
+
+func (s *TestSuite) TestStateByName() {
+
+	ctx, ctxCancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer ctxCancel()
+
+	response, err := s.storage.State.ByName(ctx, "test")
+	s.Require().NoError(err)
+	s.Require().EqualValues(1, response.ID)
+	s.Require().EqualValues(100, response.LastHeight)
+
+	_, err = s.storage.State.ByName(ctx, "unknown")
+	s.Require().Error(err)
+}
+
+func (s *TestSuite) TestTxSaveAddress() {
+	ctx, ctxCancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer ctxCancel()
+
+	tx, err := BeginTransaction(ctx, s.storage.Transactable)
+	s.Require().NoError(err)
+	defer tx.Close(ctx)
+
+	err = tx.SaveAddress(ctx, &storage.Address{
+		ID:   100,
+		Hash: []byte{0, 1, 2, 3, 4},
+	})
+	s.Require().NoError(err)
+
+	err = tx.Flush(ctx)
+	s.Require().NoError(err)
+
+	response, err := s.storage.Address.GetByID(ctx, 100)
+	s.Require().NoError(err)
+	s.Require().EqualValues(100, response.ID)
+	s.Require().Equal([]byte{0, 1, 2, 3, 4}, response.Hash)
+}
+
+func TestSuite_Run(t *testing.T) {
+	suite.Run(t, new(TestSuite))
+}