Merge pull request #2 from Financial-Times/10pc/image-set-changes

ImageSet republishing changes

docStoreClient.go

@@ -12,17 +12,13 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-type docStoreClient interface {
-	GetImageSetsModelUUID(setUUID, tid string) (found bool, modelUUID string, err error)
-}
-
 type httpDocStore struct {
 	httpClient          *http.Client
 	docStoreAddressBase string
 	authHeader          string
 }
 
-func newHTTPDocStore(httpClient *http.Client, docStoreAddressBase, authHeader string) (*httpDocStore, error) {
+func newHTTPDocStore(httpClient *http.Client, docStoreAddressBase, authHeader string) (imageSetUUIDResolver, error) {
 	return &httpDocStore{
 		httpClient:          httpClient,
 		docStoreAddressBase: docStoreAddressBase,

imageSetResolver.go

@@ -0,0 +1,22 @@
+package main
+
+import (
+	"github.com/Financial-Times/uuid-utils-go"
+)
+
+type imageSetUUIDResolver interface {
+	GetImageSetsModelUUID(setUUID, tid string) (found bool, modelUUID string, err error)
+}
+
+type uuidImageSetResolver struct {
+}
+
+func newUUIDImageSetResolver() imageSetUUIDResolver {
+	return &uuidImageSetResolver{}
+}
+
+func (r *uuidImageSetResolver) GetImageSetsModelUUID(setUUID, tid string) (found bool, modelUUID string, err error) {
+	requestedUUID, _ := uuidutils.NewUUIDFromString(setUUID)
+	derivedUUID, _ := uuidutils.NewUUIDDeriverWith(uuidutils.IMAGE_SET).From(requestedUUID)
+	return true, derivedUUID.String(), nil
+}

imageSetResolver_test.go

@@ -0,0 +1,18 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestUuidImageSetResolver(t *testing.T) {
+	r := newUUIDImageSetResolver()
+
+	imageSetUUID := "9f365884-0c25-11e8-24ad-bec2279df517"
+
+	found, imageModelUUID, err := r.GetImageSetsModelUUID(imageSetUUID, "tid_test")
+	assert.True(t, found, "found image model UUID")
+	assert.Equal(t, "9f365884-0c25-11e8-bacb-2958fde95e5e", imageModelUUID, "image model UUID")
+	assert.NoError(t, err)
+}

main.go

@@ -137,20 +137,35 @@ func main() {
 		httpClient := setupHTTPClient()
 		nativeStoreClient := newNativeStoreClient(httpClient, "https://"+*sourceEnvHost+"/__nativerw/", "Basic "+base64.StdEncoding.EncodeToString([]byte(*sourceAuth)))
 		notifierClient, err := newHTTPNotifier(httpClient, "https://"+*targetEnvHost+"/__", "Basic "+base64.StdEncoding.EncodeToString([]byte(*targetAuth)))
-		docStoreClient, err := newHTTPDocStore(httpClient, "https://"+*deliveryEnvHost+"/__document-store-api/content", "Basic "+base64.StdEncoding.EncodeToString([]byte(*deliveryAuth)))
+		var imageSetResolver imageSetUUIDResolver
+		if *deliveryEnvHost == "" || *deliveryAuth == "" {
+			imageSetResolver = newUUIDImageSetResolver()
+		} else {
+			imageSetResolver, err = newHTTPDocStore(httpClient, "https://"+*deliveryEnvHost+"/__document-store-api/content", "Basic "+base64.StdEncoding.EncodeToString([]byte(*deliveryAuth)))
+		}
 		rateLimit := time.Duration(*rateLimitMs) * time.Millisecond
 		uuidCollectionRepublisher := newNotifyingUCRepublisher(notifierClient, nativeStoreClient, rateLimit)
-		uuidRepublisher := newNotifyingUUIDRepublisher(uuidCollectionRepublisher, docStoreClient, defaultCollections)
-		parallelRepublisher := newNotifyingParallelRepublisher(uuidRepublisher, *parallelism)
+		uuidRepublisher := newNotifyingUUIDRepublisher(uuidCollectionRepublisher, imageSetResolver, defaultCollections)
+		var republisher bulkRepublisher
+		if *parallelism > 1 {
+			republisher = newNotifyingParallelRepublisher(uuidRepublisher, *parallelism)
+		} else {
+			republisher = newNotifyingSequentialRepublisher(uuidRepublisher)
+		}
+
 		if err != nil {
 			log.Fatalf("Couldn't create notifier client. %v", err)
 		}
 
 		uuids := regSplit(*uuidList, "\\s")
 		log.Infof("uuidList=%v", uuids)
-		parallelRepublisher.Republish(uuids, *republishScope, *transactionIDPrefix)
+		_, errs := republisher.Republish(uuids, *republishScope, *transactionIDPrefix)
 
 		log.Infof("Dealt with nUuids=%v in duration=%v", len(uuids), time.Duration(time.Now().UnixNano()-start.UnixNano())*time.Nanosecond)
+
+		if len(errs) > 0 {
+			os.Exit(1)
+		}
 	}
 	err := app.Run(os.Args)
 	if err != nil {

parallelRepublisher.go

@@ -7,7 +7,7 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-type parallelRepublisher interface {
+type bulkRepublisher interface {
 	Republish(uuids []string, publishScope string, tidPrefix string) ([]*okMsg, []error)
 }
 
@@ -17,7 +17,7 @@ type notifyingParallelRepublisher struct {
 	parallelism     int
 }
 
-func newNotifyingParallelRepublisher(uuidRepublisher uuidRepublisher, parallelism int) *notifyingParallelRepublisher {
+func newNotifyingParallelRepublisher(uuidRepublisher uuidRepublisher, parallelism int) bulkRepublisher {
 	return &notifyingParallelRepublisher{
 		uuidRepublisher: uuidRepublisher,
 		balancer:        workbalancer.NewChannelBalancer(parallelism),

sequentialRepublisher.go

@@ -0,0 +1,35 @@
+package main
+
+import (
+	log "github.com/sirupsen/logrus"
+)
+
+type notifyingSequentialRepublisher struct {
+	uuidRepublisher uuidRepublisher
+}
+
+func newNotifyingSequentialRepublisher(uuidRepublisher uuidRepublisher) bulkRepublisher {
+	return &notifyingSequentialRepublisher{
+		uuidRepublisher: uuidRepublisher,
+	}
+}
+
+func (r *notifyingSequentialRepublisher) Republish(uuids []string, publishScope string, tidPrefix string) ([]*okMsg, []error) {
+	var msgs []*okMsg
+	var errs []error
+
+	for _, uuid := range uuids {
+		tmsgs, terrs := r.uuidRepublisher.Republish(uuid, tidPrefix, publishScope)
+
+		for _, msg := range tmsgs {
+			log.Info(msg)
+			msgs = append(msgs, msg)
+		}
+		for _, err := range terrs {
+			log.Error(err)
+			errs = append(errs, err)
+		}
+	}
+
+	return msgs, errs
+}

sequentialRepublisher_test.go

@@ -0,0 +1,88 @@
+package main
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+func TestSequentialRepublishSingle_Ok(t *testing.T) {
+	mockedUUIDRepublisher := new(mockUUIDRepublisher)
+	msg := okMsg{
+		uuid:                     "19cf2763-90b1-40db-90e7-e813425ebe81",
+		tid:                      "prefix1",
+		collectionName:           "collection1",
+		collectionOriginSystemID: "originSystemId1",
+		sizeBytes:                1024,
+		notifierAppName:          "cms-notifier",
+	}
+	mockedUUIDRepublisher.On("Republish", "19cf2763-90b1-40db-90e7-e813425ebe81", "prefix1", scopeBoth).Return([]*okMsg{&msg}, []error{})
+
+	pRepublisher := newNotifyingSequentialRepublisher(mockedUUIDRepublisher)
+
+	pRepublisher.Republish([]string{"19cf2763-90b1-40db-90e7-e813425ebe81"}, scopeBoth, "prefix1")
+
+	mock.AssertExpectationsForObjects(t, mockedUUIDRepublisher)
+}
+
+func TestRepublishMultiple_Ok(t *testing.T) {
+	mockedUUIDRepublisher := new(mockUUIDRepublisher)
+	nOk := 10
+	nErr := 5
+	uuids := []string{}
+	for i := 0; i < nOk; i++ {
+		uuids = append(uuids, "19cf2763-90b1-40db-90e7-e813425ebe81")
+	}
+	for i := 0; i < nErr; i++ {
+		uuids = append(uuids, "70357268-04f7-4149-bb17-217d3eb56d49")
+	}
+	msg1 := okMsg{
+		uuid:                     "19cf2763-90b1-40db-90e7-e813425ebe81",
+		tid:                      "prefix1tid_123",
+		collectionName:           "collection1",
+		collectionOriginSystemID: "originSystemId1",
+		sizeBytes:                1024,
+		notifierAppName:          "cms-notifier",
+	}
+	msg2 := okMsg{
+		uuid:                     "19cf2763-90b1-40db-90e7-e813425ebe81",
+		tid:                      "prefix1tid_456",
+		collectionName:           "collection2",
+		collectionOriginSystemID: "originSystemId1",
+		sizeBytes:                1024,
+		notifierAppName:          "cms-notifier",
+	}
+	err1 := fmt.Errorf("test some error publishing 1")
+	err2 := fmt.Errorf("test some error publishing 2")
+	mockedUUIDRepublisher.On("Republish", "19cf2763-90b1-40db-90e7-e813425ebe81", "prefix1", scopeBoth).Times(nOk).Return([]*okMsg{&msg1, &msg2}, []error{})
+	mockedUUIDRepublisher.On("Republish", "70357268-04f7-4149-bb17-217d3eb56d49", "prefix1", scopeBoth).Times(nErr).Return([]*okMsg{}, []error{err1, err2})
+	pRepublisher := newNotifyingSequentialRepublisher(mockedUUIDRepublisher)
+
+	actualMsgs, actualErrs := pRepublisher.Republish(uuids, scopeBoth, "prefix1")
+
+	mock.AssertExpectationsForObjects(t, mockedUUIDRepublisher)
+	assert.Equal(t, 2*nOk, len(actualMsgs))
+	var msg1equal, msg2equal int
+	for _, actualMsg := range actualMsgs {
+		if msg1 == *actualMsg {
+			msg1equal++
+		} else if msg2 == *actualMsg {
+			msg2equal++
+		}
+	}
+	assert.Equal(t, nOk, msg1equal)
+	assert.Equal(t, nOk, msg2equal)
+	assert.Equal(t, 2*nErr, len(actualErrs))
+	var err1equal, err2equal int
+	for _, actualErr := range actualErrs {
+		if err1 == actualErr {
+			err1equal++
+		} else if err2 == actualErr {
+			err2equal++
+		}
+	}
+	assert.Equal(t, nErr, err1equal)
+	assert.Equal(t, nErr, err2equal)
+}

uuidCollectionRepublisher_test.go

@@ -59,7 +59,6 @@ func TestRepublishNotFound_NotFound(t *testing.T) {
 }
 
 func TestRepublishErrNative_Err(t *testing.T) {
-	start := time.Now()
 	mockedNativeStoreClient := new(mockNativeStoreClient)
 	mockedNativeStoreClient.On("GetNative", "methode", "f3b3b579-732b-4323-affa-a316aacad213", "tid_123").Return([]byte("native"), false, fmt.Errorf("Error 401 on native client"))
 	mockedNotifierClient := new(mockNotifierClient)
@@ -71,12 +70,14 @@ func TestRepublishErrNative_Err(t *testing.T) {
 		scope:          "content",
 	}
 
+	start := time.Now()
 	msg, wasFound, err := republisher.RepublishUUIDFromCollection("f3b3b579-732b-4323-affa-a316aacad213", "tid_123", collection)
+	now := time.Now()
 
 	assert.Error(t, fmt.Errorf("Error 401 on native client"), err)
 	assert.False(t, wasFound)
 	assert.Nil(t, msg)
-	assert.True(t, time.Now().UnixNano()-start.UnixNano() < 950000, "The time limter should have no effect on native client.")
+	assert.WithinDuration(t, start, now, time.Nanosecond * 950000, "The time limiter should have no effect on native client.")
 }
 
 func TestRepublishErrNotifier_Err(t *testing.T) {

uuidRepublisher.go

@@ -12,16 +12,16 @@ type uuidRepublisher interface {
 }
 
 type notifyingUUIDRepublisher struct {
-	ucRepublisher  uuidCollectionRepublisher
-	docStoreClient docStoreClient
-	collections    map[string]targetSystem
+	ucRepublisher    uuidCollectionRepublisher
+	imageSetResolver imageSetUUIDResolver
+	collections      map[string]targetSystem
 }
 
-func newNotifyingUUIDRepublisher(uuidCollectionRepublisher uuidCollectionRepublisher, docStoreClient docStoreClient, collections map[string]targetSystem) *notifyingUUIDRepublisher {
+func newNotifyingUUIDRepublisher(uuidCollectionRepublisher uuidCollectionRepublisher, imageSetResolver imageSetUUIDResolver, collections map[string]targetSystem) *notifyingUUIDRepublisher {
 	return &notifyingUUIDRepublisher{
-		ucRepublisher:  uuidCollectionRepublisher,
-		docStoreClient: docStoreClient,
-		collections:    collections,
+		ucRepublisher:    uuidCollectionRepublisher,
+		imageSetResolver: imageSetResolver,
+		collections:      collections,
 	}
 }
 
@@ -49,7 +49,7 @@ func (r *notifyingUUIDRepublisher) Republish(uuid, tidPrefix string, republishSc
 
 	if !isFoundInAnyCollection && isScopedInAnyCollection {
 		tid := tidPrefix + transactionidutils.NewTransactionID()
-		isFoundAsImageSet, imageModelUUID, err := r.docStoreClient.GetImageSetsModelUUID(uuid, tid)
+		isFoundAsImageSet, imageModelUUID, err := r.imageSetResolver.GetImageSetsModelUUID(uuid, tid)
 		if err != nil {
 			errs = append(errs, fmt.Errorf("couldn't check if it's an ImageSet containing an image inside because of an error uuid=%v tid=%v %v", uuid, tid, err))
 			return nil, errs