#4911: added reingest cmd integration test for datastore ingest

go.mod

@@ -99,6 +99,7 @@ require (
  github.com/nxadm/tail v1.4.8 // indirect
  github.com/opencontainers/go-digest v1.0.0 // indirect
  github.com/opencontainers/image-spec v1.1.0 // indirect
+ github.com/otiai10/copy v1.14.0 // indirect
  github.com/pelletier/go-toml/v2 v2.1.0 // indirect
  github.com/pkg/xattr v0.4.9 // indirect
  github.com/rogpeppe/go-internal v1.11.0 // indirect

go.sum

@@ -401,6 +401,8 @@ github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQ
 github.com/opencontainers/image-spec v1.1.0/go.mod h1:W4s4sFTMaBeK1BQLXbG4AdM2szdn85PY75RI83NrTrM=
 github.com/opentracing/opentracing-go v1.1.0 h1:pWlfV3Bxv7k65HYwkikxat0+s3pV4bsqf19k25Ur8rU=
 github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
+github.com/otiai10/copy v1.14.0 h1:dCI/t1iTdYGtkvCuBG2BgR6KZa83PTclw4U5n2wAllU=
+github.com/otiai10/copy v1.14.0/go.mod h1:ECfuL02W+/FkTWZWgQqXPWZgW9oeKCSQ5qVfSc4qc4w=
 github.com/pelletier/go-toml v1.9.5 h1:4yBQzkHv+7BHq2PQUZF3Mx0IYxG7LsP222s7Agd3ve8=
 github.com/pelletier/go-toml v1.9.5/go.mod h1:u1nR/EPcESfeI/szUZKdtJ0xRNbUoANCkoOuaOx1Y+c=
 github.com/pelletier/go-toml/v2 v2.1.0 h1:FnwAJ4oYMvbT/34k9zzHuZNrhlz48GB3/s6at6/MHO4=

services/horizon/cmd/db.go

@@ -101,6 +101,7 @@ func migrate(dir schema.MigrateDir, count int) error {
  if err != nil {
  return err
  }
+ defer dbConn.Close()
 
  numMigrationsRun, err := schema.Migrate(dbConn.DB.DB, dir, count)
  if err != nil {

services/horizon/cmd/db_test.go

@@ -18,7 +18,7 @@ func TestDBCommandsTestSuite(t *testing.T) {
 
 type DBCommandsTestSuite struct {
  suite.Suite
- dsn string
+ db *dbtest.DB
 }
 
 func (s *DBCommandsTestSuite) SetupSuite() {
@@ -27,18 +27,21 @@ func (s *DBCommandsTestSuite) SetupSuite() {
  return nil
  }
 
- newDB := dbtest.Postgres(s.T())
- s.dsn = newDB.DSN
+ s.db = dbtest.Postgres(s.T())
 
  RootCmd.SetArgs([]string{
- "db", "migrate", "up", "--db-url", s.dsn})
+ "db", "migrate", "up", "--db-url", s.db.DSN})
  require.NoError(s.T(), RootCmd.Execute())
 }
 
+func (s *DBCommandsTestSuite) TearDownSuite() {
+ s.db.Close()
+}
+
 func (s *DBCommandsTestSuite) TestDefaultParallelJobSizeForBufferedBackend() {
  RootCmd.SetArgs([]string{
  "db", "reingest", "range",
- "--db-url", s.dsn,
+ "--db-url", s.db.DSN,
  "--network", "testnet",
  "--parallel-workers", "2",
  "--ledgerbackend", "datastore",
@@ -53,7 +56,7 @@ func (s *DBCommandsTestSuite) TestDefaultParallelJobSizeForBufferedBackend() {
 func (s *DBCommandsTestSuite) TestDefaultParallelJobSizeForCaptiveBackend() {
  RootCmd.SetArgs([]string{
  "db", "reingest", "range",
- "--db-url", s.dsn,
+ "--db-url", s.db.DSN,
  "--network", "testnet",
  "--stellar-core-binary-path", "/test/core/bin/path",
  "--parallel-workers", "2",
@@ -68,7 +71,7 @@ func (s *DBCommandsTestSuite) TestDefaultParallelJobSizeForCaptiveBackend() {
 func (s *DBCommandsTestSuite) TestUsesParallelJobSizeWhenSetForCaptive() {
  RootCmd.SetArgs([]string{
  "db", "reingest", "range",
- "--db-url", s.dsn,
+ "--db-url", s.db.DSN,
  "--network", "testnet",
  "--stellar-core-binary-path", "/test/core/bin/path",
  "--parallel-workers", "2",
@@ -84,7 +87,7 @@ func (s *DBCommandsTestSuite) TestUsesParallelJobSizeWhenSetForCaptive() {
 func (s *DBCommandsTestSuite) TestUsesParallelJobSizeWhenSetForBuffered() {
  RootCmd.SetArgs([]string{
  "db", "reingest", "range",
- "--db-url", s.dsn,
+ "--db-url", s.db.DSN,
  "--network", "testnet",
  "--parallel-workers", "2",
  "--parallel-job-size", "5",

services/horizon/internal/integration/db_test.go

@@ -3,12 +3,16 @@ package integration
 import (
  "context"
  "fmt"
+ "os"
  "path/filepath"
  "strconv"
  "testing"
  "time"
 
+ "github.com/fsouza/fake-gcs-server/fakestorage"
+ cp "github.com/otiai10/copy"
  "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
 
  "github.com/stellar/go/clients/horizonclient"
  "github.com/stellar/go/historyarchive"
@@ -549,6 +553,67 @@ func TestReingestDB(t *testing.T) {
  tt.NoError(horizoncmd.RootCmd.Execute(), "Repeat the same reingest range against db, should not have errors.")
 }
 
+func TestReingestDatastore(t *testing.T) {
+ os.Setenv("HORIZON_INTEGRATION_TESTS_ENABLED", "true")
+
+ if os.Getenv("HORIZON_INTEGRATION_TESTS_ENABLED") == "" {
+ t.Skip("skipping integration test: HORIZON_INTEGRATION_TESTS_ENABLED not set")
+ }
+
+ newDB := dbtest.Postgres(t)
+ defer newDB.Close()
+ horizoncmd.RootCmd.SetArgs([]string{
+ "db", "migrate", "up", "--db-url", newDB.DSN})
+ require.NoError(t, horizoncmd.RootCmd.Execute())
+
+ testTempDir := t.TempDir()
+ tempSeedDataPath := filepath.Join(testTempDir, "data")
+ tempSeedBucketPath := filepath.Join(tempSeedDataPath, "path", "to", "my", "bucket")
+ tempSeedBucketFolder := filepath.Join(tempSeedBucketPath, "FFFFFFFF--0-63999")
+ if err := os.MkdirAll(tempSeedBucketFolder, 0777); err != nil {
+ t.Fatalf("unable to create seed data in temp path, %v", err)
+ }
+
+ err := cp.Copy("./testdata/testbucket", tempSeedBucketFolder)
+ if err != nil {
+ t.Fatalf("unable to copy seed data files for fake gcs, %v", err)
+ }
+
+ testWriter := &testWriter{test: t}
+ opts := fakestorage.Options{
+ Scheme: "http",
+ Host: "127.0.0.1",
+ Port: uint16(0),
+ Writer: testWriter,
+ Seed: tempSeedDataPath,
+ StorageRoot: filepath.Join(testTempDir, "bucket"),
+ PublicHost: "127.0.0.1",
+ }
+
+ gcsServer, err := fakestorage.NewServerWithOptions(opts)
+
+ if err != nil {
+ t.Fatalf("couldn't start the fake gcs http server %v", err)
+ }
+
+ defer gcsServer.Stop()
+ t.Logf("fake gcs server started at %v", gcsServer.URL())
+ t.Setenv("STORAGE_EMULATOR_HOST", gcsServer.URL())
+
+ horizoncmd.RootCmd.SetArgs([]string{"db",
+ "reingest",
+ "range",
+ "--db-url", newDB.DSN,
+ "--network", "testnet",
+ "--parallel-workers", "1",
+ "--ledgerbackend", "datastore",
+ "--datastore-config", "../../config.storagebackend.toml",
+ "997",
+ "999"})
+
+ require.NoError(t, horizoncmd.RootCmd.Execute())
+}
+
 func TestReingestDBWithFilterRules(t *testing.T) {
  itest, _ := initializeDBIntegrationTest(t)
  tt := assert.New(t)
@@ -905,3 +970,12 @@ func TestResumeFromInitializedDB(t *testing.T) {
 
  tt.Eventually(successfullyResumed, 1*time.Minute, 1*time.Second)
 }
+
+type testWriter struct {
+ test *testing.T
+}
+
+func (w *testWriter) Write(p []byte) (n int, err error) {
+ w.test.Log(string(p))
+ return len(p), nil
+}