Fix migration issues and increase test coverage

Signed-off-by: Nicko Guyer <[email protected]>

db/migrations/postgres/000116_tx_type_not_null.down.sql

@@ -0,0 +1,3 @@
+BEGIN;
+ALTER TABLE messages ALTER COLUMN tx_parent_type DROP NOT NULL;
+COMMIT;

db/migrations/postgres/000116_tx_type_not_null.up.sql

@@ -0,0 +1,5 @@
+BEGIN;
+UPDATE messages SET tx_parent_type = ''
+  WHERE tx_parent_type IS NULL;
+ALTER TABLE messages ALTER COLUMN tx_parent_type SET NOT NULL;
+COMMIT;

db/migrations/sqlite/000116_tx_type_not_null.down.sql

@@ -0,0 +1,4 @@
+ALTER TABLE messages RENAME COLUMN tx_parent_type TO tx_parent_type_temp;
+ALTER TABLE messages ADD COLUMN tx_parent_type VARCHAR(64);
+UPDATE messages SET tx_parent_type = tx_parent_type_temp;
+ALTER TABLE messages DROP COLUMN tx_parent_type_temp;

db/migrations/sqlite/000116_tx_type_not_null.up.sql

@@ -0,0 +1,5 @@
+UPDATE messages SET tx_parent_type = '' WHERE tx_parent_type IS NULL;
+ALTER TABLE messages RENAME COLUMN tx_parent_type TO tx_parent_type_temp;
+ALTER TABLE messages ADD COLUMN tx_parent_type VARCHAR(64) NOT NULL;
+UPDATE messages SET tx_parent_type = tx_parent_type_temp;
+ALTER TABLE messages DROP COLUMN tx_parent_type_temp;

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/blockchain/ethereum/ethereum_test.go

