Add tx test and in change_test.go

ingest/change.go

@@ -52,7 +52,7 @@ type Change struct {
 	Reason LedgerEntryChangeReason
 
 	// The index of the operation within the transaction that caused the change.
-	// This field is relevant only when the Reason is of type LedgerEntryChangeReasonOperation
+	// This field is relevant only when the Reason is LedgerEntryChangeReasonOperation
 	// This field cannot be relied upon when the compactingChangeReader is used.
 	OperationIndex uint32
 
@@ -64,11 +64,13 @@ type Change struct {
 
 	// The LedgerCloseMeta that precipitated the change.
 	// This is useful only when the Change is caused by an upgrade or by an eviction, i.e. outside a transaction
+	// This field is populated only when the Reason is one of:
+	// LedgerEntryChangeReasonUpgrade or LedgerEntryChangeReasonEviction
 	// For changes caused by transaction or operations, look at the Transaction field
 	Ledger *xdr.LedgerCloseMeta
 
 	// Information about the upgrade, if the change occurred as part of an upgrade
-	// This field is relevant only when the Reason is of type LedgerEntryChangeReasonUpgrade
+	// This field is relevant only when the Reason is LedgerEntryChangeReasonUpgrade
 	LedgerUpgrade *xdr.LedgerUpgrade
 }
 

services/horizon/internal/integration/change_test.go

@@ -5,7 +5,9 @@ import (
 	"github.com/stellar/go/historyarchive"
 	"github.com/stellar/go/ingest"
 	"github.com/stellar/go/ingest/ledgerbackend"
+	"github.com/stellar/go/keypair"
 	"github.com/stellar/go/services/horizon/internal/test/integration"
+	"github.com/stellar/go/txnbuild"
 	"github.com/stellar/go/xdr"
 	"github.com/stretchr/testify/assert"
 	"io"
@@ -16,38 +18,111 @@ import (
 func TestProtocolUpgradeChanges(t *testing.T) {
 	tt := assert.New(t)
 	itest := integration.NewTest(t, integration.Config{SkipHorizonStart: true, SkipProtocolUpgrade: true})
-	archive, err := historyarchive.Connect(
-		integration.HistoryArchiveUrl,
-		historyarchive.ArchiveOptions{
-			NetworkPassphrase:   integration.StandaloneNetworkPassphrase,
-			CheckpointFrequency: integration.CheckpointFrequency,
-		})
+	archive, err := integration.GetHistoryArchive()
 	tt.NoError(err)
 
 	// Manually invoke command to upgrade protocol
 	itest.UpgradeProtocol(itest.Config().ProtocolVersion)
 	upgradedLedgerSeq, _ := itest.GetUpgradeLedgerSeq()
 
-	var latestCheckpoint uint32
-	publishedNextCheckpoint := func() bool {
-		has, requestErr := archive.GetRootHAS()
-		if requestErr != nil {
-			t.Logf("request to fetch checkpoint failed: %v", requestErr)
-			return false
-		}
-		latestCheckpoint = has.CurrentLedger
-		return latestCheckpoint >= upgradedLedgerSeq
-	}
+	publishedNextCheckpoint := publishedNextCheckpoint(archive, upgradedLedgerSeq, t)
 	// Ensure that a checkpoint has been created with the ledgerNumber you want in it
 	tt.Eventually(publishedNextCheckpoint, 15*time.Second, time.Second)
 
 	prevLedgerToUpgrade := upgradedLedgerSeq - 1
 	ledgerSeqToLedgers := getLedgersFromArchive(itest, prevLedgerToUpgrade, upgradedLedgerSeq)
-	prevLedgerChangeMap := changeMap(getChangesFromLedger(itest, ledgerSeqToLedgers[prevLedgerToUpgrade]))
-	upgradedLedgerChangeMap := changeMap(getChangesFromLedger(itest, ledgerSeqToLedgers[upgradedLedgerSeq]))
+
+	prevLedgerChanges := getChangesFromLedger(itest, ledgerSeqToLedgers[prevLedgerToUpgrade])
+	prevLedgerChangeMap := changeReasonCountMap(prevLedgerChanges)
+	upgradedLedgerChanges := getChangesFromLedger(itest, ledgerSeqToLedgers[upgradedLedgerSeq])
+	upgradedLedgerChangeMap := changeReasonCountMap(upgradedLedgerChanges)
 
 	tt.Zero(prevLedgerChangeMap[ingest.LedgerEntryChangeReasonUpgrade])
 	tt.NotZero(upgradedLedgerChangeMap[ingest.LedgerEntryChangeReasonUpgrade])
+	for _, change := range upgradedLedgerChanges {
+		tt.Equal(change.Ledger.LedgerSequence(), upgradedLedgerSeq)
+		tt.Empty(change.Transaction)
+		tt.NotEmpty(change.LedgerUpgrade)
+	}
+}
+
+func TestOneTxOneOperationChanges(t *testing.T) {
+	tt := assert.New(t)
+	itest := integration.NewTest(t, integration.Config{})
+
+	master := itest.Master()
+	keys, _ := itest.CreateAccounts(2, "1000")
+	keyA, keyB := keys[0], keys[1]
+
+	operation := txnbuild.Payment{
+		SourceAccount: keyA.Address(),
+		Destination:   keyB.Address(),
+		Asset:         txnbuild.NativeAsset{},
+		Amount:        "900",
+	}
+	txResp, err := itest.SubmitMultiSigOperations(itest.MasterAccount(), []*keypair.Full{master, keyA}, &operation)
+	if err != nil {
+		t.Fatalf("failed to submit transaction: %v", err)
+	}
+	ledgerSeq := uint32(txResp.Ledger)
+
+	archive, err := integration.GetHistoryArchive()
+	tt.NoError(err)
+
+	publishedNextCheckpoint := publishedNextCheckpoint(archive, ledgerSeq, t)
+	// Ensure that a checkpoint has been created with the ledgerNumber you want in it
+	tt.Eventually(publishedNextCheckpoint, 15*time.Second, time.Second)
+
+	ledger := getLedgersFromArchive(itest, ledgerSeq, ledgerSeq)[ledgerSeq]
+	changes := getChangesFromLedger(itest, ledger)
+
+	reasonCntMap := changeReasonCountMap(changes)
+	tt.Equal(2, reasonCntMap[ingest.LedgerEntryChangeReasonOperation])
+	tt.Equal(1, reasonCntMap[ingest.LedgerEntryChangeReasonTransaction])
+	tt.Equal(1, reasonCntMap[ingest.LedgerEntryChangeReasonFee])
+
+	reasonToChangeMap := changeReasonToChangeMap(changes)
+	// Assert Transaction Hash and Ledger Sequence are accurate in all changes
+	for _, change := range changes {
+		tt.Equal(change.Transaction.Hash.HexString(), txResp.Hash)
+		tt.Equal(change.Transaction.Ledger.LedgerSequence(), ledgerSeq)
+	}
+
+	tt.Equal(
+		ledgerKey(reasonToChangeMap[ingest.LedgerEntryChangeReasonFee][0]).MustAccount().AccountId.Address(),
+		master.Address())
+	tt.Equal(
+		ledgerKey(reasonToChangeMap[ingest.LedgerEntryChangeReasonTransaction][0]).MustAccount().AccountId.Address(),
+		master.Address())
+	tt.True(containsAccount(t, reasonToChangeMap[ingest.LedgerEntryChangeReasonOperation], keyA.Address()))
+	tt.True(containsAccount(t, reasonToChangeMap[ingest.LedgerEntryChangeReasonOperation], keyB.Address()))
+	tt.False(containsAccount(t, reasonToChangeMap[ingest.LedgerEntryChangeReasonOperation], master.Address()))
+}
+
+// Helper function to check if a specific XX exists in the list
+func containsAccount(t *testing.T, slice []ingest.Change, target string) bool {
+	t.Logf("Target: %v", target)
+	for _, change := range slice {
+		addr := ledgerKey(change).MustAccount().AccountId.Address()
+		t.Logf("Comparing address: %v with target: %v", addr, target)
+		if addr == target {
+			return true
+		}
+	}
+	return false
+}
+
+func ledgerKey(c ingest.Change) xdr.LedgerKey {
+	var l xdr.LedgerKey
+	var err error
+	if c.Pre != nil {
+		l, err = c.Pre.LedgerKey()
+	}
+	l, err = c.Post.LedgerKey()
+	if err != nil {
+		panic(err)
+	}
+	return l
 }
 
 func getChangesFromLedger(itest *integration.Test, ledger xdr.LedgerCloseMeta) []ingest.Change {
@@ -104,10 +179,31 @@ func getLedgersFromArchive(itest *integration.Test, startingLedger uint32, endLe
 	return seqToLedgersMap
 }
 
-func changeMap(changes []ingest.Change) map[ingest.LedgerEntryChangeReason]int {
+func changeReasonCountMap(changes []ingest.Change) map[ingest.LedgerEntryChangeReason]int {
 	changeMap := make(map[ingest.LedgerEntryChangeReason]int)
 	for _, change := range changes {
 		changeMap[change.Reason]++
 	}
 	return changeMap
 }
+
+func publishedNextCheckpoint(archive *historyarchive.Archive, ledgerSeq uint32, t *testing.T) func() bool {
+	return func() bool {
+		var latestCheckpoint uint32
+		has, requestErr := archive.GetRootHAS()
+		if requestErr != nil {
+			t.Logf("Request to fetch checkpoint failed: %v", requestErr)
+			return false
+		}
+		latestCheckpoint = has.CurrentLedger
+		return latestCheckpoint >= ledgerSeq
+	}
+}
+
+func changeReasonToChangeMap(changes []ingest.Change) map[ingest.LedgerEntryChangeReason][]ingest.Change {
+	changeMap := make(map[ingest.LedgerEntryChangeReason][]ingest.Change)
+	for _, change := range changes {
+		changeMap[change.Reason] = append(changeMap[change.Reason], change)
+	}
+	return changeMap
+}

services/horizon/internal/integration/db_test.go

@@ -15,7 +15,6 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/stellar/go/clients/horizonclient"
-	"github.com/stellar/go/historyarchive"
 	"github.com/stellar/go/keypair"
 	hProtocol "github.com/stellar/go/protocols/horizon"
 	horizoncmd "github.com/stellar/go/services/horizon/cmd"
@@ -507,12 +506,7 @@ func TestReingestDB(t *testing.T) {
 	// cannot ingest past the most recent checkpoint ledger when using captive
 	// core
 	toLedger := uint32(reachedLedger)
-	archive, err := historyarchive.Connect(
-		horizonConfig.HistoryArchiveURLs[0],
-		historyarchive.ArchiveOptions{
-			NetworkPassphrase:   horizonConfig.NetworkPassphrase,
-			CheckpointFrequency: horizonConfig.CheckpointFrequency,
-		})
+	archive, err := integration.GetHistoryArchive()
 	tt.NoError(err)
 
 	// make sure a full checkpoint has elapsed otherwise there will be nothing to reingest
@@ -635,12 +629,7 @@ func TestReingestDBWithFilterRules(t *testing.T) {
 	itest, _ := initializeDBIntegrationTest(t)
 	tt := assert.New(t)
 
-	archive, err := historyarchive.Connect(
-		itest.GetHorizonIngestConfig().HistoryArchiveURLs[0],
-		historyarchive.ArchiveOptions{
-			NetworkPassphrase:   itest.GetHorizonIngestConfig().NetworkPassphrase,
-			CheckpointFrequency: itest.GetHorizonIngestConfig().CheckpointFrequency,
-		})
+	archive, err := integration.GetHistoryArchive()
 	tt.NoError(err)
 
 	// make sure one full checkpoint has elapsed before making ledger entries
@@ -879,12 +868,7 @@ func TestFillGaps(t *testing.T) {
 	// cap reachedLedger to the nearest checkpoint ledger because reingest range cannot ingest past the most
 	// recent checkpoint ledger when using captive core
 	toLedger := uint32(reachedLedger)
-	archive, err := historyarchive.Connect(
-		horizonConfig.HistoryArchiveURLs[0],
-		historyarchive.ArchiveOptions{
-			NetworkPassphrase:   horizonConfig.NetworkPassphrase,
-			CheckpointFrequency: horizonConfig.CheckpointFrequency,
-		})
+	archive, err := integration.GetHistoryArchive()
 	tt.NoError(err)
 
 	t.Run("validate parallel range", func(t *testing.T) {

services/horizon/internal/test/integration/integration.go

@@ -4,6 +4,7 @@ package integration
 import (
 	"context"
 	"fmt"
+	"github.com/stellar/go/historyarchive"
 	"github.com/stellar/go/ingest/ledgerbackend"
 	"io/ioutil"
 	"os"
@@ -1492,3 +1493,12 @@ func GetCoreMaxSupportedProtocol() uint32 {
 func (i *Test) GetEffectiveProtocolVersion() uint32 {
 	return i.config.ProtocolVersion
 }
+
+func GetHistoryArchive() (*historyarchive.Archive, error) {
+	return historyarchive.Connect(
+		HistoryArchiveUrl,
+		historyarchive.ArchiveOptions{
+			NetworkPassphrase:   StandaloneNetworkPassphrase,
+			CheckpointFrequency: CheckpointFrequency,
+		})
+}