test: Expand tests for Peer subscribe actions

net/peer.go

@@ -758,7 +758,10 @@ type EvtPubSub struct {
 	Peer peer.ID
 }
 
-// AddP2PCollectionTopic adds the collectionID to the pubsup topics
+// AddP2PCollections adds the given collectionIDs to the pubsup topics.
+//
+// It will error if any of the given collectionIDs are invalid, in such a case some of the
+// changes to the server may still be applied.
 func (p *Peer) AddP2PCollections(collections []string) error {
 	txn, err := p.db.NewTxn(p.ctx, false)
 	if err != nil {
@@ -803,7 +806,10 @@ func (p *Peer) AddP2PCollections(collections []string) error {
 	return txn.Commit(p.ctx)
 }
 
-// RemoveP2PCollectionTopics adds the collectionID from the pubsup topics
+// RemoveP2PCollections removes the given collectionIDs from the pubsup topics.
+//
+// It will error if any of the given collectionIDs are invalid, in such a case some of the
+// changes to the server may still be applied.
 func (p *Peer) RemoveP2PCollections(collections []string) error {
 	txn, err := p.db.NewTxn(p.ctx, false)
 	if err != nil {
@@ -861,7 +867,7 @@ func (p *Peer) GetAllP2PCollections() ([]client.P2PCollection, error) {
 		return nil, err
 	}
 
-	var p2pCols []client.P2PCollection
+	p2pCols := []client.P2PCollection{}
 	for _, colID := range collections {
 		col, err := store.GetCollectionBySchemaID(p.ctx, colID)
 		if err != nil {

tests/integration/net/state/simple/peer/subscribe/with_add_get_remove_test.go

@@ -0,0 +1,89 @@
+// Copyright 2022 Democratized Data Foundation
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package subscribe_test
+
+import (
+	"testing"
+
+	testUtils "github.com/sourcenetwork/defradb/tests/integration"
+)
+
+func TestP2PSubscribeAddRemoveGetSingle(t *testing.T) {
+	test := testUtils.TestCase{
+		Actions: []any{
+			testUtils.RandomNetworkingConfig(),
+			testUtils.RandomNetworkingConfig(),
+			testUtils.SchemaUpdate{
+				Schema: `
+					type Users {
+						name: String
+					}
+				`,
+			},
+			testUtils.ConnectPeers{
+				SourceNodeID: 1,
+				TargetNodeID: 0,
+			},
+			testUtils.SubscribeToCollection{
+				NodeID:        1,
+				CollectionIDs: []int{0},
+			},
+			testUtils.UnsubscribeToCollection{
+				NodeID:        1,
+				CollectionIDs: []int{0},
+			},
+			testUtils.GetAllP2PCollections{
+				NodeID:                1,
+				ExpectedCollectionIDs: []int{},
+			},
+		},
+	}
+
+	testUtils.ExecuteTestCase(t, []string{"Users"}, test)
+}
+
+func TestP2PSubscribeAddRemoveGetMultiple(t *testing.T) {
+	test := testUtils.TestCase{
+		Actions: []any{
+			testUtils.RandomNetworkingConfig(),
+			testUtils.RandomNetworkingConfig(),
+			testUtils.SchemaUpdate{
+				Schema: `
+					type Users {
+						name: String
+					}
+					type Giraffes {
+						name: String
+					}
+				`,
+			},
+			testUtils.ConnectPeers{
+				SourceNodeID: 1,
+				TargetNodeID: 0,
+			},
+			testUtils.SubscribeToCollection{
+				NodeID:        1,
+				CollectionIDs: []int{0, 1},
+			},
+			testUtils.UnsubscribeToCollection{
+				NodeID: 1,
+				// Unsubscribe from Users, but remain subscribed to Giraffes
+				CollectionIDs: []int{0},
+			},
+			testUtils.GetAllP2PCollections{
+				NodeID:                1,
+				ExpectedCollectionIDs: []int{1},
+			},
+		},
+	}
+
+	testUtils.ExecuteTestCase(t, []string{"Users", "Giraffes"}, test)
+}

tests/integration/net/state/simple/peer/subscribe/with_add_get_test.go

@@ -0,0 +1,83 @@
+// Copyright 2022 Democratized Data Foundation
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package subscribe_test
+
+import (
+	"testing"
+
+	testUtils "github.com/sourcenetwork/defradb/tests/integration"
+)
+
+func TestP2PSubscribeAddGetSingle(t *testing.T) {
+	test := testUtils.TestCase{
+		Actions: []any{
+			testUtils.RandomNetworkingConfig(),
+			testUtils.RandomNetworkingConfig(),
+			testUtils.SchemaUpdate{
+				Schema: `
+					type Users {
+						name: String
+					}
+				`,
+			},
+			testUtils.ConnectPeers{
+				SourceNodeID: 1,
+				TargetNodeID: 0,
+			},
+			testUtils.SubscribeToCollection{
+				NodeID:        1,
+				CollectionIDs: []int{0},
+			},
+			testUtils.GetAllP2PCollections{
+				NodeID:                1,
+				ExpectedCollectionIDs: []int{0},
+			},
+		},
+	}
+
+	testUtils.ExecuteTestCase(t, []string{"Users"}, test)
+}
+
+func TestP2PSubscribeAddGetMultiple(t *testing.T) {
+	test := testUtils.TestCase{
+		Actions: []any{
+			testUtils.RandomNetworkingConfig(),
+			testUtils.RandomNetworkingConfig(),
+			testUtils.SchemaUpdate{
+				Schema: `
+					type Users {
+						name: String
+					}
+					type Giraffes {
+						name: String
+					}
+					type Bears {
+						name: String
+					}
+				`,
+			},
+			testUtils.ConnectPeers{
+				SourceNodeID: 1,
+				TargetNodeID: 0,
+			},
+			testUtils.SubscribeToCollection{
+				NodeID:        1,
+				CollectionIDs: []int{0, 2},
+			},
+			testUtils.GetAllP2PCollections{
+				NodeID:                1,
+				ExpectedCollectionIDs: []int{2, 0},
+			},
+		},
+	}
+
+	testUtils.ExecuteTestCase(t, []string{"Users", "Giraffes", "Bears"}, test)
+}