@@ -398,6 +398,119 @@ func TestStartStopNamespace(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func TestStartStopNamespaceOldEventstream(t *testing.T) {
+	e, cancel := newTestEthereum()
+	defer cancel()
+
+	toServer, _, wsURL, done := wsclient.NewTestWSServer(nil)
+	defer done()
+
+	mockedClient := &http.Client{}
+	httpmock.ActivateNonDefault(mockedClient)
+	defer httpmock.DeactivateAndReset()
+
+	u, _ := url.Parse(wsURL)
+	u.Scheme = "http"
+	httpURL := u.String()
+
+	httpmock.RegisterResponder("GET", fmt.Sprintf("%s/eventstreams", httpURL),
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{
+			{
+				ID:   "es12345",
+				Name: "topic1",
+			},
+		}))
+	httpmock.RegisterResponder("DELETE", fmt.Sprintf("%s/eventstreams/es12345", httpURL),
+		httpmock.NewJsonResponderOrPanic(204, ""))
+	httpmock.RegisterResponder("POST", fmt.Sprintf("%s/eventstreams", httpURL),
+		httpmock.NewJsonResponderOrPanic(200, eventStream{ID: "es12345"}))
+
+	resetConf(e)
+	utEthconnectConf.Set(ffresty.HTTPConfigURL, httpURL)
+	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
+	utEthconnectConf.Set(EthconnectConfigInstanceDeprecated, "/instances/0x71C7656EC7ab88b098defB751B7401B5f6d8976F")
+	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
+	utFFTMConf.Set(ffresty.HTTPConfigURL, httpURL)
+
+	cmi := &cachemocks.Manager{}
+	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
+	err := e.Init(e.ctx, e.cancelCtx, utConfig, e.metrics, cmi)
+	assert.NoError(t, err)
+
+	err = e.StartNamespace(e.ctx, "ns1")
+	assert.NoError(t, err)
+
+	<-toServer
+
+	err = e.StopNamespace(e.ctx, "ns1")
+	assert.NoError(t, err)
+}
+
+func TestEnsureEventStreamDeleteFail(t *testing.T) {
+	e, cancel := newTestEthereum()
+	defer cancel()
+
+	mockedClient := &http.Client{}
+	httpmock.ActivateNonDefault(mockedClient)
+	defer httpmock.DeactivateAndReset()
+
+	httpURL := "http://localhost:12345"
+
+	httpmock.RegisterResponder("GET", fmt.Sprintf("%s/eventstreams", httpURL),
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{
+			{
+				ID:   "es12345",
+				Name: "topic1",
+			},
+		}))
+	httpmock.RegisterResponder("DELETE", fmt.Sprintf("%s/eventstreams/es12345", httpURL),
+		httpmock.NewJsonResponderOrPanic(500, "pop"))
+
+	resetConf(e)
+	utEthconnectConf.Set(ffresty.HTTPConfigURL, httpURL)
+	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
+	utEthconnectConf.Set(EthconnectConfigInstanceDeprecated, "/instances/0x71C7656EC7ab88b098defB751B7401B5f6d8976F")
+	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
+	utFFTMConf.Set(ffresty.HTTPConfigURL, httpURL)
+
+	cmi := &cachemocks.Manager{}
+	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
+	err := e.Init(e.ctx, e.cancelCtx, utConfig, e.metrics, cmi)
+	assert.NoError(t, err)
+
+	_, err = e.streams.ensureEventStream(context.Background(), "topic1/ns1", "topic1")
+	assert.Regexp(t, "pop", err)
+}
+
+func TestDeleteStreamOKNotFound(t *testing.T) {
+	e, cancel := newTestEthereum()
+	defer cancel()
+
+	mockedClient := &http.Client{}
+	httpmock.ActivateNonDefault(mockedClient)
+	defer httpmock.DeactivateAndReset()
+
+	httpURL := "http://localhost:12345"
+
+	httpmock.RegisterResponder("DELETE", fmt.Sprintf("%s/eventstreams/es12345", httpURL),
+		httpmock.NewJsonResponderOrPanic(404, "pop"))
+
+	resetConf(e)
+	utEthconnectConf.Set(ffresty.HTTPConfigURL, httpURL)
+	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
+	utEthconnectConf.Set(EthconnectConfigInstanceDeprecated, "/instances/0x71C7656EC7ab88b098defB751B7401B5f6d8976F")
+	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
+	utFFTMConf.Set(ffresty.HTTPConfigURL, httpURL)
+
+	cmi := &cachemocks.Manager{}
+	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
+	err := e.Init(e.ctx, e.cancelCtx, utConfig, e.metrics, cmi)
+	assert.NoError(t, err)
+
+	err = e.streams.deleteEventStream(context.Background(), "es12345", true)
+	assert.NoError(t, err)
+}
+
 func TestStartNamespaceWSCreateFail(t *testing.T) {
 	e, cancel := newTestEthereum()
 	defer cancel()

internal/blockchain/fabric/fabric_test.go

@@ -893,6 +893,108 @@ func TestStreamCreateError(t *testing.T) {
 
 }
 
+func TestEnsureStreamDelete(t *testing.T) {
+
+	e, cancel := newTestFabric()
+	defer cancel()
+
+	mockedClient := &http.Client{}
+	httpmock.ActivateNonDefault(mockedClient)
+	defer httpmock.DeactivateAndReset()
+
+	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{
+			{
+				ID:   "es12345",
+				Name: "topic1",
+			},
+		}))
+	httpmock.RegisterResponder("DELETE", "http://localhost:12345/eventstreams/es12345",
+		httpmock.NewJsonResponderOrPanic(204, ""))
+	httpmock.RegisterResponder("POST", "http://localhost:12345/eventstreams",
+		httpmock.NewStringResponder(200, ""))
+	resetConf(e)
+	utFabconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
+	utFabconnectConf.Set(ffresty.HTTPConfigRetryEnabled, false)
+	utFabconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
+	utFabconnectConf.Set(FabconnectConfigChaincodeDeprecated, "firefly")
+	utFabconnectConf.Set(FabconnectConfigSigner, "signer001")
+	utFabconnectConf.Set(FabconnectConfigTopic, "topic1")
+
+	cmi := &cachemocks.Manager{}
+	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
+	err := e.Init(e.ctx, e.cancelCtx, utConfig, &metricsmocks.Manager{}, cmi)
+	assert.NoError(t, err)
+
+	_, err = e.streams.ensureEventStream(context.Background(), "topic1/ns1", "topic1")
+	assert.NoError(t, err)
+}
+
+func TestEnsureStreamDeleteFail(t *testing.T) {
+
+	e, cancel := newTestFabric()
+	defer cancel()
+
+	mockedClient := &http.Client{}
+	httpmock.ActivateNonDefault(mockedClient)
+	defer httpmock.DeactivateAndReset()
+
+	httpmock.RegisterResponder("GET", "http://localhost:12345/eventstreams",
+		httpmock.NewJsonResponderOrPanic(200, []eventStream{
+			{
+				ID:   "es12345",
+				Name: "topic1",
+			},
+		}))
+	httpmock.RegisterResponder("DELETE", "http://localhost:12345/eventstreams/es12345",
+		httpmock.NewJsonResponderOrPanic(500, "pop"))
+	resetConf(e)
+	utFabconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
+	utFabconnectConf.Set(ffresty.HTTPConfigRetryEnabled, false)
+	utFabconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
+	utFabconnectConf.Set(FabconnectConfigChaincodeDeprecated, "firefly")
+	utFabconnectConf.Set(FabconnectConfigSigner, "signer001")
+	utFabconnectConf.Set(FabconnectConfigTopic, "topic1")
+
+	cmi := &cachemocks.Manager{}
+	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
+	err := e.Init(e.ctx, e.cancelCtx, utConfig, &metricsmocks.Manager{}, cmi)
+	assert.NoError(t, err)
+
+	_, err = e.streams.ensureEventStream(context.Background(), "topic1/ns1", "topic1")
+	assert.Regexp(t, "FF10284.*pop", err)
+}
+
+func TestDeleteStreamOKNotFound(t *testing.T) {
+	e, cancel := newTestFabric()
+	defer cancel()
+
+	mockedClient := &http.Client{}
+	httpmock.ActivateNonDefault(mockedClient)
+	defer httpmock.DeactivateAndReset()
+
+	httpURL := "http://localhost:12345"
+
+	httpmock.RegisterResponder("DELETE", fmt.Sprintf("%s/eventstreams/es12345", httpURL),
+		httpmock.NewJsonResponderOrPanic(404, "pop"))
+
+	resetConf(e)
+	utFabconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
+	utFabconnectConf.Set(ffresty.HTTPConfigRetryEnabled, false)
+	utFabconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
+	utFabconnectConf.Set(FabconnectConfigChaincodeDeprecated, "firefly")
+	utFabconnectConf.Set(FabconnectConfigSigner, "signer001")
+	utFabconnectConf.Set(FabconnectConfigTopic, "topic1")
+
+	cmi := &cachemocks.Manager{}
+	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
+	err := e.Init(e.ctx, e.cancelCtx, utConfig, e.metrics, cmi)
+	assert.NoError(t, err)
+
+	err = e.streams.deleteEventStream(context.Background(), "es12345", true)
+	assert.NoError(t, err)
+}
+
 func TestSubQueryCreateError(t *testing.T) {
 
 	e, cancel := newTestFabric()