mirror of
https://dev.azure.com/schwarzit/schwarzit.stackit-public/_git/audit-go
synced 2026-02-08 00:57:24 +00:00
404 lines
14 KiB
Go
404 lines
14 KiB
Go
package messaging
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/mock"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
type connectionPoolMock struct {
|
|
mock.Mock
|
|
}
|
|
|
|
func (m *connectionPoolMock) Close() error {
|
|
return m.Called().Error(0)
|
|
}
|
|
|
|
func (m *connectionPoolMock) NewHandle() *ConnectionPoolHandle {
|
|
return m.Called().Get(0).(*ConnectionPoolHandle)
|
|
}
|
|
|
|
func (m *connectionPoolMock) GetConnection(handle *ConnectionPoolHandle) (*AmqpConnection, error) {
|
|
return m.Called(handle).Get(0).(*AmqpConnection), m.Called(handle).Error(1)
|
|
}
|
|
|
|
var _ ConnectionPool = (*connectionPoolMock)(nil)
|
|
|
|
func Test_NewAmqpMessagingApi(t *testing.T) {
|
|
_, err := NewAmqpApi(
|
|
AmqpConnectionPoolConfig{
|
|
Parameters: AmqpConnectionConfig{BrokerUrl: "not-handled-protocol://localhost:5672"},
|
|
PoolSize: 1,
|
|
})
|
|
assert.EqualError(t, err, "new amqp connection pool: initialize connections: new connection: failed to connect to amqp broker: internal connection connect: dial: unsupported scheme \"not-handled-protocol\"")
|
|
}
|
|
|
|
func Test_AmqpMessagingApi_Send(t *testing.T) {
|
|
// Specify test timeout
|
|
ctx, cancelFn := context.WithTimeout(context.Background(), 120*time.Second)
|
|
defer cancelFn()
|
|
|
|
// Start solace docker container
|
|
solaceContainer, err := NewSolaceContainer(context.Background())
|
|
assert.NoError(t, err)
|
|
defer solaceContainer.Stop()
|
|
|
|
t.Run("Missing topic prefix", func(t *testing.T) {
|
|
defer solaceContainer.StopOnError()
|
|
|
|
api, err := NewAmqpApi(AmqpConnectionPoolConfig{
|
|
Parameters: AmqpConnectionConfig{BrokerUrl: solaceContainer.AmqpConnectionString},
|
|
PoolSize: 1,
|
|
})
|
|
assert.NoError(t, err)
|
|
|
|
err = api.Send(ctx, "topic-name", []byte{}, "application/json", make(map[string]any))
|
|
assert.EqualError(t, err, "send: topic \"topic-name\" name lacks mandatory prefix \"topic://\"")
|
|
})
|
|
|
|
t.Run("send successfully", func(t *testing.T) {
|
|
defer solaceContainer.StopOnError()
|
|
|
|
// Initialize the solace queue
|
|
topicSubscriptionTopicPattern := "auditlog/>"
|
|
queueName := "send-successfully"
|
|
assert.NoError(t, solaceContainer.QueueCreate(ctx, queueName))
|
|
assert.NoError(t, solaceContainer.TopicSubscriptionCreate(ctx, queueName, topicSubscriptionTopicPattern))
|
|
topicName := fmt.Sprintf("topic://auditlog/%s", "amqp-send-successfully")
|
|
assert.NoError(t, solaceContainer.ValidateTopicName(topicSubscriptionTopicPattern, topicName))
|
|
|
|
api, err := NewAmqpApi(AmqpConnectionPoolConfig{
|
|
Parameters: AmqpConnectionConfig{BrokerUrl: solaceContainer.AmqpConnectionString},
|
|
PoolSize: 1,
|
|
})
|
|
assert.NoError(t, err)
|
|
|
|
data := []byte("data")
|
|
applicationProperties := make(map[string]interface{})
|
|
applicationProperties["key"] = "value"
|
|
|
|
err = api.Send(ctx, topicName, data, "application/json", applicationProperties)
|
|
assert.NoError(t, err)
|
|
|
|
message, err := solaceContainer.NextMessage(ctx, fmt.Sprintf("queue://%s", queueName), true)
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, "data", string(message.Data[0]))
|
|
assert.Equal(t, topicName, *message.Properties.To)
|
|
assert.Equal(t, "application/json", *message.Properties.ContentType)
|
|
assert.Equal(t, applicationProperties, message.ApplicationProperties)
|
|
|
|
err = api.Close(ctx)
|
|
assert.NoError(t, err)
|
|
})
|
|
}
|
|
|
|
func Test_AmqpMessagingApi_Send_Special_Cases(t *testing.T) {
|
|
|
|
channelReceiver := func(channel chan struct{}) <-chan struct{} {
|
|
return channel
|
|
}
|
|
|
|
newActiveConnection := func() *AmqpConnection {
|
|
channel := make(chan struct{})
|
|
conn := &amqpConnMock{}
|
|
conn.On("Done", mock.Anything).Return(channelReceiver(channel))
|
|
return &AmqpConnection{
|
|
connectionName: "test",
|
|
lock: sync.RWMutex{},
|
|
conn: conn,
|
|
}
|
|
}
|
|
|
|
newClosedConnection := func() *AmqpConnection {
|
|
channel := make(chan struct{})
|
|
close(channel)
|
|
|
|
conn := &amqpConnMock{}
|
|
conn.On("Done", mock.Anything).Return(channelReceiver(channel))
|
|
return &AmqpConnection{
|
|
connectionName: "test",
|
|
lock: sync.RWMutex{},
|
|
conn: conn,
|
|
}
|
|
}
|
|
|
|
t.Run("connection nil sender nil", func(t *testing.T) {
|
|
sender := &amqpSenderMock{}
|
|
sender.On("Send", mock.Anything, mock.Anything, mock.Anything).Return(nil)
|
|
|
|
session := &amqpSessionMock{}
|
|
session.On("NewSender", mock.Anything, mock.Anything, mock.Anything).Return(sender, nil)
|
|
|
|
connection := newActiveConnection()
|
|
conn := connection.conn.(*amqpConnMock)
|
|
conn.On("NewSession", mock.Anything, mock.Anything).Return(session, nil)
|
|
|
|
pool := &connectionPoolMock{}
|
|
pool.On("GetConnection", mock.Anything).Return(connection, nil)
|
|
|
|
amqpApi := &AmqpApi{config: AmqpConnectionPoolConfig{},
|
|
connectionPool: pool,
|
|
connectionPoolHandle: &ConnectionPoolHandle{connectionOffset: 0},
|
|
senderCache: make(map[string]*AmqpSenderSession),
|
|
}
|
|
|
|
err := amqpApi.Send(context.Background(), "topic://some-topic", []byte("data"), "application/json", make(map[string]any))
|
|
assert.NoError(t, err)
|
|
|
|
sender.AssertNumberOfCalls(t, "Send", 1)
|
|
session.AssertNumberOfCalls(t, "NewSender", 1)
|
|
pool.AssertNumberOfCalls(t, "GetConnection", 2)
|
|
})
|
|
|
|
t.Run("connection closed sender nil", func(t *testing.T) {
|
|
sender := &amqpSenderMock{}
|
|
sender.On("Send", mock.Anything, mock.Anything, mock.Anything).Return(nil)
|
|
|
|
session := &amqpSessionMock{}
|
|
session.On("NewSender", mock.Anything, mock.Anything, mock.Anything).Return(sender, nil)
|
|
|
|
connection := newActiveConnection()
|
|
conn := connection.conn.(*amqpConnMock)
|
|
conn.On("NewSession", mock.Anything, mock.Anything).Return(session, nil)
|
|
|
|
pool := &connectionPoolMock{}
|
|
pool.On("GetConnection", mock.Anything).Return(connection, nil)
|
|
|
|
closedConnection := newClosedConnection()
|
|
closedConnMock := closedConnection.conn.(*amqpConnMock)
|
|
amqpApi := &AmqpApi{config: AmqpConnectionPoolConfig{},
|
|
connection: closedConnection,
|
|
connectionPool: pool,
|
|
connectionPoolHandle: &ConnectionPoolHandle{connectionOffset: 0},
|
|
senderCache: make(map[string]*AmqpSenderSession),
|
|
}
|
|
|
|
err := amqpApi.Send(context.Background(), "topic://some-topic", []byte("data"), "application/json", make(map[string]any))
|
|
assert.NoError(t, err)
|
|
|
|
sender.AssertNumberOfCalls(t, "Send", 1)
|
|
session.AssertNumberOfCalls(t, "NewSender", 1)
|
|
pool.AssertNumberOfCalls(t, "GetConnection", 2)
|
|
closedConnMock.AssertNumberOfCalls(t, "Done", 1)
|
|
})
|
|
|
|
t.Run("connection nil get connection fail", func(t *testing.T) {
|
|
var connection *AmqpConnection = nil
|
|
|
|
pool := &connectionPoolMock{}
|
|
pool.On("GetConnection", mock.Anything).Return(connection, errors.New("connection error"))
|
|
|
|
amqpApi := &AmqpApi{config: AmqpConnectionPoolConfig{},
|
|
connectionPool: pool,
|
|
connectionPoolHandle: &ConnectionPoolHandle{connectionOffset: 0},
|
|
senderCache: make(map[string]*AmqpSenderSession),
|
|
}
|
|
|
|
err := amqpApi.Send(context.Background(), "topic://some-topic", []byte("data"), "application/json", make(map[string]any))
|
|
assert.EqualError(t, err, "get connection: connection error")
|
|
|
|
pool.AssertNumberOfCalls(t, "GetConnection", 2)
|
|
})
|
|
|
|
t.Run("connection active sender nil", func(t *testing.T) {
|
|
sender := &amqpSenderMock{}
|
|
sender.On("Send", mock.Anything, mock.Anything, mock.Anything).Return(nil)
|
|
|
|
session := &amqpSessionMock{}
|
|
session.On("NewSender", mock.Anything, mock.Anything, mock.Anything).Return(sender, nil)
|
|
|
|
connection := newActiveConnection()
|
|
conn := connection.conn.(*amqpConnMock)
|
|
conn.On("NewSession", mock.Anything, mock.Anything).Return(session, nil)
|
|
|
|
amqpApi := &AmqpApi{config: AmqpConnectionPoolConfig{},
|
|
connection: connection,
|
|
senderCache: make(map[string]*AmqpSenderSession),
|
|
}
|
|
|
|
err := amqpApi.Send(context.Background(), "topic://some-topic", []byte("data"), "application/json", make(map[string]any))
|
|
assert.NoError(t, err)
|
|
|
|
sender.AssertNumberOfCalls(t, "Send", 1)
|
|
session.AssertNumberOfCalls(t, "NewSender", 1)
|
|
})
|
|
|
|
t.Run("connection active new sender fail", func(t *testing.T) {
|
|
var sender *amqpSenderMock = nil
|
|
|
|
session := &amqpSessionMock{}
|
|
session.On("NewSender", mock.Anything, mock.Anything, mock.Anything).Return(sender, errors.New("new sender error"))
|
|
session.On("Close", mock.Anything).Return(nil)
|
|
|
|
connection := newActiveConnection()
|
|
conn := connection.conn.(*amqpConnMock)
|
|
conn.On("NewSession", mock.Anything, mock.Anything).Return(session, nil)
|
|
|
|
amqpApi := &AmqpApi{config: AmqpConnectionPoolConfig{},
|
|
connection: connection,
|
|
senderCache: make(map[string]*AmqpSenderSession),
|
|
}
|
|
|
|
err := amqpApi.Send(context.Background(), "topic://some-topic", []byte("data"), "application/json", make(map[string]any))
|
|
assert.EqualError(t, err, "new sender: new internal sender: new sender error")
|
|
|
|
session.AssertNumberOfCalls(t, "NewSender", 1)
|
|
session.AssertNumberOfCalls(t, "Close", 1)
|
|
})
|
|
|
|
t.Run("connection active sender set", func(t *testing.T) {
|
|
sender := &amqpSenderMock{}
|
|
sender.On("Send", mock.Anything, mock.Anything, mock.Anything).Return(nil)
|
|
|
|
topic := "topic://some-topic"
|
|
amqpApi := &AmqpApi{config: AmqpConnectionPoolConfig{},
|
|
connection: newActiveConnection(),
|
|
senderCache: map[string]*AmqpSenderSession{topic: {sender: sender}},
|
|
}
|
|
|
|
err := amqpApi.Send(context.Background(), topic, []byte("data"), "application/json", make(map[string]any))
|
|
assert.NoError(t, err)
|
|
|
|
sender.AssertNumberOfCalls(t, "Send", 1)
|
|
})
|
|
|
|
t.Run("send fail", func(t *testing.T) {
|
|
sender := &amqpSenderMock{}
|
|
sender.On("Send", mock.Anything, mock.Anything, mock.Anything).Return(errors.New("send error"))
|
|
|
|
topic := "topic://some-topic"
|
|
amqpApi := &AmqpApi{config: AmqpConnectionPoolConfig{},
|
|
connection: newActiveConnection(),
|
|
senderCache: map[string]*AmqpSenderSession{topic: {sender: sender}},
|
|
}
|
|
|
|
err := amqpApi.Send(context.Background(), topic, []byte("data"), "application/json", make(map[string]any))
|
|
assert.EqualError(t, err, "send: send error")
|
|
|
|
sender.AssertNumberOfCalls(t, "Send", 1)
|
|
})
|
|
}
|
|
|
|
func Test_AmqpMessagingApi_Close(t *testing.T) {
|
|
|
|
t.Run("close without cached senders", func(t *testing.T) {
|
|
pool := &connectionPoolMock{}
|
|
pool.On("Close").Return(nil)
|
|
|
|
amqpApi := &AmqpApi{config: AmqpConnectionPoolConfig{},
|
|
connectionPool: pool,
|
|
connectionPoolHandle: &ConnectionPoolHandle{connectionOffset: 0},
|
|
senderCache: make(map[string]*AmqpSenderSession),
|
|
}
|
|
|
|
err := amqpApi.Close(context.Background())
|
|
assert.NoError(t, err)
|
|
|
|
pool.AssertNumberOfCalls(t, "Close", 1)
|
|
})
|
|
|
|
t.Run("close fail without cached senders", func(t *testing.T) {
|
|
pool := &connectionPoolMock{}
|
|
pool.On("Close").Return(errors.New("close error"))
|
|
|
|
amqpApi := &AmqpApi{config: AmqpConnectionPoolConfig{},
|
|
connectionPool: pool,
|
|
connectionPoolHandle: &ConnectionPoolHandle{connectionOffset: 0},
|
|
senderCache: make(map[string]*AmqpSenderSession),
|
|
}
|
|
|
|
err := amqpApi.Close(context.Background())
|
|
assert.EqualError(t, err, "close: close pool: close error")
|
|
|
|
pool.AssertNumberOfCalls(t, "Close", 1)
|
|
})
|
|
|
|
t.Run("close with cached senders", func(t *testing.T) {
|
|
pool := &connectionPoolMock{}
|
|
pool.On("Close").Return(nil)
|
|
|
|
session := &amqpSessionMock{}
|
|
session.On("Close", mock.Anything).Return(nil)
|
|
sender := &amqpSenderMock{}
|
|
sender.On("Close", mock.Anything).Return(nil)
|
|
senderSession := &AmqpSenderSession{
|
|
session: session,
|
|
sender: sender,
|
|
}
|
|
|
|
amqpApi := &AmqpApi{config: AmqpConnectionPoolConfig{},
|
|
connectionPool: pool,
|
|
connectionPoolHandle: &ConnectionPoolHandle{connectionOffset: 0},
|
|
senderCache: map[string]*AmqpSenderSession{"key": senderSession},
|
|
}
|
|
|
|
err := amqpApi.Close(context.Background())
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, 0, len(amqpApi.senderCache))
|
|
|
|
pool.AssertNumberOfCalls(t, "Close", 1)
|
|
session.AssertNumberOfCalls(t, "Close", 1)
|
|
sender.AssertNumberOfCalls(t, "Close", 1)
|
|
})
|
|
|
|
t.Run("close fail with cached senders", func(t *testing.T) {
|
|
pool := &connectionPoolMock{}
|
|
pool.On("Close").Return(nil)
|
|
|
|
session := &amqpSessionMock{}
|
|
session.On("Close", mock.Anything).Return(nil)
|
|
sender := &amqpSenderMock{}
|
|
sender.On("Close", mock.Anything).Return(errors.New("close sender error"))
|
|
senderSession := &AmqpSenderSession{
|
|
session: session,
|
|
sender: sender,
|
|
}
|
|
|
|
amqpApi := &AmqpApi{config: AmqpConnectionPoolConfig{},
|
|
connectionPool: pool,
|
|
connectionPoolHandle: &ConnectionPoolHandle{connectionOffset: 0},
|
|
senderCache: map[string]*AmqpSenderSession{"key": senderSession},
|
|
}
|
|
|
|
err := amqpApi.Close(context.Background())
|
|
assert.EqualError(t, err, "close: close session: close sender error")
|
|
assert.Equal(t, 0, len(amqpApi.senderCache))
|
|
|
|
pool.AssertNumberOfCalls(t, "Close", 1)
|
|
session.AssertNumberOfCalls(t, "Close", 1)
|
|
sender.AssertNumberOfCalls(t, "Close", 1)
|
|
})
|
|
|
|
t.Run("close fail", func(t *testing.T) {
|
|
pool := &connectionPoolMock{}
|
|
pool.On("Close").Return(errors.New("close pool error"))
|
|
|
|
session := &amqpSessionMock{}
|
|
session.On("Close", mock.Anything).Return(errors.New("close session error"))
|
|
sender := &amqpSenderMock{}
|
|
sender.On("Close", mock.Anything).Return(errors.New("close sender error"))
|
|
senderSession := &AmqpSenderSession{
|
|
session: session,
|
|
sender: sender,
|
|
}
|
|
|
|
amqpApi := &AmqpApi{config: AmqpConnectionPoolConfig{},
|
|
connectionPool: pool,
|
|
connectionPoolHandle: &ConnectionPoolHandle{connectionOffset: 0},
|
|
senderCache: map[string]*AmqpSenderSession{"key": senderSession},
|
|
}
|
|
|
|
err := amqpApi.Close(context.Background())
|
|
assert.EqualError(t, err, "close: close session: close sender error\nclose session error\nclose pool: close pool error")
|
|
assert.Equal(t, 0, len(amqpApi.senderCache))
|
|
|
|
pool.AssertNumberOfCalls(t, "Close", 1)
|
|
session.AssertNumberOfCalls(t, "Close", 1)
|
|
sender.AssertNumberOfCalls(t, "Close", 1)
|
|
})
|
|
}
|