#4418: changed integration tests to spin up one horizon web node with…

… r/o db connection and one horizon ingest node with r/w db conn

services/horizon/CHANGELOG.md

@@ -16,7 +16,7 @@ file. This project adheres to [Semantic Versioning](https://semver.org/).
  * 'whitelist by account id' ([4221](https://github.com/stellar/go/issues/4221))
  * 'whitelist by canonical asset id' ([4222](https://github.com/stellar/go/issues/4222))
 
- The filters and their configuration are optional features and must be enabled with horizon command line parameters `admin-port=4200` and `enable-ingestion-filtering=true`
+ The filters and their configuration are optional features and must be enabled with horizon command line parameters `admin-port=4200` and `exp-enable-ingestion-filtering=true`
 
  Once set, filter configurations and their rules are initially empty and the filters are disabled by default. To enable filters, update the configuration settings, refer to the Admin API Docs which are published on the Admin Port at https://localhost:<admin_port>/, follow details and examples for endpoints:
  * `/ingestion/filters/account`

services/horizon/internal/integration/db_test.go

@@ -444,7 +444,7 @@ func TestReingestDB(t *testing.T) {
  itest, reachedLedger := initializeDBIntegrationTest(t)
  tt := assert.New(t)
 
- horizonConfig := itest.GetHorizonConfig()
+ horizonConfig := itest.GetHorizonIngestConfig()
  t.Run("validate parallel range", func(t *testing.T) {
  horizoncmd.RootCmd.SetArgs(command(horizonConfig,
  "db",
@@ -537,11 +537,8 @@ func TestFillGaps(t *testing.T) {
 
  // Create a fresh Horizon database
  newDB := dbtest.Postgres(t)
- // TODO: Unfortunately Horizon's ingestion System leaves open sessions behind,leading to
- // a "database is being accessed by other users" error when trying to drop it
- // defer newDB.Close()
  freshHorizonPostgresURL := newDB.DSN
- horizonConfig := itest.GetHorizonConfig()
+ horizonConfig := itest.GetHorizonIngestConfig()
  horizonConfig.DatabaseURL = freshHorizonPostgresURL
  // Initialize the DB schema
  dbConn, err := db.Open("postgres", freshHorizonPostgresURL)

services/horizon/internal/integration/ingestion_filtering_test.go

@@ -17,7 +17,7 @@ func TestFilteringAccountWhiteList(t *testing.T) {
  tt := assert.New(t)
  const adminPort uint16 = 6000
  itest := integration.NewTest(t, integration.Config{
- HorizonParameters: map[string]string{
+ HorizonIngestParameters: map[string]string{
  "admin-port": strconv.Itoa(int(adminPort)),
  "exp-enable-ingestion-filtering": "true",
  },
@@ -93,7 +93,7 @@ func TestFilteringAssetWhiteList(t *testing.T) {
  tt := assert.New(t)
  const adminPort uint16 = 6000
  itest := integration.NewTest(t, integration.Config{
- HorizonParameters: map[string]string{
+ HorizonIngestParameters: map[string]string{
  "admin-port": strconv.Itoa(int(adminPort)),
  "exp-enable-ingestion-filtering": "true",
  },

services/horizon/internal/integration/parameters_test.go

@@ -46,10 +46,10 @@ func NewParameterTest(t *testing.T, params map[string]string) *integration.Test
 
 func NewParameterTestWithEnv(t *testing.T, params, envvars map[string]string) *integration.Test {
  config := integration.Config{
- ProtocolVersion: 17,
- SkipHorizonStart: true,
- HorizonParameters:  params,
- HorizonEnvironment: envvars,
+ ProtocolVersion:  17,
+ SkipHorizonStart:  true,
+ HorizonIngestParameters: params,
+ HorizonEnvironment:  envvars,
  }
  return integration.NewTest(t, config)
 }
@@ -153,15 +153,15 @@ func TestMaxAssetsForPathRequests(t *testing.T) {
  err := test.StartHorizon()
  assert.NoError(t, err)
  test.WaitForHorizon()
- assert.Equal(t, test.Horizon().Config().MaxAssetsPerPathRequest, 15)
+ assert.Equal(t, test.HorizonIngest().Config().MaxAssetsPerPathRequest, 15)
  test.Shutdown()
  })
  t.Run("set to 2", func(t *testing.T) {
  test := NewParameterTest(t, map[string]string{"max-assets-per-path-request": "2"})
  err := test.StartHorizon()
  assert.NoError(t, err)
  test.WaitForHorizon()
- assert.Equal(t, test.Horizon().Config().MaxAssetsPerPathRequest, 2)
+ assert.Equal(t, test.HorizonIngest().Config().MaxAssetsPerPathRequest, 2)
  test.Shutdown()
  })
 }
@@ -172,8 +172,8 @@ func TestMaxPathFindingRequests(t *testing.T) {
  err := test.StartHorizon()
  assert.NoError(t, err)
  test.WaitForHorizon()
- assert.Equal(t, test.Horizon().Config().MaxPathFindingRequests, uint(0))
- _, ok := test.Horizon().Paths().(simplepath.InMemoryFinder)
+ assert.Equal(t, test.HorizonIngest().Config().MaxPathFindingRequests, uint(0))
+ _, ok := test.HorizonIngest().Paths().(simplepath.InMemoryFinder)
  assert.True(t, ok)
  test.Shutdown()
  })
@@ -182,8 +182,8 @@ func TestMaxPathFindingRequests(t *testing.T) {
  err := test.StartHorizon()
  assert.NoError(t, err)
  test.WaitForHorizon()
- assert.Equal(t, test.Horizon().Config().MaxPathFindingRequests, uint(5))
- finder, ok := test.Horizon().Paths().(*paths.RateLimitedFinder)
+ assert.Equal(t, test.HorizonIngest().Config().MaxPathFindingRequests, uint(5))
+ finder, ok := test.HorizonIngest().Paths().(*paths.RateLimitedFinder)
  assert.True(t, ok)
  assert.Equal(t, finder.Limit(), 5)
  test.Shutdown()
@@ -196,8 +196,8 @@ func TestDisablePathFinding(t *testing.T) {
  err := test.StartHorizon()
  assert.NoError(t, err)
  test.WaitForHorizon()
- assert.Equal(t, test.Horizon().Config().MaxPathFindingRequests, uint(0))
- _, ok := test.Horizon().Paths().(simplepath.InMemoryFinder)
+ assert.Equal(t, test.HorizonIngest().Config().MaxPathFindingRequests, uint(0))
+ _, ok := test.HorizonIngest().Paths().(simplepath.InMemoryFinder)
  assert.True(t, ok)
  test.Shutdown()
  })
@@ -206,7 +206,7 @@ func TestDisablePathFinding(t *testing.T) {
  err := test.StartHorizon()
  assert.NoError(t, err)
  test.WaitForHorizon()
- assert.Nil(t, test.Horizon().Paths())
+ assert.Nil(t, test.HorizonIngest().Paths())
  test.Shutdown()
  })
 }

services/horizon/internal/integration/state_verifier_test.go

@@ -120,7 +120,7 @@ func TestStateVerifier(t *testing.T) {
  }
 
  // Trigger state rebuild to check if ingesting from history archive works
- session := itest.Horizon().HistoryQ().Clone()
+ session := itest.HorizonIngest().HistoryQ().Clone()
  q := &history.Q{session}
  err = q.Begin()
  assert.NoError(t, err)

services/horizon/internal/integration/trade_aggregations_test.go

@@ -18,7 +18,7 @@ import (
 func TestTradeAggregations(t *testing.T) {
  itest := integration.NewTest(t, integration.Config{})
  ctx := context.Background()
- historyQ := itest.Horizon().HistoryQ()
+ historyQ := itest.HorizonIngest().HistoryQ()
 
  // Insert some trades
  now := strtime.Now().RoundDown(60_000)