chore: more tests

connection_test.go

@@ -268,6 +268,12 @@ func Test_ExchangeDeclare(t *testing.T) {
 	require.Equal(t, ExchangeDeclaration{name: "name", kind: "topic", durable: true, autoDelete: false, noWait: false, args: nil}, channel.ExchangeDeclarations[0])
 }
 
+func Test_Publish_Fail(t *testing.T) {
+	channel := NewMockAmqpChannel()
+	err := publishMessage(context.Background(), channel, Message{true}, "failed", "exchange", nil)
+	require.EqualError(t, err, "failed")
+}
+
 func Test_Publish(t *testing.T) {
 	channel := NewMockAmqpChannel()
 	headers := amqp.Table{}

consumer_test.go

@@ -33,6 +33,14 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func Test_Invalid_Payload(t *testing.T) {
+	err := newWrappedHandler(func(ctx context.Context, event ConsumableEvent[string]) error {
+		return nil
+	})(context.TODO(), unmarshalEvent{Payload: []byte(`{"a":}`)})
+	require.ErrorIs(t, err, ErrParseJSON)
+	require.ErrorContains(t, err, "invalid character '}' looking for beginning of value")
+}
+
 func Test_Consume(t *testing.T) {
 	consumer := queueConsumer{
 		queue:    "aQueue",
@@ -107,6 +115,62 @@ func Test_ConsumerLoop(t *testing.T) {
 	require.Len(t, acker.Acks, 2)
 }
 
+func Test_HandleDelivery(t *testing.T) {
+	tests := []struct {
+		name            string
+		error           error
+		numberOfAcks    int
+		numberOfNacks   int
+		numberOfRejects int
+		notification    string
+	}{
+		{
+			name:         "ok",
+			notification: "event handler for key succeeded",
+			numberOfAcks: 1,
+		},
+		{
+			name:          "invalid JSON",
+			error:         ErrParseJSON,
+			notification:  "error: failed to parse",
+			numberOfNacks: 1,
+		},
+		{
+			name:            "no match for routingkey",
+			error:           ErrNoMessageTypeForRouteKey,
+			notification:    "error: no message type for routingkey configured",
+			numberOfRejects: 1,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			notifications := make(chan Notification, 1)
+			consumer := queueConsumer{
+				notificationCh: notifications,
+			}
+			deliveryInfo := DeliveryInfo{
+				RoutingKey: "key",
+			}
+			acker := MockAcknowledger{
+				Acks:    make(chan Ack, 1),
+				Nacks:   make(chan Nack, 1),
+				Rejects: make(chan Reject, 1),
+			}
+			handler := func(ctx context.Context, event unmarshalEvent) error {
+				return tt.error
+			}
+			d := delivery(acker, "routingKey", true)
+			consumer.handleDelivery(handler, d, deliveryInfo)
+			notification := <-notifications
+			require.Contains(t, notification.Message, tt.notification)
+
+			require.Len(t, acker.Acks, tt.numberOfAcks)
+			require.Len(t, acker.Nacks, tt.numberOfNacks)
+			require.Len(t, acker.Rejects, tt.numberOfRejects)
+		})
+	}
+}
+
 func delivery(acker MockAcknowledger, routingKey string, success bool) amqp.Delivery {
 	body, _ := json.Marshal(Message{success})
 

mocks_test.go

@@ -181,10 +181,6 @@ func (m *MockAmqpChannel) ExchangeDeclare(name, kind string, durable, autoDelete
 	return nil
 }
 
-func (m *MockAmqpChannel) Publish(exchange, key string, mandatory, immediate bool, msg amqp.Publishing) error {
-	return m.PublishWithContext(context.Background(), exchange, key, mandatory, immediate, msg)
-}
-
 func (m *MockAmqpChannel) PublishWithContext(ctx context.Context, exchange, key string, mandatory, immediate bool, msg amqp.Publishing) error {
 	if key == "failed" {
 		return errors.New("failed")
@@ -236,14 +232,6 @@ func NewMockAmqpChannel() *MockAmqpChannel {
 	}
 }
 
-func NewMockAcknowledger() MockAcknowledger {
-	return MockAcknowledger{
-		Acks:    make(chan Ack, 2),
-		Nacks:   make(chan Nack, 2),
-		Rejects: make(chan Reject, 2),
-	}
-}
-
 var (
 	_ amqpConnection = &MockAmqpConnection{}
 	_ AmqpChannel    = &MockAmqpChannel{}

request_response_test.go

@@ -60,6 +60,9 @@ func Test_RequestResponseHandler(t *testing.T) {
 	require.Len(t, *conn.queueConsumers, 1)
 	handler, _ := conn.queueConsumers.get("svc.direct.exchange.request.queue", "key")
 	require.Equal(t, "github.com/sparetimecoders/goamqp.ServiceRequestConsumer[...].func1", runtime.FuncForPC(reflect.ValueOf(handler).Pointer()).Name())
+	missing, exists := conn.queueConsumers.get("miggins", "key")
+	require.Nil(t, missing)
+	require.False(t, exists)
 
 	msg, _ := json.Marshal(Message{Ok: true})
 	err = handler(context.TODO(), unmarshalEvent{

setup_test.go

@@ -127,8 +127,11 @@ func Test_Start_WithPrefetchLimit_Resets_Qos(t *testing.T) {
 		channel:        mockChannel,
 		queueConsumers: &queueConsumers{},
 	}
+	notifications := make(chan<- Notification)
 	err := conn.Start(context.Background(),
 		WithPrefetchLimit(1),
+		WithNotificationChannel(notifications),
 	)
 	require.NoError(t, err)
+	require.Equal(t, notifications, conn.notificationCh)
 }