Integration Tests: Subscriptions (dapr#7578)

* Adds operator informer v1 and v2 tests

Signed-off-by: joshvanl <[email protected]>

* Adds more v1alpha1 operator sub tests

Signed-off-by: joshvanl <[email protected]>

* int test subs dec v1alpha1 scopes

Signed-off-by: joshvanl <[email protected]>

* Complete v1alpha1 scope tests

Signed-off-by: joshvanl <[email protected]>

* wip v2alpha1 operator

Signed-off-by: joshvanl <[email protected]>

* Wire up v2alpha1 declarative subscription tests

Signed-off-by: joshvanl <[email protected]>

* Adds grpc programmatic subscription tests

Signed-off-by: joshvanl <[email protected]>

* Adds tests/integration/suite/daprd/subscriptions/programmatic/http/v1alpha1

Signed-off-by: joshvanl <[email protected]>

* linting

Signed-off-by: joshvanl <[email protected]>

* Changes `Subset` checks for `Contains`

Signed-off-by: joshvanl <[email protected]>

* Adds subscriptions/programmatic/http/v2alpha1

Signed-off-by: joshvanl <[email protected]>

* Adds subscriptions/programmatic/http/mixed

Signed-off-by: joshvanl <[email protected]>

* Adds tests/integration/suite/daprd/subscriptions/declarative/selfhosted/mixed

Signed-off-by: joshvanl <[email protected]>

* Adds tests/integration/suite/daprd/subscriptions/mixed

Signed-off-by: joshvanl <[email protected]>

* Set correct control plane trust domain in subscription int tests

Signed-off-by: joshvanl <[email protected]>

---------

Signed-off-by: joshvanl <[email protected]>

Author: JoshVanL

tests/integration/framework/process/grpc/app/app.go

@@ -14,6 +14,7 @@ limitations under the License.
 package app
 
 import (
+	"context"
 	"testing"
 
 	"google.golang.org/grpc"
@@ -27,7 +28,7 @@ type Option func(*options)
 
 // App is a wrapper around a grpc.Server that implements a Dapr App.
 type App struct {
-	*procgrpc.GRPC
+	grpc *procgrpc.GRPC
 }
 
 func New(t *testing.T, fopts ...Option) *App {
@@ -39,7 +40,7 @@ func New(t *testing.T, fopts ...Option) *App {
 	}
 
 	return &App{
-		GRPC: procgrpc.New(t, append(opts.grpcopts, procgrpc.WithRegister(func(s *grpc.Server) {
+		grpc: procgrpc.New(t, append(opts.grpcopts, procgrpc.WithRegister(func(s *grpc.Server) {
 			srv := &server{
 				onInvokeFn:         opts.onInvokeFn,
 				onTopicEventFn:     opts.onTopicEventFn,
@@ -58,3 +59,15 @@ func New(t *testing.T, fopts ...Option) *App {
 		}))...),
 	}
 }
+
+func (a *App) Run(t *testing.T, ctx context.Context) {
+	a.grpc.Run(t, ctx)
+}
+
+func (a *App) Cleanup(t *testing.T) {
+	a.grpc.Cleanup(t)
+}
+
+func (a *App) Port(t *testing.T) int {
+	return a.grpc.Port(t)
+}

tests/integration/framework/process/grpc/app/options.go

@@ -24,7 +24,8 @@ import (
 	procgrpc "github.com/dapr/dapr/tests/integration/framework/process/grpc"
 )
 
-// options contains the options for running a GRPC server in integration tests.
+// options contains the options for running a GRPC server app in integration
+// tests.
 type options struct {
 	grpcopts           []procgrpc.Option
 	withRegister       func(s *grpc.Server)

tests/integration/framework/process/http/subscriber/json.go

@@ -0,0 +1,42 @@
+/*
+Copyright 2024 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package subscriber
+
+type (
+	SubscriptionJSON struct {
+		PubsubName      string            `json:"pubsubname"`
+		Topic           string            `json:"topic"`
+		DeadLetterTopic string            `json:"deadLetterTopic"`
+		Metadata        map[string]string `json:"metadata,omitempty"`
+		Route           string            `json:"route"`  // Single route from v1alpha1
+		Routes          RoutesJSON        `json:"routes"` // Multiple routes from v2alpha1
+		BulkSubscribe   BulkSubscribeJSON `json:"bulkSubscribe,omitempty"`
+	}
+
+	RoutesJSON struct {
+		Rules   []*RuleJSON `json:"rules,omitempty"`
+		Default string      `json:"default,omitempty"`
+	}
+
+	BulkSubscribeJSON struct {
+		Enabled            bool  `json:"enabled"`
+		MaxMessagesCount   int32 `json:"maxMessagesCount,omitempty"`
+		MaxAwaitDurationMs int32 `json:"maxAwaitDurationMs,omitempty"`
+	}
+
+	RuleJSON struct {
+		Match string `json:"match"`
+		Path  string `json:"path"`
+	}
+)

tests/integration/framework/process/http/subscriber/options.go

@@ -23,6 +23,7 @@ type options struct {
 	routes       []string
 	bulkRoutes   []string
 	handlerFuncs []app.Option
+	progSubs     *[]SubscriptionJSON
 }
 
 func WithRoutes(routes ...string) Option {
@@ -42,3 +43,12 @@ func WithHandlerFunc(path string, fn http.HandlerFunc) Option {
 		o.handlerFuncs = append(o.handlerFuncs, app.WithHandlerFunc(path, fn))
 	}
 }
+
+func WithProgrammaticSubscriptions(subs ...SubscriptionJSON) Option {
+	return func(o *options) {
+		if o.progSubs == nil {
+			o.progSubs = new([]SubscriptionJSON)
+		}
+		*o.progSubs = append(*o.progSubs, subs...)
+	}
+}

tests/integration/framework/process/http/subscriber/subscriber.go

@@ -119,6 +119,12 @@ func New(t *testing.T, fopts ...Option) *Subscriber {
 		}))
 	}
 
+	if opts.progSubs != nil {
+		appOpts = append(appOpts, app.WithHandlerFunc("/dapr/subscribe", func(w http.ResponseWriter, r *http.Request) {
+			require.NoError(t, json.NewEncoder(w).Encode(*opts.progSubs))
+		}))
+	}
+
 	appOpts = append(appOpts, opts.handlerFuncs...)
 
 	return &Subscriber{

tests/integration/framework/process/kubernetes/options.go

@@ -31,7 +31,8 @@ import (
 	configapi "github.com/dapr/dapr/pkg/apis/configuration/v1alpha1"
 	httpendapi "github.com/dapr/dapr/pkg/apis/httpEndpoint/v1alpha1"
 	resapi "github.com/dapr/dapr/pkg/apis/resiliency/v1alpha1"
-	subapi "github.com/dapr/dapr/pkg/apis/subscriptions/v1alpha1"
+	subv1api "github.com/dapr/dapr/pkg/apis/subscriptions/v1alpha1"
+	subv2api "github.com/dapr/dapr/pkg/apis/subscriptions/v2alpha1"
 	"github.com/dapr/dapr/tests/integration/framework/process/kubernetes/store"
 )
 
@@ -54,7 +55,17 @@ func WithClusterDaprResiliencyList(t *testing.T, res *resapi.ResiliencyList) Opt
 	return handleClusterListResource(t, "/apis/dapr.io/v1alpha1/resiliencies", res)
 }
 
-func WithClusterDaprSubscriptionList(t *testing.T, subs *subapi.SubscriptionList) Option {
+func WithClusterDaprSubscriptionList(t *testing.T, subs *subv1api.SubscriptionList) Option {
+	subv2List := new(subv2api.SubscriptionList)
+	for _, s := range subs.Items {
+		subv2 := new(subv2api.Subscription)
+		require.NoError(t, subv2.ConvertFrom(s.DeepCopy()))
+		subv2List.Items = append(subv2List.Items, *subv2)
+	}
+	return handleClusterListResource(t, "/apis/dapr.io/v2alpha1/subscriptions", subv2List)
+}
+
+func WithClusterDaprSubscriptionListV2(t *testing.T, subs *subv2api.SubscriptionList) Option {
 	return handleClusterListResource(t, "/apis/dapr.io/v2alpha1/subscriptions", subs)
 }
 
@@ -66,6 +77,10 @@ func WithClusterDaprComponentListFromStore(t *testing.T, store *store.Store) Opt
 	return handleClusterListResourceFromStore(t, "/apis/dapr.io/v1alpha1/components", store)
 }
 
+func WithClusterDaprSubscriptionListFromStore(t *testing.T, store *store.Store) Option {
+	return handleClusterListResourceFromStore(t, "/apis/dapr.io/v1alpha1/subscriptions", store)
+}
+
 func WithClusterDaprHTTPEndpointList(t *testing.T, endpoints *httpendapi.HTTPEndpointList) Option {
 	return handleClusterListResource(t, "/apis/dapr.io/v1alpha1/httpendpoints", endpoints)
 }

tests/integration/suite/daprd/daprd.go

@@ -31,5 +31,6 @@ import (
 	_ "github.com/dapr/dapr/tests/integration/suite/daprd/serviceinvocation"
 	_ "github.com/dapr/dapr/tests/integration/suite/daprd/shutdown"
 	_ "github.com/dapr/dapr/tests/integration/suite/daprd/state"
+	_ "github.com/dapr/dapr/tests/integration/suite/daprd/subscriptions"
 	_ "github.com/dapr/dapr/tests/integration/suite/daprd/workflow"
 )

tests/integration/suite/daprd/pubsub/pubsub.go

@@ -17,4 +17,5 @@ import (
 	_ "github.com/dapr/dapr/tests/integration/suite/daprd/pubsub/contentlength"
 	_ "github.com/dapr/dapr/tests/integration/suite/daprd/pubsub/grpc"
 	_ "github.com/dapr/dapr/tests/integration/suite/daprd/pubsub/http"
+	_ "github.com/dapr/dapr/tests/integration/suite/daprd/pubsub/scopes"
 )