Fix unit tests

Signed-off-by: Nicko Guyer <nicko.guyer@kaleido.io>

go.sum

@@ -457,7 +457,6 @@ github.com/go-latex/latex v0.0.0-20210118124228-b3d85cf34e07/go.mod h1:CO1AlKB2C
 github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A=
-github.com/go-logfmt/logfmt v0.5.1/go.mod h1:WYhtIu8zTZfxdn5+rREduYbwxfcBr/Vr6KEVveWlfTs=
 github.com/go-logr/logr v0.1.0/go.mod h1:ixOQHD9gLJUVQQ2ZOR7zLEifBX6tGkNJF4QyIY7sIas=
 github.com/go-logr/logr v0.2.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU=
 github.com/go-logr/logr v0.4.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU=

internal/assets/manager_test.go

@@ -18,6 +18,7 @@ package assets
 import (
 	"context"
 	"errors"
+	"fmt"
 	"testing"
 	"time"
 
@@ -160,3 +161,86 @@ func TestGetTokenConnectors(t *testing.T) {
 	assert.Equal(t, 1, len(connectors))
 	assert.Equal(t, "magic-tokens", connectors[0].Name)
 }
+
+func TestStart(t *testing.T) {
+	coreconfig.Reset()
+	mdi := &databasemocks.Plugin{}
+	mdm := &datamocks.Manager{}
+	mim := &identitymanagermocks.Manager{}
+	msa := &syncasyncmocks.Bridge{}
+	mbm := &broadcastmocks.Manager{}
+	mpm := &privatemessagingmocks.Manager{}
+	mti := &tokenmocks.Plugin{}
+	mm := &metricsmocks.Manager{}
+	mom := &operationmocks.Manager{}
+	mcm := &contractmocks.Manager{}
+	cmi := &cachemocks.Manager{}
+	cmi.On("GetCache", mock.Anything).Return(nil, nil)
+	mom.On("RegisterHandler", mock.Anything, mock.Anything, mock.Anything)
+	mdi.On("GetTokenPools", mock.Anything, mock.Anything, mock.Anything).Return([]*core.TokenPool{
+		{
+			Connector: "hot_tokens",
+			Active:    true,
+		},
+	}, nil, nil)
+	mti.On("StartNamespace", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	mti.On("ConnectorName").Return("hot_tokens")
+	txHelper, _ := txcommon.NewTransactionHelper(context.Background(), "ns1", mdi, mdm, cmi)
+	am, err := NewAssetManager(context.Background(), "ns1", "blockchain_plugin", mdi, map[string]tokens.Plugin{"magic-tokens": mti}, mim, msa, mbm, mpm, mm, mom, mcm, txHelper, cmi)
+	assert.NoError(t, err)
+	err = am.Start(context.Background())
+	assert.NoError(t, err)
+}
+
+func TestStartDBError(t *testing.T) {
+	coreconfig.Reset()
+	mdi := &databasemocks.Plugin{}
+	mdm := &datamocks.Manager{}
+	mim := &identitymanagermocks.Manager{}
+	msa := &syncasyncmocks.Bridge{}
+	mbm := &broadcastmocks.Manager{}
+	mpm := &privatemessagingmocks.Manager{}
+	mti := &tokenmocks.Plugin{}
+	mm := &metricsmocks.Manager{}
+	mom := &operationmocks.Manager{}
+	mcm := &contractmocks.Manager{}
+	cmi := &cachemocks.Manager{}
+	cmi.On("GetCache", mock.Anything).Return(nil, nil)
+	mom.On("RegisterHandler", mock.Anything, mock.Anything, mock.Anything)
+	mdi.On("GetTokenPools", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil, fmt.Errorf("pop"))
+	txHelper, _ := txcommon.NewTransactionHelper(context.Background(), "ns1", mdi, mdm, cmi)
+	am, err := NewAssetManager(context.Background(), "ns1", "blockchain_plugin", mdi, map[string]tokens.Plugin{"magic-tokens": mti}, mim, msa, mbm, mpm, mm, mom, mcm, txHelper, cmi)
+	assert.NoError(t, err)
+	err = am.Start(context.Background())
+	assert.Regexp(t, "pop", err)
+}
+
+func TestStartError(t *testing.T) {
+	coreconfig.Reset()
+	mdi := &databasemocks.Plugin{}
+	mdm := &datamocks.Manager{}
+	mim := &identitymanagermocks.Manager{}
+	msa := &syncasyncmocks.Bridge{}
+	mbm := &broadcastmocks.Manager{}
+	mpm := &privatemessagingmocks.Manager{}
+	mti := &tokenmocks.Plugin{}
+	mm := &metricsmocks.Manager{}
+	mom := &operationmocks.Manager{}
+	mcm := &contractmocks.Manager{}
+	cmi := &cachemocks.Manager{}
+	cmi.On("GetCache", mock.Anything).Return(nil, nil)
+	mom.On("RegisterHandler", mock.Anything, mock.Anything, mock.Anything)
+	mdi.On("GetTokenPools", mock.Anything, mock.Anything, mock.Anything).Return([]*core.TokenPool{
+		{
+			Connector: "hot_tokens",
+			Active:    true,
+		},
+	}, nil, nil)
+	mti.On("StartNamespace", mock.Anything, mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
+	mti.On("ConnectorName").Return("hot_tokens")
+	txHelper, _ := txcommon.NewTransactionHelper(context.Background(), "ns1", mdi, mdm, cmi)
+	am, err := NewAssetManager(context.Background(), "ns1", "blockchain_plugin", mdi, map[string]tokens.Plugin{"magic-tokens": mti}, mim, msa, mbm, mpm, mm, mom, mcm, txHelper, cmi)
+	assert.NoError(t, err)
+	err = am.Start(context.Background())
+	assert.Regexp(t, "pop", err)
+}

internal/blockchain/ethereum/ethereum.go

@@ -454,6 +454,7 @@ func (e *Ethereum) eventLoop(namespace string) {
 	defer close(e.closed[namespace])
 	l := log.L(e.ctx).WithField("role", "event-loop")
 	ctx := log.WithLogger(e.ctx, l)
+	log.L(ctx).Debugf("Starting event loop for namespace '%s'", namespace)
 	for {
 		select {
 		case <-ctx.Done():

internal/orchestrator/orchestrator_test.go

@@ -231,7 +231,7 @@ func TestInitOK(t *testing.T) {
 	or.mbi.On("SetHandler", "ns", mock.Anything).Return()
 	or.mbi.On("SetOperationHandler", "ns", mock.Anything).Return()
 	or.mbi.On("StartNamespace", mock.Anything, "ns").Return(nil)
-	or.mti.On("StartNamespace", mock.Anything, "ns").Return(nil)
+	// or.mti.On("StartNamespace", mock.Anything, "ns", mock.Anything).Return(nil)
 	or.mdi.On("GetIdentities", mock.Anything, "ns", mock.Anything).Return([]*core.Identity{node}, nil, nil)
 	or.mdx.On("SetHandler", "ns2", "node1", mock.Anything).Return()
 	or.mdx.On("SetOperationHandler", "ns", mock.Anything).Return()
@@ -301,19 +301,19 @@ func TestInitMessagingComponentFail(t *testing.T) {
 	or.messaging = nil
 	or.mbi.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	or.mmp.On("ConfigureContract", mock.Anything, mock.Anything).Return(nil)
-	or.mti.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	err := or.initComponents(context.Background())
 	assert.Regexp(t, "FF10128", err)
 }
 
 func TestInitTokensFail(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
-	or.plugins.Database.Plugin = nil
-	or.messaging = nil
-	or.mbi.On("StartNamespace", mock.Anything, "ns").Return(nil)
-	or.mti.On("StartNamespace", mock.Anything, "ns").Return(fmt.Errorf("pop"))
-	err := or.initComponents(context.Background())
+	or.assets = nil
+	or.mom.On("RegisterHandler", mock.Anything, mock.Anything, mock.Anything)
+	or.mmp.On("ConfigureContract", mock.Anything).Return(nil)
+	or.mdi.On("GetTokenPools", mock.Anything, "ns", mock.Anything).Return([]*core.TokenPool{}, nil, nil)
+	or.mti.On("StartNamespace", mock.Anything, "ns", mock.Anything).Return(fmt.Errorf("pop"))
+	err := or.initManagers(context.Background())
 	assert.Regexp(t, "pop", err)
 }
 
@@ -324,7 +324,6 @@ func TestInitEventsComponentFail(t *testing.T) {
 	or.events = nil
 	or.mbi.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	or.mmp.On("ConfigureContract", mock.Anything, mock.Anything).Return(nil)
-	or.mti.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	err := or.initComponents(context.Background())
 	assert.Regexp(t, "FF10128", err)
 }
@@ -346,7 +345,6 @@ func TestInitNetworkMapComponentFail(t *testing.T) {
 	or.networkmap = nil
 	or.mbi.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	or.mmp.On("ConfigureContract", mock.Anything, mock.Anything).Return(nil)
-	or.mti.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	err := or.initComponents(context.Background())
 	assert.Regexp(t, "FF10128", err)
 }
@@ -355,7 +353,6 @@ func TestInitMultipartyComponentFail(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
 	or.mbi.On("StartNamespace", mock.Anything, "ns").Return(nil)
-	or.mti.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	or.plugins.Database.Plugin = nil
 	or.multiparty = nil
 	err := or.initComponents(context.Background())
@@ -367,7 +364,6 @@ func TestInitMultipartyComponentConfigureFail(t *testing.T) {
 	defer or.cleanup(t)
 	or.mbi.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	or.mmp.On("ConfigureContract", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
-	or.mti.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	err := or.initComponents(context.Background())
 	assert.EqualError(t, err, "pop")
 }
@@ -379,7 +375,6 @@ func TestInitSharedStorageDownloadComponentFail(t *testing.T) {
 	or.sharedDownload = nil
 	or.mbi.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	or.mmp.On("ConfigureContract", mock.Anything, mock.Anything).Return(nil)
-	or.mti.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	err := or.initComponents(context.Background())
 	assert.Regexp(t, "FF10128", err)
 }
@@ -391,7 +386,6 @@ func TestInitBatchComponentFail(t *testing.T) {
 	or.batch = nil
 	or.mbi.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	or.mmp.On("ConfigureContract", mock.Anything, mock.Anything).Return(nil)
-	or.mti.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	err := or.initComponents(context.Background())
 	assert.Regexp(t, "FF10128", err)
 }
@@ -403,7 +397,6 @@ func TestInitBroadcastComponentFail(t *testing.T) {
 	or.broadcast = nil
 	or.mbi.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	or.mmp.On("ConfigureContract", mock.Anything, mock.Anything).Return(nil)
-	or.mti.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	err := or.initComponents(context.Background())
 	assert.Regexp(t, "FF10128", err)
 }
@@ -424,7 +417,6 @@ func TestInitDataComponentFail(t *testing.T) {
 	or.plugins.Database.Plugin = nil
 	or.data = nil
 	or.mbi.On("StartNamespace", mock.Anything, "ns").Return(nil)
-	or.mti.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	err := or.initComponents(context.Background())
 	assert.Regexp(t, "FF10128", err)
 }
@@ -436,7 +428,6 @@ func TestInitIdentityComponentFail(t *testing.T) {
 	or.identity = nil
 	or.mbi.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	or.mmp.On("ConfigureContract", mock.Anything, mock.Anything).Return(nil)
-	or.mti.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	err := or.initComponents(context.Background())
 	assert.Regexp(t, "FF10128", err)
 }
@@ -448,7 +439,6 @@ func TestInitAssetsComponentFail(t *testing.T) {
 	or.assets = nil
 	or.mbi.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	or.mmp.On("ConfigureContract", mock.Anything, mock.Anything).Return(nil)
-	or.mti.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	err := or.initComponents(context.Background())
 	assert.Regexp(t, "FF10128", err)
 }
@@ -460,7 +450,6 @@ func TestInitContractsComponentFail(t *testing.T) {
 	or.contracts = nil
 	or.mbi.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	or.mmp.On("ConfigureContract", mock.Anything, mock.Anything).Return(nil)
-	or.mti.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	err := or.initComponents(context.Background())
 	assert.Regexp(t, "FF10128", err)
 }
@@ -472,7 +461,6 @@ func TestInitOperationsComponentFail(t *testing.T) {
 	or.operations = nil
 	or.txHelper = nil
 	or.mbi.On("StartNamespace", mock.Anything, "ns").Return(nil)
-	or.mti.On("StartNamespace", mock.Anything, "ns").Return(nil)
 	err := or.initComponents(context.Background())
 	assert.Regexp(t, "FF10128", err)
 }

internal/tokens/fftokens/fftokens.go

@@ -666,7 +666,7 @@ func (ft *FFTokens) handleMessage(ctx context.Context, namespace string, msgByte
 	case messageStarted:
 		err = ft.handleNamespaceStarted(ctx, msg.Data)
 	case messageActivated:
-		err = ft.handlePoolActivated(ctx, msg.Data)
+		err = ft.handleTokenPoolActivated(ctx, msg.Data)
 	default:
 		log.L(ctx).Errorf("Message unexpected: %s", msg.Event)
 		// do not set error here - we will never be able to process this message so log+swallow it.
@@ -702,17 +702,19 @@ func (ft *FFTokens) handleNamespaceStarted(ctx context.Context, data fftypes.JSO
 	namespace := data.GetString("namespace")
 	log.L(ctx).Debugf("Token connector '%s' started namespace '%s'. Ensuring all token pools active.", ft.Name(), namespace)
 	for _, pool := range ft.poolsToActivate[namespace] {
-		if _, err := ft.ActivateTokenPool(ctx, pool); err != nil {
+		if _, err := ft.EnsureTokenPoolActive(ctx, pool); err == nil {
+			log.L(ctx).Debugf("Ensured token pool active '%s'", pool.ID)
+		} else {
 			// Log the error and continue trying to activate pools
 			// At this point we've already started
-			log.L(ctx).Errorf("Error auto re-activating token pool '%s': %s", pool.ID, err.Error())
+			log.L(ctx).Errorf("Error ensuring token pool active '%s': %s", pool.ID, err.Error())
 		}
-		log.L(ctx).Debugf("Activated token pool '%s'", pool.ID)
+
 	}
 	return nil
 }
 
-func (ft *FFTokens) handlePoolActivated(ctx context.Context, data fftypes.JSONObject) error {
+func (ft *FFTokens) handleTokenPoolActivated(ctx context.Context, data fftypes.JSONObject) error {
 	// NOOP
 	return nil
 }
@@ -800,9 +802,9 @@ func (ft *FFTokens) CreateTokenPool(ctx context.Context, nsOpID string, pool *co
 	return core.OpPhasePending, nil
 }
 
-func (ft *FFTokens) ActivateTokenPool(ctx context.Context, pool *core.TokenPool) (phase core.OpPhase, err error) {
+func (ft *FFTokens) EnsureTokenPoolActive(ctx context.Context, pool *core.TokenPool) (res *resty.Response, err error) {
 	var errRes tokenError
-	res, err := ft.client.R().SetContext(ctx).
+	res, err = ft.client.R().SetContext(ctx).
 		SetBody(&activatePool{
 			Namespace:   pool.Namespace,
 			PoolData:    packPoolData(pool.Namespace, pool.ID),
@@ -812,7 +814,15 @@ func (ft *FFTokens) ActivateTokenPool(ctx context.Context, pool *core.TokenPool)
 		SetError(&errRes).
 		Post("/api/v1/activatepool")
 	if err != nil || !res.IsSuccess() {
-		return core.OpPhaseInitializing, wrapError(ctx, &errRes, res, err)
+		return res, wrapError(ctx, &errRes, res, err)
+	}
+	return res, nil
+}
+
+func (ft *FFTokens) ActivateTokenPool(ctx context.Context, pool *core.TokenPool) (phase core.OpPhase, err error) {
+	res, err := ft.EnsureTokenPoolActive(ctx, pool)
+	if err != nil || !res.IsSuccess() {
+		return core.OpPhaseInitializing, err
 	}
 	if res.StatusCode() == 200 {
 		// HTTP 200: Activation was successful, and pool details are in response body

internal/tokens/fftokens/fftokens_test.go

@@ -1857,3 +1857,43 @@ func TestErrorWrappingBodyErr(t *testing.T) {
 	assert.True(t, ok)
 	assert.True(t, errInterface.IsConflictError())
 }
+
+func TestHandleNamespaceStartedEnsureActive(t *testing.T) {
+	h, _, _, httpURL, done := newTestFFTokens(t)
+	defer done()
+	h.poolsToActivate = map[string][]*core.TokenPool{
+		"ns1": {{Active: true}},
+	}
+	httpmock.RegisterResponder("POST", fmt.Sprintf("%s/api/v1/activatepool", httpURL),
+		httpmock.NewJsonResponderOrPanic(200, fftypes.JSONObject{}))
+
+	_, err := h.handleMessage(context.Background(), "ns1", []byte(`{"event":"started","data":{"namespace": "ns1"}}`))
+	assert.NoError(t, err)
+}
+
+func TestHandleNamespaceStartedEnsureActiveError(t *testing.T) {
+	h, _, _, httpURL, done := newTestFFTokens(t)
+	defer done()
+	h.poolsToActivate = map[string][]*core.TokenPool{
+		"ns1": {{Active: true}},
+	}
+	httpmock.RegisterResponder("POST", fmt.Sprintf("%s/api/v1/activatepool", httpURL),
+		httpmock.NewJsonResponderOrPanic(500, fftypes.JSONObject{}))
+
+	_, err := h.handleMessage(context.Background(), "ns1", []byte(`{"event":"started","data":{"namespace": "ns1"}}`))
+	assert.NoError(t, err)
+}
+
+func TestHandlePoolActivated(t *testing.T) {
+	h, _, _, _, done := newTestFFTokens(t)
+	defer done()
+	_, err := h.handleMessage(context.Background(), "ns1", []byte(`{"event":"activated","dat":{"namespace": "ns1"}}`))
+	assert.NoError(t, err)
+}
+
+func TestConnectorName(t *testing.T) {
+	h := &FFTokens{
+		configuredName: "bob",
+	}
+	assert.Equal(t, h.ConnectorName(), "bob")
+}