grpc-gateway/test: manage TestCreateAndDeleteResource via env

.vscode/settings.json

@@ -20,6 +20,7 @@
         "TEST_OAUTH_SERVER_ID_TOKEN_PRIVATE_KEY": "${workspaceFolder}/.tmp/privKeys/idTokenKey.pem",
         "TEST_OAUTH_SERVER_ACCESS_TOKEN_PRIVATE_KEY": "${workspaceFolder}/.tmp/privKeys/accessTokenKey.pem",
         "TEST_HTTP_GW_WWW_ROOT": "${workspaceFolder}/.tmp/usr/local/www",
+        "ENABLE_RECEIVE_CANCELLATION_NOTIFICATION": false,
         // "GOMAXPROCS": 1,
         // "GOFLAGS":"-mod=vendor",
         // "GRPC_VERBOSITY":"DEBUG",

Makefile

@@ -139,6 +139,7 @@ define RUN-DOCKER
 		-e TEST_OAUTH_SERVER_ID_TOKEN_PRIVATE_KEY=/privKeys/idTokenKey.pem \
 		-e TEST_OAUTH_SERVER_ACCESS_TOKEN_PRIVATE_KEY=/privKeys/accessTokenKey.pem \
 		-e TEST_HTTP_GW_WWW_ROOT=/usr/local/www \
+		-e ENABLE_RECEIVE_CANCELLATION_NOTIFICATION=false \
 		hub-test \
 		$(1) ;
 endef

coap-gateway/service/client.go

