feat: [indexer] add test cases to block_update_indexer_block_process_strategy

ingest/indexer/domain/mocks/pair_publisher_mock.go

@@ -0,0 +1,34 @@
+package mocks
+
+import (
+	sdk "github.com/cosmos/cosmos-sdk/types"
+
+	commondomain "github.com/osmosis-labs/osmosis/v26/ingest/common/domain"
+	indexerdomain "github.com/osmosis-labs/osmosis/v26/ingest/indexer/domain"
+	poolmanagertypes "github.com/osmosis-labs/osmosis/v26/x/poolmanager/types"
+)
+
+var _ indexerdomain.PairPublisher = &MockPairPublisher{}
+
+// MockPairPublisher is a mock implementation of the PairPublisherI interface.
+type MockPairPublisher struct {
+	PublishPoolPairsError       error
+	PublishPoolPairsCalled      bool
+	CalledWithPools             []poolmanagertypes.PoolI
+	CalledWithCreatedPoolIDs    map[uint64]commondomain.PoolCreation
+	NumPoolPairPublished        int
+	NumPoolPairWithCreationData int
+}
+
+func (m *MockPairPublisher) PublishPoolPairs(ctx sdk.Context, pools []poolmanagertypes.PoolI, createdPoolIDs map[uint64]commondomain.PoolCreation) error {
+	m.PublishPoolPairsCalled = true
+	m.CalledWithPools = pools
+	m.CalledWithCreatedPoolIDs = createdPoolIDs
+	m.NumPoolPairPublished += len(pools)
+	for _, pool := range pools {
+		if _, ok := createdPoolIDs[pool.GetId()]; ok {
+			m.NumPoolPairWithCreationData++
+		}
+	}
+	return m.PublishPoolPairsError
+}

ingest/indexer/domain/mocks/publisher_mock.go

