stellar#4222: backfilling tests on filters

services/horizon/internal/db2/history/filter_rules.go

@@ -31,16 +31,16 @@ type QFilter interface {
 
 func (q *Q) GetAllFilters(ctx context.Context) ([]FilterConfig, error) {
 	var filterConfigs []FilterConfig
-	sql := sq.Select().From(filterRulesTableName)
-	err := q.Select(ctx, filterConfigs, sql)
+	sql := sq.Select("*").From(filterRulesTableName)
+	err := q.Select(ctx, &filterConfigs, sql)
 
 	return filterConfigs, err
 }
 
 func (q *Q) GetFilterByName(ctx context.Context, name string) (FilterConfig, error) {
-	var filterConfig FilterConfig
-	sql := sq.Select().From(filterRulesTableName).Where(sq.Eq{filterRulesTypeColumnName: name})
-	err := q.Select(ctx, filterConfig, sql)
+	filterConfig := FilterConfig{}
+	sql := sq.Select("*").From(filterRulesTableName).Where(sq.Eq{filterRulesTypeColumnName: name})
+	err := q.Get(ctx, &filterConfig, sql)
 
 	return filterConfig, err
 }

services/horizon/internal/db2/history/filter_rules_test.go

@@ -0,0 +1,89 @@
+package history
+
+import (
+	"testing"
+
+	"github.com/stellar/go/services/horizon/internal/test"
+	"github.com/stretchr/testify/assert"
+)
+
+var (
+	fc1 = FilterConfig{
+		Rules:   "{}",
+		Name:    "test data",
+		Enabled: false,
+	}
+)
+
+func TestInsertConfig(t *testing.T) {
+	tt := test.Start(t)
+	defer tt.Finish()
+	test.ResetHorizonDB(t, tt.HorizonDB)
+	q := &Q{tt.HorizonSession()}
+
+	err := q.UpsertFilterConfig(tt.Ctx, fc1)
+	assert.NoError(t, err)
+	fc1, err = q.GetFilterByName(tt.Ctx, "test data")
+	assert.NoError(t, err)
+	tt.Assert.True(fc1.LastModified > 0)
+	tt.Assert.Equal(fc1.Name, "test data")
+	tt.Assert.Equal(fc1.Enabled, false)
+	tt.Assert.Equal(fc1.Rules, "{}")
+}
+
+func TestGetAll(t *testing.T) {
+	tt := test.Start(t)
+	defer tt.Finish()
+	test.ResetHorizonDB(t, tt.HorizonDB)
+	q := &Q{tt.HorizonSession()}
+
+	err := q.UpsertFilterConfig(tt.Ctx, fc1)
+	assert.NoError(t, err)
+	results, err := q.GetAllFilters(tt.Ctx)
+	assert.NoError(t, err)
+	tt.Assert.Len(results, 1)
+
+	tt.Assert.Equal(results[0].Name, "test data")
+}
+func TestRemoveFilterConfig(t *testing.T) {
+	tt := test.Start(t)
+	defer tt.Finish()
+	test.ResetHorizonDB(t, tt.HorizonDB)
+	q := &Q{tt.HorizonSession()}
+
+	err := q.UpsertFilterConfig(tt.Ctx, fc1)
+	assert.NoError(t, err)
+
+	err = q.DeleteFilterByName(tt.Ctx, "not found")
+	assert.Error(t, err)
+
+	err = q.DeleteFilterByName(tt.Ctx, "test data")
+	assert.NoError(t, err)
+
+	fc1, err = q.GetFilterByName(tt.Ctx, "test data")
+	assert.Error(t, err)
+}
+
+func TestUpdateExisting(t *testing.T) {
+	tt := test.Start(t)
+	defer tt.Finish()
+	test.ResetHorizonDB(t, tt.HorizonDB)
+	q := &Q{tt.HorizonSession()}
+
+	err := q.UpsertFilterConfig(tt.Ctx, fc1)
+	assert.NoError(t, err)
+	fc1, err = q.GetFilterByName(tt.Ctx, "test data")
+	assert.NoError(t, err)
+	tt.Assert.Equal(fc1.Enabled, false)
+	tt.Assert.Equal(fc1.Rules, "{}")
+
+	fc1.Enabled = true
+	fc1.Rules = `{"abc": "123"}`
+	err = q.UpsertFilterConfig(tt.Ctx, fc1)
+	assert.NoError(t, err)
+	fc1, err = q.GetFilterByName(tt.Ctx, "test data")
+	assert.NoError(t, err)
+	tt.Assert.Equal(fc1.Name, "test data")
+	tt.Assert.Equal(fc1.Enabled, true)
+	tt.Assert.Equal(fc1.Rules, `{"abc": "123"}`)
+}

services/horizon/internal/ingest/filters/account.go

@@ -35,12 +35,12 @@ func (filter *accountFilter) RefreshAccountFilter(filterConfig *history.FilterCo
 	// only need to re-initialize the filter config state(rules) if it's cached version(in  memory)
 	// is older than the incoming config version based on lastModified epoch timestamp
 	if filterConfig.LastModified > filter.lastModified {
-		var assetFilterRules AssetFilterRules
-		if err := json.Unmarshal([]byte(filterConfig.Rules), &assetFilterRules); err != nil {
-			return errors.Wrap(err, "unable to serialize asset filter rules")
+		var accountFilterRules AccountFilterRules
+		if err := json.Unmarshal([]byte(filterConfig.Rules), &accountFilterRules); err != nil {
+			return errors.Wrap(err, "unable to serialize account filter rules")
 		}
 
-		filter.whitelistedAccountsSet = listToMap(assetFilterRules.CanonicalWhitelist)
+		filter.whitelistedAccountsSet = listToMap(accountFilterRules.CanonicalWhitelist)
 		filter.lastModified = filterConfig.LastModified
 
 	}

services/horizon/internal/ingest/mock_filter.go

@@ -8,11 +8,10 @@ import (
 	"github.com/stretchr/testify/mock"
 )
 
-type mockFilters struct {
+type MockFilters struct {
 	mock.Mock
 }
 
-func (m *mockFilters) GetFilters(filterQ history.QFilter, ctx context.Context) []processors.LedgerTransactionFilterer {
-	a := m.Called(filterQ, ctx)
-	return a.Get(0).([]processors.LedgerTransactionFilterer)
+func (m *MockFilters) GetFilters(filterQ history.QFilter, ctx context.Context) []processors.LedgerTransactionFilterer {
+	return []processors.LedgerTransactionFilterer{}
 }

services/horizon/internal/ingest/processor_runner_test.go

@@ -54,7 +54,7 @@ func TestProcessorRunnerRunHistoryArchiveIngestionGenesis(t *testing.T) {
 			NetworkPassphrase: network.PublicNetworkPassphrase,
 		},
 		historyQ: q,
-		filters:  &mockFilters{},
+		filters:  &MockFilters{},
 	}
 
 	_, err := runner.RunGenesisStateIngestion()
@@ -118,7 +118,7 @@ func TestProcessorRunnerRunHistoryArchiveIngestionHistoryArchive(t *testing.T) {
 		config:         config,
 		historyQ:       q,
 		historyAdapter: historyAdapter,
-		filters:        &mockFilters{},
+		filters:        &MockFilters{},
 	}
 
 	_, err := runner.RunHistoryArchiveIngestion(63, MaxSupportedProtocolVersion, bucketListHash)
@@ -153,7 +153,7 @@ func TestProcessorRunnerRunHistoryArchiveIngestionProtocolVersionNotSupported(t
 		config:         config,
 		historyQ:       q,
 		historyAdapter: historyAdapter,
-		filters:        &mockFilters{},
+		filters:        &MockFilters{},
 	}
 
 	_, err := runner.RunHistoryArchiveIngestion(100, 200, xdr.Hash{})
@@ -173,7 +173,7 @@ func TestProcessorRunnerBuildChangeProcessor(t *testing.T) {
 	runner := ProcessorRunner{
 		ctx:      ctx,
 		historyQ: q,
-		filters:  &mockFilters{},
+		filters:  &MockFilters{},
 	}
 
 	stats := &ingest.StatsChangeProcessor{}
@@ -195,7 +195,7 @@ func TestProcessorRunnerBuildChangeProcessor(t *testing.T) {
 	runner = ProcessorRunner{
 		ctx:      ctx,
 		historyQ: q,
-		filters:  &mockFilters{},
+		filters:  &MockFilters{},
 	}
 
 	processor = buildChangeProcessor(runner.historyQ, stats, historyArchiveSource, 456)
@@ -247,6 +247,60 @@ func TestProcessorRunnerBuildTransactionProcessor(t *testing.T) {
 	assert.IsType(t, &processors.TransactionProcessor{}, processor.processors[6])
 }
 
+func TestProcessorRunnerWithFilterEnabled(t *testing.T) {
+	ctx := context.Background()
+	maxBatchSize := 100000
+
+	config := Config{
+		NetworkPassphrase:        network.PublicNetworkPassphrase,
+		EnableIngestionFiltering: true,
+	}
+
+	q := &mockDBQ{}
+	defer mock.AssertExpectationsForObjects(t, q)
+
+	ledger := xdr.LedgerCloseMeta{
+		V0: &xdr.LedgerCloseMetaV0{
+			LedgerHeader: xdr.LedgerHeaderHistoryEntry{
+				Header: xdr.LedgerHeader{
+					BucketListHash: xdr.Hash([32]byte{0, 1, 2}),
+				},
+			},
+		},
+	}
+
+	// Batches
+	mockAccountSignersBatchInsertBuilder := &history.MockAccountSignersBatchInsertBuilder{}
+	defer mock.AssertExpectationsForObjects(t, mockAccountSignersBatchInsertBuilder)
+	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
+		Return(mockAccountSignersBatchInsertBuilder).Once()
+
+	mockOperationsBatchInsertBuilder := &history.MockOperationsBatchInsertBuilder{}
+	defer mock.AssertExpectationsForObjects(t, mockOperationsBatchInsertBuilder)
+	mockOperationsBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
+	q.MockQOperations.On("NewOperationBatchInsertBuilder", maxBatchSize).
+		Return(mockOperationsBatchInsertBuilder).Twice()
+
+	mockTransactionsBatchInsertBuilder := &history.MockTransactionsBatchInsertBuilder{}
+	defer mock.AssertExpectationsForObjects(t, mockTransactionsBatchInsertBuilder)
+	mockTransactionsBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
+	q.MockQTransactions.On("NewTransactionBatchInsertBuilder", maxBatchSize).
+		Return(mockTransactionsBatchInsertBuilder).Twice()
+
+	q.MockQLedgers.On("InsertLedger", ctx, ledger.V0.LedgerHeader, 0, 0, 0, 0, CurrentVersion).
+		Return(int64(1), nil).Once()
+
+	runner := ProcessorRunner{
+		ctx:      ctx,
+		config:   config,
+		historyQ: q,
+		filters:  &MockFilters{},
+	}
+
+	_, err := runner.RunAllProcessorsOnLedger(ledger)
+	assert.NoError(t, err)
+}
+
 func TestProcessorRunnerRunAllProcessorsOnLedger(t *testing.T) {
 	ctx := context.Background()
 	maxBatchSize := 100000
@@ -293,7 +347,7 @@ func TestProcessorRunnerRunAllProcessorsOnLedger(t *testing.T) {
 		ctx:      ctx,
 		config:   config,
 		historyQ: q,
-		filters:  &mockFilters{},
+		filters:  &MockFilters{},
 	}
 
 	_, err := runner.RunAllProcessorsOnLedger(ledger)
@@ -342,7 +396,7 @@ func TestProcessorRunnerRunAllProcessorsOnLedgerProtocolVersionNotSupported(t *t
 		ctx:      ctx,
 		config:   config,
 		historyQ: q,
-		filters:  &mockFilters{},
+		filters:  &MockFilters{},
 	}
 
 	_, err := runner.RunAllProcessorsOnLedger(ledger)