Add Fred's changes

tests/integration/acp.go

@@ -113,7 +113,7 @@ func addPolicyACP(
 	nodeIDs, nodes := getNodesWithIDs(action.NodeID, s.nodes)
 	for index, node := range nodes {
 		ctx := getContextWithIdentity(s.ctx, s, action.Identity, nodeIDs[index])
-		policyResult, err := node.DB.AddPolicy(ctx, action.Policy)
+		policyResult, err := node.AddPolicy(ctx, action.Policy)
 
 		expectedErrorRaised := AssertError(s.t, s.testCase.Description, err, action.ExpectedError)
 		assertExpectedErrorRaised(s.t, s.testCase.Description, action.ExpectedError, expectedErrorRaised)
@@ -190,7 +190,7 @@ func addDocActorRelationshipACP(
 		var collectionName string
 		collectionName, docID = getCollectionAndDocInfo(s, action.CollectionID, action.DocID, nodeID)
 
-		exists, err := node.DB.AddDocActorRelationship(
+		exists, err := node.AddDocActorRelationship(
 			getContextWithIdentity(s.ctx, s, action.RequestorIdentity, nodeID),
 			collectionName,
 			docID,
@@ -276,7 +276,7 @@ func deleteDocActorRelationshipACP(
 
 		collectionName, docID := getCollectionAndDocInfo(s, action.CollectionID, action.DocID, nodeID)
 
-		deleteDocActorRelationshipResult, err := node.DB.DeleteDocActorRelationship(
+		deleteDocActorRelationshipResult, err := node.DeleteDocActorRelationship(
 			getContextWithIdentity(s.ctx, s, action.RequestorIdentity, nodeID),
 			collectionName,
 			docID,
@@ -617,7 +617,7 @@ func getNodeAudience(s *state, nodeIndex int) immutable.Option[string] {
 	if nodeIndex >= len(s.nodes) {
 		return immutable.None[string]()
 	}
-	switch client := s.nodes[nodeIndex].DB.(type) {
+	switch client := s.nodes[nodeIndex].Client.(type) {
 	case *http.Wrapper:
 		return immutable.Some(strings.TrimPrefix(client.Host(), "http://"))
 	case *cli.Wrapper:

tests/integration/db.go

@@ -205,11 +205,6 @@ func setupNode(s *state, opts ...node.Option) (*nodeState, error) {
 	}
 
 	if s.isNetworkEnabled {
-		var addresses []string
-		for _, node := range s.nodes {
-			addresses = append(addresses, node.Peer.PeerInfo().String())
-		}
-		netOpts = append(netOpts, net.WithListenAddresses(addresses...))
 		opts = append(opts, node.WithDisableP2P(false))
 	}
 
@@ -230,12 +225,16 @@ func setupNode(s *state, opts ...node.Option) (*nodeState, error) {
 	require.NoError(s.t, err)
 
 	st := &nodeState{
-		Node:    node,
+		Client:  c,
 		event:   eventState,
 		p2p:     newP2PState(),
 		dbPath:  path,
 		netOpts: netOpts,
 	}
 
+	if node.Peer != nil {
+		st.peerInfo = node.Peer.PeerInfo()
+	}
+
 	return st, nil
 }

tests/integration/events.go

@@ -28,10 +28,10 @@ const eventTimeout = 1 * time.Second
 // waitForNetworkSetupEvents waits for p2p topic completed and
 // replicator completed events to be published on the local node event bus.
 func waitForNetworkSetupEvents(s *state, nodeID int) {
-	cols, err := s.nodes[nodeID].DB.GetAllP2PCollections(s.ctx)
+	cols, err := s.nodes[nodeID].GetAllP2PCollections(s.ctx)
 	require.NoError(s.t, err)
 
-	reps, err := s.nodes[nodeID].DB.GetAllReplicators(s.ctx)
+	reps, err := s.nodes[nodeID].GetAllReplicators(s.ctx)
 	require.NoError(s.t, err)
 
 	replicatorEvents := len(reps)

tests/integration/explain.go

@@ -135,7 +135,7 @@ func executeExplainRequest(
 
 	_, nodes := getNodesWithIDs(action.NodeID, s.nodes)
 	for _, node := range nodes {
-		result := node.DB.ExecRequest(
+		result := node.ExecRequest(
 			s.ctx,
 			action.Request,
 		)

tests/integration/lens.go

@@ -59,10 +59,10 @@ func configureMigration(
 ) {
 	_, nodes := getNodesWithIDs(action.NodeID, s.nodes)
 	for _, node := range nodes {
-		txn := getTransaction(s, node.DB, action.TransactionID, action.ExpectedError)
+		txn := getTransaction(s, node.Client, action.TransactionID, action.ExpectedError)
 		ctx := db.SetContextTxn(s.ctx, txn)
 
-		err := node.DB.SetMigration(ctx, action.LensConfig)
+		err := node.SetMigration(ctx, action.LensConfig)
 		expectedErrorRaised := AssertError(s.t, s.testCase.Description, err, action.ExpectedError)
 
 		assertExpectedErrorRaised(s.t, s.testCase.Description, action.ExpectedError, expectedErrorRaised)

tests/integration/p2p.go

@@ -150,10 +150,10 @@ func connectPeers(
 	targetNode := s.nodes[cfg.TargetNodeID]
 
 	log.InfoContext(s.ctx, "Connect peers",
-		corelog.Any("Source", sourceNode.DB.PeerInfo()),
-		corelog.Any("Target", targetNode.DB.PeerInfo()))
+		corelog.Any("Source", sourceNode.PeerInfo()),
+		corelog.Any("Target", targetNode.PeerInfo()))
 
-	err := sourceNode.Peer.Connect(s.ctx, targetNode.DB.PeerInfo())
+	err := sourceNode.Connect(s.ctx, targetNode.PeerInfo())
 	require.NoError(s.t, err)
 
 	s.nodes[cfg.SourceNodeID].p2p.connections[cfg.TargetNodeID] = struct{}{}
@@ -177,8 +177,8 @@ func configureReplicator(
 	sourceNode := s.nodes[cfg.SourceNodeID]
 	targetNode := s.nodes[cfg.TargetNodeID]
 
-	err := sourceNode.DB.SetReplicator(s.ctx, client.ReplicatorParams{
-		Info: targetNode.DB.PeerInfo(),
+	err := sourceNode.SetReplicator(s.ctx, client.ReplicatorParams{
+		Info: targetNode.PeerInfo(),
 	})
 
 	expectedErrorRaised := AssertError(s.t, s.testCase.Description, err, cfg.ExpectedError)
@@ -196,8 +196,8 @@ func deleteReplicator(
 	sourceNode := s.nodes[cfg.SourceNodeID]
 	targetNode := s.nodes[cfg.TargetNodeID]
 
-	err := sourceNode.DB.DeleteReplicator(s.ctx, client.ReplicatorParams{
-		Info: targetNode.DB.PeerInfo(),
+	err := sourceNode.DeleteReplicator(s.ctx, client.ReplicatorParams{
+		Info: targetNode.PeerInfo(),
 	})
 	require.NoError(s.t, err)
 	waitForReplicatorDeleteEvent(s, cfg)
@@ -223,7 +223,7 @@ func subscribeToCollection(
 		schemaRoots = append(schemaRoots, col.SchemaRoot())
 	}
 
-	err := n.DB.AddP2PCollections(s.ctx, schemaRoots)
+	err := n.AddP2PCollections(s.ctx, schemaRoots)
 	if err == nil {
 		waitForSubscribeToCollectionEvent(s, action)
 	}
@@ -257,7 +257,7 @@ func unsubscribeToCollection(
 		schemaRoots = append(schemaRoots, col.SchemaRoot())
 	}
 
-	err := n.DB.RemoveP2PCollections(s.ctx, schemaRoots)
+	err := n.RemoveP2PCollections(s.ctx, schemaRoots)
 	if err == nil {
 		waitForUnsubscribeToCollectionEvent(s, action)
 	}
@@ -286,7 +286,7 @@ func getAllP2PCollections(
 	}
 
 	n := s.nodes[action.NodeID]
-	cols, err := n.DB.GetAllP2PCollections(s.ctx)
+	cols, err := n.GetAllP2PCollections(s.ctx)
 	require.NoError(s.t, err)
 
 	assert.Equal(s.t, expectedCollections, cols)
@@ -300,10 +300,10 @@ func reconnectPeers(s *state) {
 			targetNode := s.nodes[j]
 
 			log.InfoContext(s.ctx, "Connect peers",
-				corelog.Any("Source", sourceNode.DB.PeerInfo()),
-				corelog.Any("Target", targetNode.DB.PeerInfo()))
+				corelog.Any("Source", sourceNode.PeerInfo()),
+				corelog.Any("Target", targetNode.PeerInfo()))
 
-			err := sourceNode.Peer.Connect(s.ctx, targetNode.DB.PeerInfo())
+			err := sourceNode.Connect(s.ctx, targetNode.PeerInfo())
 			require.NoError(s.t, err)
 		}
 	}

tests/integration/state.go

@@ -15,12 +15,14 @@ import (
 	"testing"
 
 	"github.com/ipfs/go-cid"
+	"github.com/libp2p/go-libp2p/core/peer"
 
 	"github.com/sourcenetwork/defradb/client"
 	"github.com/sourcenetwork/defradb/datastore"
 	"github.com/sourcenetwork/defradb/event"
 	"github.com/sourcenetwork/defradb/net"
 	"github.com/sourcenetwork/defradb/node"
+	"github.com/sourcenetwork/defradb/tests/clients"
 )
 
 // p2pState contains all p2p related testing state.
@@ -114,8 +116,8 @@ func newEventState(bus *event.Bus) (*eventState, error) {
 
 // nodeState contains all testing state for a node.
 type nodeState struct {
-	// The node active in this test.
-	*node.Node
+	// The node's client active in this test.
+	clients.Client
 	// event contains all event node subscriptions.
 	event *eventState
 	// p2p contains p2p states for the node.
@@ -131,6 +133,8 @@ type nodeState struct {
 	indexes [][]client.IndexDescription
 	// indicates if the node is closed.
 	closed bool
+	// peerInfo contains the peer information for the node.
+	peerInfo peer.AddrInfo
 }
 
 // state contains all testing state.