Added certification tests to rabbitMQ bindings (#1919)
* Added certification tests to rabbitMQ bindings Signed-off-by: Amulya Varote <amulyavarote@microsoft.com> * Resolving merge conflicts Signed-off-by: Amulya Varote <amulyavarote@microsoft.com> * Keep queue names different Signed-off-by: Amulya Varote <amulyavarote@microsoft.com> * go mod tidy changes Signed-off-by: Amulya Varote <amulyavarote@microsoft.com> * Branch update Signed-off-by: Amulya Varote <amulyavarote@microsoft.com> * Added a test case for network interruption Signed-off-by: Amulya Varote <amulyavarote@microsoft.com> * Fixed network interruption test case Signed-off-by: Amulya Varote <amulyavarote@microsoft.com> * Changed go mod file Signed-off-by: Amulya Varote <amulyavarote@microsoft.com> * Added a test case for exclusive attribute Signed-off-by: Amulya Varote <amulyavarote@microsoft.com> * go mod tidy execution changes Signed-off-by: Amulya Varote <amulyavarote@microsoft.com> * Modified test case for rabbitmq Signed-off-by: Amulya Varote <amulyavarote@microsoft.com> * changed exclusive mode test case Signed-off-by: Amulya Varote <amulyavarote@microsoft.com> * go mod tidy fix Signed-off-by: Amulya Varote <amulyavarote@microsoft.com> * remove unnecessary imports, use go1.19 and modtidy Signed-off-by: Bernd Verst <4535280+berndverst@users.noreply.github.com> * Update tests/certification/bindings/rabbitmq/rabbitmq_test.go * Changed README file Signed-off-by: Amulya Varote <amulyavarote@microsoft.com> Signed-off-by: Amulya Varote <amulyavarote@microsoft.com> Signed-off-by: Bernd Verst <4535280+berndverst@users.noreply.github.com> Co-authored-by: Bernd Verst <4535280+berndverst@users.noreply.github.com>
This commit is contained in:
amulyavarote
GitHub
parent
aad920f307
commit
a69b9b946c
|
|
@ -48,6 +48,7 @@ jobs:
|
|||
- bindings.kafka
|
||||
- secretstores.local.env
|
||||
- secretstores.local.file
|
||||
- bindings.rabbitmq
|
||||
EOF
|
||||
)
|
||||
echo "::set-output name=pr-components::$PR_COMPONENTS"
|
||||
|
|
|
|||
|
|
@ -0,0 +1,51 @@
|
|||
# RabbitMQ Binding Certification
|
||||
|
||||
The purpose of this module is to provide tests that certify the RabbitMQ Binding as a stable component.
|
||||
|
||||
## Test plan
|
||||
|
||||
* Verify the queue is created/present.
|
||||
* Create component spec.
|
||||
* Run dapr application with component.
|
||||
* Ensure the queue is created/present.
|
||||
* Verify the connection is established to RabbitMQ.
|
||||
* Create component spec.
|
||||
* Run dapr application with component.
|
||||
* Ensure that you have access to the queue and connection to the queue is established.
|
||||
* Verify data is getting stored in the queue.
|
||||
* Create component spec with the data to be stored.
|
||||
* Run dapr application with component to store data in the queue as output binding.
|
||||
* Read stored data from the queue as input binding.
|
||||
* Ensure that read data is same as the data that was stored.
|
||||
* Verify Data level TTL is regarded.
|
||||
* Create component spec with the field `ttlInSeconds`.
|
||||
* Run dapr application with component.
|
||||
* Send a message, wait TTL seconds, and verify the message is deleted/expired.
|
||||
* Verify durable attribute is regarded.
|
||||
* Create component spec with the field `durable` set true.
|
||||
* Run dapr application with component.
|
||||
* Send a message to the queue.
|
||||
* Ensure that the message is stored in the storage.
|
||||
* Verify deleteWhenUnused attribute is regarded.
|
||||
* Create component spec with the field `deleteWhenUnused` set true.
|
||||
* Run dapr application with component.
|
||||
* Send a message to the queue.
|
||||
* Ensure that the message is deleted.
|
||||
* Verify maxPriority attribute is regarded.
|
||||
* Create component spec with the field `maxPriority`.
|
||||
* Run dapr application with component.
|
||||
* Ensure that the priority queue is created.
|
||||
* Verify exclusive attribute is regarded.
|
||||
* Create component spec with the field `exclusive`.
|
||||
* Run dapr application with component.
|
||||
* Ensure that the topic is exclusive.
|
||||
* Verify reconnection to the queue for output binding.
|
||||
* Simulate a network error before sending any messages.
|
||||
* Run dapr application with the component.
|
||||
* After the reconnection, send messages to the queue.
|
||||
* Ensure that the messages sent after the reconnection are sent to the queue.
|
||||
* Verify reconnection to the queue for input binding.
|
||||
* Simulate a network error before reading any messages.
|
||||
* Run dapr application with the component.
|
||||
* After the reconnection, read messages from the queue.
|
||||
* Ensure that the messages after the reconnection are read.
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
apiVersion: dapr.io/v1alpha1
|
||||
kind: Component
|
||||
metadata:
|
||||
name: options-binding
|
||||
spec:
|
||||
type: bindings.rabbitmq
|
||||
version: v1
|
||||
metadata:
|
||||
- name: queueName
|
||||
value: optionsQueue
|
||||
- name: host
|
||||
value: "amqp://test:test@localhost:5672"
|
||||
- name: durable
|
||||
value: true
|
||||
- name: deleteWhenUnused
|
||||
value: true
|
||||
- name: prefetchCount
|
||||
value: 0
|
||||
- name: exclusive
|
||||
value: false
|
||||
# - name: exclusive
|
||||
# value: true
|
||||
- name: maxPriority
|
||||
value: 5
|
||||
- name: contentType
|
||||
value: "text/plain"
|
||||
|
|
@ -0,0 +1,18 @@
|
|||
apiVersion: dapr.io/v1alpha1
|
||||
kind: Component
|
||||
metadata:
|
||||
name: retry-binding
|
||||
spec:
|
||||
type: bindings.rabbitmq
|
||||
version: v1
|
||||
metadata:
|
||||
- name: queueName
|
||||
value: retryQueue
|
||||
- name: host
|
||||
value: "amqp://test:test@localhost:5672"
|
||||
- name: prefetchCount
|
||||
value: 0
|
||||
- name: exclusive
|
||||
value: false
|
||||
- name: contentType
|
||||
value: "text/plain"
|
||||
|
|
@ -0,0 +1,18 @@
|
|||
apiVersion: dapr.io/v1alpha1
|
||||
kind: Component
|
||||
metadata:
|
||||
name: standard-binding
|
||||
spec:
|
||||
type: bindings.rabbitmq
|
||||
version: v1
|
||||
metadata:
|
||||
- name: queueName
|
||||
value: standardQueue
|
||||
- name: host
|
||||
value: "amqp://test:test@localhost:5672"
|
||||
- name: prefetchCount
|
||||
value: 0
|
||||
- name: exclusive
|
||||
value: false
|
||||
- name: contentType
|
||||
value: "text/plain"
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
apiVersion: dapr.io/v1alpha1
|
||||
kind: Component
|
||||
metadata:
|
||||
name: msg-ttl-binding
|
||||
namespace: default
|
||||
spec:
|
||||
type: bindings.rabbitmq
|
||||
version: v1
|
||||
metadata:
|
||||
- name: queueName
|
||||
value: msgttlqueue
|
||||
- name: host
|
||||
value: "amqp://test:test@localhost:5672"
|
||||
- name: prefetchCount
|
||||
value: 0
|
||||
- name: exclusive
|
||||
value: false
|
||||
- name: contentType
|
||||
value: "text/plain"
|
||||
|
|
@ -0,0 +1,21 @@
|
|||
apiVersion: dapr.io/v1alpha1
|
||||
kind: Component
|
||||
metadata:
|
||||
name: overwrite-ttl-binding
|
||||
namespace: default
|
||||
spec:
|
||||
type: bindings.rabbitmq
|
||||
version: v1
|
||||
metadata:
|
||||
- name: queueName
|
||||
value: overwritettlqueue
|
||||
- name: host
|
||||
value: "amqp://test:test@localhost:5672"
|
||||
- name: prefetchCount
|
||||
value: 0
|
||||
- name: exclusive
|
||||
value: false
|
||||
- name: contentType
|
||||
value: "text/plain"
|
||||
- name: ttlInSeconds
|
||||
value: 30 # Short TTL for easier testing
|
||||
|
|
@ -0,0 +1,21 @@
|
|||
apiVersion: dapr.io/v1alpha1
|
||||
kind: Component
|
||||
metadata:
|
||||
name: queue-ttl-binding
|
||||
namespace: default
|
||||
spec:
|
||||
type: bindings.rabbitmq
|
||||
version: v1
|
||||
metadata:
|
||||
- name: queueName
|
||||
value: queuettlqueue
|
||||
- name: host
|
||||
value: "amqp://test:test@localhost:5672"
|
||||
- name: prefetchCount
|
||||
value: 0
|
||||
- name: exclusive
|
||||
value: false
|
||||
- name: contentType
|
||||
value: "text/plain"
|
||||
- name: ttlInSeconds
|
||||
value: 10 # Short TTL for easier testing
|
||||
|
|
@ -0,0 +1,4 @@
|
|||
apiVersion: dapr.io/v1alpha1
|
||||
kind: Configuration
|
||||
metadata:
|
||||
name: bindings
|
||||
|
|
@ -0,0 +1,18 @@
|
|||
version: "3.7"
|
||||
services:
|
||||
rabbitmq:
|
||||
image: rabbitmq:3-management-alpine
|
||||
container_name: 'rabbitmq'
|
||||
ports:
|
||||
- 5672:5672
|
||||
- 15672:15672
|
||||
networks:
|
||||
- rabbitmq_go_net
|
||||
environment:
|
||||
RABBITMQ_DEFAULT_USER: test
|
||||
RABBITMQ_DEFAULT_PASS: test
|
||||
hostname: rmq
|
||||
|
||||
networks:
|
||||
rabbitmq_go_net:
|
||||
driver: bridge
|
||||
|
|
@ -0,0 +1,137 @@
|
|||
module github.com/dapr/components-contrib/tests/certification/binding/rabbitmq
|
||||
|
||||
go 1.19
|
||||
|
||||
require (
|
||||
github.com/dapr/components-contrib v1.8.0-rc.6
|
||||
github.com/dapr/components-contrib/tests/certification v0.0.0-20211130185200-4918900c09e1
|
||||
github.com/dapr/dapr v1.8.4-0.20220822154328-9797b4006e23
|
||||
github.com/dapr/go-sdk v1.4.0
|
||||
github.com/dapr/kit v0.0.2
|
||||
github.com/rabbitmq/amqp091-go v1.3.4
|
||||
github.com/stretchr/testify v1.8.0
|
||||
go.uber.org/multierr v1.8.0
|
||||
)
|
||||
|
||||
require (
|
||||
contrib.go.opencensus.io/exporter/prometheus v0.4.1 // indirect
|
||||
github.com/AdhityaRamadhanus/fasthttpcors v0.0.0-20170121111917-d4c07198763a // indirect
|
||||
github.com/PuerkitoBio/purell v1.1.1 // indirect
|
||||
github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 // indirect
|
||||
github.com/andybalholm/brotli v1.0.2 // indirect
|
||||
github.com/antlr/antlr4 v0.0.0-20200503195918-621b933c7a7f // indirect
|
||||
github.com/armon/go-metrics v0.3.10 // indirect
|
||||
github.com/beorn7/perks v1.0.1 // indirect
|
||||
github.com/cenkalti/backoff v2.2.1+incompatible // indirect
|
||||
github.com/cenkalti/backoff/v4 v4.1.3 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.1.2 // indirect
|
||||
github.com/davecgh/go-spew v1.1.1 // indirect
|
||||
github.com/evanphx/json-patch v4.12.0+incompatible // indirect
|
||||
github.com/fasthttp/router v1.3.8 // indirect
|
||||
github.com/fatih/color v1.13.0 // indirect
|
||||
github.com/fsnotify/fsnotify v1.5.4 // indirect
|
||||
github.com/ghodss/yaml v1.0.0 // indirect
|
||||
github.com/go-kit/log v0.2.0 // indirect
|
||||
github.com/go-logfmt/logfmt v0.5.1 // indirect
|
||||
github.com/go-logr/logr v1.2.3 // indirect
|
||||
github.com/go-logr/stdr v1.2.2 // indirect
|
||||
github.com/gogo/protobuf v1.3.2 // indirect
|
||||
github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
|
||||
github.com/golang/mock v1.6.0 // indirect
|
||||
github.com/golang/protobuf v1.5.2 // indirect
|
||||
github.com/google/cel-go v0.9.0 // indirect
|
||||
github.com/google/go-cmp v0.5.8 // indirect
|
||||
github.com/google/gofuzz v1.2.0 // indirect
|
||||
github.com/google/uuid v1.3.0 // indirect
|
||||
github.com/googleapis/gnostic v0.5.5 // indirect
|
||||
github.com/gorilla/mux v1.8.0 // indirect
|
||||
github.com/grandcat/zeroconf v0.0.0-20190424104450-85eadb44205c // indirect
|
||||
github.com/grpc-ecosystem/go-grpc-middleware v1.3.0 // indirect
|
||||
github.com/grpc-ecosystem/grpc-gateway/v2 v2.7.0 // indirect
|
||||
github.com/hashicorp/consul/api v1.11.0 // indirect
|
||||
github.com/hashicorp/errwrap v1.1.0 // indirect
|
||||
github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
|
||||
github.com/hashicorp/go-hclog v1.2.1 // indirect
|
||||
github.com/hashicorp/go-immutable-radix v1.3.1 // indirect
|
||||
github.com/hashicorp/go-multierror v1.1.1 // indirect
|
||||
github.com/hashicorp/go-rootcerts v1.0.2 // indirect
|
||||
github.com/hashicorp/golang-lru v0.5.4 // indirect
|
||||
github.com/hashicorp/serf v0.9.6 // indirect
|
||||
github.com/imdario/mergo v0.3.12 // indirect
|
||||
github.com/json-iterator/go v1.1.12 // indirect
|
||||
github.com/klauspost/compress v1.15.1 // indirect
|
||||
github.com/mattn/go-colorable v0.1.12 // indirect
|
||||
github.com/mattn/go-isatty v0.0.14 // indirect
|
||||
github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
|
||||
github.com/miekg/dns v1.1.50 // indirect
|
||||
github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1 // indirect
|
||||
github.com/mitchellh/go-homedir v1.1.0 // indirect
|
||||
github.com/mitchellh/mapstructure v1.5.1-0.20220423185008-bf980b35cac4 // indirect
|
||||
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
|
||||
github.com/modern-go/reflect2 v1.0.2 // indirect
|
||||
github.com/openzipkin/zipkin-go v0.4.0 // indirect
|
||||
github.com/pkg/errors v0.9.1 // indirect
|
||||
github.com/pmezard/go-difflib v1.0.0 // indirect
|
||||
github.com/prometheus/client_golang v1.12.2 // indirect
|
||||
github.com/prometheus/client_model v0.2.0 // indirect
|
||||
github.com/prometheus/common v0.35.0 // indirect
|
||||
github.com/prometheus/procfs v0.7.3 // indirect
|
||||
github.com/prometheus/statsd_exporter v0.22.3 // indirect
|
||||
github.com/savsgio/gotils v0.0.0-20210217112953-d4a072536008 // indirect
|
||||
github.com/sirupsen/logrus v1.9.0 // indirect
|
||||
github.com/sony/gobreaker v0.4.2-0.20210216022020-dd874f9dd33b // indirect
|
||||
github.com/spf13/pflag v1.0.5 // indirect
|
||||
github.com/stoewer/go-strcase v1.2.0 // indirect
|
||||
github.com/stretchr/objx v0.4.0 // indirect
|
||||
github.com/tylertreat/comcast v1.0.1 // indirect
|
||||
github.com/valyala/bytebufferpool v1.0.0 // indirect
|
||||
github.com/valyala/fasthttp v1.31.1-0.20211216042702-258a4c17b4f4 // indirect
|
||||
go.opencensus.io v0.23.0 // indirect
|
||||
go.opentelemetry.io/otel v1.7.0 // indirect
|
||||
go.opentelemetry.io/otel/exporters/otlp/internal/retry v1.7.0 // indirect
|
||||
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.7.0 // indirect
|
||||
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.7.0 // indirect
|
||||
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.7.0 // indirect
|
||||
go.opentelemetry.io/otel/exporters/zipkin v1.7.0 // indirect
|
||||
go.opentelemetry.io/otel/sdk v1.7.0 // indirect
|
||||
go.opentelemetry.io/otel/trace v1.7.0 // indirect
|
||||
go.opentelemetry.io/proto/otlp v0.16.0 // indirect
|
||||
go.uber.org/atomic v1.9.0 // indirect
|
||||
golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect
|
||||
golang.org/x/net v0.0.0-20220630215102-69896b714898 // indirect
|
||||
golang.org/x/oauth2 v0.0.0-20220309155454-6242fa91716a // indirect
|
||||
golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 // indirect
|
||||
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 // indirect
|
||||
golang.org/x/text v0.3.7 // indirect
|
||||
golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11 // indirect
|
||||
golang.org/x/tools v0.1.11 // indirect
|
||||
gomodules.xyz/jsonpatch/v2 v2.2.0 // indirect
|
||||
google.golang.org/appengine v1.6.7 // indirect
|
||||
google.golang.org/genproto v0.0.0-20220622171453-ea41d75dfa0f // indirect
|
||||
google.golang.org/grpc v1.47.0 // indirect
|
||||
google.golang.org/protobuf v1.28.0 // indirect
|
||||
gopkg.in/inf.v0 v0.9.1 // indirect
|
||||
gopkg.in/yaml.v2 v2.4.0 // indirect
|
||||
gopkg.in/yaml.v3 v3.0.1 // indirect
|
||||
k8s.io/api v0.23.0 // indirect
|
||||
k8s.io/apiextensions-apiserver v0.23.0 // indirect
|
||||
k8s.io/apimachinery v0.23.0 // indirect
|
||||
k8s.io/client-go v0.23.0 // indirect
|
||||
k8s.io/component-base v0.23.0 // indirect
|
||||
k8s.io/klog/v2 v2.30.0 // indirect
|
||||
k8s.io/kube-openapi v0.0.0-20211115234752-e816edb12b65 // indirect
|
||||
k8s.io/utils v0.0.0-20210930125809-cb0fa318a74b // indirect
|
||||
sigs.k8s.io/controller-runtime v0.11.0 // indirect
|
||||
sigs.k8s.io/json v0.0.0-20211020170558-c049b76a60c6 // indirect
|
||||
sigs.k8s.io/structured-merge-diff/v4 v4.2.0 // indirect
|
||||
sigs.k8s.io/yaml v1.3.0 // indirect
|
||||
)
|
||||
|
||||
replace github.com/dapr/components-contrib/tests/certification => ../../
|
||||
|
||||
replace github.com/dapr/components-contrib => ../../../../
|
||||
|
||||
// Uncomment for local development for testing with changes
|
||||
// in the Dapr runtime. Don't commit with this uncommented!
|
||||
//
|
||||
// replace github.com/dapr/dapr => ../../../../../dapr
|
||||
File diff suppressed because it is too large
Load Diff
|
|
@ -0,0 +1,559 @@
|
|||
/*
|
||||
Copyright 2022 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 rabbitmq_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
amqp "github.com/rabbitmq/amqp091-go"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/multierr"
|
||||
|
||||
"github.com/dapr/components-contrib/bindings"
|
||||
binding_rabbitmq "github.com/dapr/components-contrib/bindings/rabbitmq"
|
||||
binding_loader "github.com/dapr/dapr/pkg/components/bindings"
|
||||
"github.com/dapr/dapr/pkg/runtime"
|
||||
dapr_testing "github.com/dapr/dapr/pkg/testing"
|
||||
daprClient "github.com/dapr/go-sdk/client"
|
||||
"github.com/dapr/go-sdk/service/common"
|
||||
"github.com/dapr/kit/logger"
|
||||
|
||||
"github.com/dapr/components-contrib/tests/certification/embedded"
|
||||
"github.com/dapr/components-contrib/tests/certification/flow"
|
||||
"github.com/dapr/components-contrib/tests/certification/flow/app"
|
||||
"github.com/dapr/components-contrib/tests/certification/flow/dockercompose"
|
||||
"github.com/dapr/components-contrib/tests/certification/flow/network"
|
||||
"github.com/dapr/components-contrib/tests/certification/flow/retry"
|
||||
"github.com/dapr/components-contrib/tests/certification/flow/sidecar"
|
||||
"github.com/dapr/components-contrib/tests/certification/flow/simulate"
|
||||
"github.com/dapr/components-contrib/tests/certification/flow/watcher"
|
||||
)
|
||||
|
||||
const (
|
||||
rabbitMQURL = "amqp://test:test@localhost:5672"
|
||||
clusterName = "rabbitmqcertification"
|
||||
dockerComposeYAML = "docker-compose.yml"
|
||||
numOfMessages = 10
|
||||
sidecarName1 = "dapr-1"
|
||||
sidecarName2 = "dapr-2"
|
||||
)
|
||||
|
||||
func amqpReady(url string) flow.Runnable {
|
||||
return func(ctx flow.Context) error {
|
||||
conn, err := amqp.Dial(url)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
ch, err := conn.Channel()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer ch.Close()
|
||||
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func TestRabbitMQ(t *testing.T) {
|
||||
log := logger.NewLogger("dapr-components")
|
||||
messages := watcher.NewUnordered()
|
||||
|
||||
ports, _ := dapr_testing.GetFreePorts(3)
|
||||
grpcPort := ports[0]
|
||||
httpPort := ports[1]
|
||||
appPort := ports[2]
|
||||
|
||||
test := func(ctx flow.Context) error {
|
||||
client, err := daprClient.NewClientWithPort(fmt.Sprintf("%d", grpcPort))
|
||||
require.NoError(t, err, "Could not initialize dapr client.")
|
||||
|
||||
// Declare the expected data.
|
||||
msgs := make([]string, numOfMessages)
|
||||
|
||||
for i := 0; i < numOfMessages; i++ {
|
||||
msgs[i] = fmt.Sprintf("standard-binding: Message %03d", i)
|
||||
}
|
||||
|
||||
messages.ExpectStrings(msgs...)
|
||||
|
||||
metadata := make(map[string]string)
|
||||
|
||||
ctx.Log("Invoking binding!")
|
||||
for _, msg := range msgs {
|
||||
ctx.Logf("Sending: %q", msg)
|
||||
|
||||
req := &daprClient.InvokeBindingRequest{Name: "standard-binding", Operation: "create", Data: []byte(msg), Metadata: metadata}
|
||||
err := client.InvokeOutputBinding(ctx, req)
|
||||
require.NoError(ctx, err, "error publishing message")
|
||||
}
|
||||
|
||||
// Assertion on the data.
|
||||
messages.Assert(ctx, time.Minute)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
application := func(ctx flow.Context, s common.Service) (err error) {
|
||||
// Setup the input binding endpoints.
|
||||
err = multierr.Combine(err,
|
||||
s.AddBindingInvocationHandler("standard-binding", func(_ context.Context, in *common.BindingEvent) ([]byte, error) {
|
||||
messages.Observe(string(in.Data))
|
||||
ctx.Logf("Got message: %s", string(in.Data))
|
||||
return []byte("{}"), nil
|
||||
}))
|
||||
return err
|
||||
}
|
||||
|
||||
flow.New(t, "rabbitmq certification").
|
||||
// Run the application logic above.
|
||||
Step(dockercompose.Run(clusterName, dockerComposeYAML)).
|
||||
Step("wait for rabbitmq readiness",
|
||||
retry.Do(time.Second, 30, amqpReady(rabbitMQURL))).
|
||||
Step(app.Run("standardApp", fmt.Sprintf(":%d", appPort), application)).
|
||||
Step(sidecar.Run("standardSidecar",
|
||||
embedded.WithAppProtocol(runtime.HTTPProtocol, appPort),
|
||||
embedded.WithDaprGRPCPort(grpcPort),
|
||||
embedded.WithDaprHTTPPort(httpPort),
|
||||
embedded.WithComponentsPath("./components/standard"),
|
||||
runtime.WithOutputBindings(
|
||||
binding_loader.NewOutput("rabbitmq", func() bindings.OutputBinding {
|
||||
return binding_rabbitmq.NewRabbitMQ(log)
|
||||
}),
|
||||
),
|
||||
runtime.WithInputBindings(
|
||||
binding_loader.NewInput("rabbitmq", func() bindings.InputBinding {
|
||||
return binding_rabbitmq.NewRabbitMQ(log)
|
||||
}),
|
||||
))).
|
||||
Step("send and wait", test).
|
||||
Run()
|
||||
}
|
||||
|
||||
func TestRabbitMQForOptions(t *testing.T) {
|
||||
log := logger.NewLogger("dapr-components")
|
||||
messages := watcher.NewUnordered()
|
||||
|
||||
ports, _ := dapr_testing.GetFreePorts(3)
|
||||
grpcPort := ports[0]
|
||||
httpPort := ports[1]
|
||||
appPort := ports[2]
|
||||
|
||||
test := func(ctx flow.Context) error {
|
||||
client, err := daprClient.NewClientWithPort(fmt.Sprintf("%d", grpcPort))
|
||||
require.NoError(t, err, "Could not initialize dapr client.")
|
||||
|
||||
// Declare the expected data.
|
||||
msgs := make([]string, numOfMessages)
|
||||
|
||||
for i := 0; i < numOfMessages; i++ {
|
||||
msgs[i] = fmt.Sprintf("options-binding: Message %03d", i)
|
||||
}
|
||||
|
||||
messages.ExpectStrings(msgs...)
|
||||
|
||||
metadata := make(map[string]string)
|
||||
|
||||
ctx.Log("Invoking binding!")
|
||||
for _, msg := range msgs {
|
||||
ctx.Logf("Sending: %q", msg)
|
||||
|
||||
req := &daprClient.InvokeBindingRequest{Name: "options-binding", Operation: "create", Data: []byte(msg), Metadata: metadata}
|
||||
err := client.InvokeOutputBinding(ctx, req)
|
||||
require.NoError(ctx, err, "error publishing message")
|
||||
}
|
||||
|
||||
// Assertion on the data.
|
||||
messages.Assert(ctx, time.Minute)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
application := func(ctx flow.Context, s common.Service) (err error) {
|
||||
// Setup the input binding endpoints.
|
||||
err = multierr.Combine(err,
|
||||
s.AddBindingInvocationHandler("options-binding", func(_ context.Context, in *common.BindingEvent) ([]byte, error) {
|
||||
messages.Observe(string(in.Data))
|
||||
ctx.Logf("Got message: %s", string(in.Data))
|
||||
return []byte("{}"), nil
|
||||
}))
|
||||
return err
|
||||
}
|
||||
|
||||
flow.New(t, "rabbitmq options certification").
|
||||
// Run the application logic above.
|
||||
Step(dockercompose.Run(clusterName, dockerComposeYAML)).
|
||||
Step("wait for rabbitmq readiness",
|
||||
retry.Do(time.Second, 30, amqpReady(rabbitMQURL))).
|
||||
Step(app.Run("optionsApp", fmt.Sprintf(":%d", appPort), application)).
|
||||
Step(sidecar.Run("optionsSidecar",
|
||||
embedded.WithAppProtocol(runtime.HTTPProtocol, appPort),
|
||||
embedded.WithDaprGRPCPort(grpcPort),
|
||||
embedded.WithDaprHTTPPort(httpPort),
|
||||
embedded.WithComponentsPath("./components/options"),
|
||||
runtime.WithOutputBindings(
|
||||
binding_loader.NewOutput("rabbitmq", func() bindings.OutputBinding {
|
||||
return binding_rabbitmq.NewRabbitMQ(log)
|
||||
}),
|
||||
),
|
||||
runtime.WithInputBindings(
|
||||
binding_loader.NewInput("rabbitmq", func() bindings.InputBinding {
|
||||
return binding_rabbitmq.NewRabbitMQ(log)
|
||||
}),
|
||||
))).
|
||||
Step("send and wait", test).
|
||||
Run()
|
||||
}
|
||||
|
||||
func TestRabbitMQTTLs(t *testing.T) {
|
||||
log := logger.NewLogger("dapr-components")
|
||||
ttlMessages := watcher.NewUnordered()
|
||||
|
||||
ports, _ := dapr_testing.GetFreePorts(3)
|
||||
grpcPort := ports[0]
|
||||
httpPort := ports[1]
|
||||
appPort := ports[2]
|
||||
|
||||
ttlTest := func(ctx flow.Context) error {
|
||||
client, err := daprClient.NewClientWithPort(fmt.Sprintf("%d", grpcPort))
|
||||
require.NoError(t, err, "Could not initialize dapr client.")
|
||||
|
||||
ctx.Logf("Sending messages for expiration.")
|
||||
for i := 0; i < numOfMessages; i++ {
|
||||
msg := fmt.Sprintf("Expiring message %d", i)
|
||||
|
||||
metadata := make(map[string]string)
|
||||
|
||||
// Send to the queue with TTL.
|
||||
queueTTLReq := &daprClient.InvokeBindingRequest{Name: "queue-ttl-binding", Operation: "create", Data: []byte(msg), Metadata: metadata}
|
||||
err := client.InvokeOutputBinding(ctx, queueTTLReq)
|
||||
require.NoError(ctx, err, "error publishing message")
|
||||
|
||||
// Send message with TTL set in yaml file
|
||||
messageTTLReq := &daprClient.InvokeBindingRequest{Name: "msg-ttl-binding", Operation: "create", Data: []byte(msg), Metadata: metadata}
|
||||
messageTTLReq.Metadata["ttlInSeconds"] = "20"
|
||||
err = client.InvokeOutputBinding(ctx, messageTTLReq)
|
||||
require.NoError(ctx, err, "error publishing message")
|
||||
|
||||
// Send message with TTL to ensure it overwrites Queue TTL.
|
||||
mixedTTLReq := &daprClient.InvokeBindingRequest{Name: "overwrite-ttl-binding", Operation: "create", Data: []byte(msg), Metadata: metadata}
|
||||
mixedTTLReq.Metadata["ttlInSeconds"] = "10"
|
||||
err = client.InvokeOutputBinding(ctx, mixedTTLReq)
|
||||
require.NoError(ctx, err, "error publishing message")
|
||||
}
|
||||
|
||||
// Wait for double the TTL after sending the last message.
|
||||
time.Sleep(time.Second * 20)
|
||||
return nil
|
||||
}
|
||||
|
||||
ttlApplication := func(ctx flow.Context, s common.Service) (err error) {
|
||||
// Setup the input binding endpoints.
|
||||
err = multierr.Combine(err,
|
||||
s.AddBindingInvocationHandler("queue-ttl-binding", func(_ context.Context, in *common.BindingEvent) ([]byte, error) {
|
||||
ctx.Logf("Got message: %s", string(in.Data))
|
||||
ttlMessages.FailIfNotExpected(t, string(in.Data))
|
||||
return []byte("{}"), nil
|
||||
}),
|
||||
s.AddBindingInvocationHandler("msg-ttl-binding", func(_ context.Context, in *common.BindingEvent) ([]byte, error) {
|
||||
ctx.Logf("Got message: %s", string(in.Data))
|
||||
ttlMessages.FailIfNotExpected(t, string(in.Data))
|
||||
return []byte("{}"), nil
|
||||
}),
|
||||
s.AddBindingInvocationHandler("overwrite-ttl-binding", func(_ context.Context, in *common.BindingEvent) ([]byte, error) {
|
||||
ctx.Logf("Got message: %s", string(in.Data))
|
||||
ttlMessages.FailIfNotExpected(t, string(in.Data))
|
||||
return []byte("{}"), nil
|
||||
}))
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
freshPorts, _ := dapr_testing.GetFreePorts(2)
|
||||
|
||||
flow.New(t, "rabbitmq ttl certification").
|
||||
// Run the application logic above.
|
||||
Step(dockercompose.Run(clusterName, dockerComposeYAML)).
|
||||
Step("wait for rabbitmq readiness",
|
||||
retry.Do(time.Second, 30, amqpReady(rabbitMQURL))).
|
||||
Step(app.Run("ttlApp", fmt.Sprintf(":%d", appPort), ttlApplication)).
|
||||
Step(sidecar.Run("ttlSidecar",
|
||||
embedded.WithAppProtocol(runtime.HTTPProtocol, appPort),
|
||||
embedded.WithDaprGRPCPort(grpcPort),
|
||||
embedded.WithDaprHTTPPort(httpPort),
|
||||
embedded.WithComponentsPath("./components/ttl"),
|
||||
runtime.WithOutputBindings(
|
||||
binding_loader.NewOutput("rabbitmq", func() bindings.OutputBinding {
|
||||
return binding_rabbitmq.NewRabbitMQ(log)
|
||||
}),
|
||||
),
|
||||
runtime.WithInputBindings(
|
||||
binding_loader.NewInput("rabbitmq", func() bindings.InputBinding {
|
||||
return binding_rabbitmq.NewRabbitMQ(log)
|
||||
}),
|
||||
))).
|
||||
Step("send ttl messages", ttlTest).
|
||||
Step("stop initial sidecar", sidecar.Stop("ttlSidecar")).
|
||||
Step(app.Run("ttlApp", fmt.Sprintf(":%d", appPort), ttlApplication)).
|
||||
Step(sidecar.Run("appSidecar",
|
||||
embedded.WithAppProtocol(runtime.HTTPProtocol, appPort),
|
||||
embedded.WithDaprGRPCPort(freshPorts[0]),
|
||||
embedded.WithDaprHTTPPort(freshPorts[1]),
|
||||
runtime.WithOutputBindings(
|
||||
binding_loader.NewOutput("rabbitmq", func() bindings.OutputBinding {
|
||||
return binding_rabbitmq.NewRabbitMQ(log)
|
||||
}),
|
||||
),
|
||||
runtime.WithInputBindings(
|
||||
binding_loader.NewInput("rabbitmq", func() bindings.InputBinding {
|
||||
return binding_rabbitmq.NewRabbitMQ(log)
|
||||
}),
|
||||
))).
|
||||
Step("verify no messages", func(ctx flow.Context) error {
|
||||
// Assertion on the data.
|
||||
ttlMessages.Assert(t, time.Minute)
|
||||
return nil
|
||||
}) //.
|
||||
//Run()
|
||||
}
|
||||
|
||||
func TestRabbitMQRetriesOnError(t *testing.T) {
|
||||
log := logger.NewLogger("dapr.components")
|
||||
messages := watcher.NewUnordered()
|
||||
|
||||
ports, _ := dapr_testing.GetFreePorts(3)
|
||||
grpcPort := ports[0]
|
||||
httpPort := ports[1]
|
||||
appPort := ports[2]
|
||||
|
||||
testRetry := func(ctx flow.Context) error {
|
||||
client, err := daprClient.NewClientWithPort(fmt.Sprintf("%d", grpcPort))
|
||||
require.NoError(t, err, "Could not initialize dapr client.")
|
||||
|
||||
// Declare the expected data.
|
||||
msgs := make([]string, numOfMessages)
|
||||
for i := 0; i < numOfMessages; i++ {
|
||||
msgs[i] = fmt.Sprintf("Message %03d", i)
|
||||
}
|
||||
|
||||
messages.ExpectStrings(msgs...)
|
||||
|
||||
metadata := make(map[string]string)
|
||||
|
||||
// Send events that the application above will observe.
|
||||
ctx.Log("Invoking binding!")
|
||||
for _, msg := range msgs {
|
||||
ctx.Logf("Sending: %q", msg)
|
||||
|
||||
req := &daprClient.InvokeBindingRequest{Name: "retry-binding", Operation: "create", Data: []byte(msg), Metadata: metadata}
|
||||
err := client.InvokeOutputBinding(ctx, req)
|
||||
require.NoError(ctx, err, "error publishing message")
|
||||
}
|
||||
|
||||
// Assertion on the data.
|
||||
messages.Assert(ctx, time.Minute)
|
||||
|
||||
return nil
|
||||
}
|
||||
// Application logic that tracks messages from a topic.
|
||||
retryApplication := func(ctx flow.Context, s common.Service) (err error) {
|
||||
// Simulate periodic errors.
|
||||
sim := simulate.PeriodicError(ctx, 10)
|
||||
|
||||
// Setup the input binding endpoint.
|
||||
err = multierr.Combine(err,
|
||||
s.AddBindingInvocationHandler("retry-binding", func(_ context.Context, in *common.BindingEvent) ([]byte, error) {
|
||||
if err := sim(); err != nil {
|
||||
ctx.Logf("Failing message: %s", string(in.Data))
|
||||
return nil, err
|
||||
}
|
||||
messages.Observe(string(in.Data))
|
||||
ctx.Logf("Got message: %s", string(in.Data))
|
||||
return []byte("{}"), nil
|
||||
}))
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
flow.New(t, "rabbitmq retry certification").
|
||||
// Run the application logic above.
|
||||
Step(dockercompose.Run(clusterName, dockerComposeYAML)).
|
||||
Step("wait for rabbitmq readiness",
|
||||
retry.Do(time.Second, 30, amqpReady(rabbitMQURL))).
|
||||
Step(app.Run("retryApp", fmt.Sprintf(":%d", appPort), retryApplication)).
|
||||
Step(sidecar.Run("retrySidecar",
|
||||
embedded.WithAppProtocol(runtime.HTTPProtocol, appPort),
|
||||
embedded.WithDaprGRPCPort(grpcPort),
|
||||
embedded.WithDaprHTTPPort(httpPort),
|
||||
embedded.WithComponentsPath("./components/retry"),
|
||||
runtime.WithOutputBindings(
|
||||
binding_loader.NewOutput("rabbitmq", func() bindings.OutputBinding {
|
||||
return binding_rabbitmq.NewRabbitMQ(log)
|
||||
}),
|
||||
),
|
||||
runtime.WithInputBindings(
|
||||
binding_loader.NewInput("rabbitmq", func() bindings.InputBinding {
|
||||
return binding_rabbitmq.NewRabbitMQ(log)
|
||||
}),
|
||||
))).
|
||||
Step("send and wait", testRetry).
|
||||
Run()
|
||||
}
|
||||
|
||||
func TestRabbitMQNetworkError(t *testing.T) {
|
||||
log := logger.NewLogger("dapr-components")
|
||||
messages := watcher.NewUnordered()
|
||||
|
||||
ports, _ := dapr_testing.GetFreePorts(3)
|
||||
grpcPort := ports[0]
|
||||
httpPort := ports[1]
|
||||
appPort := ports[2]
|
||||
|
||||
test := func(ctx flow.Context) error {
|
||||
client, err := daprClient.NewClientWithPort(fmt.Sprintf("%d", grpcPort))
|
||||
require.NoError(t, err, "Could not initialize dapr client.")
|
||||
|
||||
// Declare the expected data.
|
||||
msgs := make([]string, numOfMessages)
|
||||
|
||||
for i := 0; i < numOfMessages; i++ {
|
||||
msgs[i] = fmt.Sprintf("standard-binding: Message %03d", i)
|
||||
}
|
||||
|
||||
messages.ExpectStrings(msgs...)
|
||||
|
||||
metadata := make(map[string]string)
|
||||
|
||||
ctx.Log("Invoking binding!")
|
||||
for _, msg := range msgs {
|
||||
ctx.Logf("Sending: %q", msg)
|
||||
|
||||
req := &daprClient.InvokeBindingRequest{Name: "standard-binding", Operation: "create", Data: []byte(msg), Metadata: metadata}
|
||||
err := client.InvokeOutputBinding(ctx, req)
|
||||
require.NoError(ctx, err, "error publishing message")
|
||||
}
|
||||
|
||||
// Assertion on the data.
|
||||
messages.Assert(ctx, time.Minute)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
application := func(ctx flow.Context, s common.Service) (err error) {
|
||||
// Setup the input binding endpoints.
|
||||
err = multierr.Combine(err,
|
||||
s.AddBindingInvocationHandler("standard-binding", func(_ context.Context, in *common.BindingEvent) ([]byte, error) {
|
||||
messages.Observe(string(in.Data))
|
||||
ctx.Logf("Got message: %s", string(in.Data))
|
||||
return []byte("{}"), nil
|
||||
}))
|
||||
return err
|
||||
}
|
||||
|
||||
flow.New(t, "rabbitmq certification").
|
||||
// Run the application logic above.
|
||||
Step(dockercompose.Run(clusterName, dockerComposeYAML)).
|
||||
Step("wait for rabbitmq readiness",
|
||||
retry.Do(time.Second, 30, amqpReady(rabbitMQURL))).
|
||||
Step(app.Run("standardApp", fmt.Sprintf(":%d", appPort), application)).
|
||||
Step(sidecar.Run("standardSidecar",
|
||||
embedded.WithAppProtocol(runtime.HTTPProtocol, appPort),
|
||||
embedded.WithDaprGRPCPort(grpcPort),
|
||||
embedded.WithDaprHTTPPort(httpPort),
|
||||
embedded.WithComponentsPath("./components/standard"),
|
||||
runtime.WithOutputBindings(
|
||||
binding_loader.NewOutput("rabbitmq", func() bindings.OutputBinding {
|
||||
return binding_rabbitmq.NewRabbitMQ(log)
|
||||
}),
|
||||
),
|
||||
runtime.WithInputBindings(
|
||||
binding_loader.NewInput("rabbitmq", func() bindings.InputBinding {
|
||||
return binding_rabbitmq.NewRabbitMQ(log)
|
||||
}),
|
||||
))).
|
||||
Step("send and wait", test).
|
||||
Step("interrupt network", network.InterruptNetwork(30*time.Second, nil, nil, "5672")).
|
||||
Run()
|
||||
}
|
||||
|
||||
func TestRabbitMQExclusive(t *testing.T) {
|
||||
log := logger.NewLogger("dapr-components")
|
||||
messages := watcher.NewUnordered()
|
||||
|
||||
ports, _ := dapr_testing.GetFreePorts(3)
|
||||
grpcPort := ports[0]
|
||||
httpPort := ports[1]
|
||||
appPort := ports[2]
|
||||
|
||||
test := func(ctx flow.Context) error {
|
||||
client, err := daprClient.NewClientWithPort(fmt.Sprintf("%d", grpcPort))
|
||||
require.NoError(t, err, "Could not initialize dapr client.")
|
||||
|
||||
// Declare the expected data.
|
||||
msgs := make([]string, numOfMessages)
|
||||
|
||||
for i := 0; i < numOfMessages; i++ {
|
||||
msgs[i] = fmt.Sprintf("exclusive-binding: Message %03d", i)
|
||||
}
|
||||
|
||||
messages.ExpectStrings(msgs...)
|
||||
|
||||
metadata := make(map[string]string)
|
||||
|
||||
ctx.Log("Invoking binding!")
|
||||
for _, msg := range msgs {
|
||||
ctx.Logf("Sending: %q", msg)
|
||||
|
||||
req := &daprClient.InvokeBindingRequest{Name: "exclusive-binding", Operation: "create", Data: []byte(msg), Metadata: metadata}
|
||||
err := client.InvokeOutputBinding(ctx, req)
|
||||
require.Error(ctx, err, "error publishing message")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
application := func(ctx flow.Context, s common.Service) (err error) {
|
||||
// Setup the input binding endpoints.
|
||||
err = multierr.Combine(err,
|
||||
s.AddBindingInvocationHandler("exclusive-binding", func(_ context.Context, in *common.BindingEvent) ([]byte, error) {
|
||||
messages.Observe(string(in.Data))
|
||||
ctx.Logf("Got message: %s", string(in.Data))
|
||||
return []byte("{}"), nil
|
||||
}))
|
||||
return err
|
||||
}
|
||||
|
||||
flow.New(t, "rabbitmq certification").
|
||||
// Run the application logic above.
|
||||
Step(dockercompose.Run(clusterName, dockerComposeYAML)).
|
||||
Step("wait for rabbitmq readiness",
|
||||
retry.Do(time.Second, 30, amqpReady(rabbitMQURL))).
|
||||
Step(app.Run("standardApp", fmt.Sprintf(":%d", appPort), application)).
|
||||
Step(sidecar.Run("standardSidecar",
|
||||
embedded.WithAppProtocol(runtime.HTTPProtocol, appPort),
|
||||
embedded.WithDaprGRPCPort(grpcPort),
|
||||
embedded.WithDaprHTTPPort(httpPort),
|
||||
embedded.WithComponentsPath("./components/exclusive"),
|
||||
runtime.WithInputBindings(
|
||||
binding_loader.NewInput("rabbitmq", func() bindings.InputBinding {
|
||||
return binding_rabbitmq.NewRabbitMQ(log)
|
||||
}),
|
||||
))).
|
||||
// TODO: The following test function will always fail as expected because the sidecar didn't initialize the component (expected). This should be updated to look for a much more specific error signature however by reading the sidecar's stderr.
|
||||
Step("send and wait", test).
|
||||
Run()
|
||||
}
|
||||
Loading…
Reference in New Issue