CBG-4255 expand interface for CRUD operations (#7139)

rest/blip_api_attachment_test.go

@@ -293,7 +293,7 @@ func TestBlipPushPullNewAttachmentCommonAncestor(t *testing.T) {
 		rt := NewRestTester(t, &rtConfig)
 		defer rt.Close()
 
-		opts := &BlipTesterClientOpts{SupportedBLIPProtocols: SupportedBLIPProtocols}
+		opts := &BlipTesterClientOpts{SupportedBLIPProtocols: SupportedBLIPProtocols, SourceID: "abc"}
 		btc := btcRunner.NewBlipTesterClientOptsWithRT(rt, opts)
 		defer btc.Close()
 
@@ -375,7 +375,7 @@ func TestBlipPushPullNewAttachmentNoCommonAncestor(t *testing.T) {
 		rt := NewRestTester(t, &rtConfig)
 		defer rt.Close()
 
-		opts := &BlipTesterClientOpts{SupportedBLIPProtocols: SupportedBLIPProtocols}
+		opts := &BlipTesterClientOpts{SupportedBLIPProtocols: SupportedBLIPProtocols, SourceID: "abc"}
 		btc := btcRunner.NewBlipTesterClientOptsWithRT(rt, opts)
 		defer btc.Close()
 		btcRunner.StartPull(btc.id)

rest/utilities_testing_blip_client.go

@@ -56,6 +56,9 @@ type BlipTesterClientOpts struct {
 
 	// sendReplacementRevs opts into the replacement rev behaviour in the event that we do not find the requested one.
 	sendReplacementRevs bool
+
+	// SourceID is used to define the SourceID for the blip client
+	SourceID string
 }
 
 // BlipTesterClient is a fully fledged client to emulate CBL behaviour on both push and pull replications through methods on this type.
@@ -653,6 +656,9 @@ func (btcRunner *BlipTestClientRunner) NewBlipTesterClientOptsWithRT(rt *RestTes
 	if !opts.AllowCreationWithoutBlipTesterClientRunner && !btcRunner.initialisedInsideRunnerCode {
 		require.FailNow(btcRunner.TB(), "must initialise BlipTesterClient inside Run() method")
 	}
+	if opts.SourceID == "" {
+		opts.SourceID = "blipclient"
+	}
 	id, err := uuid.NewRandom()
 	require.NoError(btcRunner.TB(), err)
 
@@ -964,7 +970,7 @@ func (btc *BlipTesterCollectionClient) PushRevWithHistory(docID, parentRev strin
 			startValue = parentVersion.Value
 			revisionHistory = append(revisionHistory, parentRev)
 		}
-		newVersion := db.DecodedVersion{SourceID: "abc", Value: startValue + uint64(revCount)}
+		newVersion := db.DecodedVersion{SourceID: btc.parent.SourceID, Value: startValue + uint64(revCount)}
 		newRevID = newVersion.String()
 
 	} else {
@@ -975,14 +981,14 @@ func (btc *BlipTesterCollectionClient) PushRevWithHistory(docID, parentRev strin
 		revGen = parentRevGen + revCount + prunedRevCount
 
 		for i := revGen - 1; i > parentRevGen; i-- {
-			rev := fmt.Sprintf("%d-%s", i, "abc")
+			rev := fmt.Sprintf("%d-%s", i, btc.parent.SourceID)
 			revisionHistory = append(revisionHistory, rev)
 		}
 		if parentRev != "" {
 
 			revisionHistory = append(revisionHistory, parentRev)
 		}
-		newRevID = fmt.Sprintf("%d-%s", revGen, "abc")
+		newRevID = fmt.Sprintf("%d-%s", revGen, btc.parent.SourceID)
 	}
 
 	// Inline attachment processing

topologytest/couchbase_lite_mock_peer_test.go

@@ -60,6 +60,11 @@ func (p *CouchbaseLiteMockPeer) getSingleBlipClient() *PeerBlipTesterClient {
 	return nil
 }
 
+// CreateDocument creates a document on the peer. The test will fail if the document already exists.
+func (p *CouchbaseLiteMockPeer) CreateDocument(dsName sgbucket.DataStoreName, docID string, body []byte) rest.DocVersion {
+	return rest.EmptyDocVersion()
+}
+
 // WriteDocument writes a document to the peer. The test will fail if the write does not succeed.
 func (p *CouchbaseLiteMockPeer) WriteDocument(dsName sgbucket.DataStoreName, docID string, body []byte) rest.DocVersion {
 	// this isn't yet collection aware, using single default collection
@@ -70,6 +75,11 @@ func (p *CouchbaseLiteMockPeer) WriteDocument(dsName sgbucket.DataStoreName, doc
 	return docVersion
 }
 
+// DeleteDocument deletes a document on the peer. The test will fail if the document does not exist.
+func (p *CouchbaseLiteMockPeer) DeleteDocument(dsName sgbucket.DataStoreName, docID string) rest.DocVersion {
+	return rest.EmptyDocVersion()
+}
+
 // WaitForDocVersion waits for a document to reach a specific version. The test will fail if the document does not reach the expected version in 20s.
 func (p *CouchbaseLiteMockPeer) WaitForDocVersion(dsName sgbucket.DataStoreName, docID string, expected rest.DocVersion) db.Body {
 	// this isn't yet collection aware, using single default collection
@@ -111,6 +121,7 @@ func (p *CouchbaseLiteMockPeer) CreateReplication(peer Peer, config PeerReplicat
 		Channels:               []string{"*"},
 		SupportedBLIPProtocols: []string{db.CBMobileReplicationV4.SubprotocolString()},
 		AllowCreationWithoutBlipTesterClientRunner: true,
+		SourceID: peer.SourceID(),
 	},
 	)
 	p.blipClients[sg.String()] = &PeerBlipTesterClient{
@@ -120,6 +131,11 @@ func (p *CouchbaseLiteMockPeer) CreateReplication(peer Peer, config PeerReplicat
 	return replication
 }
 
+// SourceID returns the source ID for the peer used in <val>@sourceID.
+func (r *CouchbaseLiteMockPeer) SourceID() string {
+	return r.name
+}
+
 // GetBackingBucket returns the backing bucket for the peer. This is always nil.
 func (p *CouchbaseLiteMockPeer) GetBackingBucket() base.Bucket {
 	return nil

topologytest/couchbase_server_peer_test.go

@@ -27,6 +27,7 @@ import (
 type CouchbaseServerPeer struct {
 	tb               testing.TB
 	bucket           base.Bucket
+	sourceID         string
 	pullReplications map[Peer]xdcr.Manager
 	pushReplications map[Peer]xdcr.Manager
 	name             string
@@ -85,13 +86,23 @@ func (p *CouchbaseServerPeer) GetDocument(dsName sgbucket.DataStoreName, docID s
 	return rest.EmptyDocVersion(), body
 }
 
+// CreateDocument creates a document on the peer. The test will fail if the document already exists.
+func (p *CouchbaseServerPeer) CreateDocument(dsName sgbucket.DataStoreName, docID string, body []byte) rest.DocVersion {
+	return rest.EmptyDocVersion()
+}
+
 // WriteDocument writes a document to the peer. The test will fail if the write does not succeed.
 func (p *CouchbaseServerPeer) WriteDocument(dsName sgbucket.DataStoreName, docID string, body []byte) rest.DocVersion {
 	err := p.getCollection(dsName).Set(docID, 0, nil, body)
 	require.NoError(p.tb, err)
 	return rest.EmptyDocVersion()
 }
 
+// DeleteDocument deletes a document on the peer. The test will fail if the document does not exist.
+func (p *CouchbaseServerPeer) DeleteDocument(dsName sgbucket.DataStoreName, docID string) rest.DocVersion {
+	return rest.EmptyDocVersion()
+}
+
 // WaitForDocVersion waits for a document to reach a specific version. The test will fail if the document does not reach the expected version in 20s.
 func (p *CouchbaseServerPeer) WaitForDocVersion(dsName sgbucket.DataStoreName, docID string, expected rest.DocVersion) db.Body {
 	var docBytes []byte
@@ -162,6 +173,11 @@ func (p *CouchbaseServerPeer) CreateReplication(passivePeer Peer, config PeerRep
 	return nil
 }
 
+// SourceID returns the source ID for the peer used in <val>@sourceID.
+func (r *CouchbaseServerPeer) SourceID() string {
+	return r.sourceID
+}
+
 // GetBackingBucket returns the backing bucket for the peer.
 func (p *CouchbaseServerPeer) GetBackingBucket() base.Bucket {
 	return p.bucket

topologytest/peer_test.go

@@ -25,8 +25,12 @@ import (
 type Peer interface {
 	// GetDocument returns the latest version of a document. The test will fail the document does not exist.
 	GetDocument(dsName sgbucket.DataStoreName, docID string) (rest.DocVersion, db.Body)
-	// WriteDocument writes a document to the peer. The test will fail if the write does not succeed.
+	// CreateDocument creates a document on the peer. The test will fail if the document already exists.
+	CreateDocument(dsName sgbucket.DataStoreName, docID string, body []byte) rest.DocVersion
+	// WriteDocument upserts a document to the peer. The test will fail if the write does not succeed. Reasons for failure might be sync function rejections for Sync Gateway rejections.
 	WriteDocument(dsName sgbucket.DataStoreName, docID string, body []byte) rest.DocVersion
+	// DeleteDocument deletes a document on the peer. The test will fail if the document does not exist.
+	DeleteDocument(dsName sgbucket.DataStoreName, docID string) rest.DocVersion
 
 	// WaitForDocVersion waits for a document to reach a specific version. The test will fail if the document does not reach the expected version in 20s.
 	WaitForDocVersion(dsName sgbucket.DataStoreName, docID string, expected rest.DocVersion) db.Body
@@ -40,6 +44,9 @@ type Peer interface {
 	// Close will shut down the peer and close any active replications on the peer.
 	Close()
 
+	// SourceID returns the source ID for the peer used in <val>@sourceID.
+	SourceID() string
+
 	// GetBackingBucket returns the backing bucket for the peer. This is nil when the peer is a Couchbase Lite peer.
 	GetBackingBucket() base.Bucket
 }
@@ -133,10 +140,13 @@ func NewPeer(t *testing.T, name string, buckets map[PeerBucketID]*base.TestBucke
 	case PeerTypeCouchbaseServer:
 		bucket, ok := buckets[opts.BucketID]
 		require.True(t, ok, "bucket not found for bucket ID %d", opts.BucketID)
+		sourceID, err := xdcr.GetSourceID(base.TestCtx(t), bucket)
+		require.NoError(t, err)
 		return &CouchbaseServerPeer{
 			name:             name,
 			tb:               t,
 			bucket:           bucket,
+			sourceID:         sourceID,
 			pullReplications: make(map[Peer]xdcr.Manager),
 			pushReplications: make(map[Peer]xdcr.Manager),
 		}

topologytest/sync_gateway_peer_test.go

@@ -50,12 +50,22 @@ func (p *SyncGatewayPeer) GetDocument(dsName sgbucket.DataStoreName, docID strin
 	return p.rt.GetDoc(docID)
 }
 
+// CreateDocument creates a document on the peer. The test will fail if the document already exists.
+func (p *SyncGatewayPeer) CreateDocument(dsName sgbucket.DataStoreName, docID string, body []byte) rest.DocVersion {
+	return rest.EmptyDocVersion()
+}
+
 // WriteDocument writes a document to the peer. The test will fail if the write does not succeed.
 func (p *SyncGatewayPeer) WriteDocument(dsName sgbucket.DataStoreName, docID string, body []byte) rest.DocVersion {
 	// this function is not yet collections aware
 	return p.rt.PutDoc(docID, string(body))
 }
 
+// DeleteDocument deletes a document on the peer. The test will fail if the document does not exist.
+func (p *SyncGatewayPeer) DeleteDocument(dsName sgbucket.DataStoreName, docID string) rest.DocVersion {
+	return rest.EmptyDocVersion()
+}
+
 // WaitForDocVersion waits for a document to reach a specific version. The test will fail if the document does not reach the expected version in 20s.
 func (p *SyncGatewayPeer) WaitForDocVersion(dsName sgbucket.DataStoreName, docID string, expected rest.DocVersion) db.Body {
 	// this function is not yet collections aware
@@ -88,6 +98,11 @@ func (p *SyncGatewayPeer) CreateReplication(peer Peer, config PeerReplicationCon
 	return nil
 }
 
+// SourceID returns the source ID for the peer used in <val>@sourceID.
+func (r *SyncGatewayPeer) SourceID() string {
+	return r.rt.GetDatabase().EncodedSourceID
+}
+
 // GetBackingBucket returns the backing bucket for the peer.
 func (p *SyncGatewayPeer) GetBackingBucket() base.Bucket {
 	return p.rt.Bucket()

xdcr/replication.go

@@ -77,8 +77,8 @@ func NewXDCR(ctx context.Context, fromBucket, toBucket base.Bucket, opts XDCROpt
 	return newCouchbaseServerManager(ctx, gocbFromBucket, gocbToBucket, opts)
 }
 
-// getSourceID returns the source ID for a bucket.
-func getSourceID(ctx context.Context, bucket base.Bucket) (string, error) {
+// GetSourceID returns the source ID for a bucket.
+func GetSourceID(ctx context.Context, bucket base.Bucket) (string, error) {
 	serverUUID, err := db.GetServerUUID(ctx, bucket)
 	if err != nil {
 		return "", err

xdcr/rosmar_xdcr.go

@@ -42,7 +42,7 @@ func newRosmarManager(ctx context.Context, fromBucket, toBucket *rosmar.Bucket,
 	if opts.Mobile != MobileOn {
 		return nil, errors.New("Only sgbucket.XDCRMobileOn is supported in rosmar")
 	}
-	fromBucketSourceID, err := getSourceID(ctx, fromBucket)
+	fromBucketSourceID, err := GetSourceID(ctx, fromBucket)
 	if err != nil {
 		return nil, fmt.Errorf("Could not get source ID for %s: %w", fromBucket.GetName(), err)
 	}

xdcr/xdcr_test.go

@@ -69,7 +69,7 @@ func TestMobileXDCRNoSyncDataCopied(t *testing.T) {
 		fromDs = fromBucket.DefaultDataStore()
 		toDs = toBucket.DefaultDataStore()
 	}
-	fromBucketSourceID, err := getSourceID(ctx, fromBucket)
+	fromBucketSourceID, err := GetSourceID(ctx, fromBucket)
 	require.NoError(t, err)
 	docCas := make(map[string]uint64)
 	for _, doc := range []string{syncDoc, attachmentDoc, normalDoc} {
@@ -149,7 +149,7 @@ func getTwoBucketDataStores(t *testing.T) (base.Bucket, sgbucket.DataStore, base
 func TestReplicateVV(t *testing.T) {
 	fromBucket, fromDs, toBucket, toDs := getTwoBucketDataStores(t)
 	ctx := base.TestCtx(t)
-	fromBucketSourceID, err := getSourceID(ctx, fromBucket)
+	fromBucketSourceID, err := GetSourceID(ctx, fromBucket)
 	require.NoError(t, err)
 
 	testCases := []struct {
@@ -233,7 +233,7 @@ func TestReplicateVV(t *testing.T) {
 func TestVVWriteTwice(t *testing.T) {
 	fromBucket, fromDs, toBucket, toDs := getTwoBucketDataStores(t)
 	ctx := base.TestCtx(t)
-	fromBucketSourceID, err := getSourceID(ctx, fromBucket)
+	fromBucketSourceID, err := GetSourceID(ctx, fromBucket)
 	require.NoError(t, err)
 
 	docID := "doc1"
@@ -267,7 +267,7 @@ func TestVVWriteTwice(t *testing.T) {
 func TestLWWAfterInitialReplication(t *testing.T) {
 	fromBucket, fromDs, toBucket, toDs := getTwoBucketDataStores(t)
 	ctx := base.TestCtx(t)
-	fromBucketSourceID, err := getSourceID(ctx, fromBucket)
+	fromBucketSourceID, err := GetSourceID(ctx, fromBucket)
 	require.NoError(t, err)
 
 	docID := "doc1"