internal/webhooks/consumer_scope_test.go

Rama main · réplica de lectura del repositorio original.

package webhooks

import (
	"context"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/riverqueue/river/rivertype"
	"github.com/stretchr/testify/mock"

	svcerrors "github.com/sarathsp06/sparrow/internal/errors"
	"github.com/sarathsp06/sparrow/internal/tenant"
	"github.com/sarathsp06/sparrow/internal/webhooks/store"
)

func (m *mockRepo) GetBatchJob(ctx context.Context, tenantID uuid.UUID, batchID uuid.UUID) (*store.BatchJob, error) {
	args := m.Called(ctx, tenantID, batchID)
	res := args.Get(0)
	if res == nil {
		return nil, args.Error(1)
	}
	return res.(*store.BatchJob), args.Error(1)
}

func (m *mockRepo) UpdateBatchJobStatus(ctx context.Context, _ uuid.UUID, batchID uuid.UUID, from, to store.BatchJobStatus) error {
	return m.Called(ctx, batchID, from, to).Error(0)
}

func (m *mockRepo) GetDeliveryAttempts(ctx context.Context, tenantID uuid.UUID, deliveryID uuid.UUID) ([]*store.WebhookHealthEvent, error) {
	args := m.Called(ctx, tenantID, deliveryID)
	return args.Get(0).([]*store.WebhookHealthEvent), args.Error(1)
}

// A batch job is reachable only from its own consumer's path. A job
// snapshotted across all consumers is reachable from any consumer path by a
// tenant-wide caller, never by a consumer-scoped one (portal token).
func TestBatchJobsStayInTheirConsumer(t *testing.T) {
	cases := []struct {
		name        string
		jobConsumer string
		scope       BatchScope
		reachable   bool
	}{
		{"own consumer", "acme", BatchScope{Consumer: "acme"}, true},
		{"own consumer, portal", "acme", BatchScope{Consumer: "acme", ConsumerOnly: true}, true},
		{"other consumer", "acme", BatchScope{Consumer: "globex"}, false},
		{"other consumer, portal", "acme", BatchScope{Consumer: "globex", ConsumerOnly: true}, false},
		{"all-consumers job, tenant-wide caller", "", BatchScope{Consumer: "default"}, true},
		{"all-consumers job, portal", "", BatchScope{Consumer: "acme", ConsumerOnly: true}, false},
	}
	ops := []struct {
		name string
		call func(svc *WebhookService, scope BatchScope, id string) error
	}{
		{"GetRetryStatus", func(svc *WebhookService, scope BatchScope, id string) error {
			_, err := svc.GetRetryStatus(testContext(), scope, id)
			return err
		}},
		{"CancelRetry", func(svc *WebhookService, scope BatchScope, id string) error {
			return svc.CancelRetry(testContext(), scope, id)
		}},
		{"RetryDeliveries", func(svc *WebhookService, scope BatchScope, id string) error {
			return svc.RetryDeliveries(testContext(), scope, id)
		}},
	}
	for _, op := range ops {
		for _, tc := range cases {
			t.Run(op.name+"/"+tc.name, func(t *testing.T) {
				repo := &mockRepo{}
				inserter := &mockJobInserter{}
				svc := NewWebhookService(inserter, repo, nil)
				id := uuid.New()
				repo.On("GetBatchJob", mock.Anything, tenant.DefaultTenantID, id).Return(&store.BatchJob{
					ID:        id,
					Consumer:  tc.jobConsumer,
					JobType:   store.BatchTypeDeliveryRetry,
					Status:    store.BatchStatusPending,
					ExpiresAt: time.Now().Add(time.Hour),
				}, nil)
				repo.On("UpdateBatchJobStatus", mock.Anything, id, mock.Anything, mock.Anything).Return(nil)
				inserter.On("Insert", mock.Anything, mock.Anything).Return(&rivertype.JobInsertResult{}, nil)

				err := op.call(svc, tc.scope, id.String())
				if !tc.reachable {
					requireStatus(t, err, svcerrors.NotFound)
					repo.AssertNotCalled(t, "UpdateBatchJobStatus", mock.Anything, mock.Anything, mock.Anything, mock.Anything)
					inserter.AssertNotCalled(t, "Insert", mock.Anything, mock.Anything)
					return
				}
				if err != nil {
					t.Fatalf("in-scope job: %v", err)
				}
			})
		}
	}
}

// Attempts of another consumer's delivery answer not found and are never read.
func TestGetDeliveryAttempts_StaysInTheConsumer(t *testing.T) {
	id := uuid.New()
	attempts := []*store.WebhookHealthEvent{{DeliveryID: id}}

	t.Run("other consumer", func(t *testing.T) {
		repo := &mockRepo{}
		svc := NewWebhookService(&mockJobInserter{}, repo, nil)
		repo.On("GetDeliveryByID", mock.Anything, tenant.DefaultTenantID, id, "globex").Return(nil, nil)

		_, err := svc.GetDeliveryAttempts(testContext(), id.String(), "globex")
		requireStatus(t, err, svcerrors.NotFound)
		repo.AssertNotCalled(t, "GetDeliveryAttempts", mock.Anything, mock.Anything, mock.Anything)
	})

	t.Run("own consumer", func(t *testing.T) {
		repo := &mockRepo{}
		svc := NewWebhookService(&mockJobInserter{}, repo, nil)
		repo.On("GetDeliveryByID", mock.Anything, tenant.DefaultTenantID, id, "acme").Return(&store.WebhookDelivery{ID: id}, nil)
		repo.On("GetDeliveryAttempts", mock.Anything, tenant.DefaultTenantID, id).Return(attempts, nil)

		got, err := svc.GetDeliveryAttempts(testContext(), id.String(), "acme")
		if err != nil || len(got) != 1 {
			t.Fatalf("got %v, %v; want the attempts", got, err)
		}
	})

	t.Run("no consumer looks up by id alone", func(t *testing.T) {
		repo := &mockRepo{}
		svc := NewWebhookService(&mockJobInserter{}, repo, nil)
		repo.On("GetDeliveryAttempts", mock.Anything, tenant.DefaultTenantID, id).Return(attempts, nil)

		if _, err := svc.GetDeliveryAttempts(testContext(), id.String(), ""); err != nil {
			t.Fatal(err)
		}
		repo.AssertNotCalled(t, "GetDeliveryByID", mock.Anything, mock.Anything, mock.Anything, mock.Anything)
	})
}