Testing vhost feature

pkg/adapter/adapter.go

@@ -114,8 +114,8 @@ func vhostHandler(brokers string, vhost string) string {
 	return fmt.Sprintf("%s%s", brokers, vhost)
 }
 
-func (a *Adapter) CreateConn(User string, Password string, logger *zap.Logger) (*amqp.Conn, error) {
-	if User != "" && Password != "" {
+func (a *Adapter) CreateConn(user string, password string, logger *zap.Logger) (*amqp.Conn, error) {
+	if user != "" && password != "" {
 		a.config.Brokers = fmt.Sprintf(
 			"amqp://%s:%s@%s",
 			a.config.User,

pkg/adapter/adapter_test.go

@@ -22,13 +22,17 @@ import (
 	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
+	"reflect"
 	"testing"
+	"time"
 
+	"github.com/NeowayLabs/wabbit"
 	"github.com/NeowayLabs/wabbit/amqp"
 	"github.com/NeowayLabs/wabbit/amqptest"
 	"github.com/NeowayLabs/wabbit/amqptest/server"
 	origamqp "github.com/streadway/amqp"
 	"go.uber.org/zap"
+	"knative.dev/eventing/pkg/adapter/v2"
 	"knative.dev/eventing/pkg/kncloudevents"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/source"
@@ -477,6 +481,11 @@ func TestAdapterVhostHandler(t *testing.T) {
 		brokers: "amqp://localhost:5672",
 		vhost:   "test-vhost",
 		want:    "amqp://localhost:5672/test-vhost",
+	}, {
+		name:    "brokers and vhost without separating slash but vhost with ending slash",
+		brokers: "amqp://localhost:5672",
+		vhost:   "test-vhost/",
+		want:    "amqp://localhost:5672/test-vhost/",
 	}, {
 		name:    "brokers with trailing slash and vhost without the slash",
 		brokers: "amqp://localhost:5672/",
@@ -494,6 +503,7 @@ func TestAdapterVhostHandler(t *testing.T) {
 		want:    "amqp://localhost:5672//test-vhost",
 	}} {
 		t.Run(tt.name, func(t *testing.T) {
+			tt := tt
 			t.Parallel()
 			got := vhostHandler(tt.brokers, tt.vhost)
 			if got != tt.want {
@@ -502,3 +512,118 @@ func TestAdapterVhostHandler(t *testing.T) {
 		})
 	}
 }
+
+func TestAdapter_PollForMessages(t *testing.T) {
+	fakeServer := server.NewServer("amqp://localhost:5672/%2f")
+	err := fakeServer.Start()
+	if err != nil {
+		t.Errorf("%s: %s", "Failed to connect to RabbitMQ", err)
+	}
+
+	conn, err := amqptest.Dial("amqp://localhost:5672/%2f")
+	if err != nil {
+		t.Errorf("%s: %s", "Failed to connect to RabbitMQ", err)
+	}
+
+	channel, err := conn.Channel()
+	if err != nil {
+		t.Errorf("Failed to open a channel")
+	}
+
+	statsReporter, _ := source.NewStatsReporter()
+
+	a := &Adapter{
+		config: &adapterConfig{
+			Topic:   "topic",
+			Brokers: "amqp://guest:guest@localhost:5672/",
+			ExchangeConfig: ExchangeConfig{
+				Name:        "Test-exchange",
+				TypeOf:      "topic",
+				Durable:     true,
+				AutoDeleted: false,
+				Internal:    false,
+				NoWait:      false,
+			},
+			QueueConfig: QueueConfig{
+				Name:             "",
+				Durable:          false,
+				DeleteWhenUnused: false,
+				Exclusive:        true,
+				NoWait:           false,
+			},
+		},
+		context:  context.TODO(),
+		logger:   zap.NewNop(),
+		reporter: statsReporter,
+	}
+
+	err = channel.ExchangeDeclare(a.config.ExchangeConfig.Name, a.config.ExchangeConfig.TypeOf, nil)
+
+	if err != nil {
+		t.Errorf("Failed to declare an exchange")
+	}
+
+	queue, err := channel.QueueDeclare("", wabbit.Option{
+		"durable":   a.config.QueueConfig.Durable,
+		"delete":    a.config.QueueConfig.DeleteWhenUnused,
+		"exclusive": a.config.QueueConfig.Exclusive,
+		"noWait":    a.config.QueueConfig.NoWait,
+	})
+	if err != nil {
+		t.Errorf(err.Error())
+	}
+
+	if err := channel.Confirm(false); err != nil {
+		t.Fatalf("[x] Channel could not be put into confirm mode: %s", err)
+	}
+
+	ctx, cancelFunc := context.WithDeadline(context.TODO(), time.Now().Add(100*time.Millisecond))
+	defer cancelFunc()
+
+	err = a.PollForMessages(&channel, &queue, ctx.Done())
+
+	if err != nil {
+		t.Errorf("testing err %s", err)
+	}
+
+	channel.Close()
+	fakeServer.Stop()
+}
+
+func TestAdapter_NewEnvConfig(t *testing.T) {
+	env := NewEnvConfig()
+	var envPlaceholder adapter.EnvConfigAccessor
+	if reflect.TypeOf(env) == reflect.TypeOf(envPlaceholder) {
+		t.Errorf("Error in NewnvConfig return Type")
+	}
+}
+
+func TestAdapter_NewAdapter(t *testing.T) {
+	ctx := context.TODO()
+	env := NewEnvConfig()
+	h := &fakeHandler{
+		handler: sinkAccepted,
+	}
+
+	sinkServer := httptest.NewServer(h)
+	defer sinkServer.Close()
+
+	s, err := kncloudevents.NewHTTPMessageSenderWithTarget(sinkServer.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	statsReporter, _ := source.NewStatsReporter()
+	a := NewAdapter(ctx, env, s, statsReporter)
+	cmpA := &Adapter{
+		config:            env.(*adapterConfig),
+		httpMessageSender: s,
+		reporter:          statsReporter,
+		logger:            logging.FromContext(ctx).Desugar(),
+		context:           ctx,
+	}
+
+	if a == cmpA {
+		t.Errorf("Error in NewnvConfig return Type")
+	}
+}

test/README.md

@@ -0,0 +1,3 @@
+# Tests
+
+For more info on the tests see [Eventing Tests](https://github.com/knative/eventing/blob/main/test/README.md)

test/e2e/config/source/rabbitmq-source.yaml

@@ -1,3 +1,17 @@
+# Copyright 2021 The Knative 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
+#
+#     https://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.
+
 apiVersion: sources.knative.dev/v1alpha1
 kind: RabbitmqSource
 metadata:

test/e2e/config/sourcevhost/rabbitmq-source-vhost.yaml

@@ -0,0 +1,44 @@
+# Copyright 2021 The Knative 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
+#
+#     https://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.
+
+apiVersion: sources.knative.dev/v1alpha1
+kind: RabbitmqSource
+metadata:
+  name: rabbitmq-source-vhost
+  namespace: {{ .namespace }}
+spec:
+  brokers: "rabbitmqc.{{ .namespace }}:5672/"
+  vhost: "test-vhost"
+  topic: ""
+  user:
+    secretKeyRef:
+      name: "rabbitmqc-default-user"
+      key: "username"
+  password:
+    secretKeyRef:
+      name: "rabbitmqc-default-user"
+      key: "password"
+  exchange_config:
+    type: "fanout"
+    durable: true
+    auto_deleted: false
+    internal: false
+    nowait: false
+  queue_config:
+    name: ""
+    routing_key: ""
+    durable: false
+    delete_when_unused: false
+    exclusive: false
+    nowait: false

test/e2e/config/sourcevhost/sourcevhost.go

@@ -0,0 +1,37 @@
+/*
+Copyright 2021 The Knative 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 sourcevhost
+
+import (
+	"context"
+
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+func init() {
+	environment.RegisterPackage(manifest.ImagesLocalYaml()...)
+}
+
+func Install() feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		if _, err := manifest.InstallLocalYaml(ctx, nil); err != nil {
+			t.Fatal(err)
+		}
+	}
+}

test/e2e/main_test.go

@@ -147,3 +147,13 @@ func TestSourceDirect(t *testing.T) {
 	env.Test(ctx, t, DirectSourceTest())
 	env.Finish()
 }
+
+func TestSourceVhostSetup(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+	)
+	env.Test(ctx, t, VHostSourceTest())
+	env.Finish()
+}

test/e2e/source_test.go

@@ -23,6 +23,8 @@ import (
 
 	"knative.dev/eventing-rabbitmq/test/e2e/config/source"
 	"knative.dev/eventing-rabbitmq/test/e2e/config/sourceproducer"
+	"knative.dev/eventing-rabbitmq/test/e2e/config/sourcevhost"
+
 	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/feature"
 
@@ -53,3 +55,13 @@ func DirectSourceTest() *feature.Feature {
 
 	return f
 }
+
+// VhostSourceTest makes sure an RabbitMQ Source is created on the desired vhost.
+func VHostSourceTest() *feature.Feature {
+	f := new(feature.Feature)
+
+	f.Setup("install RabbitMQ source on test-vhost", sourcevhost.Install())
+	// TODO: still needs some more logic to test all of it
+
+	return f
+}