fix(MAGNETO-7807): Fixed rabbitmq sonar issues

go.mod

@@ -6,7 +6,7 @@ require (
 	github.com/AdguardTeam/dnsproxy v0.41.2
 	github.com/aws/aws-sdk-go v1.38.71
 	github.com/cucumber/godog v0.12.0
-	github.com/cucumber/messages-go/v16 v16.0.1 // indirect
+	github.com/cucumber/messages-go/v16 v16.0.1
 	github.com/elastic/go-elasticsearch/v7 v7.12.0
 	github.com/go-redis/redis/v8 v8.7.1
 	github.com/google/uuid v1.2.0

steps/rabbit/amqp.go

@@ -0,0 +1,74 @@
+// Copyright 2021 Telefonica Cybersecurity & Cloud Tech SL
+//
+// 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 rabbit
+
+import (
+	"github.com/streadway/amqp"
+)
+
+type AMQPServiceFunctions interface {
+	Dial(url string) (*amqp.Connection, error)
+	ConnectionChannel(c *amqp.Connection) (*amqp.Channel, error)
+	ChannelExchangeDeclare(channel *amqp.Channel, name string) error
+	ChannelQueueDeclare(channel *amqp.Channel) (amqp.Queue, error)
+	ChannelQueueBind(channel *amqp.Channel, name, exchange string) error
+	ChannelConsume(channel *amqp.Channel, queue string,
+	) (<-chan amqp.Delivery, error)
+	ChannelClose(channel *amqp.Channel) error
+	ChannelPublish(channel *amqp.Channel,
+		exchange string,
+		msg amqp.Publishing,
+	) error
+}
+
+type AMQPService struct{}
+
+func NewAMQPService() *AMQPService {
+	return &AMQPService{}
+}
+
+func (a AMQPService) Dial(url string) (*amqp.Connection, error) {
+	return amqp.Dial(url)
+}
+
+func (a AMQPService) ConnectionChannel(connection *amqp.Connection) (*amqp.Channel, error) {
+	return connection.Channel()
+}
+func (a AMQPService) ChannelExchangeDeclare(channel *amqp.Channel, name string) error {
+	return channel.ExchangeDeclare(name, "fanout", true, false, false, false, nil)
+}
+
+func (a AMQPService) ChannelQueueDeclare(channel *amqp.Channel) (amqp.Queue, error) {
+	return channel.QueueDeclare("", false, true, true, false, nil)
+}
+func (a AMQPService) ChannelQueueBind(channel *amqp.Channel, name, exchange string) error {
+	return channel.QueueBind(name, "", exchange, false, nil)
+}
+
+func (a AMQPService) ChannelConsume(channel *amqp.Channel, queue string,
+) (<-chan amqp.Delivery, error) {
+	return channel.Consume(queue, "", true, false, false, false, nil)
+}
+
+func (a AMQPService) ChannelClose(channel *amqp.Channel) error {
+	return channel.Close()
+}
+
+func (a AMQPService) ChannelPublish(channel *amqp.Channel,
+	exchange string,
+	msg amqp.Publishing,
+) error {
+	return channel.Publish(exchange, "", false, false, msg)
+}

steps/rabbit/amqp_mock.go

@@ -0,0 +1,68 @@
+// Copyright 2021 Telefonica Cybersecurity & Cloud Tech SL
+//
+// 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 rabbit
+
+import (
+	"github.com/streadway/amqp"
+)
+
+var (
+	DialError                   error
+	ConnectionChannelError      error
+	ChannelExchangeDeclareError error
+	ChannelQueueDeclareError    error
+	ChannelQueueBindError       error
+	ChannelConsumeError         error
+	MockSubCh                   <-chan amqp.Delivery
+	ChannelPublishError         error
+)
+
+type AMQPServiceFuncMock struct{}
+
+func (a AMQPServiceFuncMock) Dial(url string) (*amqp.Connection, error) {
+	return nil, DialError
+}
+
+func (a AMQPServiceFuncMock) ConnectionChannel(c *amqp.Connection) (*amqp.Channel, error) {
+	return nil, ConnectionChannelError
+}
+
+func (a AMQPServiceFuncMock) ChannelExchangeDeclare(channel *amqp.Channel, name string) error {
+	return ChannelExchangeDeclareError
+}
+
+func (a AMQPServiceFuncMock) ChannelQueueDeclare(channel *amqp.Channel) (amqp.Queue, error) {
+	amqpQueue := amqp.Queue{}
+	return amqpQueue, ChannelQueueDeclareError
+}
+
+func (a AMQPServiceFuncMock) ChannelQueueBind(channel *amqp.Channel, name, exchange string) error {
+	return ChannelQueueBindError
+}
+
+func (a AMQPServiceFuncMock) ChannelConsume(channel *amqp.Channel, queue string,
+) (<-chan amqp.Delivery, error) {
+	return MockSubCh, ChannelConsumeError
+}
+func (a AMQPServiceFuncMock) ChannelClose(channel *amqp.Channel) error {
+	return nil
+}
+
+func (a AMQPServiceFuncMock) ChannelPublish(channel *amqp.Channel,
+	exchange string,
+	msg amqp.Publishing,
+) error {
+	return ChannelPublishError
+}

steps/rabbit/context.go

@@ -26,7 +26,7 @@ const contextKey ContextKey = "rabbitSession"
 // InitializeContext adds the rabbit session to the context.
 // The new context is returned because context is immutable.
 func InitializeContext(ctx context.Context) context.Context {
-	return context.WithValue(ctx, contextKey, &Session{})
+	return context.WithValue(ctx, contextKey, &Session{AMQPService: *NewAMQPService()})
 }
 
 // GetSession returns the rabbit session stored in context.