@@ -280,15 +280,15 @@ func (c *Client) onGetResourceContent(ctx context.Context, deviceID, href string
 	notification.Hijack()
 	err := c.server.taskQueue.Submit(func() {
 		defer c.server.messagePool.ReleaseMessage(notification)
+		if notification.Code() == codes.NotFound {
+			c.unpublishResourceLinks(c.getUserAuthorizedContext(ctx), []string{href}, nil)
+		}
 		err2 := c.notifyContentChanged(deviceID, href, false, notification)
 		if err2 != nil {
 			// cloud is unsynchronized against device. To recover cloud state, client need to reconnect to cloud.
 			c.Errorf("%w", cannotGetResourceContentError(deviceID, href, err2))
 			c.Close()
 		}
-		if notification.Code() == codes.NotFound {
-			c.unpublishResourceLinks(c.getUserAuthorizedContext(ctx), []string{href}, nil)
-		}
 	})
 	if err != nil {
 		defer c.server.messagePool.ReleaseMessage(notification)
@@ -311,15 +311,15 @@ func (c *Client) onObserveResource(ctx context.Context, deviceID, href string, b
 	notification.Hijack()
 	err := c.server.taskQueue.SubmitForOneWorker(resource.GetInstanceID(deviceID+href), func() {
 		defer c.server.messagePool.ReleaseMessage(notification)
+		if notification.Code() == codes.NotFound {
+			c.unpublishResourceLinks(c.getUserAuthorizedContext(notification.Context()), []string{href}, nil)
+		}
 		err2 := c.notifyContentChanged(deviceID, href, batch, notification)
 		if err2 != nil {
 			// cloud is unsynchronized against device. To recover cloud state, client need to reconnect to cloud.
 			c.Errorf("%w", cannotObserResourceError(err2))
 			c.Close()
 		}
-		if notification.Code() == codes.NotFound {
-			c.unpublishResourceLinks(c.getUserAuthorizedContext(notification.Context()), []string{href}, nil)
-		}
 	})
 	if err != nil {
 		defer c.server.messagePool.ReleaseMessage(notification)

grpc-gateway/service/createAndDeleteResource_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"os"
 	"testing"
 
 	"github.com/plgd-dev/device/schema"
@@ -21,7 +22,6 @@ import (
 	oauthTest "github.com/plgd-dev/hub/v2/test/oauth-server/test"
 	pbTest "github.com/plgd-dev/hub/v2/test/pb"
 	"github.com/plgd-dev/hub/v2/test/service"
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
@@ -216,12 +216,25 @@ func deleteSwitchResourceExpectedEvents(t *testing.T, deviceID, subID, correlati
 			},
 		}
 		e[pbTest.GetEventID(changedRes)] = changedRes
+	} else if os.Getenv("ENABLE_RECEIVE_CANCELLATION_NOTIFICATION") == "" {
+		res := pbTest.MakeResourceChanged(t, deviceID, test.TestResourceSwitchesInstanceHref(switchID), "", nil)
+		res.Status = commands.Status_NOT_FOUND
+		res.Content.CoapContentFormat = -1
+		res.Content.ContentType = ""
+		changedRes := &pb.Event{
+			SubscriptionId: subID,
+			CorrelationId:  correlationID,
+			Type: &pb.Event_ResourceChanged{
+				ResourceChanged: res,
+			},
+		}
+		e[pbTest.GetEventID(changedRes)] = changedRes
 	}
 
 	return e
 }
 
-func validateEvents(t *testing.T, subClient pb.GrpcGateway_SubscribeToEventsClient, expectedEvents map[string]*pb.Event, optionalEvents map[string]*pb.Event) {
+func validateEvents(t *testing.T, subClient pb.GrpcGateway_SubscribeToEventsClient, expectedEvents map[string]*pb.Event) {
 	for {
 		ev, err := subClient.Recv()
 		if kitNetGrpc.IsContextDeadlineExceeded(err) {
@@ -230,19 +243,6 @@ func validateEvents(t *testing.T, subClient pb.GrpcGateway_SubscribeToEventsClie
 		require.NoError(t, err)
 
 		eventID := pbTest.GetEventID(ev)
-		optional, ok := optionalEvents[eventID]
-		if ok {
-			pbTest.CleanUpEvent(t, optional)
-			pbTest.CleanUpEvent(t, ev)
-			equal := false
-			test.CheckProtobufs(t, optional, ev, func(t *testing.T, expected, actual interface{}, msgAndArgs ...interface{}) {
-				equal = assert.ObjectsAreEqual(expected, actual)
-			})
-			if equal {
-				continue
-			}
-		}
-
 		expected, ok := expectedEvents[eventID]
 		if !ok {
 			require.Failf(t, "unexpected event", "invalid event: %+v", ev)
@@ -255,25 +255,6 @@ func validateEvents(t *testing.T, subClient pb.GrpcGateway_SubscribeToEventsClie
 	}
 }
 
-func getOptionalEvents(t *testing.T, deviceID, subID, correlationID, switchID string, isDiscoveryResourceBatchObservable bool) map[string]*pb.Event {
-	e := make(map[string]*pb.Event)
-	if !isDiscoveryResourceBatchObservable {
-		res := pbTest.MakeResourceChanged(t, deviceID, test.TestResourceSwitchesInstanceHref(switchID), "", nil)
-		res.Status = commands.Status_NOT_FOUND
-		res.Content.CoapContentFormat = -1
-		res.Content.ContentType = ""
-		changedRes := &pb.Event{
-			SubscriptionId: subID,
-			CorrelationId:  correlationID,
-			Type: &pb.Event_ResourceChanged{
-				ResourceChanged: res,
-			},
-		}
-		e[pbTest.GetEventID(changedRes)] = changedRes
-	}
-	return e
-}
-
 func TestCreateAndDeleteResource(t *testing.T) {
 	deviceID := test.MustFindDeviceByName(test.TestDeviceName)
 	ctx, cancel := context.WithTimeout(context.Background(), config.TEST_TIMEOUT)
@@ -302,15 +283,13 @@ func TestCreateAndDeleteResource(t *testing.T) {
 
 	isDiscoveryResourceBatchObservable := test.IsDiscoveryResourceBatchObservable(ctx, t, deviceID)
 
-	optionalEvents := getOptionalEvents(t, deviceID, subID, correlationID, switchID, isDiscoveryResourceBatchObservable)
-
 	for i := 0; i < 5; i++ {
 		fmt.Printf("iteration %v\n", i)
 		createSwitchResource(t, ctx, c, deviceID, switchID)
 		expectedCreateEvents := createSwitchResourceExpectedEvents(t, deviceID, subID, correlationID, switchID)
-		validateEvents(t, subClient, expectedCreateEvents, optionalEvents)
+		validateEvents(t, subClient, expectedCreateEvents)
 		deleteSwitchResource(t, ctx, c, deviceID, switchID)
 		expectedDeleteEvents := deleteSwitchResourceExpectedEvents(t, deviceID, subID, correlationID, switchID, isDiscoveryResourceBatchObservable)
-		validateEvents(t, subClient, expectedDeleteEvents, optionalEvents)
+		validateEvents(t, subClient, expectedDeleteEvents)
 	}
 }