@@ -8,45 +8,55 @@ import (
 
 // PublisherMock is a mock for Publisher.
 type PublisherMock struct {
-	CalledWithPair  indexerdomain.Pair
-	CalledWithBlock indexerdomain.Block
-	// CalledWithPools             []types.PoolI
-	CalledWithTokenSupply       indexerdomain.TokenSupply
-	CalledWithTokenSupplyOffset indexerdomain.TokenSupplyOffset
-	CalledWithTransaction       indexerdomain.Transaction
-	ForcePairError              error
-	ForceBlockError             error
-	// ForcePoolError              error
-	// ForcePoolsError             error
-	ForceTokenSupplyError       error
-	ForceTokenSupplyOffsetError error
-	ForceTransactionError       error
+	CalledWithPair                   indexerdomain.Pair
+	CalledWithBlock                  indexerdomain.Block
+	CalledWithTokenSupply            indexerdomain.TokenSupply
+	CalledWithTokenSupplyOffset      indexerdomain.TokenSupplyOffset
+	CalledWithTransaction            indexerdomain.Transaction
+	NumPublishPairCalls              int
+	NumPublishBlockCalls             int
+	NumPublishTokenSupplyCalls       int
+	NumPublishTokenSupplyOffsetCalls int
+	NumPublishTransactionCalls       int
+	ForcePairError                   error
+	ForceBlockError                  error
+	ForceTokenSupplyError            error
+	ForceTokenSupplyOffsetError      error
+	ForceTransactionError            error
 }
 
 // PublishPair implements domain.Publisher.
 func (p *PublisherMock) PublishPair(ctx context.Context, pair indexerdomain.Pair) error {
+	p.CalledWithPair = pair
+	p.NumPublishPairCalls++
 	return p.ForcePairError
 }
 
 // PublishBlock implements domain.Publisher.
 func (p *PublisherMock) PublishBlock(ctx context.Context, block indexerdomain.Block) error {
+	p.CalledWithBlock = block
+	p.NumPublishBlockCalls++
 	return p.ForceBlockError
 }
 
 // PublishTokenSupply implements domain.Publisher.
 func (p *PublisherMock) PublishTokenSupply(ctx context.Context, tokenSupply indexerdomain.TokenSupply) error {
 	p.CalledWithTokenSupply = tokenSupply
+	p.NumPublishTokenSupplyCalls++
 	return p.ForceTokenSupplyError
 }
 
 // PublishTokenSupplyOffset implements domain.Publisher.
 func (p *PublisherMock) PublishTokenSupplyOffset(ctx context.Context, tokenSupplyOffset indexerdomain.TokenSupplyOffset) error {
 	p.CalledWithTokenSupplyOffset = tokenSupplyOffset
+	p.NumPublishTokenSupplyOffsetCalls++
 	return p.ForceTokenSupplyOffsetError
 }
 
 // PublishTransaction implements domain.Publisher.
 func (p *PublisherMock) PublishTransaction(ctx context.Context, txn indexerdomain.Transaction) error {
+	p.CalledWithTransaction = txn
+	p.NumPublishTransactionCalls++
 	return p.ForceTransactionError
 }
 

ingest/indexer/service/blockprocessor/block_updates_indexer_block_process_strategy.go

@@ -5,6 +5,7 @@ import (
 
 	commondomain "github.com/osmosis-labs/osmosis/v26/ingest/common/domain"
 	"github.com/osmosis-labs/osmosis/v26/ingest/indexer/domain"
+	poolmanagertypes "github.com/osmosis-labs/osmosis/v26/x/poolmanager/types"
 )
 
 type blockUpdatesIndexerBlockProcessStrategy struct {
@@ -50,8 +51,21 @@ func (f *blockUpdatesIndexerBlockProcessStrategy) publishCreatedPools(ctx types.
 		return nil
 	}
 
+	// Filter pools to include only those with pool IDs found in createdPoolIDs
+	filteredPools := []poolmanagertypes.PoolI{}
+	for _, pool := range pools {
+		if _, exists := createdPoolIDs[pool.GetId()]; exists {
+			filteredPools = append(filteredPools, pool)
+		}
+	}
+
+	// Do nothing if no pools are left after filtering
+	if len(filteredPools) == 0 {
+		return nil
+	}
+
 	// Publish pool pairs
-	if err := f.poolPairPublisher.PublishPoolPairs(ctx, pools, createdPoolIDs); err != nil {
+	if err := f.poolPairPublisher.PublishPoolPairs(ctx, filteredPools, createdPoolIDs); err != nil {
 		return err
 	}
 

ingest/indexer/service/blockprocessor/block_updates_indexer_block_process_strategy_test.go

@@ -0,0 +1,154 @@
+package blockprocessor_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/suite"
+
+	"github.com/osmosis-labs/osmosis/v26/app/apptesting"
+	commondomain "github.com/osmosis-labs/osmosis/v26/ingest/common/domain"
+	commonmocks "github.com/osmosis-labs/osmosis/v26/ingest/common/domain/mocks"
+	indexermocks "github.com/osmosis-labs/osmosis/v26/ingest/indexer/domain/mocks"
+	"github.com/osmosis-labs/osmosis/v26/ingest/indexer/service/blockprocessor"
+	sqsmocks "github.com/osmosis-labs/osmosis/v26/ingest/sqs/domain/mocks"
+)
+
+type BlockUpdateIndexerBlockProcessStrategyTestSuite struct {
+	apptesting.ConcentratedKeeperTestHelper
+}
+
+// TestBlockUpdateIndexerBlockProcessStrategyTestSuite verifies the block update indexer strategy for processing created pools.
+// The test suite initializes all supported pools (concentrated, cfmm, cosmwasm) via s.App.PrepareAllSupportedPools(), creating pool IDs 1-5.
+// Test cases inject pool creation data and validate expected behavior of the pair publisher.
+//
+// Scenarios tested:
+// - Happy path: single pool creation: should perform publishing
+// - Happy path: multiple pool creation: should perform publishing
+// - No pool creation: nothing is published
+// - Pool creation data without a match: nothing is published
+func TestBlockUpdateIndexerBlockProcessStrategyTestSuite(t *testing.T) {
+	suite.Run(t, new(BlockUpdateIndexerBlockProcessStrategyTestSuite))
+}
+
+func (s *BlockUpdateIndexerBlockProcessStrategyTestSuite) TestPublishCreatedPools() {
+	tests := []struct {
+		name                             string
+		createdPoolIDs                   map[uint64]commondomain.PoolCreation
+		expectedPublishPoolPairsCalled   bool
+		expectedNumPoolsPublished        int
+		expectedNumPoolsWithCreationData int
+	}{
+		{
+			name: "happy path with one pool creation",
+			createdPoolIDs: map[uint64]commondomain.PoolCreation{
+				DefaultConcentratedPoolId: {
+					PoolId:      DefaultConcentratedPoolId,
+					BlockHeight: DefaultConcentratedPoolHeight,
+					BlockTime:   DefaultConcentratedPoolTime,
+					TxnHash:     DefaultConcentratedPoolTxnHash,
+				},
+			},
+			expectedPublishPoolPairsCalled:   true,
+			expectedNumPoolsPublished:        1,
+			expectedNumPoolsWithCreationData: 1,
+		},
+		{
+			name: "happy path with multiple pool creation",
+			createdPoolIDs: map[uint64]commondomain.PoolCreation{
+				DefaultConcentratedPoolId: {
+					PoolId:      DefaultConcentratedPoolId,
+					BlockHeight: DefaultConcentratedPoolHeight,
+					BlockTime:   DefaultConcentratedPoolTime,
+					TxnHash:     DefaultConcentratedPoolTxnHash,
+				},
+				DefaultCfmmPoolId: {
+					PoolId:      DefaultCfmmPoolId,
+					BlockHeight: DefaultCfmmPoolHeight,
+					BlockTime:   DefaultCfmmPoolTime,
+					TxnHash:     DefaultCfmmPoolTxnHash,
+				},
+			},
+			expectedPublishPoolPairsCalled:   true,
+			expectedNumPoolsPublished:        2,
+			expectedNumPoolsWithCreationData: 2,
+		},
+		{
+			name:                             "should not publish when there is no pool creation",
+			createdPoolIDs:                   map[uint64]commondomain.PoolCreation{},
+			expectedPublishPoolPairsCalled:   false,
+			expectedNumPoolsPublished:        0,
+			expectedNumPoolsWithCreationData: 0,
+		},
+		{
+			name: "should not publish when pool creation data has no match in the pool list",
+			createdPoolIDs: map[uint64]commondomain.PoolCreation{
+				999: {
+					PoolId:      999,
+					BlockHeight: 12345,
+					BlockTime:   time.Now(),
+					TxnHash:     "txhash",
+				},
+			},
+			expectedPublishPoolPairsCalled:   false,
+			expectedNumPoolsPublished:        0,
+			expectedNumPoolsWithCreationData: 0,
+		},
+	}
+
+	for _, test := range tests {
+		s.Run(test.name, func() {
+			s.Setup()
+
+			// Initialized chain pools
+			s.PrepareAllSupportedPools()
+
+			// Get all chain pools from state for asserting later
+			// pool id 1 created below
+			concentratedPools, err := s.App.ConcentratedLiquidityKeeper.GetPools(s.Ctx)
+			s.Require().NoError(err)
+			// pool id 2, 3 created below
+			cfmmPools, err := s.App.GAMMKeeper.GetPools(s.Ctx)
+			s.Require().NoError(err)
+			// pool id 4, 5 created below
+			cosmWasmPools, err := s.App.CosmwasmPoolKeeper.GetPoolsWithWasmKeeper(s.Ctx)
+			s.Require().NoError(err)
+			blockPools := commondomain.BlockPools{
+				ConcentratedPools: concentratedPools,
+				CFMMPools:         cfmmPools,
+				CosmWasmPools:     cosmWasmPools,
+			}
+
+			// Mock out block updates process utils
+			blockUpdatesProcessUtilsMock := &sqsmocks.BlockUpdateProcessUtilsMock{}
+
+			// Mock out pool extractor
+			poolsExtracter := &commonmocks.PoolsExtractorMock{
+				BlockPools:     blockPools,
+				CreatedPoolIDs: test.createdPoolIDs,
+			}
+
+			// Mock out publisher
+			publisherMock := &indexermocks.PublisherMock{}
+
+			// Mock out pair publisher
+			pairPublisherMock := &indexermocks.MockPairPublisher{}
+
+			bprocess := blockprocessor.NewBlockUpdatesIndexerBlockProcessStrategy(blockUpdatesProcessUtilsMock, publisherMock, poolsExtracter, pairPublisherMock)
+
+			err = bprocess.PublishCreatedPools(s.Ctx)
+			s.Require().NoError(err)
+
+			// Check that the pair publisher is called correctly
+			s.Require().Equal(test.expectedPublishPoolPairsCalled, pairPublisherMock.PublishPoolPairsCalled)
+			if test.expectedPublishPoolPairsCalled {
+				// Check that the number of pools published
+				s.Require().Equal(test.expectedNumPoolsPublished, pairPublisherMock.NumPoolPairPublished)
+				// Check that the pools and created pool IDs are set correctly
+				s.Require().Equal(test.createdPoolIDs, pairPublisherMock.CalledWithCreatedPoolIDs)
+				// Check that the number of pools with creation data
+				s.Require().Equal(test.expectedNumPoolsWithCreationData, pairPublisherMock.NumPoolPairWithCreationData)
+			}
+		})
+	}
+}

ingest/indexer/service/blockprocessor/export_test.go

@@ -0,0 +1,42 @@
+package blockprocessor
+
+import (
+	"github.com/cosmos/cosmos-sdk/types"
+
+	commondomain "github.com/osmosis-labs/osmosis/v26/ingest/common/domain"
+	"github.com/osmosis-labs/osmosis/v26/ingest/indexer/domain"
+)
+
+func NewBlockUpdatesIndexerBlockProcessStrategy(blockUpdateProcessUtils commondomain.BlockUpdateProcessUtilsI, client domain.Publisher, poolExtractor commondomain.PoolExtractor, poolPairPublisher domain.PairPublisher) *blockUpdatesIndexerBlockProcessStrategy {
+	return &blockUpdatesIndexerBlockProcessStrategy{
+		blockUpdateProcessUtils: blockUpdateProcessUtils,
+		client:                  client,
+		poolExtractor:           poolExtractor,
+		poolPairPublisher:       poolPairPublisher,
+	}
+}
+
+type BlockUpdatesIndexerBlockProcessStrategy = blockUpdatesIndexerBlockProcessStrategy
+
+func (s *blockUpdatesIndexerBlockProcessStrategy) PublishCreatedPools(ctx types.Context) error {
+	return s.publishCreatedPools(ctx)
+}
+
+func NewFullIndexerBlockProcessStrategy(client domain.Publisher, keepers domain.Keepers, poolExtractor commondomain.PoolExtractor, poolPairPublisher domain.PairPublisher) *fullIndexerBlockProcessStrategy {
+	return &fullIndexerBlockProcessStrategy{
+		client:            client,
+		keepers:           keepers,
+		poolExtractor:     poolExtractor,
+		poolPairPublisher: poolPairPublisher,
+	}
+}
+
+type FullIndexerBlockProcessStrategy = fullIndexerBlockProcessStrategy
+
+func (s *fullIndexerBlockProcessStrategy) PublishAllSupplies(ctx types.Context) {
+	s.publishAllSupplies(ctx)
+}
+
+func (s *fullIndexerBlockProcessStrategy) ProcessPools(ctx types.Context) error {
+	return s.processPools(ctx)
+}