kmsvc-sdk/messages_test.go

package kmsvc

import (
	"bytes"
	"context"
	"errors"
	"testing"

	kafkamgmtv1 "forgejo.riotpiao.com/rock/kmsvc-proto/gen/kafkamgmt/v1"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

func TestSendMessageRejectsOversizedBody(t *testing.T) {
	client := newTestClient(t, &fakeQueueService{})

	_, err := client.SendMessage(context.Background(), SendMessageInput{
		QueueName: "q",
		Body:      bytes.Repeat([]byte("x"), MaxMessageBodyBytes+1),
	})
	if !errors.Is(err, ErrMessageTooLarge) {
		t.Fatalf("err = %v, want ErrMessageTooLarge", err)
	}
}

func TestSendMessageMapsRequestAndResponse(t *testing.T) {
	var gotReq *kafkamgmtv1.SendMessageRequest
	fake := &fakeQueueService{
		sendMessage: func(ctx context.Context, req *kafkamgmtv1.SendMessageRequest) (*kafkamgmtv1.SendMessageResponse, error) {
			gotReq = req
			return &kafkamgmtv1.SendMessageResponse{MessageId: "m1", SequenceNumber: "seq1"}, nil
		},
	}
	client := newTestClient(t, fake)

	out, err := client.SendMessage(context.Background(), SendMessageInput{
		QueueName:      "orders",
		Body:           []byte("payload"),
		Attributes:     map[string]string{"k": "v"},
		MessageGroupID: "group-1",
		DelaySeconds:   5,
	})
	if err != nil {
		t.Fatalf("SendMessage: %v", err)
	}

	if out.MessageID != "m1" || out.SequenceNumber != "seq1" {
		t.Errorf("unexpected output: %+v", out)
	}
	if gotReq.QueueName != "orders" || string(gotReq.MessageBody) != "payload" || gotReq.MessageGroupId != "group-1" || gotReq.DelaySeconds != 5 {
		t.Errorf("unexpected request: %+v", gotReq)
	}
	if gotReq.MessageAttributes.Values["k"] != "v" {
		t.Errorf("unexpected attributes: %+v", gotReq.MessageAttributes)
	}
}

func TestSendMessageMapsNotFoundError(t *testing.T) {
	fake := &fakeQueueService{
		sendMessage: func(ctx context.Context, req *kafkamgmtv1.SendMessageRequest) (*kafkamgmtv1.SendMessageResponse, error) {
			return nil, status.Error(codes.NotFound, "queue does not exist")
		},
	}
	client := newTestClient(t, fake)

	_, err := client.SendMessage(context.Background(), SendMessageInput{QueueName: "missing", Body: []byte("x")})
	if !errors.Is(err, ErrQueueNotFound) {
		t.Fatalf("err = %v, want ErrQueueNotFound", err)
	}
}

func TestDeleteMessage(t *testing.T) {
	var gotReq *kafkamgmtv1.DeleteMessageRequest
	fake := &fakeQueueService{
		deleteMessage: func(ctx context.Context, req *kafkamgmtv1.DeleteMessageRequest) (*kafkamgmtv1.DeleteMessageResponse, error) {
			gotReq = req
			return &kafkamgmtv1.DeleteMessageResponse{}, nil
		},
	}
	client := newTestClient(t, fake)

	if err := client.DeleteMessage(context.Background(), "q", "rh-1"); err != nil {
		t.Fatalf("DeleteMessage: %v", err)
	}
	if gotReq.QueueName != "q" || gotReq.ReceiptHandle != "rh-1" {
		t.Errorf("unexpected request: %+v", gotReq)
	}
}

func TestChangeMessageVisibility(t *testing.T) {
	var gotReq *kafkamgmtv1.ChangeMessageVisibilityRequest
	fake := &fakeQueueService{
		changeMessageVisibility: func(ctx context.Context, req *kafkamgmtv1.ChangeMessageVisibilityRequest) (*kafkamgmtv1.ChangeMessageVisibilityResponse, error) {
			gotReq = req
			return &kafkamgmtv1.ChangeMessageVisibilityResponse{}, nil
		},
	}
	client := newTestClient(t, fake)

	if err := client.ChangeMessageVisibility(context.Background(), "q", "rh-1", 30); err != nil {
		t.Fatalf("ChangeMessageVisibility: %v", err)
	}
	if gotReq.VisibilityTimeoutSeconds != 30 {
		t.Errorf("unexpected request: %+v", gotReq)
	}
}

func TestSendMessageBatchRejectsOversizedEntry(t *testing.T) {
	client := newTestClient(t, &fakeQueueService{})

	_, err := client.SendMessageBatch(context.Background(), "q", []SendMessageBatchEntry{
		{ID: "1", Body: []byte("ok")},
		{ID: "2", Body: bytes.Repeat([]byte("x"), MaxMessageBodyBytes+1)},
	})
	if !errors.Is(err, ErrMessageTooLarge) {
		t.Fatalf("err = %v, want ErrMessageTooLarge", err)
	}
}

func TestSendMessageBatchMapsResults(t *testing.T) {
	fake := &fakeQueueService{
		sendMessageBatch: func(ctx context.Context, req *kafkamgmtv1.SendMessageBatchRequest) (*kafkamgmtv1.SendMessageBatchResponse, error) {
			return &kafkamgmtv1.SendMessageBatchResponse{
				Successful: []*kafkamgmtv1.BatchResultEntry{{Id: "1", MessageId: "m1"}},
				Failed:     []*kafkamgmtv1.BatchResultEntry{{Id: "2", Error: "boom"}},
			}, nil
		},
	}
	client := newTestClient(t, fake)

	out, err := client.SendMessageBatch(context.Background(), "q", []SendMessageBatchEntry{
		{ID: "1", Body: []byte("a")},
		{ID: "2", Body: []byte("b")},
	})
	if err != nil {
		t.Fatalf("SendMessageBatch: %v", err)
	}
	if len(out.Successful) != 1 || out.Successful[0].MessageID != "m1" {
		t.Errorf("unexpected successful: %+v", out.Successful)
	}
	if len(out.Failed) != 1 || out.Failed[0].Error != "boom" {
		t.Errorf("unexpected failed: %+v", out.Failed)
	}
}